1

server/scripts/build_tags_from_titles.py

@@ -4,79 +4,25 @@
     cd /root/project/video/server
     .venv/bin/python scripts/build_tags_from_titles.py [--dry-run]
 
-提取规则：
-    1. 番号（如 CAWD-601）及其系列前缀（CAWD）
-    2. 书名号/方括号内容（≤12 字，去掉「无码破解」「中文字幕」等标记词）
-    3. jieba TF-IDF 关键词 topK=6（过滤停用词、纯数字、单字）
+提取规则见 app/services/title_tags.py（幂等可重复执行）。
 """
 import argparse
 import asyncio
 import logging
-import re
 import sys
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
-import jieba.analyse  # noqa: E402
 from sqlalchemy import select  # noqa: E402
 
 from app.database import SessionFactory, engine  # noqa: E402
 from app.models.video import Tag, Video, VideoTag  # noqa: E402
+from app.services.title_tags import extract_keywords  # noqa: E402
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
 log = logging.getLogger(__name__)
 
-STOP = set(
-    "第 集 部 篇 全集 完结 中文字幕 无码 破解 无码破解 字幕 高清 最新 精选 视频 在线 观看 "
-    "一个 一起 没有 为了 开始 结果 只能 直接 就是 没想到 真爽".split()
-)
-
-CODE_RE = re.compile(r"[A-Za-z]{2,6}-\d{2,4}")
-EPISODE_RE = re.compile(r"第[0-9一二三四五六七八九十百千零]+[集部期话卷章]")
-PUNCT_RE = re.compile(r"[《》「」『』\[\]【】()（）_\-:：,，.。!！?？…、/\\|~～\s]+")
-BRACKET_RE = re.compile(r"[《「『\[【(（]([^《》「」『』\[\]【】()（）]{2,30})[》」』\]】)）]")
-
-
-def extract_keywords(title: str) -> list[str]:
-    kws: list[str] = []
-
-    # 1. 番号 + 系列前缀
-    for code in CODE_RE.findall(title.upper()):
-        kws.append(code)
-        prefix = code.split("-")[0]
-        if len(prefix) >= 2:
-            kws.append(prefix)
-
-    # 2. 书名号/括号内容（剧名/主题）
-    for inner in BRACKET_RE.findall(title):
-        inner = inner.strip()
-        if 2 <= len(inner) <= 12 and inner not in STOP and not CODE_RE.search(inner):
-            kws.append(inner)
-
-    # 3. jieba TF-IDF
-    t = EPISODE_RE.sub(" ", title)
-    t = PUNCT_RE.sub(" ", t)
-    for w in jieba.analyse.extract_tags(t, topK=6):
-        w = w.strip()
-        if not (2 <= len(w) <= 12):
-            continue
-        if w in STOP or re.fullmatch(r"\d+", w):
-            continue
-        kws.append(w)
-
-    # 去重保序（按 casefold 键，避免 MySQL 不区分大小写的唯一键冲突）
-    seen: set[str] = set()
-    out: list[str] = []
-    for k in kws:
-        if k.isascii():
-            k = k.upper()
-        key = k.casefold()
-        if key not in seen:
-            seen.add(key)
-            out.append(k)
-    return out
-
 
 async def main(dry_run: bool) -> None:
     async with SessionFactory() as db: