daily-robots/daily/board_history.py

"""企微展示历史：读写 data.wecom_shown_keys，与 movement_baseline 严格分离。"""

from __future__ import annotations

import json
import logging
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

from daily.config import OUTPUT_DIR, board_dedup_days
from daily.delta import RECENT_BOARD_KEYS, skill_id

logger = logging.getLogger(__name__)

BOARD_KEYS = RECENT_BOARD_KEYS

_GITHUB_REPO_RE = re.compile(r"github\.com/([\w.-]+/[\w.-]+)", re.I)
_SKILL_SH_RE = re.compile(r"skills\.sh/([\w.-]+/[\w.-]+(?:/[\w.-]+)?)", re.I)

# 与企微正文榜单标题对齐；顺序用于切分相邻 section
_WECOM_SECTION_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("skills_trending", re.compile(r"Skills\s+Trending", re.I)),
    ("skills_hot", re.compile(r"Skills\s+Hot", re.I)),
    ("github_trending", re.compile(r"GitHub\s+Trending", re.I)),
    ("github_emerging", re.compile(r"GitHub\s+新兴", re.I)),
    ("github_topic", re.compile(r"Topic\s+", re.I)),
)


def extract_shown_keys(board: str, items: list[dict[str, Any]]) -> list[str]:
    """从最终展示 items 抽取稳定 identity key。

    Skills 榜同时写入 skill id 与 source，便于周去重按仓屏蔽。
    """
    keys: list[str] = []
    seen: set[str] = set()
    for item in items:
        if board.startswith("skills_"):
            candidates = [skill_id(item), str(item.get("source") or "").strip()]
        else:
            candidates = [str(item.get("repo") or "")]
        for key in candidates:
            if not key or key in seen:
                continue
            seen.add(key)
            keys.append(key)
    return keys


def _keys_from_board_items(board: str, data: dict[str, Any]) -> set[str]:
    if board == "github_topic":
        topic = data.get("github_topic") or {}
        items = topic.get("repos") if isinstance(topic, dict) else []
    else:
        items = data.get(board) or []
    if not isinstance(items, list):
        return set()
    return set(extract_shown_keys(board, items))


def parse_wecom_shown_keys(md: str) -> dict[str, set[str]]:
    """从企微 Markdown 按榜单 section 解析已展示 keys（冷启动兼容）。"""
    out: dict[str, set[str]] = {board: set() for board in BOARD_KEYS}
    if not (md or "").strip():
        return out

    hits: list[tuple[int, str]] = []
    for board, pattern in _WECOM_SECTION_PATTERNS:
        for match in pattern.finditer(md):
            hits.append((match.start(), board))
    if not hits:
        return out
    hits.sort(key=lambda x: x[0])

    for idx, (start, board) in enumerate(hits):
        end = hits[idx + 1][0] if idx + 1 < len(hits) else len(md)
        chunk = md[start:end]
        if board.startswith("skills_"):
            out[board].update(_SKILL_SH_RE.findall(chunk))
        else:
            out[board].update(_GITHUB_REPO_RE.findall(chunk))
    return out


def _load_shown_keys_for_day(path: Path, date_str: str) -> dict[str, set[str]] | None:
    """读一日历史：优先 wecom_shown_keys；缺省则回退 wecom.md，再回退 data 榜字段。"""
    empty = {board: set() for board in BOARD_KEYS}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        logger.warning("读取 wecom_shown_keys %s 失败：%s", path, exc)
        return None

    data = payload.get("data") if isinstance(payload, dict) else None
    if not isinstance(data, dict):
        return empty

    out: dict[str, set[str]] = {board: set() for board in BOARD_KEYS}
    shown = data.get("wecom_shown_keys")
    if isinstance(shown, dict):
        for board in BOARD_KEYS:
            keys = shown.get(board) or []
            if isinstance(keys, list):
                out[board].update(str(k) for k in keys if k)
        if any(out.values()):
            return out

    wecom_path = OUTPUT_DIR / f"{date_str}.wecom.md"
    if wecom_path.exists():
        try:
            md = wecom_path.read_text(encoding="utf-8")
        except OSError as exc:
            logger.warning("读取 wecom.md 回退 %s 失败：%s", wecom_path, exc)
        else:
            parsed = parse_wecom_shown_keys(md)
            if any(parsed.values()):
                return parsed

    for board in BOARD_KEYS:
        out[board].update(_keys_from_board_items(board, data))
    return out


def load_recent_shown_keys(
    date_str: str,
    *,
    lookback_days: int | None = None,
) -> dict[str, set[str]]:
    """近 N 日已展示 keys 并集（不含当日）。缺省或读失败视为空集。"""
    empty = {board: set() for board in BOARD_KEYS}
    try:
        dt = datetime.strptime(date_str, "%Y-%m-%d")
    except ValueError:
        return empty

    days = lookback_days if lookback_days is not None else board_dedup_days()
    out: dict[str, set[str]] = {board: set() for board in BOARD_KEYS}

    for day_offset in range(1, days + 1):
        prev_date = (dt - timedelta(days=day_offset)).strftime("%Y-%m-%d")
        path = OUTPUT_DIR / f"{prev_date}.data.json"
        if not path.exists():
            continue
        day_keys = _load_shown_keys_for_day(path, prev_date)
        if day_keys is None:
            continue
        for board in BOARD_KEYS:
            out[board].update(day_keys.get(board) or set())
    return out


def merge_wecom_shown_into_data(
    data: dict[str, Any],
    shown: dict[str, list[str]],
) -> dict[str, Any]:
    """写入 wecom_shown_keys，不修改 movement_baseline。"""
    merged = dict(data)
    merged["wecom_shown_keys"] = {
        board: list(keys) for board, keys in shown.items()
    }
    return merged