application / scraper /engine.py
Krixter's picture
Upload 10 files
18b6a95 verified
Raw History Blame Contribute Delete
10.6 kB
"""
Core scraping engine with async downloading, retry logic, and chapter management.
"""
from __future__ import annotations
import asyncio
import hashlib
import json
import logging
import random
import re
import time
from dataclasses import dataclass, field, asdict
from enum import Enum
from pathlib import Path
from typing import AsyncIterator, Callable, Optional
from urllib.parse import urljoin, urlparse
import aiohttp
import aiofiles
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Data models
# ---------------------------------------------------------------------------
class ChapterStatus(str, Enum):
PENDING = "pending"
DOWNLOADING = "downloading"
DONE = "done"
FAILED = "failed"
SKIPPED = "skipped"
LOCKED = "locked"
@dataclass
class Chapter:
index: int
title: str
url: str
status: ChapterStatus = ChapterStatus.PENDING
retries: int = 0
error: str = ""
content: str = ""
word_count: int = 0
def to_dict(self) -> dict:
d = asdict(self)
d["status"] = self.status.value
return d
@dataclass
class NovelJob:
url: str
title: str = ""
chapters: list[Chapter] = field(default_factory=list)
status: str = "queued" # queued | running | paused | done | cancelled
start_chapter: int = 1
end_chapter: int = 0 # 0 = all
specific_chapters: list[int] = field(default_factory=list)
output_dir: Path = field(default_factory=lambda: Path("output"))
created_at: float = field(default_factory=time.time)
started_at: float = 0.0
finished_at: float = 0.0
bytes_downloaded: int = 0
# Progress helpers
@property
def total(self) -> int:
return len(self.chapters)
@property
def done_count(self) -> int:
return sum(1 for c in self.chapters if c.status == ChapterStatus.DONE)
@property
def failed_count(self) -> int:
return sum(1 for c in self.chapters if c.status == ChapterStatus.FAILED)
@property
def progress(self) -> float:
if not self.total:
return 0.0
return self.done_count / self.total
def chapter_in_range(self, ch: Chapter) -> bool:
if self.specific_chapters:
return ch.index in self.specific_chapters
end = self.end_chapter if self.end_chapter else float("inf")
return self.start_chapter <= ch.index <= end
# ---------------------------------------------------------------------------
# User-agent pool
# ---------------------------------------------------------------------------
_USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:124.0) Gecko/20100101 Firefox/124.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 14.4; rv:124.0) Gecko/20100101 Firefox/124.0",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0",
]
def random_ua() -> str:
return random.choice(_USER_AGENTS)
def build_headers(extra: dict | None = None) -> dict:
h = {
"User-Agent": random_ua(),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
}
if extra:
h.update(extra)
return h
# ---------------------------------------------------------------------------
# Session factory
# ---------------------------------------------------------------------------
async def create_session(
cookies: dict | None = None,
timeout_seconds: int = 30,
) -> aiohttp.ClientSession:
connector = aiohttp.TCPConnector(
limit=20,
limit_per_host=5,
ssl=False,
force_close=False,
enable_cleanup_closed=True,
)
timeout = aiohttp.ClientTimeout(total=timeout_seconds, connect=10)
session = aiohttp.ClientSession(
connector=connector,
timeout=timeout,
headers=build_headers(),
cookies=cookies or {},
)
return session
# ---------------------------------------------------------------------------
# Fetch helpers
# ---------------------------------------------------------------------------
_CF_SIGNALS = [
"checking your browser",
"cloudflare",
"ddos-guard",
"ray id",
"enable javascript",
"__cf_bm",
"cf-challenge",
"403 forbidden",
"access denied",
"bot detection",
"please wait",
"browser check",
]
def _is_cloudflare(html: str, status: int) -> bool:
if status in (403, 503, 429):
return True
low = html.lower()
return any(sig in low for sig in _CF_SIGNALS)
def _is_login_page(html: str) -> bool:
low = html.lower()
return any(k in low for k in ["please log in", "sign in to continue", "login required", "members only"])
def _is_locked_chapter(html: str) -> bool:
low = html.lower()
return any(k in low for k in ["this chapter is locked", "premium chapter", "unlock chapter", "coins required", "vip chapter"])
async def fetch_html(
session: aiohttp.ClientSession,
url: str,
retries: int = 3,
base_delay: float = 2.0,
extra_headers: dict | None = None,
) -> tuple[str, int]:
"""Fetch URL, return (html, status_code). Raises on permanent failure."""
headers = build_headers(extra_headers)
for attempt in range(retries):
try:
async with session.get(url, headers=headers, allow_redirects=True) as resp:
# Try utf-8 first, fallback to detected encoding
raw = await resp.read()
try:
html = raw.decode("utf-8")
except UnicodeDecodeError:
html = raw.decode("latin-1", errors="replace")
return html, resp.status
except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
if attempt == retries - 1:
raise
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
logger.debug(f"Fetch attempt {attempt+1} failed for {url}: {exc}. Retry in {delay:.1f}s")
await asyncio.sleep(delay)
return "", 0
# ---------------------------------------------------------------------------
# Cloudscraper fallback
# ---------------------------------------------------------------------------
def _cloudscraper_fetch(url: str, cookies: dict | None = None) -> tuple[str, int]:
try:
import cloudscraper
scraper = cloudscraper.create_scraper(browser={"browser": "chrome", "platform": "windows", "mobile": False})
if cookies:
scraper.cookies.update(cookies)
resp = scraper.get(url, timeout=30)
return resp.text, resp.status_code
except ImportError:
raise RuntimeError("cloudscraper not installed")
except Exception as exc:
raise RuntimeError(f"cloudscraper failed: {exc}")
# ---------------------------------------------------------------------------
# Playwright fallback
# ---------------------------------------------------------------------------
async def _playwright_fetch(url: str, wait_seconds: float = 3.0) -> tuple[str, int]:
try:
from playwright.async_api import async_playwright
async with async_playwright() as pw:
browser = await pw.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage",
"--disable-gpu", "--no-first-run", "--no-zygote", "--single-process"],
)
ctx = await browser.new_context(user_agent=random_ua())
page = await ctx.new_page()
response = await page.goto(url, wait_until="networkidle", timeout=30000)
await asyncio.sleep(wait_seconds)
html = await page.content()
status = response.status if response else 200
await browser.close()
return html, status
except ImportError:
raise RuntimeError("playwright not installed")
except Exception as exc:
raise RuntimeError(f"playwright failed: {exc}")
# ---------------------------------------------------------------------------
# Smart fetch with fallback chain
# ---------------------------------------------------------------------------
class FetchMode(str, Enum):
FAST = "fast"
BALANCED = "balanced"
COMPAT = "compat"
async def smart_fetch(
session: aiohttp.ClientSession,
url: str,
mode: FetchMode = FetchMode.BALANCED,
cookies: dict | None = None,
log_cb: Callable[[str], None] | None = None,
) -> tuple[str, int]:
"""Try aiohttp → cloudscraper → playwright depending on mode and CF detection."""
def log(msg: str):
if log_cb:
log_cb(msg)
logger.info(msg)
# 1. Normal aiohttp
try:
html, status = await fetch_html(session, url)
if _is_cloudflare(html, status):
log(f"⚠️ Cloudflare detected on {url}")
if mode == FetchMode.FAST:
return html, status
else:
return html, status
except Exception as exc:
log(f"⚠️ aiohttp failed for {url}: {exc}")
if mode == FetchMode.FAST:
raise
# 2. Cloudscraper
if mode in (FetchMode.BALANCED, FetchMode.COMPAT):
try:
log(f"🔄 Trying cloudscraper for {url}")
loop = asyncio.get_event_loop()
html, status = await loop.run_in_executor(None, _cloudscraper_fetch, url, cookies)
if not _is_cloudflare(html, status):
return html, status
log(f"⚠️ Cloudscraper still blocked on {url}")
except Exception as exc:
log(f"⚠️ Cloudscraper failed: {exc}")
# 3. Playwright
if mode == FetchMode.COMPAT:
log(f"🌐 Activating Playwright browser mode for {url}")
try:
html, status = await _playwright_fetch(url)
return html, status
except Exception as exc:
log(f"❌ Playwright failed: {exc}")
raise
raise RuntimeError(f"All fetch methods failed for {url}")