scraper.py

"""
OpenLibrary 数据爬取器
使用 Scrapling 提取书籍/作者/列表等数据
优先使用 JSON API (在 URL 后加 .json), 回退 HTML 解析
"""
import json
import time
import logging
from pathlib import Path
from typing import Any

from config import (
    USER_AGENT,
    REQUEST_DELAY,
    MAX_RETRIES,
    RETRY_DELAY,
    DATA_DIR,
    DEFAULT_SEARCH_LIMIT,
    DEFAULT_PAGE_SIZE,
)
from cookie_manager import CookieManager

logger = logging.getLogger(__name__)


class OpenLibraryScraper:
    """OpenLibrary 数据爬取核心类"""

    BASE_URL = "https://openlibrary.org"
    COVERS_URL = "https://covers.openlibrary.org"

    def __init__(self, cookie_mgr: CookieManager):
        self.cookie_mgr = cookie_mgr
        self._last_request_time: float = 0
        self._fetcher = None
        self._session = None

    # ----------------------------------------------------------
    # 速率控制
    # ----------------------------------------------------------
    def _rate_limit(self):
        """遵守 OpenLibrary 速率限制"""
        elapsed = time.time() - self._last_request_time
        if elapsed < REQUEST_DELAY:
            time.sleep(REQUEST_DELAY - elapsed)
        self._last_request_time = time.time()

    # ----------------------------------------------------------
    # HTTP 请求 (用 Fetcher, 轻量快速)
    # ----------------------------------------------------------
    def _get(self, url: str, retries: int = MAX_RETRIES) -> Any:
        """发起 GET 请求, 返回 Scrapling 响应对象"""
        from scrapling.fetchers import Fetcher

        headers = {"User-Agent": USER_AGENT}

        # 附加 cookie (如果有)
        cookies = {}
        if self.cookie_mgr.has_cookies:
            cookies = {
                c["name"]: c["value"]
                for c in self.cookie_mgr.cookies
                if "openlibrary.org" in c.get("domain", "")
            }

        for attempt in range(retries):
            self._rate_limit()
            try:
                resp = Fetcher.get(
                    url,
                    headers=headers,
                    cookies=cookies if cookies else None,
                    follow_redirects=True,
                )
                if resp.status == 200:
                    return resp
                elif resp.status == 429:
                    wait = RETRY_DELAY * (attempt + 1) * 2
                    logger.warning(
                        f"速率限制 429, 等待 {wait}s 后重试..."
                    )
                    time.sleep(wait)
                elif resp.status == 403:
                    logger.error(f"被禁止访问 403: {url}")
                    return None
                else:
                    logger.warning(
                        f"HTTP {resp.status} for {url} "
                        f"(attempt {attempt + 1}/{retries})"
                    )
            except Exception as e:
                logger.error(f"请求失败: {url} -> {e}")
                if attempt < retries - 1:
                    time.sleep(RETRY_DELAY)

        return None

    def _get_json(self, url: str) -> dict | None:
        """请求 JSON API 并解析"""
        resp = self._get(url)
        if resp is None:
            return None
        try:
            return json.loads(resp.body)
        except json.JSONDecodeError:
            logger.error(f"JSON 解析失败: {url}")
            return None

    # ==========================================================
    # 搜索 API
    # ==========================================================

    def search_books(
        self,
        query: str,
        limit: int = DEFAULT_SEARCH_LIMIT,
        page_size: int = DEFAULT_PAGE_SIZE,
        fields: str = "key,title,author_name,first_publish_year,"
                       "edition_count,cover_i,isbn,language,subject",
    ) -> list[dict]:
        """
        搜索书籍, 返回结果列表

        参数:
            query:    搜索关键词
            limit:    最大返回数
            page_size: 每页大小
            fields:   返回字段 (逗号分隔)
        """
        results = []
        offset = 0

        while offset < limit:
            batch_size = min(page_size, limit - offset)
            url = (
                f"{self.BASE_URL}/search.json"
                f"?q={query}"
                f"&limit={batch_size}"
                f"&offset={offset}"
                f"&fields={fields}"
            )
            logger.info(f"搜索: {query} (offset={offset})")
            data = self._get_json(url)
            if not data:
                break

            docs = data.get("docs", [])
            if not docs:
                break

            results.extend(docs)
            offset += len(docs)

            total = data.get("numFound", 0)
            logger.info(f"  获取 {len(docs)} 条, 共 {total} 条")
            if offset >= total:
                break

        return results[:limit]

    def search_authors(
        self, query: str, limit: int = 20
    ) -> list[dict]:
        """搜索作者"""
        url = (
            f"{self.BASE_URL}/search/authors.json"
            f"?q={query}&limit={limit}"
        )
        data = self._get_json(url)
        if data:
            return data.get("docs", [])
        return []

    # ==========================================================
    # 详情 API
    # ==========================================================

    def get_work(self, work_id: str) -> dict | None:
        """
        获取作品详情
        work_id: 如 "OL45804W" 或 "/works/OL45804W"
        """
        if not work_id.startswith("/"):
            work_id = f"/works/{work_id}"
        return self._get_json(f"{self.BASE_URL}{work_id}.json")

    def get_work_editions(
        self, work_id: str, limit: int = 50, offset: int = 0
    ) -> list[dict]:
        """获取作品的所有版本"""
        if not work_id.startswith("/"):
            work_id = f"/works/{work_id}"
        data = self._get_json(
            f"{self.BASE_URL}{work_id}/editions.json"
            f"?limit={limit}&offset={offset}"
        )
        if data:
            return data.get("entries", [])
        return []

    def get_work_ratings(self, work_id: str) -> dict | None:
        """获取作品评分"""
        if not work_id.startswith("/"):
            work_id = f"/works/{work_id}"
        return self._get_json(f"{self.BASE_URL}{work_id}/ratings.json")

    def get_edition(self, edition_id: str) -> dict | None:
        """
        获取版本详情
        edition_id: 如 "OL7353617M" 或 "/books/OL7353617M"
        """
        if not edition_id.startswith("/"):
            edition_id = f"/books/{edition_id}"
        return self._get_json(f"{self.BASE_URL}{edition_id}.json")

    def get_by_isbn(self, isbn: str) -> dict | None:
        """通过 ISBN 查询"""
        return self._get_json(f"{self.BASE_URL}/isbn/{isbn}.json")

    def get_author(self, author_id: str) -> dict | None:
        """
        获取作者详情
        author_id: 如 "OL34184A"
        """
        if not author_id.startswith("/"):
            author_id = f"/authors/{author_id}"
        return self._get_json(f"{self.BASE_URL}{author_id}.json")

    def get_author_works(
        self, author_id: str, limit: int = 50, offset: int = 0
    ) -> list[dict]:
        """获取作者的所有作品"""
        if not author_id.startswith("/"):
            author_id = f"/authors/{author_id}"
        data = self._get_json(
            f"{self.BASE_URL}{author_id}/works.json"
            f"?limit={limit}&offset={offset}"
        )
        if data:
            return data.get("entries", [])
        return []

    def get_subject(
        self, subject: str, limit: int = 50, details: bool = True
    ) -> dict | None:
        """获取主题下的书籍"""
        detail_param = "&details=true" if details else ""
        return self._get_json(
            f"{self.BASE_URL}/subjects/{subject}.json"
            f"?limit={limit}{detail_param}"
        )

    # ==========================================================
    # 用户数据 (需要登录)
    # ==========================================================

    def get_my_lists(self, username: str) -> list[dict]:
        """获取用户的书单列表"""
        data = self._get_json(
            f"{self.BASE_URL}/people/{username}/lists.json"
        )
        if data:
            return data.get("entries", [])
        return []

    def get_list_detail(
        self, username: str, list_id: str
    ) -> dict | None:
        """获取书单详情"""
        return self._get_json(
            f"{self.BASE_URL}/people/{username}/lists/{list_id}.json"
        )

    def get_reading_log(
        self,
        username: str,
        shelf: str = "want-to-read",
        limit: int = 50,
        page: int = 1,
    ) -> dict | None:
        """
        获取用户阅读记录
        shelf: want-to-read / currently-reading / already-read
        """
        return self._get_json(
            f"{self.BASE_URL}/people/{username}/books/{shelf}.json"
            f"?limit={limit}&page={page}"
        )

    # ==========================================================
    # 封面 URL 构建
    # ==========================================================

    @staticmethod
    def cover_url(
        cover_id: int | str,
        size: str = "L",
        key: str = "id",
    ) -> str:
        """
        构建书籍封面 URL
        key: id / isbn / olid / oclc / lccn
        size: S / M / L
        """
        return (
            f"https://covers.openlibrary.org/b/{key}/{cover_id}-{size}.jpg"
        )

    @staticmethod
    def author_photo_url(
        author_id: str, size: str = "L", key: str = "olid"
    ) -> str:
        """构建作者照片 URL"""
        return (
            f"https://covers.openlibrary.org/a/{key}/{author_id}-{size}.jpg"
        )

    # ==========================================================
    # 数据保存
    # ==========================================================

    def save_results(
        self, data: Any, filename: str, subdir: str = ""
    ) -> Path:
        """把数据保存为 JSON 文件"""
        save_dir = DATA_DIR / subdir if subdir else DATA_DIR
        save_dir.mkdir(parents=True, exist_ok=True)

        filepath = save_dir / filename
        filepath.write_text(
            json.dumps(data, indent=2, ensure_ascii=False),
            encoding="utf-8",
        )
        logger.info(f"数据已保存: {filepath}")
        return filepath

    # ==========================================================
    # 批量爬取 (HTML 渲染模式, 用于需要 JS 的页面)
    # ==========================================================

    def fetch_with_browser(self, url: str) -> Any:
        """
        用 StealthyFetcher 渲染 JS 后获取页面,
        适用于需要 JavaScript 渲染的内容
        """
        from scrapling.fetchers import StealthyFetcher

        cookies = self.cookie_mgr.cookies if self.cookie_mgr.has_cookies else None

        self._rate_limit()
        page = StealthyFetcher.fetch(
            url,
            headless=True,
            cookies=cookies,
            network_idle=True,
        )
        return page