mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-09-24 22:05:20 +01:00
CaliBrainandmik593
3a92c5de78
The Calibre dependency was due to the script testing for validity of the downloaded file, as often they would be corrupted from aa. But CWA is already doing that, so we are just having redundant code here. For the cloudflarebypasser, I basically run my own version now, instead of depending on an external library, this way we have better control for debugging and on the docker image. Fixes #18, #33, #27, #48, #65, #78, #86, #88, #89 --------- Co-authored-by: mik593 <91991279+mik593@users.noreply.github.com>
129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""Network operations manager for the book downloader application."""
|
|
|
|
import requests
|
|
import time
|
|
from io import BytesIO
|
|
import urllib.request
|
|
from typing import Optional
|
|
from urllib.parse import urlparse
|
|
from tqdm import tqdm
|
|
|
|
import cloudflare_bypasser
|
|
from logger import setup_logger
|
|
from config import MAX_RETRY, DEFAULT_SLEEP, USE_CF_BYPASS, PROXIES
|
|
|
|
logger = setup_logger(__name__)
|
|
"""Configure urllib opener with appropriate headers."""
|
|
opener = urllib.request.build_opener()
|
|
opener.addheaders = [
|
|
('User-agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
|
|
'AppleWebKit/537.36 (KHTML, like Gecko) '
|
|
'Chrome/129.0.0.0 Safari/537.3')
|
|
]
|
|
urllib.request.install_opener(opener)
|
|
|
|
|
|
def html_get_page(url: str, retry: int = MAX_RETRY, use_bypasser: bool = False) -> str:
|
|
"""Fetch HTML content from a URL with retry mechanism.
|
|
|
|
Args:
|
|
url: Target URL
|
|
retry: Number of retry attempts
|
|
skip_404: Whether to skip 404 errors
|
|
|
|
Returns:
|
|
str: HTML content if successful, None otherwise
|
|
"""
|
|
try:
|
|
|
|
if use_bypasser and USE_CF_BYPASS:
|
|
logger.info(f"GET Using Cloudflare Bypasser for: {url}")
|
|
response = cloudflare_bypasser.get(url)
|
|
logger.debug(f"Cloudflare Bypasser response: {response}")
|
|
if response:
|
|
return str(response.html)
|
|
else:
|
|
raise requests.exceptions.RequestException("Failed to bypass Cloudflare")
|
|
|
|
logger.info(f"GET: {url}")
|
|
response = requests.get(url, proxies=PROXIES)
|
|
response.raise_for_status()
|
|
logger.debug(f"Success getting: {url}")
|
|
time.sleep(1)
|
|
return str(response.text)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
if retry == 0:
|
|
logger.error(f"Failed to fetch page: {url}, error: {e}")
|
|
return ""
|
|
|
|
if response.status_code == 404:
|
|
logger.warning(f"404 error for URL: {url}")
|
|
return ""
|
|
|
|
if response.status_code == 403:
|
|
logger.warning(f"403 error for URL: {url}. Should retry using cloudflare bypass.")
|
|
if use_bypasser:
|
|
return ""
|
|
use_bypasser = True
|
|
|
|
|
|
sleep_time = DEFAULT_SLEEP * (MAX_RETRY - retry + 1)
|
|
logger.warning(
|
|
f"Retrying GET {url} in {sleep_time} seconds due to error: {e}"
|
|
)
|
|
time.sleep(sleep_time)
|
|
return html_get_page(url, retry - 1, use_bypasser)
|
|
|
|
def download_url(link: str, size: str = "") -> Optional[BytesIO]:
|
|
"""Download content from URL into a BytesIO buffer.
|
|
|
|
Args:
|
|
link: URL to download from
|
|
|
|
Returns:
|
|
BytesIO: Buffer containing downloaded content if successful
|
|
"""
|
|
try:
|
|
logger.info(f"Downloading from: {link}")
|
|
response = requests.get(link, stream=True, proxies=PROXIES)
|
|
response.raise_for_status()
|
|
|
|
total_size : float = 0.0
|
|
try:
|
|
# we assume size is in MB
|
|
total_size = float(size.strip().replace(" ", "").replace(",", ".").upper()[:-2].strip()) * 1024 * 1024
|
|
except:
|
|
total_size = float(response.headers.get('content-length', 0))
|
|
|
|
buffer = BytesIO()
|
|
|
|
# Initialize the progress bar with your guess
|
|
pbar = tqdm(total=total_size, unit='B', unit_scale=True, desc='Downloading')
|
|
for chunk in response.iter_content(chunk_size=1000):
|
|
buffer.write(chunk)
|
|
pbar.update(len(chunk))
|
|
|
|
pbar.close()
|
|
return buffer
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Failed to download from {link}: {e}")
|
|
return None
|
|
|
|
def get_absolute_url(base_url: str, url: str) -> str:
|
|
"""Get absolute URL from relative URL and base URL.
|
|
|
|
Args:
|
|
base_url: Base URL
|
|
url: Relative URL
|
|
"""
|
|
if url.strip() == "":
|
|
return ""
|
|
if url.startswith("http"):
|
|
return url
|
|
parsed_url = urlparse(url)
|
|
parsed_base = urlparse(base_url)
|
|
if parsed_url.netloc == "" or parsed_url.scheme == "":
|
|
parsed_url = parsed_url._replace(netloc=parsed_base.netloc, scheme=parsed_base.scheme)
|
|
return parsed_url.geturl()
|