daily-robots/daily/news/rank.py

"""RSS 新闻去重、打分与排序。"""

from __future__ import annotations

import logging
import re
from datetime import datetime, timezone
from typing import Any

from daily.config import env_int
from daily.delta import find_previous_data
from daily.news.fetch import _entry_datetime, _normalize_link, _normalize_title

logger = logging.getLogger(__name__)

_WS = re.compile(r"\s+")
_TOKEN = re.compile(r"[\w]{2,}", re.UNICODE)
_CJK = re.compile(r"[\u4e00-\u9fff]")

CATEGORY_TIER: dict[str, int] = {
    "official": 20,
    "developer": 16,
    "research": 14,
    "media": 12,
    "newsletter": 10,
    "community": 8,
    "tech": 12,
    "dev": 10,
}


def title_similarity_threshold() -> float:
    raw = env_int("DAILY_NEWS_TITLE_SIM", 55)
    return max(0, min(raw, 95)) / 100.0


def _title_tokens(title: str) -> set[str]:
    normalized = _normalize_title(title)
    tokens = set(_TOKEN.findall(normalized))
    cjk = "".join(_CJK.findall(title))
    for i in range(max(0, len(cjk) - 1)):
        tokens.add(cjk[i : i + 2])
    if not tokens and normalized:
        tokens.add(normalized)
    return tokens


def jaccard_similarity(left: set[str], right: set[str]) -> float:
    if not left or not right:
        return 0.0
    inter = len(left & right)
    union = len(left | right)
    return inter / union if union else 0.0


def fuzzy_dedupe_by_title(
    items: list[dict[str, Any]],
    *,
    threshold: float | None = None,
) -> list[dict[str, Any]]:
    """按标题相似度合并重复报道，保留 score 更高（或更靠前）的条目。"""
    if not items:
        return []
    limit = threshold if threshold is not None else title_similarity_threshold()
    kept: list[dict[str, Any]] = []
    kept_tokens: list[set[str]] = []
    for item in items:
        tokens = _title_tokens(str(item.get("title") or ""))
        duplicate_idx: int | None = None
        for idx, existing_tokens in enumerate(kept_tokens):
            if jaccard_similarity(tokens, existing_tokens) >= limit:
                duplicate_idx = idx
                break
        if duplicate_idx is None:
            kept.append(item)
            kept_tokens.append(tokens)
            continue
        existing = kept[duplicate_idx]
        if float(item.get("score") or 0) > float(existing.get("score") or 0):
            kept[duplicate_idx] = item
            kept_tokens[duplicate_idx] = tokens
    return kept


def _freshness_points(item: dict[str, Any], *, now: datetime | None = None) -> int:
    now = now or datetime.now(timezone.utc)
    dt = _entry_datetime(item)
    if dt is None:
        return 4
    age_hours = max(0.0, (now - dt).total_seconds() / 3600.0)
    if age_hours <= 6:
        return 30
    if age_hours <= 24:
        return 22
    if age_hours <= 72:
        return 8
    return 0


def _source_tier(item: dict[str, Any]) -> int:
    category_id = str(item.get("category_id") or "")
    return CATEGORY_TIER.get(category_id, 8)


def _novelty_points(item: dict[str, Any], yesterday_links: set[str]) -> int:
    link = _normalize_link(str(item.get("link") or ""))
    if link and link in yesterday_links:
        return -20
    return 5


def score_news_item(
    item: dict[str, Any],
    *,
    yesterday_links: set[str] | None = None,
    now: datetime | None = None,
) -> int:
    links = yesterday_links or set()
    total = _source_tier(item) + _freshness_points(item, now=now) + _novelty_points(item, links)
    return max(0, total)


def load_yesterday_news_links(date_str: str) -> set[str]:
    baseline = find_previous_data(date_str)
    if not baseline:
        return set()
    _, data = baseline
    links: set[str] = set()
    for key in ("ai_news", "cn_ai_news"):
        for item in data.get(key) or []:
            if not isinstance(item, dict):
                continue
            link = _normalize_link(str(item.get("link") or ""))
            if link:
                links.add(link)
    return links


def apply_news_ranking(news: dict[str, Any], *, date_str: str) -> dict[str, Any]:
    """对 categories / flat 打分、模糊去重并写回 score 字段。"""
    if not news.get("enabled"):
        return news

    yesterday_links = load_yesterday_news_links(date_str)
    threshold = title_similarity_threshold()
    now = datetime.now(timezone.utc)

    categories = news.get("categories") or []
    for category in categories:
        items = list(category.get("items") or [])
        for item in items:
            item["score"] = score_news_item(item, yesterday_links=yesterday_links, now=now)
        items.sort(key=lambda x: float(x.get("score") or 0), reverse=True)
        category["items"] = fuzzy_dedupe_by_title(items, threshold=threshold)

    flat: list[dict[str, Any]] = []
    for category in categories:
        flat.extend(category.get("items") or [])
    flat.sort(key=lambda x: float(x.get("score") or 0), reverse=True)
    news["flat"] = fuzzy_dedupe_by_title(flat, threshold=threshold)

    stats = dict(news.get("stats") or {})
    stats["ranked_items"] = len(news["flat"])
    stats["yesterday_links"] = len(yesterday_links)
    news["stats"] = stats
    logger.info(
        "新闻排序完成：%d 条 flat，昨日链接基准 %d",
        len(news["flat"]),
        len(yesterday_links),
    )
    return news