# -*- 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: """去掉搜索结果里的 ```` 高亮标签。""" 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。 NEGATIVE_PER_WORD = 2 DATE_SCORE = 4 SELF_SCORE = 3 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 计算候选分数。 正向按**命中次数**超线性加权:每命中一个正向词(``require_title`` / ``reaction`` / ``charge`` / ``weak_positive``)记 1 次,正向得分 = 命中次数²(1→1、2→4、 3→9……),使多信号同现的真 reaction 不会被单个负向词打穿(1+1>2、1+1+1>5)。 ``reaction`` 词表含 ``re`` 这类通用子串(用于兼容 reaction/reactio/reation 等 拼写),因此必须与 ``require_title`` 或 ``充电`` 同现才计一次命中。标题不含 ``require_title`` 直接 0 分;时长只在此基础上已有命中时加权。 负向按命中词数递减:每个词 -2,命中 2 个额外 -2、3 个额外 -6(k*(k-1)), 带日期再 -4。 """ 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 hits = 0 if require_title: hits += 1 hits += reaction_hit hits += len(matched_words(title, classify.get("charge"))) hits += len(matched_words(title, classify.get("weak_positive"))) 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 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