scraper/engine.py

"""
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}")