#!/usr/bin/env python3
"""Incrementally collect Hongcan articles into SQLite using a real browser."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sqlite3
import sys
import time
from contextlib import closing
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import urljoin, urlsplit, urlunsplit

from playwright.sync_api import BrowserContext, Page, sync_playwright


DEFAULT_LIST_URL = "https://m.canyin88.com/zixun/"
ARTICLE_PATH_RE = re.compile(r"/zixun/\d{4}/\d{1,2}/\d{1,2}/\d+\.html$")
DATE_RE = re.compile(r"(20\d{2})[-/.年](\d{1,2})[-/.月](\d{1,2})(?:日)?(?:\s+(\d{1,2}):(\d{2})(?::(\d{2}))?)?")


def now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def normalize_url(raw_url: str, base_url: str = DEFAULT_LIST_URL) -> str | None:
    absolute = urljoin(base_url, raw_url.strip())
    parts = urlsplit(absolute)
    host = parts.netloc.lower().removeprefix("m.").removeprefix("www.")
    if host != "canyin88.com" or not ARTICLE_PATH_RE.search(parts.path):
        return None
    return urlunsplit(("https", "www.canyin88.com", parts.path, "", ""))


def clean_text(value: str | None) -> str:
    return re.sub(r"\s+", " ", value or "").strip().lstrip("\ufeff")


def parse_date(value: str | None) -> str | None:
    if not value:
        return None
    match = DATE_RE.search(value)
    if not match:
        return None
    year, month, day, hour, minute, second = match.groups()
    dt = datetime(
        int(year), int(month), int(day), int(hour or 0), int(minute or 0), int(second or 0)
    )
    return dt.isoformat(timespec="seconds")


def date_from_article_url(url: str) -> datetime | None:
    match = re.search(r"/zixun/(20\d{2})/(\d{1,2})/(\d{1,2})/", url)
    if not match:
        return None
    return datetime(*(int(part) for part in match.groups()))


def init_db(conn: sqlite3.Connection, schema_path: Path) -> None:
    conn.executescript(schema_path.read_text(encoding="utf-8"))


def discover_urls(page: Page, list_url: str, scrolls: int, pause_ms: int) -> list[str]:
    page.goto(list_url, wait_until="domcontentloaded", timeout=60_000)
    page.wait_for_timeout(2_000)
    found: set[str] = set()
    stagnant = 0
    for _ in range(scrolls + 1):
        hrefs = page.locator("a[href]").evaluate_all("els => els.map(e => e.href)")
        before = len(found)
        for href in hrefs:
            normalized = normalize_url(href, list_url)
            if normalized:
                found.add(normalized)
        stagnant = stagnant + 1 if len(found) == before else 0
        if stagnant >= 3:
            break
        page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
        page.wait_for_timeout(pause_ms)
    return sorted(found, reverse=True)


def meta(page: Page, selector: str) -> str:
    locator = page.locator(selector).first
    return clean_text(locator.get_attribute("content")) if locator.count() else ""


def first_text(page: Page, selectors: list[str]) -> str:
    for selector in selectors:
        locator = page.locator(selector).first
        if locator.count():
            text = clean_text(locator.inner_text(timeout=3_000))
            if text:
                return text
    return ""


def extract_article(page: Page, url: str) -> dict[str, str | None]:
    page.goto(url, wait_until="domcontentloaded", timeout=60_000)
    page.wait_for_timeout(800)
    document_title = clean_text(page.title())
    document_title = re.sub(r"(?:[_-]红餐网.*)$", "", document_title).strip()
    title = meta(page, 'meta[property="og:title"]') or document_title
    if not title or title == "相关推荐":
        title = first_text(page, [".title", "h1"])
    summary = (
        meta(page, 'meta[name="description"]')
        or meta(page, 'meta[property="og:description"]')
    )
    author = (
        meta(page, 'meta[name="author"]')
        or first_text(page, [".author", ".source", "[class*=author]"])
    )
    published_raw = (
        meta(page, 'meta[property="article:published_time"]')
        or first_text(page, ["time", ".time", "[class*=time]", "[class*=date]"])
        or page.locator("body").inner_text(timeout=5_000)[:1000]
    )
    content = first_text(
        page,
        [
            "article",
            ".article-content",
            ".article_content",
            ".content",
            "[class*=detail-content]",
            "[class*=article] [class*=content]",
        ],
    )
    if not content:
        paragraphs = page.locator("p").all_inner_texts()
        content = "\n\n".join(clean_text(p) for p in paragraphs if len(clean_text(p)) >= 20)
    if not title or len(content) < 80:
        raise ValueError(f"article extraction incomplete: title={bool(title)}, content_length={len(content)}")
    published_at = parse_date(published_raw)
    url_date = date_from_article_url(url)
    if url_date and (
        not published_at or datetime.fromisoformat(published_at).date() != url_date.date()
    ):
        published_at = url_date.isoformat(timespec="seconds")
    content_hash = hashlib.sha256(f"{title}\n{content}".encode("utf-8")).hexdigest()
    return {
        "canonical_url": url,
        "title": title,
        "author": author or None,
        "published_at": published_at,
        "summary": summary or None,
        "content": content,
        "content_hash": content_hash,
    }


def pending_urls(conn: sqlite3.Connection, discovered: list[str], max_retries: int) -> list[str]:
    rows = conn.execute(
        "SELECT canonical_url FROM articles WHERE crawl_status != 'success' AND retry_count < ?",
        (max_retries,),
    ).fetchall()
    return list(dict.fromkeys(discovered + [row[0] for row in rows]))


def insert_discoveries(conn: sqlite3.Connection, urls: list[str]) -> int:
    timestamp = now_iso()
    inserted = 0
    for url in urls:
        cursor = conn.execute(
            """
            INSERT OR IGNORE INTO articles
                (canonical_url, discovered_at, updated_at)
            VALUES (?, ?, ?)
            """,
            (url, timestamp, timestamp),
        )
        inserted += cursor.rowcount
    conn.commit()
    return inserted


def save_success(conn: sqlite3.Connection, article: dict[str, str | None]) -> None:
    timestamp = now_iso()
    conn.execute(
        """
        UPDATE articles SET
            title = ?, author = ?, published_at = ?, summary = ?, content = ?,
            content_hash = ?, crawl_status = 'success', last_error = NULL,
            crawled_at = ?, updated_at = ?
        WHERE canonical_url = ?
        """,
        (
            article["title"], article["author"], article["published_at"], article["summary"],
            article["content"], article["content_hash"], timestamp, timestamp,
            article["canonical_url"],
        ),
    )
    conn.commit()


def save_failure(conn: sqlite3.Connection, url: str, error: Exception) -> None:
    conn.execute(
        """
        UPDATE articles SET
            crawl_status = 'failed', retry_count = retry_count + 1,
            last_error = ?, updated_at = ?
        WHERE canonical_url = ?
        """,
        (str(error)[:1000], now_iso(), url),
    )
    conn.commit()


def new_context(playwright, headless: bool) -> tuple[BrowserContext, Page]:
    browser = playwright.chromium.launch(headless=headless)
    context = browser.new_context(
        locale="zh-CN",
        timezone_id="Asia/Shanghai",
        viewport={"width": 1280, "height": 900},
        user_agent=(
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
            "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"
        ),
    )
    context.set_default_timeout(15_000)
    return context, context.new_page()


def run(args: argparse.Namespace) -> int:
    base_dir = Path(__file__).resolve().parent
    db_path = Path(args.db).expanduser().resolve()
    db_path.parent.mkdir(parents=True, exist_ok=True)
    with closing(sqlite3.connect(db_path)) as conn:
        init_db(conn, base_dir / "schema.sql")
        started_at = now_iso()
        run_id = conn.execute(
            "INSERT INTO crawl_runs(started_at) VALUES (?)", (started_at,)
        ).lastrowid
        conn.commit()
        discovered_count = inserted_count = updated_count = failed_count = 0
        try:
            with sync_playwright() as playwright:
                context, page = new_context(playwright, args.headed is False)
                try:
                    urls = discover_urls(page, args.list_url, args.scrolls, args.pause_ms)
                    discovered_count = len(urls)
                    inserted_count = insert_discoveries(conn, urls)
                    queue = pending_urls(conn, urls, args.max_retries)
                    cutoff = datetime.now() - timedelta(days=args.lookback_days)
                    for index, url in enumerate(queue, start=1):
                        path_date = parse_date(url.replace("/", "-"))
                        if path_date and datetime.fromisoformat(path_date) < cutoff:
                            continue
                        try:
                            article = extract_article(page, url)
                            save_success(conn, article)
                            updated_count += 1
                            print(f"[{index}/{len(queue)}] OK {article['title']}")
                        except Exception as exc:
                            failed_count += 1
                            save_failure(conn, url, exc)
                            print(f"[{index}/{len(queue)}] FAIL {url}: {exc}", file=sys.stderr)
                        time.sleep(args.article_pause)
                finally:
                    context.close()
            status = "success" if failed_count == 0 else "partial"
            message = json.dumps({"db": str(db_path)}, ensure_ascii=False)
        except Exception as exc:
            status, message = "failed", str(exc)[:1000]
            print(f"crawl failed: {exc}", file=sys.stderr)
        conn.execute(
            """
            UPDATE crawl_runs SET finished_at=?, status=?, discovered_count=?,
                inserted_count=?, updated_count=?, failed_count=?, message=?
            WHERE id=?
            """,
            (now_iso(), status, discovered_count, inserted_count, updated_count, failed_count, message, run_id),
        )
        conn.commit()
        print(json.dumps({
            "status": status, "discovered": discovered_count, "inserted": inserted_count,
            "updated": updated_count, "failed": failed_count, "database": str(db_path),
        }, ensure_ascii=False))
        return 0 if status in {"success", "partial"} else 1


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="红餐网文章每日增量采集入库")
    parser.add_argument("--db", default="hongcan.db", help="SQLite 数据库路径")
    parser.add_argument("--list-url", default=DEFAULT_LIST_URL)
    parser.add_argument("--lookback-days", type=int, default=7, help="回溯天数")
    parser.add_argument("--scrolls", type=int, default=12, help="列表页最大下拉次数")
    parser.add_argument("--pause-ms", type=int, default=1200, help="列表页下拉等待毫秒")
    parser.add_argument("--article-pause", type=float, default=0.8, help="文章间隔秒数")
    parser.add_argument("--max-retries", type=int, default=3)
    parser.add_argument("--headed", action="store_true", help="显示浏览器，便于排查验证码")
    return parser


if __name__ == "__main__":
    raise SystemExit(run(build_parser().parse_args()))