Files
re-xxm/reaction_board/classify.py
T
kairee 337e379514 feat(classify): super-linear positive weighting by hit count
正向改为按命中次数 n(require_title + reaction + charge + weak)做平方加权,使多信号同现的真 reaction 不被单个负向词打穿(1+1>2、1+1+1>5);标题不含 require_title 直接 0 分。库内 796 条有效视频复测:线以下误伤由 47 降至 14。
2026-11-01 19:08:59 +08:00

131 lines
4.3 KiB
Python

# -*- 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)。正向改为「命中次数」超线性加权,见 score_item。
NEGATIVE_SCORE = 6
DATE_SCORE = 6
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)。
标题不含 ``require_title`` 直接 0 分;时长只在此基础上已有命中时加权。
"""
require_title = str(classify.get("require_title") or "")
if require_title and not passes_required_title(title, require_title):
return 0
hits = 0
if require_title:
hits += 1
hits += len(matched_words(title, classify.get("reaction")))
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
if matched_words(title, classify.get("negative")):
score -= NEGATIVE_SCORE
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