re-xxm/reaction_board/classify.py

# -*- coding: utf-8 -*-
"""候选分类与时长解析。

规则见 project.md §3.3：分类只决定“是否通知”，不决定是否入库。
"""

from __future__ import annotations

import re
import unicodedata
from typing import Any, Mapping

_TAG_RE = re.compile(r"<[^>]+>")


def normalize_title(text: str) -> str:
    """全半角归一化并转小写，便于标题关键词匹配。"""
    return unicodedata.normalize("NFKC", text or "").lower()


def clean_title(text: str) -> str:
    """去掉搜索结果里的 ``<em class="keyword">`` 高亮标签。"""
    return _TAG_RE.sub("", text or "").strip()


def parse_duration(value: Any) -> int:
    """解析 B 站搜索结果时长。

    支持 ``105:5``（分钟:秒）、``1:02:03``（时:分:秒）和整数秒。
    """
    if isinstance(value, bool):
        return 0
    if isinstance(value, int):
        return max(0, value)
    if not isinstance(value, str):
        return 0
    parts = value.strip().split(":")
    if not parts or any(not part.isdigit() for part in parts):
        return 0
    numbers = [int(part) for part in parts]
    if len(numbers) == 1:
        return numbers[0]
    if len(numbers) == 2:
        return numbers[0] * 60 + numbers[1]
    if len(numbers) == 3:
        return numbers[0] * 3600 + numbers[1] * 60 + numbers[2]
    return 0


def matched_words(title: str, words: Any) -> list[str]:
    """返回标题命中（归一化子串）的词，保持 ``words`` 顺序、跳过空词。"""
    normalized = normalize_title(title)
    if not isinstance(words, (list, tuple)):
        return []
    return [str(word) for word in words if word and normalize_title(str(word)) in normalized]


def contains_any(title: str, words: Any) -> bool:
    return bool(matched_words(title, words))


def passes_required_title(title: str, require_title: str) -> bool:
    if not require_title:
        return True
    return normalize_title(require_title) in normalize_title(title)


# 标题带日期通常意味着录播/回放/录像（按具体某场直播命名），作为强负向信号。
# 覆盖 `2026-10-28` / `2026/10/25` / `2026.10.26` / `2024年11月24日` / `10月25日`。
_DATE_PATTERNS = (
    r"\d{4}\s*[-/.]\s*\d{1,2}\s*[-/.]\s*\d{1,2}",
    r"\d{4}\s*年\s*\d{1,2}\s*月(?:\s*\d{1,2}\s*日)?",
    r"\d{1,2}\s*月\s*\d{1,2}\s*日",
)
_DATE_RE = re.compile("|".join(_DATE_PATTERNS))

# 负向权重（见 project.md §3.3）：每个负向词 -2，命中多条额外叠加；日期单独 -4；
# 时长 <10 分钟视作切片，-1（时长未知=0 时不扣）。
NEGATIVE_PER_WORD = 2
DATE_SCORE = 4
SELF_SCORE = 3
SHORT_DURATION_SECONDS = 600
SHORT_DURATION_SCORE = 1


def date_hits(title: str) -> list[str]:
    """返回标题中的日期片段，用于判定「带日期≈录播/回放」。"""
    return _DATE_RE.findall(title or "")


def score_item(
    *,
    title: str,
    up_mid: int | None,
    duration_seconds: int,
    classify: Mapping[str, Any],
) -> int:
    """按 project.md §3.3 计算候选分数。

    正向按**命中次数**超线性加权：正向词（``reaction`` / ``charge`` / ``weak_positive``）
    每命中一个记 1 次；仅在已有正向命中时，``require_title`` 才额外记 1 次。得分 =
    命中次数²（1→1、2→4、3→9……）。也就是说：**只含 ``require_title`` 的标题 0 分**
    （"咻咻满"单独出现和 "reaction" 单独出现一样没有正向意义），必须
    "require_title + 任一正向词" 才得分；多信号同现时不会被单个负向词打穿。

    ``reaction`` 词表含 ``re`` 这类通用子串（用于兼容 reaction/reactio/reation 等
    拼写），因此必须与 ``require_title`` 或 ``充电`` 同现才计一次命中。标题不含
    ``require_title`` 直接 0 分；时长只在已有命中时加权。

    负向按命中词数递减：每个词 -2，命中 2 个额外 -2、3 个额外 -6（k*(k-1)），
    带日期再 -4；时长 <10 分钟（切片/片段）再 -1。
    """
    require_title = str(classify.get("require_title") or "")
    if require_title and not passes_required_title(title, require_title):
        return 0

    # reaction 类通用词必须与 require_title 或 "充电" 同现才算命中（避免通用词加分）。
    reaction_gate = passes_required_title(title, require_title) or contains_any(title, ["充电"])
    reaction_hit = 1 if (matched_words(title, classify.get("reaction")) and reaction_gate) else 0

    positive_hits = (
        reaction_hit
        + len(matched_words(title, classify.get("charge")))
        + len(matched_words(title, classify.get("weak_positive")))
    )
    # 只过标题关（标题仅含 require_title）不给分；require_title 仅在已有正向命中时才算 1 次。
    hits = positive_hits + (1 if (require_title and positive_hits) else 0)

    score = hits * hits

    if hits:
        boost = classify.get("duration_boost") or {}
        gt_60 = int(boost.get("gt_60min", 2))
        gt_30 = int(boost.get("gt_30min", 1))
        if duration_seconds > 60 * 60:
            score += gt_60
        elif duration_seconds > 30 * 60:
            score += gt_30

    # 时长 <10 分钟视作切片/片段，扣 1 分（时长未知 0 不扣）。
    if 0 < duration_seconds < SHORT_DURATION_SECONDS:
        score -= SHORT_DURATION_SCORE

    neg_words = matched_words(title, classify.get("negative"))
    k = len(neg_words)
    if k:
        score -= NEGATIVE_PER_WORD * k
        score -= k * (k - 1)  # 命中 2 个额外 -2、3 个额外 -6、4 个额外 -12
    if classify.get("date_negative", True) and date_hits(title):
        score -= DATE_SCORE
    if up_mid is not None and int(up_mid) == int(classify.get("self_mid", 37754047)):
        score -= SELF_SCORE
    return score