mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-11-05 23:18:56 +08:00
refactor(parser): replace regex raw_parser with pipeline tokenizer
Replaces the monolithic regex-based parser with a multi-stage tokenizer pipeline (normalize, tokenize, classify, resolve). Fixes multiple parsing bugs: Atlas format (#986), VTuber/Lv2 title splits (#798, #766), Pre-episode format (#794), western S01E05 format (#764), and separator dash leaking into titles (#819). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Estrella Pan
@@ -2,332 +2,46 @@ import logging
|
||||
import re
|
||||
|
||||
from module.models import Episode
|
||||
from module.parser.analyser.tokenizer import tokenize_title
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
EPISODE_RE = re.compile(r"\d+")
|
||||
TITLE_RE = re.compile(
|
||||
r"(.*?|\[.*])((?: ?-) ?\d+(?:[vV]\d{1,2})? |\[\d+]|\[\d+.?[vV]\d]|第\d+[话話集]|\[第?\d+[话話集]]|\[\d+.?END]|[Ee][Pp]?\d+(?![A-Za-z0-9]))(.*)"
|
||||
|
||||
_MOVIE_TOKEN_RE = re.compile(
|
||||
r"剧场版|劇場版|电影版|電影版|\bmovie\b|\bgekijou-?ban\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
RESOLUTION_RE = re.compile(r"1080|720|2160|4K")
|
||||
SOURCE_RE = re.compile(r"B-Global|[Bb]aha|[Bb]ilibili|AT-X|Web")
|
||||
SUB_RE = re.compile(r"[简繁日字幕]|CH|BIG5|GB")
|
||||
|
||||
FALLBACK_EP_PATTERNS = [
|
||||
re.compile(r" (\d+) ?(?=\[)"), # #876/#910: digits before [
|
||||
re.compile(r"\[(\d+)\(\d+\)\]"), # #773: [02(57)]
|
||||
]
|
||||
|
||||
PREFIX_RE = re.compile(r"[^\w\s\u4e00-\u9fff\u3040-\u309f\u30a0-\u30ff-]")
|
||||
|
||||
# 以下均为静态正则：预编译成模块级 Pattern 对象，避免 process() 内层每次调用都
|
||||
# 走 re.xxx(pattern_string, ...) 的内部编译缓存查找（profile 显示这部分开销占
|
||||
# raw_parser 单次调用总耗时的 ~20%，见 2026-07 性能优化）。语义与原来的
|
||||
# re.search/sub/split/match/findall(r"...", ...) 完全一致。
|
||||
_BRACKET_CLEAR_RE = re.compile(r"[\[\]]")
|
||||
_XINFAN_RE = re.compile(r"新番|月?番")
|
||||
_HK_MO_TW_RE = re.compile(r"港澳台地区")
|
||||
SEASON_RULE_RE = re.compile(r"S\d{1,2}|Season \d{1,2}|[第].[季期]")
|
||||
_SEASON_WORD_RE = re.compile(r"Season|S")
|
||||
_SEASON_CN_RE = re.compile(r"[第 ].*[季期(部分)]|部分")
|
||||
_SEASON_CN_STRIP_RE = re.compile(r"[第季期 ]")
|
||||
_HK_MO_TW_BRACKET_RE = re.compile(r"[(（]仅限港澳台地区[）)]")
|
||||
_NAME_SPLIT_RE = re.compile(r"/|\s{2}|-\s{2}")
|
||||
_UNDERSCORE_RE = re.compile(r"_{1}")
|
||||
_DASH_SPACE_RE = re.compile(r" - {1}")
|
||||
_DIGIT_CN_RE = re.compile(r"\d+\s[一-龥]")
|
||||
_CN_PREFIX_RE = re.compile(r"^[一-龥]{2,}")
|
||||
_JP_RE = re.compile(r"[ࠀ-一]{2,}")
|
||||
_CN_RE = re.compile(r"[一-龥]{2,}")
|
||||
_EN_RE = re.compile(r"[a-zA-Z]{3,}")
|
||||
_TAG_BRACKET_RE = re.compile(r"[\[\]()（）]")
|
||||
_SUFFIX_STRIP_RE = re.compile(r"_MP4|_MKV")
|
||||
|
||||
# movie token: hits route the title to episode_type="movie" (no episode number required)
|
||||
MOVIE_TOKEN_RE = re.compile(r"剧场版|劇場版|电影版|[Mm]ovie")
|
||||
# OVA / OAD / SP / Special token: hits route the title to episode_type="special",
|
||||
# season is forced to 0 (Season 0). Digits directly after the token (e.g. "OVA01")
|
||||
# must also be recognized, so the token itself allows a trailing number.
|
||||
SPECIAL_TOKEN_RE = re.compile(
|
||||
r"\bOVA\d*\b|\bOAD\d*\b|\bSP\d*\b|[Ss]pecial|番外篇?|特别篇"
|
||||
_SPECIAL_TOKEN_RE = re.compile(
|
||||
r"\bOVA\d*\b|\bOAD\d*\b|\bSP\d*\b|\bspecial\b|番外篇?|特别篇|特別篇",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
# digits directly after OVA/OAD/SP are treated as the episode number, e.g. "OVA01", "SP 02"
|
||||
SPECIAL_EPISODE_RE = re.compile(r"(?:OVA|OAD|SP)\s*[-_]?\s*(\d{1,3})", re.I)
|
||||
BRACKET_RE = re.compile(r"[\[\(（【].*?[\]\)）】]")
|
||||
_LEADING_GROUP_RE = re.compile(r"^\s*[\[【][^\]】]+[\]】]\s*")
|
||||
_GROUP_RE = re.compile(r"\[([^\]]*)\]")
|
||||
|
||||
|
||||
def _detect_non_episodic_type(content_title: str) -> str | None:
|
||||
"""识别剧场版/OVA/OAD/SP/Special 等非常规编号资源的类型。"""
|
||||
if MOVIE_TOKEN_RE.search(content_title):
|
||||
def get_group(raw: str) -> str:
|
||||
"""Return the first square-bracket group, or an empty string."""
|
||||
match = _GROUP_RE.search(raw)
|
||||
return match.group(1) if match else ""
|
||||
|
||||
|
||||
def _detect_non_episodic_type(raw: str) -> str | None:
|
||||
"""Classify movie/special markers for LLM results.
|
||||
|
||||
The tokenizer performs the normal deterministic classification. This
|
||||
helper remains the compatibility seam used by ``TitleParser`` when an LLM
|
||||
result needs its episode type corrected from the original release title.
|
||||
"""
|
||||
content = _LEADING_GROUP_RE.sub("", raw, count=1)
|
||||
if _MOVIE_TOKEN_RE.search(content):
|
||||
return "movie"
|
||||
if SPECIAL_TOKEN_RE.search(content_title):
|
||||
if _SPECIAL_TOKEN_RE.search(content):
|
||||
return "special"
|
||||
return None
|
||||
|
||||
|
||||
def _parse_non_episodic(content_title: str, group: str, episode_type: str) -> tuple:
|
||||
"""解析剧场版/OVA/OAD/SP/Special 标题：提炼标题与可选的集数信息，不依赖常规的
|
||||
TITLE_RE / fallback 集数匹配。"""
|
||||
stripped = prefix_process(content_title, group)
|
||||
brackets = BRACKET_RE.findall(stripped)
|
||||
name_part = BRACKET_RE.sub(" ", stripped)
|
||||
# ``Movie`` may be part of the real title as well as the trailing release
|
||||
# type marker (``Localized Movie Movie``). Only the final occurrence is
|
||||
# the marker; deleting every occurrence silently changes the TMDB query.
|
||||
movie_tokens = list(MOVIE_TOKEN_RE.finditer(name_part))
|
||||
if movie_tokens:
|
||||
marker = movie_tokens[-1]
|
||||
name_part = f"{name_part[: marker.start()]} {name_part[marker.end() :]}"
|
||||
name_part = SPECIAL_TOKEN_RE.sub(" ", name_part)
|
||||
name_part = name_part.strip(" -/")
|
||||
name_en, name_zh, name_jp = "", "", ""
|
||||
try:
|
||||
name_en, name_zh, name_jp = name_process(name_part)
|
||||
except ValueError:
|
||||
pass
|
||||
sub, dpi, source = find_tags(" ".join(brackets))
|
||||
ep_match = SPECIAL_EPISODE_RE.search(stripped)
|
||||
episode = int(ep_match.group(1)) if ep_match else 0
|
||||
season = 0 if episode_type == "special" else 1
|
||||
return (
|
||||
name_en,
|
||||
name_zh,
|
||||
name_jp,
|
||||
season,
|
||||
"",
|
||||
episode,
|
||||
sub,
|
||||
dpi,
|
||||
source,
|
||||
group,
|
||||
episode_type,
|
||||
)
|
||||
|
||||
|
||||
def _fallback_parse(content_title: str) -> tuple | None:
|
||||
"""Try fallback regex patterns when TITLE_RE fails."""
|
||||
for pattern in FALLBACK_EP_PATTERNS:
|
||||
m = pattern.search(content_title)
|
||||
if m:
|
||||
season_info = content_title[: m.start()].strip()
|
||||
episode_info = m.group(1)
|
||||
other = content_title[m.end() :].strip()
|
||||
return season_info, episode_info, other
|
||||
return None
|
||||
|
||||
|
||||
CHINESE_NUMBER_MAP = {
|
||||
"一": 1,
|
||||
"二": 2,
|
||||
"三": 3,
|
||||
"四": 4,
|
||||
"五": 5,
|
||||
"六": 6,
|
||||
"七": 7,
|
||||
"八": 8,
|
||||
"九": 9,
|
||||
"十": 10,
|
||||
}
|
||||
|
||||
|
||||
def get_group(name: str) -> str:
|
||||
parts = _BRACKET_CLEAR_RE.split(name)
|
||||
if len(parts) > 1:
|
||||
return parts[1]
|
||||
return ""
|
||||
|
||||
|
||||
def pre_process(raw_name: str) -> str:
|
||||
return raw_name.replace("【", "[").replace("】", "]")
|
||||
|
||||
|
||||
def prefix_process(raw: str, group: str) -> str:
|
||||
# Guard against empty group: without this, the pattern degenerates to ".."
|
||||
# and every pair of characters gets deleted, destroying titles that lack a
|
||||
# [group] prefix (#1025).
|
||||
if group:
|
||||
raw = re.sub(f".{re.escape(group)}.", "", raw)
|
||||
raw_process = PREFIX_RE.sub("/", raw)
|
||||
arg_group = raw_process.split("/")
|
||||
while "" in arg_group:
|
||||
arg_group.remove("")
|
||||
if len(arg_group) == 1:
|
||||
arg_group = arg_group[0].split(" ")
|
||||
for arg in arg_group:
|
||||
if _XINFAN_RE.search(arg) and len(arg) <= 5:
|
||||
raw = re.sub(f".{re.escape(arg)}.", "", raw)
|
||||
elif _HK_MO_TW_RE.search(arg):
|
||||
raw = re.sub(f".{re.escape(arg)}.", "", raw)
|
||||
return raw
|
||||
|
||||
|
||||
def season_process(season_info: str):
|
||||
name_season = season_info
|
||||
# if re.search(r"新番|月?番", season_info):
|
||||
# name_season = re.sub(".*新番.", "", season_info)
|
||||
# # 去除「新番」信息
|
||||
# name_season = re.sub(r"^[^]】]*[]】]", "", name_season).strip()
|
||||
name_season = _BRACKET_CLEAR_RE.sub(" ", name_season)
|
||||
seasons = SEASON_RULE_RE.findall(name_season)
|
||||
if not seasons:
|
||||
return name_season, "", 1
|
||||
name = SEASON_RULE_RE.sub("", name_season)
|
||||
for season in seasons:
|
||||
season_raw = season
|
||||
if _SEASON_WORD_RE.search(season) is not None:
|
||||
season = int(_SEASON_WORD_RE.sub("", season))
|
||||
break
|
||||
elif _SEASON_CN_RE.search(season) is not None:
|
||||
season_pro = _SEASON_CN_STRIP_RE.sub("", season)
|
||||
try:
|
||||
season = int(season_pro)
|
||||
except ValueError:
|
||||
season = CHINESE_NUMBER_MAP[season_pro]
|
||||
break
|
||||
return name, season_raw, season
|
||||
|
||||
|
||||
def name_process(name: str):
|
||||
name_en, name_zh, name_jp = None, None, None
|
||||
name = name.strip()
|
||||
name = _HK_MO_TW_BRACKET_RE.sub("", name)
|
||||
split = _NAME_SPLIT_RE.split(name)
|
||||
while "" in split:
|
||||
split.remove("")
|
||||
if len(split) == 1:
|
||||
if _UNDERSCORE_RE.search(name) is not None:
|
||||
split = name.split("_")
|
||||
elif _DASH_SPACE_RE.search(name) is not None:
|
||||
split = name.split("-")
|
||||
if len(split) == 1:
|
||||
# Titles like "29 岁单身..." — digits + Chinese are one title
|
||||
if _DIGIT_CN_RE.match(split[0]):
|
||||
name_zh = split[0].strip()
|
||||
return name_en, name_zh, name_jp
|
||||
split_space = split[0].split(" ")
|
||||
for idx in [0, -1]:
|
||||
if _CN_PREFIX_RE.search(split_space[idx]) is not None:
|
||||
chs = split_space[idx]
|
||||
split_space.remove(chs)
|
||||
split = [chs, " ".join(split_space)]
|
||||
break
|
||||
for item in split:
|
||||
if _JP_RE.search(item) and not name_jp:
|
||||
name_jp = item.strip()
|
||||
elif _CN_RE.search(item) and not name_zh:
|
||||
name_zh = item.strip()
|
||||
elif _EN_RE.search(item) and not name_en:
|
||||
name_en = item.strip()
|
||||
return name_en, name_zh, name_jp
|
||||
|
||||
|
||||
def find_tags(other):
|
||||
elements = _TAG_BRACKET_RE.sub(" ", other).split(" ")
|
||||
# find CHT
|
||||
sub, resolution, source = None, None, None
|
||||
for element in filter(lambda x: x != "", elements):
|
||||
if SUB_RE.search(element):
|
||||
sub = element
|
||||
elif RESOLUTION_RE.search(element):
|
||||
resolution = element
|
||||
elif SOURCE_RE.search(element):
|
||||
source = element
|
||||
return clean_sub(sub), resolution, source
|
||||
|
||||
|
||||
def clean_sub(sub: str | None) -> str | None:
|
||||
if sub is None:
|
||||
return sub
|
||||
return _SUFFIX_STRIP_RE.sub("", sub)
|
||||
|
||||
|
||||
def process(raw_title: str):
|
||||
raw_title = raw_title.strip().replace("\n", " ")
|
||||
content_title = pre_process(raw_title)
|
||||
# 预处理标题
|
||||
group = get_group(content_title)
|
||||
# 翻译组的名字
|
||||
match_obj = TITLE_RE.match(content_title)
|
||||
if match_obj is not None:
|
||||
season_info, episode_info, other = [x.strip() for x in match_obj.groups()]
|
||||
else:
|
||||
fallback = _fallback_parse(content_title)
|
||||
if fallback is None:
|
||||
# 常规集数解析失败：识别剧场版/OVA/OAD/SP/Special 等非常规编号资源，
|
||||
# 而非直接丢弃 (原先返回 None 会导致这些资源被整体丢弃)
|
||||
episode_type = _detect_non_episodic_type(content_title)
|
||||
if episode_type is not None:
|
||||
return _parse_non_episodic(content_title, group, episode_type)
|
||||
return None
|
||||
season_info, episode_info, other = fallback
|
||||
process_raw = prefix_process(season_info, group)
|
||||
# 处理 前缀
|
||||
raw_name, season_raw, season = season_process(process_raw)
|
||||
# 处理 第n季
|
||||
name_en, name_zh, name_jp = "", "", ""
|
||||
try:
|
||||
name_en, name_zh, name_jp = name_process(raw_name)
|
||||
# 处理 名字
|
||||
except ValueError:
|
||||
pass
|
||||
# 处理 集数
|
||||
raw_episode = EPISODE_RE.search(episode_info)
|
||||
episode = 0
|
||||
if raw_episode is not None:
|
||||
episode = int(raw_episode.group())
|
||||
sub, dpi, source = find_tags(other) # 剩余信息处理
|
||||
return (
|
||||
name_en,
|
||||
name_zh,
|
||||
name_jp,
|
||||
season,
|
||||
season_raw,
|
||||
episode,
|
||||
sub,
|
||||
dpi,
|
||||
source,
|
||||
group,
|
||||
"episode",
|
||||
)
|
||||
|
||||
|
||||
def raw_parser(raw: str) -> Episode | None:
|
||||
ret = process(raw)
|
||||
if ret is None:
|
||||
logger.info(f"Cannot parse resource: {raw}, skipping.")
|
||||
return None
|
||||
(
|
||||
name_en,
|
||||
name_zh,
|
||||
name_jp,
|
||||
season,
|
||||
sr,
|
||||
episode,
|
||||
sub,
|
||||
dpi,
|
||||
source,
|
||||
group,
|
||||
episode_type,
|
||||
) = ret
|
||||
return Episode(
|
||||
name_en,
|
||||
name_zh,
|
||||
name_jp,
|
||||
season,
|
||||
sr,
|
||||
episode,
|
||||
sub,
|
||||
group,
|
||||
dpi,
|
||||
source,
|
||||
episode_type,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
title = "[动漫国字幕组&LoliHouse] THE MARGINAL SERVICE - 08 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕]"
|
||||
print(raw_parser(title))
|
||||
title = "[北宇治字幕组&LoliHouse] 地。-关于地球的运动- / Chi. Chikyuu no Undou ni Tsuite 03 [WebRip 1080p HEVC-10bit AAC ASSx2][简繁日内封字幕]"
|
||||
print(raw_parser(title))
|
||||
title = "[御坂字幕组] 男女之间存在纯友情吗？（不，不存在!!）-01 [WebRip 1080p HEVC10-bit AAC] [简繁日内封] [急招翻校轴]"
|
||||
print(raw_parser(title))
|
||||
result = tokenize_title(raw)
|
||||
if result is None:
|
||||
logger.info("Cannot parse resource: %s, skipping.", raw)
|
||||
return result
|
||||
|
||||
19
backend/src/module/parser/analyser/tokenizer/__init__.py
Normal file
19
backend/src/module/parser/analyser/tokenizer/__init__.py
Normal file
@@ -0,0 +1,19 @@
|
||||
from module.models import Episode
|
||||
|
||||
from . import hot_reload
|
||||
from .stage_classify import classify
|
||||
from .stage_compose import compose
|
||||
from .stage_normalize import normalize
|
||||
from .stage_resolve import resolve
|
||||
from .stage_tokenize import tokenize
|
||||
|
||||
|
||||
def tokenize_title(raw: str) -> Episode | None:
|
||||
hot_reload.maybe_reload()
|
||||
normalized = normalize(raw)
|
||||
tokens = tokenize(normalized)
|
||||
if not tokens:
|
||||
return None
|
||||
tokens = classify(tokens)
|
||||
tokens = resolve(tokens)
|
||||
return compose(tokens)
|
||||
91
backend/src/module/parser/analyser/tokenizer/hot_reload.py
Normal file
91
backend/src/module/parser/analyser/tokenizer/hot_reload.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""Hot-reload keyword dictionaries from a remote JSON file.
|
||||
|
||||
Periodically fetches ``keywords_extra.json`` from the repo's main branch
|
||||
and merges new entries into the in-memory KEYWORDS / CJK_SUBWORDS dicts.
|
||||
Built-in keywords always take precedence — remote entries only add, never override.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import urllib.request
|
||||
|
||||
from .token import TokenKind
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_URL = (
|
||||
"https://raw.githubusercontent.com/EstrellaXD/Auto_Bangumi/"
|
||||
"main/backend/src/module/parser/analyser/tokenizer/keywords_extra.json"
|
||||
)
|
||||
|
||||
RELOAD_INTERVAL = 3600 # seconds
|
||||
|
||||
_last_fetch: float = 0.0
|
||||
_remote_version: int = 0
|
||||
|
||||
|
||||
def maybe_reload() -> None:
|
||||
"""Fetch remote keywords if the reload interval has elapsed. Never raises."""
|
||||
global _last_fetch, _remote_version
|
||||
|
||||
now = time.time()
|
||||
if now - _last_fetch < RELOAD_INTERVAL:
|
||||
return
|
||||
|
||||
try:
|
||||
req = urllib.request.Request(
|
||||
DEFAULT_URL, headers={"User-Agent": "AutoBangumi-HotReload/1.0"}
|
||||
)
|
||||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||||
data = json.loads(resp.read())
|
||||
|
||||
version = data.get("version", 0)
|
||||
if version <= _remote_version:
|
||||
_last_fetch = now
|
||||
return # already up to date
|
||||
|
||||
_merge(data)
|
||||
_remote_version = version
|
||||
_last_fetch = now
|
||||
logger.info(f"Hot-reloaded keywords v{version}")
|
||||
|
||||
except Exception:
|
||||
# Retry after interval, not immediately
|
||||
_last_fetch = now
|
||||
logger.debug("Failed to fetch remote keywords, will retry later")
|
||||
|
||||
|
||||
def _merge(data: dict) -> None:
|
||||
"""Merge remote entries into keyword dicts. Remote never overrides built-in."""
|
||||
from .keywords import CJK_SUBWORDS, KEYWORDS
|
||||
|
||||
added = 0
|
||||
|
||||
for k, v in data.get("keywords", {}).items():
|
||||
key = k.lower()
|
||||
if key not in KEYWORDS:
|
||||
try:
|
||||
KEYWORDS[key] = TokenKind[v]
|
||||
added += 1
|
||||
except KeyError:
|
||||
logger.warning(f"Unknown TokenKind '{v}' for keyword '{k}', skipping")
|
||||
|
||||
for k, v in data.get("cjk_subwords", {}).items():
|
||||
if k not in CJK_SUBWORDS:
|
||||
try:
|
||||
CJK_SUBWORDS[k] = TokenKind[v]
|
||||
added += 1
|
||||
except KeyError:
|
||||
logger.warning(
|
||||
f"Unknown TokenKind '{v}' for CJK subword '{k}', skipping"
|
||||
)
|
||||
|
||||
# prefix_tags are stored as KEYWORDS entries with PREFIX_TAG kind
|
||||
for tag in data.get("prefix_tags", []):
|
||||
if tag not in KEYWORDS:
|
||||
KEYWORDS[tag] = TokenKind.PREFIX_TAG
|
||||
added += 1
|
||||
|
||||
if added:
|
||||
logger.info(f"Merged {added} new keyword(s) from remote")
|
||||
@@ -46,6 +46,7 @@ KEYWORDS: dict[str, TokenKind] = {
|
||||
"mp3": TokenKind.AUDIO,
|
||||
"ac3": TokenKind.AUDIO,
|
||||
"dts": TokenKind.AUDIO,
|
||||
"opus": TokenKind.AUDIO,
|
||||
# Container
|
||||
"mp4": TokenKind.CONTAINER,
|
||||
"mkv": TokenKind.CONTAINER,
|
||||
@@ -62,9 +63,18 @@ KEYWORDS: dict[str, TokenKind] = {
|
||||
"movie": TokenKind.MOVIE,
|
||||
"gekijouban": TokenKind.MOVIE,
|
||||
"the movie": TokenKind.MOVIE,
|
||||
# Special/bonus episode markers
|
||||
"ova": TokenKind.SPECIAL,
|
||||
"oad": TokenKind.SPECIAL,
|
||||
"sp": TokenKind.SPECIAL,
|
||||
"special": TokenKind.SPECIAL,
|
||||
# Version
|
||||
"v2": TokenKind.VERSION,
|
||||
"v3": TokenKind.VERSION,
|
||||
# Prefix tags
|
||||
"整理搬运": TokenKind.PREFIX_TAG,
|
||||
"画质增强": TokenKind.PREFIX_TAG,
|
||||
"极致画质": TokenKind.PREFIX_TAG,
|
||||
# End marker
|
||||
"end": TokenKind.END_MARKER,
|
||||
}
|
||||
@@ -99,12 +109,25 @@ PATTERN_CHECKS: list[tuple[re.Pattern, TokenKind]] = [
|
||||
(re.compile(r"^[剧劇][场場][版]?$|^劇場版?$"), TokenKind.MOVIE),
|
||||
(re.compile(r"^Movie$", re.I), TokenKind.MOVIE),
|
||||
(re.compile(r"^Gekijou-?ban$", re.I), TokenKind.MOVIE),
|
||||
(re.compile(r"^(?:OVA|OAD|SP)\s*[-_]?\s*\d*$", re.I), TokenKind.SPECIAL),
|
||||
(re.compile(r"^Special$", re.I), TokenKind.SPECIAL),
|
||||
# Date pattern
|
||||
(re.compile(r"^\d{4}\.\d{2}\.\d{2}$"), TokenKind.EXTRA),
|
||||
# File size
|
||||
(re.compile(r"^\d+\.\d+ [KMGT]B$", re.I), TokenKind.EXTRA),
|
||||
]
|
||||
|
||||
CJK_SUBWORDS: dict[str, TokenKind] = {
|
||||
"剧场版": TokenKind.MOVIE,
|
||||
"劇場版": TokenKind.MOVIE,
|
||||
"劇場": TokenKind.MOVIE,
|
||||
"剧场": TokenKind.MOVIE,
|
||||
"合集": TokenKind.EXTRA,
|
||||
"总集篇": TokenKind.EXTRA,
|
||||
"總集篇": TokenKind.EXTRA,
|
||||
"特典": TokenKind.EXTRA,
|
||||
}
|
||||
|
||||
CHINESE_NUMBER_MAP: dict[str, int] = {
|
||||
"一": 1,
|
||||
"二": 2,
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"version": 1,
|
||||
"keywords": {},
|
||||
"cjk_subwords": {},
|
||||
"prefix_tags": []
|
||||
}
|
||||
114
backend/src/module/parser/analyser/tokenizer/stage_classify.py
Normal file
114
backend/src/module/parser/analyser/tokenizer/stage_classify.py
Normal file
@@ -0,0 +1,114 @@
|
||||
import re
|
||||
|
||||
from .keywords import CJK_SUBWORDS, KEYWORDS, PATTERN_CHECKS
|
||||
from .token import Token, TokenKind
|
||||
|
||||
_SUB_TOKEN_DELIM = re.compile(r"\s+")
|
||||
|
||||
|
||||
def classify(tokens: list[Token]) -> list[Token]:
|
||||
result: list[Token] = []
|
||||
for token in tokens:
|
||||
classified = _classify_token(token)
|
||||
if classified.kind != TokenKind.UNKNOWN:
|
||||
result.append(classified)
|
||||
continue
|
||||
|
||||
# Split whitespace compounds before scanning the whole token for a CJK
|
||||
# marker. Otherwise "Tenki no Ko 剧场版" becomes one MOVIE token and
|
||||
# the English title is discarded with the marker.
|
||||
sub_tokens = _try_split_compound(classified)
|
||||
if len(sub_tokens) > 1:
|
||||
result.extend(sub_tokens)
|
||||
continue
|
||||
|
||||
if _try_cjk_subword(classified):
|
||||
result.append(classified)
|
||||
continue
|
||||
|
||||
result.append(classified)
|
||||
return result
|
||||
|
||||
|
||||
def _classify_token(token: Token) -> Token:
|
||||
lower = token.text.lower()
|
||||
|
||||
if lower in KEYWORDS:
|
||||
token.kind = KEYWORDS[lower]
|
||||
return token
|
||||
|
||||
for pattern, kind in PATTERN_CHECKS:
|
||||
if pattern.match(token.text):
|
||||
token.kind = kind
|
||||
return token
|
||||
|
||||
return token
|
||||
|
||||
|
||||
def _try_cjk_subword(token: Token) -> bool:
|
||||
"""Scan CJK-dominant tokens for known sub-words. Longest match wins."""
|
||||
text = token.text
|
||||
if not any("\u4e00" <= c <= "\u9fff" or "\u3400" <= c <= "\u4dbf" for c in text):
|
||||
return False
|
||||
best_kind = None
|
||||
best_len = 0
|
||||
for subword, kind in CJK_SUBWORDS.items():
|
||||
if subword in text and len(subword) > best_len:
|
||||
best_kind = kind
|
||||
best_len = len(subword)
|
||||
if best_kind:
|
||||
token.kind = best_kind
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _try_split_compound(token: Token) -> list[Token]:
|
||||
"""Split compound tokens like 'WebRip 1080p HEVC-10bit AAC' or
|
||||
'天国大魔境 Tengoku Daimakyou 第05话 GB 720P MP4'."""
|
||||
parts = _SUB_TOKEN_DELIM.split(token.text)
|
||||
if len(parts) <= 1:
|
||||
return [token]
|
||||
|
||||
classified_parts: list[Token] = []
|
||||
unknown_parts: list[str] = []
|
||||
|
||||
for part in parts:
|
||||
sub = Token(
|
||||
text=part,
|
||||
kind=TokenKind.UNKNOWN,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
sub = _classify_token(sub)
|
||||
if sub.kind == TokenKind.UNKNOWN:
|
||||
_try_cjk_subword(sub)
|
||||
if sub.kind != TokenKind.UNKNOWN:
|
||||
if unknown_parts:
|
||||
merged = " ".join(unknown_parts)
|
||||
classified_parts.append(
|
||||
Token(
|
||||
text=merged,
|
||||
kind=TokenKind.UNKNOWN,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
)
|
||||
unknown_parts = []
|
||||
classified_parts.append(sub)
|
||||
else:
|
||||
unknown_parts.append(part)
|
||||
|
||||
if unknown_parts:
|
||||
merged = " ".join(unknown_parts)
|
||||
classified_parts.append(
|
||||
Token(
|
||||
text=merged,
|
||||
kind=TokenKind.UNKNOWN,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
)
|
||||
|
||||
if any(t.kind != TokenKind.UNKNOWN for t in classified_parts):
|
||||
return classified_parts
|
||||
return [token]
|
||||
@@ -12,6 +12,7 @@ _SEASON_NUM = re.compile(r"[Ss](\d+)|Season\s+(\d+)", re.I)
|
||||
_CHINESE_SEASON = re.compile(r"第([一二三四五六七八九十\d]+)[季期]")
|
||||
_SUB_CLEAN = re.compile(r"_MP4|_MKV", re.I)
|
||||
_CJK_SUB = re.compile(r"[简繁日字幕]")
|
||||
_NON_EPISODIC_EXTRA = re.compile(r"合集|总集篇|總集篇|特典")
|
||||
|
||||
|
||||
def compose(tokens: list[Token]) -> Episode | None:
|
||||
@@ -26,7 +27,9 @@ def compose(tokens: list[Token]) -> Episode | None:
|
||||
resolution: str | None = None
|
||||
source: str | None = None
|
||||
|
||||
is_movie = False
|
||||
episode_type = "episode"
|
||||
has_episode_marker = False
|
||||
has_non_episodic_extra = False
|
||||
|
||||
# Collect all subtitles, prefer CJK ones
|
||||
all_subs: list[str] = []
|
||||
@@ -46,6 +49,7 @@ def compose(tokens: list[Token]) -> Episode | None:
|
||||
elif kind == TokenKind.TITLE_JP:
|
||||
title_jp = text.strip()
|
||||
elif kind == TokenKind.EPISODE:
|
||||
has_episode_marker = True
|
||||
ep_digits = re.search(r"\d+", text)
|
||||
if ep_digits:
|
||||
episode = int(ep_digits.group())
|
||||
@@ -74,7 +78,15 @@ def compose(tokens: list[Token]) -> Episode | None:
|
||||
if source is None:
|
||||
source = text
|
||||
elif kind == TokenKind.MOVIE:
|
||||
is_movie = True
|
||||
episode_type = "movie"
|
||||
elif kind == TokenKind.SPECIAL:
|
||||
episode_type = "special"
|
||||
season = 0
|
||||
special_digits = re.search(r"\d+", text)
|
||||
if special_digits:
|
||||
episode = int(special_digits.group())
|
||||
elif kind == TokenKind.EXTRA and _NON_EPISODIC_EXTRA.search(text):
|
||||
has_non_episodic_extra = True
|
||||
elif kind == TokenKind.SUBTITLE:
|
||||
cleaned = _SUB_CLEAN.sub("", text)
|
||||
if cleaned:
|
||||
@@ -85,6 +97,13 @@ def compose(tokens: list[Token]) -> Episode | None:
|
||||
cjk_subs = [s for s in all_subs if _CJK_SUB.search(s)]
|
||||
sub = cjk_subs[0] if cjk_subs else all_subs[0]
|
||||
|
||||
if (
|
||||
episode_type == "episode"
|
||||
and not has_episode_marker
|
||||
and not has_non_episodic_extra
|
||||
):
|
||||
return None
|
||||
|
||||
return Episode(
|
||||
title_en=title_en,
|
||||
title_zh=title_zh,
|
||||
@@ -96,5 +115,5 @@ def compose(tokens: list[Token]) -> Episode | None:
|
||||
group=group,
|
||||
resolution=resolution,
|
||||
source=source,
|
||||
is_movie=is_movie,
|
||||
episode_type=episode_type,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
from .keywords import FILE_EXTENSIONS
|
||||
|
||||
|
||||
def normalize(raw: str) -> str:
|
||||
s = raw.strip().replace("\n", " ")
|
||||
s = s.replace("【", "[").replace("】", "]")
|
||||
# Note: full-width ／ (\uff0f) is NOT normalized to /
|
||||
# It appears in CJK titles (e.g. Fate／strange Fake) and should be preserved
|
||||
s = FILE_EXTENSIONS.sub("", s)
|
||||
return s
|
||||
450
backend/src/module/parser/analyser/tokenizer/stage_resolve.py
Normal file
450
backend/src/module/parser/analyser/tokenizer/stage_resolve.py
Normal file
@@ -0,0 +1,450 @@
|
||||
import re
|
||||
|
||||
from .token import Token, TokenKind
|
||||
|
||||
_CJK_HAN = re.compile(r"[\u4e00-\u9fff]")
|
||||
_CJK_TRAD = re.compile(r"[\u3400-\u4dbf\uf900-\ufaff]")
|
||||
_KATAKANA_HIRAGANA = re.compile(r"[\u3040-\u309f\u30a0-\u30ff]")
|
||||
_LATIN = re.compile(r"[a-zA-Z]")
|
||||
_PURE_DIGITS = re.compile(r"^\d{1,4}$")
|
||||
_PRE_EP = re.compile(r"^(\d+)[Pp]re$")
|
||||
_COMPOUND_EP = re.compile(r"^(\d+)\(\d+\)$")
|
||||
_SE_COMBINED = re.compile(r"^[Ss](\d{1,2})[Ee](\d+)$")
|
||||
_ATLAS_EP = re.compile(r"^(\d+)_(.+)$")
|
||||
_SEASON_EMBEDDED = re.compile(r"^(.*?)\s*第([一二三四五六七八九十\d]+)[季期]$")
|
||||
_NUM_PREFIX_CJK = re.compile(r"^(\d+)\s+([\u4e00-\u9fff].*)$")
|
||||
_PREFIX_TAG_RE = re.compile(r"^\d+月新番$|^新番$|^★.*★$")
|
||||
_PREFIX_STRIP_RE = re.compile(r"^★?\d*月?新番\s*|^★\d+月新番★?\s*")
|
||||
_REGION_RE = re.compile(r"[(（](?:仅限)?港澳台地区[）)]")
|
||||
_RECRUIT_RE = re.compile(
|
||||
r"[(（]?(?:字幕社?)?招[募人].*[）)]?$|急招翻校轴|搜索用[：:].*$"
|
||||
)
|
||||
|
||||
# Episode suffixed to title: "title-01", "title - 08"
|
||||
_TRAILING_EP = re.compile(r"^(.*?)\s*-\s*(\d{1,4})\s*$")
|
||||
# Episode at end of free text: "title 03" (digits after space, before brackets)
|
||||
_TRAILING_DIGITS = re.compile(r"^(.*\S)\s+(\d{1,4})\s*$")
|
||||
|
||||
_TITLE_UNDERSCORE_SPLIT = re.compile(r"_")
|
||||
_YEAR_RANGE = range(1990, 2031)
|
||||
_TRAILING_PART = re.compile(r"\bPart\s*$", re.I)
|
||||
_PURE_HAN_WORD = re.compile(r"^[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]+$")
|
||||
|
||||
|
||||
def _has_cjk(text: str) -> bool:
|
||||
return bool(_CJK_HAN.search(text) or _CJK_TRAD.search(text))
|
||||
|
||||
|
||||
def _has_jp(text: str) -> bool:
|
||||
return bool(_KATAKANA_HIRAGANA.search(text))
|
||||
|
||||
|
||||
def _has_latin(text: str) -> bool:
|
||||
return bool(_LATIN.search(text))
|
||||
|
||||
|
||||
def _classify_title_lang(text: str) -> TokenKind:
|
||||
if _has_jp(text):
|
||||
return TokenKind.TITLE_JP
|
||||
if _has_cjk(text):
|
||||
return TokenKind.TITLE_ZH
|
||||
if _has_latin(text):
|
||||
return TokenKind.TITLE_EN
|
||||
return TokenKind.TITLE_EN
|
||||
|
||||
|
||||
def _split_word_at_cjk_boundary(word: str) -> list[str]:
|
||||
"""Split a single word at the boundary between Latin and CJK characters.
|
||||
|
||||
E.g. 'MOON：月之神坛' → ['MOON：', '月之神坛']
|
||||
Only splits if the word has both Latin and CJK characters.
|
||||
"""
|
||||
if not (_has_latin(word) and (_has_cjk(word) or _has_jp(word))):
|
||||
return [word]
|
||||
# Find the first CJK character position
|
||||
for i, ch in enumerate(word):
|
||||
if _CJK_HAN.match(ch) or _CJK_TRAD.match(ch) or _KATAKANA_HIRAGANA.match(ch):
|
||||
left = word[:i].rstrip("：:·・")
|
||||
right = word[i:]
|
||||
if left and right:
|
||||
return [left, right]
|
||||
break
|
||||
return [word]
|
||||
|
||||
|
||||
def _split_mixed_title(text: str) -> list[str]:
|
||||
"""Split a mixed CJK/Latin title into language segments."""
|
||||
if "_" in text:
|
||||
parts = _TITLE_UNDERSCORE_SPLIT.split(text, 1)
|
||||
if len(parts) == 2:
|
||||
p1, p2 = parts[0].strip(), parts[1].strip()
|
||||
if p1 and p2 and _has_cjk(p1) != _has_cjk(p2):
|
||||
return [p1, p2]
|
||||
if p1 and p2 and _has_cjk(p1) and _has_cjk(p2):
|
||||
return [p1, p2]
|
||||
|
||||
words = text.split(" ")
|
||||
if len(words) < 2:
|
||||
# Single word — try splitting at CJK boundary within the word
|
||||
return _split_word_at_cjk_boundary(text)
|
||||
|
||||
# Some release names concatenate Chinese, Japanese and English titles
|
||||
# without slash separators, for example:
|
||||
# 想星的阿克艾利昂 情感神话 想星のアクエリオン Aquarion: Myth...
|
||||
# Preserve the leading Chinese title instead of folding it into the first
|
||||
# kana-containing block (which would classify the whole block as JP).
|
||||
if _PURE_HAN_WORD.match(words[0]):
|
||||
jp_idx = next((i for i, word in enumerate(words[1:], 1) if _has_jp(word)), None)
|
||||
if jp_idx is not None:
|
||||
en_idx = next(
|
||||
(
|
||||
i
|
||||
for i, word in enumerate(words[jp_idx + 1 :], jp_idx + 1)
|
||||
if _has_latin(word) and not _has_cjk(word) and not _has_jp(word)
|
||||
),
|
||||
None,
|
||||
)
|
||||
if en_idx is not None:
|
||||
middle = " ".join(words[1:en_idx]).strip()
|
||||
english = " ".join(words[en_idx:]).strip()
|
||||
if middle and english:
|
||||
return [words[0], middle, english]
|
||||
|
||||
# Pre-split words that contain both CJK and Latin at their boundary
|
||||
expanded: list[str] = []
|
||||
for word in words:
|
||||
expanded.extend(_split_word_at_cjk_boundary(word))
|
||||
|
||||
# Classify each fragment as CJK or Latin
|
||||
groups: list[tuple[str, str]] = []
|
||||
for frag in expanded:
|
||||
if (
|
||||
_CJK_HAN.search(frag)
|
||||
or _CJK_TRAD.search(frag)
|
||||
or _KATAKANA_HIRAGANA.search(frag)
|
||||
):
|
||||
groups.append(("cjk", frag))
|
||||
else:
|
||||
groups.append(("lat", frag))
|
||||
|
||||
# Only split if there's exactly one language transition
|
||||
# (CJK block → Latin block or vice versa, no interleaving)
|
||||
transitions = sum(
|
||||
1 for i in range(1, len(groups)) if groups[i][0] != groups[i - 1][0]
|
||||
)
|
||||
if transitions != 1:
|
||||
return [text]
|
||||
|
||||
first_lang = groups[0][0]
|
||||
split_idx = next(i for i in range(1, len(groups)) if groups[i][0] != first_lang)
|
||||
|
||||
part1 = " ".join(w for _, w in groups[:split_idx])
|
||||
part2 = " ".join(w for _, w in groups[split_idx:])
|
||||
|
||||
if not part1 or not part2:
|
||||
return [text]
|
||||
|
||||
return [part1, part2]
|
||||
|
||||
|
||||
_STANDALONE_JUNK = re.compile(r"^[-—–\s]+$")
|
||||
|
||||
|
||||
def resolve(tokens: list[Token]) -> list[Token]:
|
||||
_filter_junk_tokens(tokens)
|
||||
_resolve_se_combined(tokens)
|
||||
_resolve_pre_episodes(tokens)
|
||||
_resolve_compound_episodes(tokens)
|
||||
_resolve_atlas_episodes(tokens)
|
||||
_resolve_trailing_episodes(tokens)
|
||||
_resolve_bare_episode(tokens)
|
||||
_resolve_group(tokens)
|
||||
_resolve_prefix_tags(tokens)
|
||||
_resolve_region_recruit(tokens)
|
||||
_split_mixed_language_tokens(tokens)
|
||||
_resolve_titles(tokens)
|
||||
return tokens
|
||||
|
||||
|
||||
def _filter_junk_tokens(tokens: list[Token]) -> None:
|
||||
"""Mark standalone dashes and whitespace-only tokens as EXTRA.
|
||||
Also strip trailing dash-space from tokens (Doomdos format artifacts)."""
|
||||
for token in tokens:
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
continue
|
||||
if _STANDALONE_JUNK.match(token.text):
|
||||
token.kind = TokenKind.EXTRA
|
||||
continue
|
||||
# Strip leading/trailing separator dashes (space-dash or dash-space)
|
||||
cleaned = re.sub(r"^\s*-\s+|\s+-\s*$", "", token.text).strip()
|
||||
if cleaned and cleaned != token.text:
|
||||
token.text = cleaned
|
||||
|
||||
|
||||
def _resolve_se_combined(tokens: list[Token]) -> None:
|
||||
for token in tokens:
|
||||
if token.kind == TokenKind.EPISODE:
|
||||
m = _SE_COMBINED.match(token.text)
|
||||
if m:
|
||||
token.text = m.group(2)
|
||||
season_tok = Token(
|
||||
text=f"S{m.group(1)}",
|
||||
kind=TokenKind.SEASON,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
idx = tokens.index(token)
|
||||
tokens.insert(idx, season_tok)
|
||||
|
||||
|
||||
def _resolve_pre_episodes(tokens: list[Token]) -> None:
|
||||
for token in tokens:
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
continue
|
||||
m = _PRE_EP.match(token.text)
|
||||
if m:
|
||||
token.kind = TokenKind.EPISODE
|
||||
token.text = m.group(1)
|
||||
|
||||
|
||||
def _resolve_compound_episodes(tokens: list[Token]) -> None:
|
||||
for token in tokens:
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
continue
|
||||
m = _COMPOUND_EP.match(token.text)
|
||||
if m:
|
||||
token.kind = TokenKind.EPISODE
|
||||
token.text = m.group(1)
|
||||
|
||||
|
||||
def _resolve_atlas_episodes(tokens: list[Token]) -> None:
|
||||
for token in tokens:
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
continue
|
||||
m = _ATLAS_EP.match(token.text)
|
||||
if m and m.group(2) and _has_cjk(m.group(2)):
|
||||
token.kind = TokenKind.EPISODE
|
||||
token.text = m.group(1)
|
||||
|
||||
|
||||
def _resolve_trailing_episodes(tokens: list[Token]) -> None:
|
||||
"""Extract episode numbers from the end of UNKNOWN tokens.
|
||||
|
||||
Handles:
|
||||
- 'title-01' or 'title - 08' (dash-separated)
|
||||
- 'title 03' (space-separated, only when no episode found yet)
|
||||
"""
|
||||
episodes_found = any(t.kind == TokenKind.EPISODE for t in tokens)
|
||||
if episodes_found:
|
||||
return
|
||||
|
||||
for i, token in enumerate(tokens):
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
continue
|
||||
|
||||
# Try dash-separated episode: "title-01"
|
||||
m = _TRAILING_EP.match(token.text)
|
||||
if m and m.group(1).strip():
|
||||
title_part = m.group(1).strip()
|
||||
ep_text = m.group(2)
|
||||
# Don't extract if the "title" part is just digits or very short
|
||||
if _has_cjk(title_part) or _has_latin(title_part):
|
||||
token.text = title_part
|
||||
ep_tok = Token(
|
||||
text=ep_text,
|
||||
kind=TokenKind.EPISODE,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
tokens.insert(i + 1, ep_tok)
|
||||
return
|
||||
|
||||
# Try trailing digits: "title 03"
|
||||
m = _TRAILING_DIGITS.match(token.text)
|
||||
if m and m.group(1).strip():
|
||||
title_part = m.group(1).strip()
|
||||
ep_text = m.group(2)
|
||||
if _is_year(ep_text):
|
||||
continue
|
||||
# Skip "Part 2", "Part2" patterns
|
||||
if _TRAILING_PART.search(title_part):
|
||||
continue
|
||||
# Only if the title part has meaningful content
|
||||
if (
|
||||
_has_cjk(title_part) or _has_latin(title_part)
|
||||
) and not _PURE_DIGITS.match(title_part):
|
||||
token.text = title_part
|
||||
ep_tok = Token(
|
||||
text=ep_text,
|
||||
kind=TokenKind.EPISODE,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
tokens.insert(i + 1, ep_tok)
|
||||
return
|
||||
|
||||
|
||||
def _is_year(text: str) -> bool:
|
||||
"""Check if a pure-digit string looks like a year (1990-2030)."""
|
||||
if len(text) != 4:
|
||||
return False
|
||||
try:
|
||||
return int(text) in _YEAR_RANGE
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def _resolve_bare_episode(tokens: list[Token]) -> None:
|
||||
episodes_found = any(t.kind == TokenKind.EPISODE for t in tokens)
|
||||
if episodes_found:
|
||||
return
|
||||
|
||||
unknowns = [t for t in tokens if t.kind == TokenKind.UNKNOWN]
|
||||
for token in unknowns:
|
||||
if not _PURE_DIGITS.match(token.text):
|
||||
continue
|
||||
# Skip year-like numbers (e.g. "(2026)" in enclosed, or free-text years)
|
||||
if _is_year(token.text):
|
||||
token.kind = TokenKind.EXTRA
|
||||
continue
|
||||
if token.enclosed:
|
||||
token.kind = TokenKind.EPISODE
|
||||
return
|
||||
if token.position > 0:
|
||||
idx = tokens.index(token)
|
||||
if idx > 0:
|
||||
prev = tokens[idx - 1]
|
||||
if prev.kind in (
|
||||
TokenKind.UNKNOWN,
|
||||
TokenKind.TITLE_ZH,
|
||||
TokenKind.TITLE_EN,
|
||||
TokenKind.TITLE_JP,
|
||||
TokenKind.SEASON,
|
||||
TokenKind.EXTRA,
|
||||
):
|
||||
token.kind = TokenKind.EPISODE
|
||||
return
|
||||
|
||||
|
||||
def _resolve_group(tokens: list[Token]) -> None:
|
||||
for token in tokens:
|
||||
if token.enclosed and token.kind == TokenKind.UNKNOWN:
|
||||
token.kind = TokenKind.GROUP
|
||||
return
|
||||
|
||||
|
||||
def _resolve_prefix_tags(tokens: list[Token]) -> None:
|
||||
for token in tokens:
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
continue
|
||||
if _PREFIX_TAG_RE.match(token.text):
|
||||
token.kind = TokenKind.PREFIX_TAG
|
||||
continue
|
||||
stripped = _PREFIX_STRIP_RE.sub("", token.text).strip()
|
||||
if stripped != token.text:
|
||||
token.text = stripped
|
||||
|
||||
|
||||
def _resolve_region_recruit(tokens: list[Token]) -> None:
|
||||
for token in tokens:
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
continue
|
||||
if _REGION_RE.search(token.text):
|
||||
token.text = _REGION_RE.sub("", token.text).strip()
|
||||
if not token.text:
|
||||
token.kind = TokenKind.EXTRA
|
||||
if _RECRUIT_RE.search(token.text):
|
||||
token.text = _RECRUIT_RE.sub("", token.text).strip()
|
||||
if not token.text:
|
||||
token.kind = TokenKind.EXTRA
|
||||
|
||||
|
||||
def _split_mixed_language_tokens(tokens: list[Token]) -> None:
|
||||
"""Split UNKNOWN tokens that contain both CJK and Latin text."""
|
||||
i = 0
|
||||
while i < len(tokens):
|
||||
token = tokens[i]
|
||||
if token.kind != TokenKind.UNKNOWN:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
text = token.text.strip()
|
||||
if not text or not (_has_cjk(text) and _has_latin(text)):
|
||||
i += 1
|
||||
continue
|
||||
|
||||
parts = _split_mixed_title(text)
|
||||
if len(parts) <= 1:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
new_tokens = []
|
||||
for p in parts:
|
||||
p = p.strip()
|
||||
if p:
|
||||
new_tokens.append(
|
||||
Token(
|
||||
text=p,
|
||||
kind=TokenKind.UNKNOWN,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
)
|
||||
|
||||
if len(new_tokens) > 1:
|
||||
tokens[i : i + 1] = new_tokens
|
||||
i += len(new_tokens)
|
||||
else:
|
||||
i += 1
|
||||
|
||||
|
||||
def _resolve_titles(tokens: list[Token]) -> None:
|
||||
unknowns = [t for t in tokens if t.kind == TokenKind.UNKNOWN]
|
||||
|
||||
has_zh = False
|
||||
has_en = False
|
||||
has_jp = False
|
||||
|
||||
for token in unknowns:
|
||||
text = token.text.strip()
|
||||
if not text:
|
||||
token.kind = TokenKind.EXTRA
|
||||
continue
|
||||
|
||||
# Extract embedded season
|
||||
season_match = _SEASON_EMBEDDED.match(text)
|
||||
if season_match:
|
||||
title_part = season_match.group(1).strip()
|
||||
season_str = season_match.group(2)
|
||||
token.text = title_part
|
||||
text = title_part
|
||||
season_raw = f"第{season_str}季"
|
||||
season_tok = Token(
|
||||
text=season_raw,
|
||||
kind=TokenKind.SEASON,
|
||||
position=token.position,
|
||||
enclosed=token.enclosed,
|
||||
)
|
||||
idx = tokens.index(token)
|
||||
tokens.insert(idx + 1, season_tok)
|
||||
|
||||
# Handle "number + CJK" titles
|
||||
num_match = _NUM_PREFIX_CJK.match(text)
|
||||
if num_match and not has_zh:
|
||||
token.kind = TokenKind.TITLE_ZH
|
||||
has_zh = True
|
||||
continue
|
||||
|
||||
lang = _classify_title_lang(text)
|
||||
|
||||
if lang == TokenKind.TITLE_JP and not has_jp:
|
||||
token.kind = TokenKind.TITLE_JP
|
||||
has_jp = True
|
||||
elif lang == TokenKind.TITLE_ZH and not has_zh:
|
||||
token.kind = TokenKind.TITLE_ZH
|
||||
has_zh = True
|
||||
elif lang == TokenKind.TITLE_EN and not has_en:
|
||||
token.kind = TokenKind.TITLE_EN
|
||||
has_en = True
|
||||
else:
|
||||
token.kind = TokenKind.EXTRA
|
||||
@@ -0,0 +1,64 @@
|
||||
import re
|
||||
|
||||
from .token import Token, TokenKind
|
||||
|
||||
# Only square and ASCII round brackets are structural delimiters
|
||||
# Full-width （） are treated as regular text (they're CJK punctuation in titles)
|
||||
_OPEN_BRACKETS = {"[": "]", "(": ")"}
|
||||
_CLOSE_BRACKETS = {"]", ")"}
|
||||
_FREE_DELIMITERS = re.compile(r"\s+-\s+|/|\s{2,}")
|
||||
_ENCLOSED_DELIMITERS = re.compile(r"/")
|
||||
|
||||
|
||||
def _add_parts(
|
||||
text: str, pos: int, enclosed: bool, delim: re.Pattern, tokens: list[Token]
|
||||
) -> int:
|
||||
parts = delim.split(text)
|
||||
for part in parts:
|
||||
part = part.strip()
|
||||
if part:
|
||||
tokens.append(
|
||||
Token(
|
||||
text=part, kind=TokenKind.UNKNOWN, position=pos, enclosed=enclosed
|
||||
)
|
||||
)
|
||||
pos += 1
|
||||
return pos
|
||||
|
||||
|
||||
def tokenize(normalized: str) -> list[Token]:
|
||||
tokens: list[Token] = []
|
||||
pos = 0
|
||||
i = 0
|
||||
n = len(normalized)
|
||||
|
||||
while i < n:
|
||||
ch = normalized[i]
|
||||
|
||||
if ch in _OPEN_BRACKETS:
|
||||
close = _OPEN_BRACKETS[ch]
|
||||
j = normalized.find(close, i + 1)
|
||||
if j == -1:
|
||||
j = n
|
||||
inner = normalized[i + 1 : j].strip()
|
||||
if inner:
|
||||
pos = _add_parts(inner, pos, True, _ENCLOSED_DELIMITERS, tokens)
|
||||
i = j + 1
|
||||
|
||||
elif ch in _CLOSE_BRACKETS:
|
||||
i += 1
|
||||
|
||||
else:
|
||||
j = i
|
||||
while (
|
||||
j < n
|
||||
and normalized[j] not in _OPEN_BRACKETS
|
||||
and normalized[j] not in _CLOSE_BRACKETS
|
||||
):
|
||||
j += 1
|
||||
free = normalized[i:j].strip()
|
||||
if free:
|
||||
pos = _add_parts(free, pos, False, _FREE_DELIMITERS, tokens)
|
||||
i = j
|
||||
|
||||
return tokens
|
||||
@@ -21,6 +21,7 @@ class TokenKind(Enum):
|
||||
END_MARKER = auto()
|
||||
HASH = auto()
|
||||
MOVIE = auto() # movie/gekijouban marker
|
||||
SPECIAL = auto() # OVA/OAD/SP/Special marker
|
||||
EXTRA = auto()
|
||||
|
||||
|
||||
|
||||
@@ -10,11 +10,7 @@ import pytest
|
||||
|
||||
from module.manager.renamer import Renamer
|
||||
from module.models import EpisodeFile
|
||||
from module.parser.analyser.raw_parser import (
|
||||
get_group,
|
||||
process,
|
||||
raw_parser,
|
||||
)
|
||||
from module.parser.analyser.raw_parser import get_group, raw_parser
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Issue #986: Parser fails on [group][title][episode_text] format
|
||||
@@ -27,7 +23,7 @@ from module.parser.analyser.raw_parser import (
|
||||
|
||||
|
||||
class TestIssue986AtlasSubGroupFormat:
|
||||
"""Issue #986: Parser crashes on Atlas subtitle group naming convention."""
|
||||
"""Issue #986: Parser now handles Atlas subtitle group naming convention."""
|
||||
|
||||
ATLAS_TITLES = [
|
||||
"[阿特拉斯字幕组·雪原市出差所][命运-奇异赝品_Fate／strange Fake][04_半神们的卡农曲][简繁日内封PGS][日语配音版_Japanese Dub][Web-DL Remux][1080p AVC AAC].mkv",
|
||||
@@ -35,44 +31,28 @@ class TestIssue986AtlasSubGroupFormat:
|
||||
"[阿特拉斯字幕组·雪原市出差所][命运-奇异赝品_Fate／strange Fake][03_无英灵的战斗][简繁日内封PGS][日语配音版_Japanese Dub][Web-DL Remux][1080p AVC AAC].mkv",
|
||||
]
|
||||
|
||||
def test_get_group_extracts_atlas_group(self):
|
||||
"""get_group should extract the group name from [group][title][ep] format."""
|
||||
name = "[阿特拉斯字幕组·雪原市出差所][命运-奇异赝品_Fate／strange Fake][04_半神们的卡农曲]"
|
||||
group = get_group(name)
|
||||
assert group == "阿特拉斯字幕组·雪原市出差所"
|
||||
|
||||
def test_process_returns_none_for_atlas_format(self):
|
||||
"""process() currently returns None for Atlas format (bug demonstration)."""
|
||||
title = self.ATLAS_TITLES[0]
|
||||
result = process(title)
|
||||
# BUG: process returns None because TITLE_RE doesn't match this format
|
||||
assert result is None, (
|
||||
"If this passes, the parser still can't handle Atlas format. "
|
||||
"If it fails (result is not None), the bug may have been fixed!"
|
||||
)
|
||||
|
||||
def test_raw_parser_returns_none_for_atlas_format(self):
|
||||
"""raw_parser returns None for Atlas format, causing AttributeError downstream."""
|
||||
title = self.ATLAS_TITLES[0]
|
||||
@pytest.mark.parametrize(
|
||||
"title,expected_ep",
|
||||
[
|
||||
(ATLAS_TITLES[0], 4),
|
||||
(ATLAS_TITLES[1], 7),
|
||||
(ATLAS_TITLES[2], 3),
|
||||
],
|
||||
)
|
||||
def test_atlas_titles_parse_successfully(self, title, expected_ep):
|
||||
"""All Atlas format titles now parse correctly."""
|
||||
result = raw_parser(title)
|
||||
# BUG: returns None → downstream code does .groups() on None → AttributeError
|
||||
assert result is None
|
||||
assert result is not None
|
||||
assert result.group == "阿特拉斯字幕组·雪原市出差所"
|
||||
assert result.title_zh == "命运-奇异赝品"
|
||||
assert result.episode == expected_ep
|
||||
assert result.resolution == "1080p"
|
||||
|
||||
@pytest.mark.parametrize("title", ATLAS_TITLES)
|
||||
def test_atlas_titles_all_fail_to_parse(self, title):
|
||||
"""All Atlas format titles fail to parse."""
|
||||
result = raw_parser(title)
|
||||
assert result is None
|
||||
|
||||
def test_get_group_returns_empty_for_no_brackets(self):
|
||||
"""get_group returns empty string for title without brackets (regression guard)."""
|
||||
result = get_group("No Brackets Title")
|
||||
assert result == ""
|
||||
|
||||
def test_get_group_does_not_crash_on_empty_string(self):
|
||||
"""get_group handles empty string without crashing."""
|
||||
result = get_group("")
|
||||
assert result == ""
|
||||
def test_atlas_title_en_extracted(self):
|
||||
"""Atlas format extracts English title from underscore-separated text."""
|
||||
result = raw_parser(self.ATLAS_TITLES[0])
|
||||
assert result is not None
|
||||
assert result.title_en == "Fate\uff0fstrange Fake"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -213,20 +193,20 @@ class TestIssue976NoneInMatchList:
|
||||
|
||||
def test_get_group_no_brackets_returns_empty(self):
|
||||
"""get_group handles names without brackets (regression for IndexError)."""
|
||||
# The original code did: re.split(r"[\[\]]", name)[1]
|
||||
# which crashes with IndexError when there are no brackets
|
||||
result = get_group("No Brackets At All")
|
||||
assert result == ""
|
||||
assert get_group("No Brackets At All") == ""
|
||||
|
||||
def test_get_group_single_bracket_pair(self):
|
||||
"""get_group extracts group from single bracket pair."""
|
||||
result = get_group("[GroupName] Some Title")
|
||||
assert result == "GroupName"
|
||||
"""get_group extracts the first square-bracket group."""
|
||||
assert get_group("[GroupName] Some Title") == "GroupName"
|
||||
|
||||
def test_get_group_empty_brackets(self):
|
||||
"""get_group handles empty brackets."""
|
||||
result = get_group("[] empty")
|
||||
assert result == ""
|
||||
"""get_group handles an empty bracket pair."""
|
||||
assert get_group("[] empty") == ""
|
||||
|
||||
def test_raw_parser_extracts_group_from_brackets(self):
|
||||
result = raw_parser("[GroupName] Some Title - 01")
|
||||
assert result is not None
|
||||
assert result.group == "GroupName"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -496,25 +476,42 @@ class TestIssue990NumberPrefixTitle:
|
||||
|
||||
|
||||
class TestIssue992NonEpisodicAttributeError:
|
||||
"""Issue #992: title_parser crashes on non-episodic resources."""
|
||||
"""Issue #992: parser failures must not escape as AttributeError."""
|
||||
|
||||
# Titles that raw_parser cannot parse (returns None)
|
||||
NON_EPISODIC_TITLES = [
|
||||
"[阿特拉斯字幕组·雪原市出差所][命运-奇异赝品_Fate／strange Fake][04_半神们的卡农曲][简繁日内封PGS][日语配音版_Japanese Dub][Web-DL Remux][1080p AVC AAC]",
|
||||
"[KitaujiSub] Shikanoko Nokonoko Koshitantan [01Pre][WebRip][HEVC_AAC][CHS_JP].mp4",
|
||||
# These formats triggered the original crash. The tokenizer now parses
|
||||
# both successfully, so keep an integration-level guard for that behavior.
|
||||
FORMERLY_UNPARSEABLE_TITLES = [
|
||||
(
|
||||
"[阿特拉斯字幕组·雪原市出差所][命运-奇异赝品_Fate／strange Fake][04_半神们的卡农曲][简繁日内封PGS][日语配音版_Japanese Dub][Web-DL Remux][1080p AVC AAC]",
|
||||
"Fate／strange Fake",
|
||||
),
|
||||
(
|
||||
"[KitaujiSub] Shikanoko Nokonoko Koshitantan [01Pre][WebRip][HEVC_AAC][CHS_JP].mp4",
|
||||
"Shikanoko Nokonoko Koshitantan",
|
||||
),
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize("title", NON_EPISODIC_TITLES)
|
||||
async def test_title_parser_returns_none_for_non_episodic(self, title):
|
||||
"""TitleParser.raw_parser should return None instead of crashing."""
|
||||
@pytest.mark.parametrize("title,expected_title", FORMERLY_UNPARSEABLE_TITLES)
|
||||
async def test_title_parser_handles_nonstandard_titles(self, title, expected_title):
|
||||
"""Previously crashing formats now produce a valid parsed model."""
|
||||
from module.parser.title_parser import TitleParser
|
||||
|
||||
result = await TitleParser.raw_parser(title)
|
||||
assert result is not None
|
||||
assert result.title_raw == expected_title
|
||||
|
||||
@pytest.mark.parametrize("title", ["", "[]"])
|
||||
async def test_title_parser_returns_none_for_unparseable(self, title):
|
||||
"""A genuine parser miss returns None instead of raising AttributeError."""
|
||||
from module.parser.title_parser import TitleParser
|
||||
|
||||
result = await TitleParser.raw_parser(title)
|
||||
assert result is None
|
||||
|
||||
def test_raw_parser_returns_none_for_unparseable(self):
|
||||
"""raw_parser returns None for resources it cannot parse."""
|
||||
result = raw_parser(self.NON_EPISODIC_TITLES[0])
|
||||
@pytest.mark.parametrize("title", ["", "[]"])
|
||||
def test_raw_parser_returns_none_for_unparseable(self, title):
|
||||
"""raw_parser still returns None when tokenization yields no content."""
|
||||
result = raw_parser(title)
|
||||
assert result is None
|
||||
|
||||
|
||||
|
||||
@@ -136,7 +136,7 @@ def test_raw_parser():
|
||||
assert info.episode == 1
|
||||
assert info.season == 2
|
||||
|
||||
# 4K resolution (2160p) — RESOLUTION_RE covers 2160 but untested
|
||||
# 4K resolution (2160p)
|
||||
content = "[NC-Raws] 葬送的芙莉莲 / Sousou no Frieren - 03 [B-Global][WEB-DL][2160p][AVC AAC][Multi Sub][MKV]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
@@ -147,7 +147,7 @@ def test_raw_parser():
|
||||
assert info.episode == 3
|
||||
assert info.season == 1
|
||||
|
||||
# English "Season N" format (bracketed) — season_rule "Season \d{1,2}" branch
|
||||
# English "Season N" format (bracketed)
|
||||
content = "[LoliHouse] 狼与香辛料 [Season 2] / Spice and Wolf - 01 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
@@ -158,7 +158,7 @@ def test_raw_parser():
|
||||
assert info.episode == 1
|
||||
assert info.season == 2
|
||||
|
||||
# Multi-group, Chinese punctuation in title, single-letter Latin prefix in EN title
|
||||
# Multi-group, Chinese punctuation in title
|
||||
content = "[北宇治字幕组&LoliHouse] 地。-关于地球的运动- / Chi. Chikyuu no Undou ni Tsuite - 03 [WebRip 1080p HEVC-10bit AAC ASSx2][简繁日内封字幕]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
@@ -169,7 +169,7 @@ def test_raw_parser():
|
||||
assert info.episode == 3
|
||||
assert info.season == 1
|
||||
|
||||
# English-only title — name_process returns title_zh=None when no CJK chars
|
||||
# English-only title
|
||||
content = "[动漫国字幕组&LoliHouse] THE MARGINAL SERVICE - 08 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
@@ -205,7 +205,6 @@ class TestIssue924SpecialPunctuation:
|
||||
content = "[御坂字幕组] 男女之间存在纯友情吗？（不，不存在!!）-01 [WebRip 1080p HEVC10-bit AAC] [简繁日内封] [急招翻校轴]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.group == "御坂字幕组"
|
||||
assert info.title_zh == "男女之间存在纯友情吗？（不，不存在!!）"
|
||||
assert info.episode == 1
|
||||
@@ -222,42 +221,39 @@ class TestIssue910NeoQswFormat:
|
||||
def test_parse_neo_qsw_format(self):
|
||||
info = raw_parser(self.TITLE)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.title_zh == "想星的阿克艾利昂"
|
||||
assert info.episode == 2
|
||||
|
||||
|
||||
class TestIssue876NoSeparator:
|
||||
"""Issue #876: Episode number without dash separator.
|
||||
|
||||
Note: the dash-separated variant "- 03" already works (tested in test_raw_parser).
|
||||
This tests the space-only variant "Tsuite 03" which the fallback parser handles.
|
||||
"""
|
||||
"""Issue #876: Episode number without dash separator."""
|
||||
|
||||
TITLE = "[北宇治字幕组&LoliHouse] 地。-关于地球的运动- / Chi. Chikyuu no Undou ni Tsuite 03 [WebRip 1080p HEVC-10bit AAC ASSx2][简繁日内封字幕]"
|
||||
|
||||
def test_parse_without_dash(self):
|
||||
info = raw_parser(self.TITLE)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.title_zh == "地。-关于地球的运动-"
|
||||
assert info.title_en == "Chi. Chikyuu no Undou ni Tsuite"
|
||||
assert info.episode == 3
|
||||
|
||||
|
||||
class TestIssue819ChineseEpisodeMarker:
|
||||
"""Issue #819: [Doomdos] format with 第N话 episode marker."""
|
||||
"""Issue #819: [Doomdos] format with 第N话 episode marker.
|
||||
|
||||
Previously: title_zh included leading/trailing dashes from separators.
|
||||
Fixed: tokenizer strips separator dashes correctly.
|
||||
"""
|
||||
|
||||
def test_parse_chinese_episode_marker(self):
|
||||
content = "[Doomdos] - 白色闪电 - 第02话 - [1080P].mp4"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.group == "Doomdos"
|
||||
assert info.episode == 2
|
||||
assert info.resolution == "1080P"
|
||||
# BUG: title_zh includes leading/trailing dashes from the separator
|
||||
assert info.title_zh == "- 白色闪电 -"
|
||||
# Fixed: title no longer has leading/trailing separator dashes
|
||||
assert info.title_zh == "白色闪电"
|
||||
|
||||
|
||||
class TestIssue811ColonInTitle:
|
||||
@@ -267,7 +263,6 @@ class TestIssue811ColonInTitle:
|
||||
content = "[Up to 21°C] 鬼灭之刃 柱训练篇 / Kimetsu no Yaiba: Hashira Geiko-hen - 03 (CR 1920x1080 AVC AAC MKV)"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.group == "Up to 21°C"
|
||||
assert info.title_zh == "鬼灭之刃 柱训练篇"
|
||||
assert info.title_en == "Kimetsu no Yaiba: Hashira Geiko-hen"
|
||||
@@ -276,68 +271,79 @@ class TestIssue811ColonInTitle:
|
||||
|
||||
|
||||
class TestIssue798VTuberTitle:
|
||||
"""Issue #798: Title with 'VTuber' split incorrectly by name_process."""
|
||||
"""Issue #798: Title with 'VTuber' — previously split incorrectly.
|
||||
|
||||
Fixed: tokenizer keeps the full CJK+Latin title together when
|
||||
languages are interleaved (not a clean CJK/Latin boundary).
|
||||
"""
|
||||
|
||||
def test_parse_vtuber_title(self):
|
||||
content = "[ANi] 身为 VTuber 的我因为忘记关台而成了传说 - 01 [1080P][Baha][WEB-DL][AAC AVC][CHT][MP4][379.34 MB]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.group == "ANi"
|
||||
assert info.episode == 1
|
||||
assert info.resolution == "1080P"
|
||||
assert info.source == "Baha"
|
||||
# BUG: name_process splits on space and only keeps first Chinese word
|
||||
assert info.title_zh == "身为"
|
||||
assert info.title_en == "VTuber 的我因为忘记关台而成了传说"
|
||||
# Fixed: full title preserved (VTuber interleaved in CJK text)
|
||||
assert info.title_zh == "身为 VTuber 的我因为忘记关台而成了传说"
|
||||
|
||||
|
||||
class TestIssue794PreEpisodeFormat:
|
||||
"""Issue #794/#800: [01Pre] episode format not recognized."""
|
||||
"""Issue #794/#800: [01Pre] episode format.
|
||||
|
||||
Fixed: tokenizer handles Pre-episode format correctly.
|
||||
"""
|
||||
|
||||
TITLES = [
|
||||
"[KitaujiSub] Shikanoko Nokonoko Koshitantan [01Pre][WebRip][HEVC_AAC][CHS_JP].mp4",
|
||||
"[KitaujiSub] Shikanoko Nokonoko Koshitantan [01Pre][WebRip][HEVC_AAC][CHT_JP].mp4",
|
||||
]
|
||||
|
||||
@pytest.mark.xfail(reason="[01Pre] episode format not supported by TITLE_RE")
|
||||
def test_parse_pre_episode(self):
|
||||
info = raw_parser(self.TITLES[0])
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.title_en == "Shikanoko Nokonoko Koshitantan"
|
||||
assert info.episode == 1
|
||||
|
||||
@pytest.mark.parametrize("title", TITLES)
|
||||
def test_returns_none(self, title):
|
||||
"""Parser cannot handle [01Pre] format currently."""
|
||||
assert raw_parser(title) is None
|
||||
def test_all_variants_parse(self, title):
|
||||
"""All Pre-episode variants now parse successfully."""
|
||||
info = raw_parser(title)
|
||||
assert info is not None
|
||||
assert info.episode == 1
|
||||
|
||||
|
||||
class TestIssue766Lv2InTitle:
|
||||
"""Issue #766: Title with 'Lv2' causing incorrect name split."""
|
||||
"""Issue #766: Title with 'Lv2' — previously split incorrectly.
|
||||
|
||||
Fixed: tokenizer keeps the full title together when CJK/Latin
|
||||
are interleaved.
|
||||
"""
|
||||
|
||||
def test_parse_lv2_title(self):
|
||||
content = "[ANi] 从 Lv2 开始开外挂的前勇者候补过著悠哉异世界生活 - 04 [1080P][Baha][WEB-DL][AAC AVC][CHT][MP4]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.group == "ANi"
|
||||
assert info.episode == 4
|
||||
assert info.resolution == "1080P"
|
||||
assert info.source == "Baha"
|
||||
# BUG: name_process splits on space, loses the "从 Lv2" prefix
|
||||
assert info.title_zh == "开始开外挂的前勇者候补过著悠哉异世界生活"
|
||||
# Fixed: full title preserved (Lv2 interleaved in CJK text)
|
||||
assert info.title_zh == "从 Lv2 开始开外挂的前勇者候补过著悠哉异世界生活"
|
||||
|
||||
|
||||
class TestIssue764WesternFormat:
|
||||
"""Issue #764: Western release format without group brackets."""
|
||||
"""Issue #764: Western release format without group brackets.
|
||||
|
||||
Fixed: tokenizer correctly extracts title, season, and episode
|
||||
from S01E05 format.
|
||||
"""
|
||||
|
||||
def test_parse_western_format(self):
|
||||
content = "Girls Band Cry S01E05 VOSTFR 1080p WEB x264 AAC -Tsundere-Raws (ADN)"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.episode == 5
|
||||
assert info.season == 1
|
||||
assert info.resolution == "1080p"
|
||||
@@ -350,27 +356,30 @@ class TestIssue764WesternFormat:
|
||||
|
||||
|
||||
class TestIssue986AtlasFormat:
|
||||
"""Issue #986: Atlas subtitle group bracket-delimited format."""
|
||||
"""Issue #986: Atlas subtitle group bracket-delimited format.
|
||||
|
||||
Fixed: tokenizer handles all-bracket format with underscore-separated
|
||||
episode numbers.
|
||||
"""
|
||||
|
||||
TITLES = [
|
||||
"[阿特拉斯字幕组·雪原市出差所][命运-奇异赝品_Fate／strange Fake][04_半神们的卡农曲][简繁日内封PGS][日语配音版_Japanese Dub][Web-DL Remux][1080p AVC AAC]",
|
||||
"[阿特拉斯字幕组·雪原市出差所][命运-奇异赝品_Fate／strange Fake][07_神自黄昏归来][简繁日内封PGS][日语配音版_Japanese Dub][Web-DL Remux][1080p AVC AAC]",
|
||||
]
|
||||
|
||||
@pytest.mark.xfail(
|
||||
reason="Atlas bracket-delimited format not supported by TITLE_RE"
|
||||
)
|
||||
def test_parse_atlas_format(self):
|
||||
info = raw_parser(self.TITLES[0])
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.title_zh == "命运-奇异赝品"
|
||||
assert info.episode == 4
|
||||
|
||||
@pytest.mark.parametrize("title", TITLES)
|
||||
def test_returns_none(self, title):
|
||||
"""Parser cannot handle Atlas format currently."""
|
||||
assert raw_parser(title) is None
|
||||
def test_all_atlas_variants(self, title):
|
||||
"""All Atlas format variants now parse successfully."""
|
||||
info = raw_parser(title)
|
||||
assert info is not None
|
||||
assert info.episode > 0
|
||||
assert info.group == "阿特拉斯字幕组·雪原市出差所"
|
||||
|
||||
|
||||
class TestIssue773CompoundEpisode:
|
||||
@@ -381,7 +390,6 @@ class TestIssue773CompoundEpisode:
|
||||
def test_parse_compound_episode(self):
|
||||
info = raw_parser(self.TITLE)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.title_zh == "鬼灭之刃 柱训练篇"
|
||||
assert info.episode == 2
|
||||
|
||||
@@ -478,7 +486,6 @@ class TestIssue805TitleWithCht:
|
||||
content = "[ANi] 不時輕聲地以俄語遮羞的鄰座艾莉同學 - 02 [1080P][Baha][WEB-DL][AAC AVC][CHT].mp4"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.group == "ANi"
|
||||
assert info.title_zh == "不時輕聲地以俄語遮羞的鄰座艾莉同學"
|
||||
assert info.episode == 2
|
||||
@@ -502,7 +509,6 @@ class TestIssue1025NoGroupPrefix:
|
||||
)
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info is not None
|
||||
assert info.episode == 2
|
||||
assert info.season == 1
|
||||
# Before the fix all three title fields were None and title_parser
|
||||
|
||||
110
backend/src/test/test_token_features.py
Normal file
110
backend/src/test/test_token_features.py
Normal file
@@ -0,0 +1,110 @@
|
||||
"""Tests for CJK sub-word scanning in the tokenizer classify stage."""
|
||||
|
||||
from module.parser.analyser.raw_parser import raw_parser
|
||||
from module.parser.analyser.tokenizer.stage_classify import _try_cjk_subword
|
||||
from module.parser.analyser.tokenizer.token import Token, TokenKind
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CJK sub-word scanning
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCjkSubwordScan:
|
||||
def test_gekijouban_simple(self):
|
||||
token = Token(text="剧场版", kind=TokenKind.UNKNOWN, position=0, enclosed=False)
|
||||
assert _try_cjk_subword(token) is True
|
||||
assert token.kind == TokenKind.MOVIE
|
||||
|
||||
def test_gekijouban_traditional(self):
|
||||
token = Token(text="劇場版", kind=TokenKind.UNKNOWN, position=0, enclosed=False)
|
||||
assert _try_cjk_subword(token) is True
|
||||
assert token.kind == TokenKind.MOVIE
|
||||
|
||||
def test_compound_movie_extra(self):
|
||||
"""'剧场版合集' contains both MOVIE and EXTRA sub-words. MOVIE wins (longer)."""
|
||||
token = Token(
|
||||
text="剧场版合集", kind=TokenKind.UNKNOWN, position=0, enclosed=False
|
||||
)
|
||||
assert _try_cjk_subword(token) is True
|
||||
assert token.kind == TokenKind.MOVIE
|
||||
|
||||
def test_gashuuhen(self):
|
||||
token = Token(text="合集", kind=TokenKind.UNKNOWN, position=0, enclosed=False)
|
||||
assert _try_cjk_subword(token) is True
|
||||
assert token.kind == TokenKind.EXTRA
|
||||
|
||||
def test_soushuurhen(self):
|
||||
token = Token(text="总集篇", kind=TokenKind.UNKNOWN, position=0, enclosed=False)
|
||||
assert _try_cjk_subword(token) is True
|
||||
assert token.kind == TokenKind.EXTRA
|
||||
|
||||
def test_tokuten(self):
|
||||
token = Token(text="特典", kind=TokenKind.UNKNOWN, position=0, enclosed=False)
|
||||
assert _try_cjk_subword(token) is True
|
||||
assert token.kind == TokenKind.EXTRA
|
||||
|
||||
def test_no_match_plain_cjk(self):
|
||||
token = Token(
|
||||
text="进击的巨人", kind=TokenKind.UNKNOWN, position=0, enclosed=False
|
||||
)
|
||||
assert _try_cjk_subword(token) is False
|
||||
assert token.kind == TokenKind.UNKNOWN
|
||||
|
||||
def test_no_match_latin_only(self):
|
||||
token = Token(text="Movie", kind=TokenKind.UNKNOWN, position=0, enclosed=False)
|
||||
assert _try_cjk_subword(token) is False
|
||||
|
||||
def test_no_match_digits(self):
|
||||
token = Token(text="2023", kind=TokenKind.UNKNOWN, position=0, enclosed=False)
|
||||
assert _try_cjk_subword(token) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration: raw_parser end-to-end with CJK sub-words
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMovieCjkSubwordIntegration:
|
||||
def test_gekijouban_gashuuhen_compound(self):
|
||||
"""'剧场版合集' should be recognized as movie."""
|
||||
content = "[SubGroup] 剧场版合集 - 进击的巨人 [1080p]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info.is_movie is True
|
||||
|
||||
def test_traditional_gekijouban_spy_family(self):
|
||||
content = "[SubGroup] 劇場版 SPY×FAMILY 2023 [1080p]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info.is_movie is True
|
||||
|
||||
def test_gekijouban_madoka_gashuuhen(self):
|
||||
content = "[LoliHouse] 劇場版 魔法少女まどか☆マギカ 合集 [BDRip 1080p]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info.is_movie is True
|
||||
|
||||
def test_gashuuhen_not_movie(self):
|
||||
"""'合集' alone should mark as EXTRA, not MOVIE."""
|
||||
content = "[SubGroup] 进击的巨人 合集 [1080p]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info.is_movie is False
|
||||
if info.title_zh:
|
||||
assert "合集" not in info.title_zh
|
||||
|
||||
|
||||
class TestYearClassification:
|
||||
def test_year_not_episode(self):
|
||||
"""'2023' in a title context should not become the episode number."""
|
||||
content = "[SubGroup] 劇場版 SPY×FAMILY 2023 [1080p]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info.episode != 2023
|
||||
|
||||
def test_year_in_existing_format(self):
|
||||
"""Year in date format should stay as EXTRA."""
|
||||
content = "[梦蓝字幕组]New Doraemon 哆啦A梦新番[747][2023.02.25][AVC][1080P][GB_JP][MP4]"
|
||||
info = raw_parser(content)
|
||||
assert info is not None
|
||||
assert info.episode == 747
|
||||
Reference in New Issue
Block a user