#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
m5_collector_skeleton.py —— 每日行业信息推送「采集骨架」

用途
    从 m1_competitor_sources.json 读取已核验的公开入口，按限速抓取列表页/详情页，
    抽取标题与正文摘要，按 JSONL 追加落盘，供 M4 判定规则做事件分类。

硬性边界（与任务包一致，已在代码中固化）
    * 只采公开页面。不实现登录、不实现验证码识别、不绕过付费墙。
    * 遇到 401/403/412/429 一律放弃该 URL 并如实记录状态码，不做任何"伪装绕过"。
    * 限速 >= 1.2 秒/请求，失败重试 2 次。
    * 启动前会读取目标站 robots.txt；被 Disallow 的路径直接跳过并记录。

本文件是骨架：不带 __main__ 自动执行副作用，需显式调用 run() 才会抓取。
    python3 m5_collector_skeleton.py --dry-run     # 只打印计划，不发请求
    python3 m5_collector_skeleton.py --limit 20     # 实际抓取前 20 个 URL
"""
from __future__ import annotations

import argparse
import json
import os
import random
import re
import time
import urllib.robotparser
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from typing import Any, Iterable

import requests
from bs4 import BeautifulSoup

# ---------------------------------------------------------------- 配置
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
SOURCE_FILE = os.path.join(BASE_DIR, "m1_competitor_sources.json")
OUTPUT_FILE = os.path.join(BASE_DIR, "output", "collected_events.jsonl")

MIN_INTERVAL = 1.2          # 限速下限（秒），不得调低
MAX_RETRY = 2               # 失败重试次数（不含首次）
TIMEOUT = 25                # 单请求超时
USER_AGENT = ("MiniMaxCode-IndustryCollector/1.0 (+daily industry push; "
              "contact: internal)")

# 遇到即放弃、不做任何绕过的状态码
BLOCK_STATUS = {401, 402, 403, 407, 412, 429, 451}

# 列表页/详情页的类型线索 -> 建议的动态类型候选（仅作默认，实际分类以 m4 规则为准）
TYPE_HINT = {
    "新闻列表页": "待判定",
    "新闻分类页": "待判定",
    "新闻详情页": "待判定",
    "产品目录": "新品发布",
    "产品详情页": "新品发布",
    "sitemap": "待判定",
    "robots": "跳过",
}


# ---------------------------------------------------------------- 数据结构
@dataclass
class Record:
    """落盘的一条记录：严格对应任务包要求的六个字段。"""
    来源URL: str
    抓取时间: str
    标题: str
    正文摘要: str
    品牌: str
    动态类型: str
    # 以下为便于排查的附加字段
    HTTP状态码: int | None = None
    备注: str = ""
    证据字段: dict[str, Any] = field(default_factory=dict)


# ---------------------------------------------------------------- 限速器
class RateLimiter:
    """全局限速，保证相邻两次请求间隔 >= MIN_INTERVAL。"""

    def __init__(self, interval: float = MIN_INTERVAL):
        self.interval = max(interval, MIN_INTERVAL)
        self._last = 0.0

    def wait(self) -> None:
        gap = time.time() - self._last
        if gap < self.interval:
            time.sleep(self.interval - gap)
        # 叠加少量抖动，降低对目标站点的压力
        time.sleep(random.uniform(0, 0.3))
        self._last = time.time()


# ---------------------------------------------------------------- robots
_ROBOT_CACHE: dict[str, urllib.robotparser.RobotFileParser | None] = {}


def _fetch_robots_text(root: str, timeout: int = 15) -> str | None:
    """用与采集一致的 UA 拉取 robots.txt。

    注意：必须显式指定 UA。Python urllib 默认 UA 会被不少站点拦截，
    导致「读不到 robots」被误当成「全站禁止」，从而静默跳过正常信源。
    """
    try:
        r = requests.get(f"{root}/robots.txt", timeout=timeout,
                         headers={"User-Agent": USER_AGENT})
        if r.status_code == 200:
            if not r.encoding or r.encoding.lower() == "iso-8859-1":
                r.encoding = r.apparent_encoding or "utf-8"
            return r.text
        # 4xx（无 robots 文件）→ 视为无限制；401/403/5xx → 视为取不到
        return "" if 400 <= r.status_code < 500 and r.status_code not in (401, 403) else None
    except requests.RequestException:
        return None


def robots_allows(url: str) -> bool | None:
    """检查 robots.txt。

    返回 True=明确允许 / False=明确禁止 / None=无法判定（读不到或被风控）。
    返回 None 时调用方应记录为「robots未知」并谨慎处理，不得当作允许或禁止。
    """
    from urllib.parse import urlparse
    p = urlparse(url)
    root = f"{p.scheme}://{p.netloc}"
    if root not in _ROBOT_CACHE:
        text = _fetch_robots_text(root)
        if text is None:
            _ROBOT_CACHE[root] = None
        else:
            rp = urllib.robotparser.RobotFileParser()
            rp.parse(text.splitlines())   # 避免 rp.read() 使用默认 UA
            _ROBOT_CACHE[root] = rp
    rp = _ROBOT_CACHE[root]
    if rp is None:
        return None
    if rp.disallow_all or rp.allow_all:
        # 解析器未能形成有效策略，视作无法判定
        return None if rp.disallow_all else True
    try:
        return rp.can_fetch(USER_AGENT, url)
    except Exception:  # noqa: BLE001
        return None


# ---------------------------------------------------------------- 抓取
class Collector:
    def __init__(self, session: requests.Session | None = None,
                 limiter: RateLimiter | None = None):
        self.session = session or requests.Session()
        self.session.headers.update({"User-Agent": USER_AGENT,
                                     "Accept-Language": "zh-CN,zh;q=0.9"})
        self.limiter = limiter or RateLimiter()
        self._seen: set[str] = set()

    def fetch(self, url: str) -> tuple[int | None, str | None, str]:
        """带重试的 GET。返回 (状态码, 正文, 备注)。不做任何绕过。"""
        last_note = ""
        for attempt in range(MAX_RETRY + 1):
            self.limiter.wait()
            try:
                r = self.session.get(url, timeout=TIMEOUT, allow_redirects=True)
            except requests.RequestException as e:
                last_note = f"请求异常: {type(e).__name__}"
                continue

            if r.status_code in BLOCK_STATUS:
                # 风控/权限类：直接放弃，如实记录，不重试、不伪装
                return r.status_code, None, f"放弃：状态码 {r.status_code}（反爬/权限），不做绕过"
            if r.status_code >= 500:
                last_note = f"服务端错误 {r.status_code}"
                time.sleep(1.5 * (attempt + 1))
                continue
            if r.status_code != 200:
                last_note = f"非200: {r.status_code}"
                continue
            if not r.encoding or r.encoding.lower() == "iso-8859-1":
                r.encoding = r.apparent_encoding or "utf-8"
            return r.status_code, r.text, ""
        return None, None, f"重试 {MAX_RETRY} 次仍失败：{last_note}"

    # ---------------------------------------------------- 抽取
    @staticmethod
    def extract_title(soup: BeautifulSoup) -> str:
        if t := soup.find("title"):
            return re_sub(t.get_text(strip=True))
        if h := soup.find(["h1", "h2"]):
            return re_sub(h.get_text(strip=True))
        return ""

    @staticmethod
    def extract_summary(soup: BeautifulSoup, limit: int = 300) -> str:
        for bad in soup(["script", "style", "nav", "header", "footer", "noscript"]):
            bad.decompose()
        text = soup.get_text(separator=" ", strip=True)
        text = " ".join(text.split())
        return text[:limit] + ("…" if len(text) > limit else "")

    @staticmethod
    def extract_article_links(soup: BeautifulSoup, base: str, same_host_only=True) -> list[str]:
        """从列表页抽取详情页链接，供下一轮下钻。"""
        from urllib.parse import urljoin, urlparse
        base_host = urlparse(base).netloc
        out: list[str] = []
        for a in soup.find_all("a", href=True):
            u = urljoin(base, a["href"])
            if same_host_only and urlparse(u).netloc != base_host:
                continue
            if re.match(r".*\.(jpg|jpeg|png|gif|css|js|pdf|zip|ico)$", u, re.I):
                continue
            if u not in out:
                out.append(u)
        return out

    # ---------------------------------------------------- 主流程
    def collect_one(self, brand: str, kind: str, url: str, hint: str) -> Record | None:
        now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
        code, text, note = self.fetch(url)

        if text is None:
            return Record(来源URL=url, 抓取时间=now, 标题="", 正文摘要="",
                          品牌=brand, 动态类型="抓取失败", HTTP状态码=code,
                          备注=note or "未取得正文")

        try:
            soup = BeautifulSoup(text, "html.parser")
        except Exception as e:  # noqa: BLE001
            return Record(来源URL=url, 抓取时间=now, 标题="", 正文摘要="",
                          品牌=brand, 动态类型="解析失败", HTTP状态码=code,
                          备注=f"HTML解析异常: {e}")

        title = self.extract_title(soup)
        summary = self.extract_summary(soup)
        links = self.extract_article_links(soup, url)
        self._seen.update(links[:200])

        return Record(
            来源URL=url, 抓取时间=now, 标题=title, 正文摘要=summary,
            品牌=brand, 动态类型=hint, HTTP状态码=code, 备注=note,
            证据字段={"页面内链接数": len(links), "链接样本": links[:10]},
        )


def re_sub(s: str) -> str:
    return " ".join(s.split())


# ---------------------------------------------------------------- 计划生成
def build_plan(source_file: str) -> list[dict]:
    """从 m1_competitor_sources.json 生成抓取计划。"""
    with open(source_file, encoding="utf-8") as f:
        data = json.load(f)

    plan = []
    for brand in data.get("品牌", []):
        name = brand.get("品牌", "")
        for src in brand.get("源", []):
            kind = src.get("类型", "")
            if kind in {"robots", "sitemap"}:
                continue  # robots 由限速器自动读取；sitemap 建议单独走全量枚举任务
            hint = TYPE_HINT.get(kind, "待判定")
            plan.append({"品牌": name, "类型": kind, "URL": src.get("URL", ""),
                         "动态类型": hint, "备注": src.get("备注", "")})
    return plan


def write_jsonl(path: str, rec: Record) -> None:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(asdict(rec), ensure_ascii=False) + "\n")


# ---------------------------------------------------------------- 入口
def run(dry_run: bool = False, limit: int | None = None,
        source_file: str = SOURCE_FILE, output_file: str = OUTPUT_FILE) -> None:
    plan = build_plan(source_file)
    if limit:
        plan = plan[:limit]

    print(f"[计划] 共 {len(plan)} 个待抓 URL，限速 {MIN_INTERVAL}s/请求，重试 {MAX_RETRY} 次")

    if dry_run:
        for i, p in enumerate(plan, 1):
            allow = robots_allows(p["URL"])
            flag = {True: "robots允许", False: "robots禁止(将跳过)", None: "robots未知"}[allow]
            print(f"  {i:>3}. [{p['品牌']}][{p['类型']}] {flag} {p['URL']}")
        print("[dry-run] 未发起任何实际请求。")
        return

    c = Collector()
    for i, p in enumerate(plan, 1):
        allow = robots_allows(p["URL"])
        if allow is False:
            print(f"  {i:>3}. 跳过（robots禁止） {p['URL']}")
            continue
        print(f"  {i:>3}. 抓取 [{p['品牌']}][{p['类型']}] {p['URL']}")
        rec = c.collect_one(p["品牌"], p["类型"], p["URL"], p["动态类型"])
        if rec:
            write_jsonl(output_file, rec)
            print(f"       -> {rec.HTTP状态码} | {rec.标题[:48]}")

    print(f"[完成] 结果写入 {output_file}（JSONL，追加模式）")


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description="行业信息推送采集骨架（不实现登录/验证码/付费墙绕过）")
    ap.add_argument("--dry-run", action="store_true", help="只打印抓取计划，不发请求")
    ap.add_argument("--limit", type=int, default=None, help="最多处理前 N 个 URL")
    args = ap.parse_args()
    run(dry_run=args.dry_run, limit=args.limit)