refactor(parser): support generic anime release titles

This commit is contained in:
Estrella Pan
2026-07-11 13:29:23 +02:00
parent fd1936fb1d
commit 3f1feb8655
18 changed files with 1595 additions and 1183 deletions

View File

@@ -126,10 +126,10 @@ class Episode:
season: int
season_raw: str
episode: int
sub: str
sub: Optional[str]
group: str
resolution: str
source: str
resolution: Optional[str]
source: Optional[str]
episode_type: str = "episode" # "episode" | "movie" | "special"
is_movie: bool = False

View File

@@ -1,28 +1,16 @@
import logging
import re
from module.models import Episode
from module.parser.analyser.tokenizer import tokenize_title
from module.parser.analyser.tokenizer import parse_release_title, tokenize_title
from module.parser.analyser.tokenizer.compat import legacy_non_episodic_type
logger = logging.getLogger(__name__)
_MOVIE_TOKEN_RE = re.compile(
r"剧场版|劇場版|电影版|電影版|\bmovie\b|\bgekijou-?ban\b",
re.IGNORECASE,
)
_SPECIAL_TOKEN_RE = re.compile(
r"\bOVA\d*\b|\bOAD\d*\b|\bSP\d*\b|\bspecial\b|番外篇?|特别篇|特別篇",
re.IGNORECASE,
)
_LEADING_GROUP_RE = re.compile(r"^\s*[\[【][^\]】]+[\]】]\s*")
_GROUP_RE = re.compile(r"\[([^\]]*)\]")
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 ""
parsed = parse_release_title(raw)
return parsed.group if parsed and parsed.group else ""
def _detect_non_episodic_type(raw: str) -> str | None:
@@ -32,12 +20,10 @@ def _detect_non_episodic_type(raw: str) -> str | None:
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):
return "special"
return None
parsed = parse_release_title(raw)
if parsed is None:
return None
return legacy_non_episodic_type(parsed.media_type)
def raw_parser(raw: str) -> Episode | None:

View File

@@ -1,19 +1,13 @@
from module.models import Episode
"""Public APIs for generic and legacy resource-title parsing."""
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
from .compat import tokenize_title
from .parser import parse_release_title
from .result import MediaType, ParsedRelease, ReleaseKind
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)
__all__ = [
"MediaType",
"ParsedRelease",
"ReleaseKind",
"parse_release_title",
"tokenize_title",
]

View File

@@ -0,0 +1,80 @@
"""Compatibility projection from generic tokenizer results to ``Episode``."""
from __future__ import annotations
from module.models import Episode
from .parser import parse_release_title
from .result import MediaType, ParsedRelease, ReleaseKind
_SPECIAL_MEDIA = {
MediaType.OVA,
MediaType.OAD,
MediaType.SPECIAL,
}
def to_legacy_episode(parsed: ParsedRelease) -> Episode | None:
"""Project a generic result onto the historical parser contract.
The old API is intentionally conservative: a plain title with neither an
episode nor a strong resource-kind marker remains unparseable. New callers
should use :func:`parse_release_title` to receive such partial results.
"""
if not parsed.primary_title:
return None
if parsed.release_kind in {ReleaseKind.RANGE, ReleaseKind.BATCH}:
return None
if parsed.media_type in {
MediaType.PV,
MediaType.OPENING,
MediaType.ENDING,
}:
return None
if isinstance(parsed.episode, float):
return None
if (
parsed.episode is None
and parsed.media_type is MediaType.UNKNOWN
and parsed.release_kind is not ReleaseKind.COLLECTION
):
return None
if parsed.media_type is MediaType.MOVIE:
episode_type = "movie"
season = parsed.season if parsed.season is not None else 1
elif parsed.media_type in _SPECIAL_MEDIA:
episode_type = "special"
season = 0
else:
episode_type = "episode"
season = parsed.season if parsed.season is not None else 1
return Episode(
title_en=parsed.title_en,
title_zh=parsed.title_zh,
title_jp=parsed.title_jp,
season=season,
season_raw=parsed.season_raw or "",
episode=parsed.episode if parsed.episode is not None else 0,
sub=parsed.subtitle,
group=parsed.group or "",
resolution=parsed.resolution,
source=parsed.source,
episode_type=episode_type,
)
def legacy_non_episodic_type(media_type: MediaType) -> str | None:
"""Map a generic media type to the legacy non-episodic classifier."""
if media_type is MediaType.MOVIE:
return "movie"
if media_type in _SPECIAL_MEDIA:
return "special"
return None
def tokenize_title(raw: str) -> Episode | None:
"""Return the historical ``Episode`` projection used by RSS callers."""
parsed = parse_release_title(raw)
return to_legacy_episode(parsed) if parsed is not None else None

View File

@@ -1,91 +0,0 @@
"""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")

View File

@@ -1,144 +0,0 @@
import re
from .token import TokenKind
KEYWORDS: dict[str, TokenKind] = {
# Resolution
"1080p": TokenKind.RESOLUTION,
"720p": TokenKind.RESOLUTION,
"2160p": TokenKind.RESOLUTION,
"4k": TokenKind.RESOLUTION,
"1080": TokenKind.RESOLUTION,
"720": TokenKind.RESOLUTION,
"2160": TokenKind.RESOLUTION,
"1920x1080": TokenKind.RESOLUTION,
"3840x2160": TokenKind.RESOLUTION,
# Source
"baha": TokenKind.SOURCE,
"b-global": TokenKind.SOURCE,
"bilibili": TokenKind.SOURCE,
"at-x": TokenKind.SOURCE,
"web-dl": TokenKind.SOURCE,
"webrip": TokenKind.SOURCE,
"webdl": TokenKind.SOURCE,
"web": TokenKind.SOURCE,
"web-dl remux": TokenKind.SOURCE,
"ttfc": TokenKind.SOURCE,
"cr": TokenKind.SOURCE,
"adn": TokenKind.SOURCE,
"iqiyi": TokenKind.SOURCE,
"abema": TokenKind.SOURCE,
# Codec
"hevc": TokenKind.CODEC,
"avc": TokenKind.CODEC,
"x264": TokenKind.CODEC,
"x265": TokenKind.CODEC,
"hevc-10bit": TokenKind.CODEC,
"hevc10-bit": TokenKind.CODEC,
"h264": TokenKind.CODEC,
"h265": TokenKind.CODEC,
"h.264": TokenKind.CODEC,
"h.265": TokenKind.CODEC,
"10bit": TokenKind.CODEC,
# Audio
"aac": TokenKind.AUDIO,
"flac": TokenKind.AUDIO,
"mp3": TokenKind.AUDIO,
"ac3": TokenKind.AUDIO,
"dts": TokenKind.AUDIO,
"opus": TokenKind.AUDIO,
# Container
"mp4": TokenKind.CONTAINER,
"mkv": TokenKind.CONTAINER,
"avi": TokenKind.CONTAINER,
# Subtitle markers
"cht": TokenKind.SUBTITLE,
"chs": TokenKind.SUBTITLE,
"big5": TokenKind.SUBTITLE,
"gb": TokenKind.SUBTITLE,
"assx2": TokenKind.SUBTITLE,
"multi sub": TokenKind.SUBTITLE,
"vostfr": TokenKind.SUBTITLE,
# Movie markers
"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,
}
PATTERN_CHECKS: list[tuple[re.Pattern, TokenKind]] = [
# Episode patterns
(re.compile(r"^[Ee][Pp]?\d+$"), TokenKind.EPISODE),
(re.compile(r"^第?\d+[话話集]$"), TokenKind.EPISODE),
(re.compile(r"^\d+[Pp]$"), TokenKind.RESOLUTION),
(re.compile(r"^\d{3,4}[xX]\d{3,4}$"), TokenKind.RESOLUTION),
# Season patterns
(re.compile(r"^[Ss]\d{1,2}$"), TokenKind.SEASON),
(re.compile(r"^Season \d+$", re.I), TokenKind.SEASON),
(re.compile(r"^第[一二三四五六七八九十\d]+[季期]$"), TokenKind.SEASON),
# S01E05 combined pattern
(re.compile(r"^[Ss]\d{1,2}[Ee]\d+$"), TokenKind.EPISODE),
# Subtitle patterns
(re.compile(r"^[简繁日字幕][简繁日字幕体文]*"), TokenKind.SUBTITLE),
(re.compile(r"^GB_?(?:JP|MP4|MKV)?$", re.I), TokenKind.SUBTITLE),
(re.compile(r"^CHS_JP$", re.I), TokenKind.SUBTITLE),
(re.compile(r"^CHT_JP$", re.I), TokenKind.SUBTITLE),
# Prefix tags
(re.compile(r"^\d+月新番$"), TokenKind.PREFIX_TAG),
(re.compile(r"^新番$"), TokenKind.PREFIX_TAG),
(re.compile(r"^★.*★$"), TokenKind.PREFIX_TAG),
# Version pattern
(re.compile(r"^[vV]\d$"), TokenKind.VERSION),
# Hash (CRC32)
(re.compile(r"^[0-9A-Fa-f]{8}$"), TokenKind.HASH),
# Movie markers
(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,
"三": 3,
"四": 4,
"五": 5,
"六": 6,
"七": 7,
"八": 8,
"九": 9,
"十": 10,
}
FILE_EXTENSIONS = re.compile(r"\.(mp4|mkv|avi|ass|srt|ssa|sub)$", re.I)

View File

@@ -1,6 +0,0 @@
{
"version": 1,
"keywords": {},
"cjk_subwords": {},
"prefix_tags": []
}

View File

@@ -0,0 +1,14 @@
"""Input normalization shared by the release-title parser."""
from __future__ import annotations
import re
_FILE_EXTENSION = re.compile(r"\.(mp4|mkv|avi|ass|srt|ssa|sub)$", re.I)
def normalize(raw: str) -> str:
value = raw.strip().replace("\n", " ")
value = value.replace("【", "[").replace("】", "]")
# Full-width slashes and parentheses are title punctuation, not structure.
return _FILE_EXTENSION.sub("", value)

View File

@@ -0,0 +1,934 @@
"""Generic, evidence-oriented parser for anime resource names.
Unlike the former ordered mutation pipeline, this module never reclassifies an
entire title fragment because it contains one marker. It records exact matched
spans, removes only those spans, and reconstructs titles from the remaining
text.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from .normalization import normalize
from .result import MediaType, ParsedRelease, ReleaseKind
_TITLE_CHAR = re.compile(r"[A-Za-z\u3040-\u30ff\u3400-\u9fff\uf900-\ufaff]")
_HAN = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]")
_PURE_HAN = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]+")
_KANA = re.compile(r"[\u3040-\u30ff]")
_LATIN = re.compile(r"[A-Za-z]")
_NUMERIC_TITLE = re.compile(r"\d{1,3}(?:\s*[/.-]\s*\d{1,3})?")
_DATE = re.compile(r"(?<!\d)(?:19|20)\d{2}[.\-/]\d{1,2}[.\-/]\d{1,2}(?!\d)")
_YEAR = re.compile(r"(?<!\d)((?:18|19|20|21)\d{2})(?!\d)")
_FILE_SIZE = re.compile(r"(?<!\w)\d+(?:\.\d+)?\s*[KMGT]i?B(?!\w)", re.I)
_HASH = re.compile(r"(?<![0-9A-F])[0-9A-F]{8}(?![0-9A-F])", re.I)
_RANGE = re.compile(
r"(?<![\w.])(?:E(?:P)?\.?\s*)?(\d{1,4}(?:\.\d+)?)\s*"
r"(?:-|~|~|—)\s*(?:E(?:P)?\.?\s*)?(\d{1,4}(?:\.\d+)?)"
r"(?:v(\d+))?(?![\w.])",
re.I,
)
_SEASON_EPISODE = re.compile(
r"(?<!\w)S(\d{1,2})\s*E(?:P)?\.?\s*(\d{1,4}(?:\.\d+)?)" r"(?:v(\d+))?(?!\w)",
re.I,
)
_SEASON_EPISODE_WORDS = re.compile(
r"\bSeason\s+(\d{1,2})\s+(?:Episode|EP?\.?)\s*" r"(\d{1,4}(?:\.\d+)?)(?:v(\d+))?\b",
re.I,
)
_SEASON = re.compile(r"(?<!\w)(S\d{1,2}|Season\s+\d{1,2})(?!\w)", re.I)
_ORDINAL_SEASON = re.compile(r"\b(\d{1,2})(?:st|nd|rd|th)\s+Season\b", re.I)
_CHINESE_SEASON = re.compile(r"第([零〇一二两三四五六七八九十百\d]+)[季期]")
_EXPLICIT_EPISODE = re.compile(
r"(?<!\w)(?:Episode|EP?\.?|#)\s*[-_. ]?\s*" r"(\d{1,4}(?:\.\d+)?)(?:v(\d+))?(?!\w)",
re.I,
)
_CHINESE_EPISODE = re.compile(r"(?:第\s*)?(\d{1,4}(?:\.\d+)?)\s*[话話集]")
_FULL_COLLECTION = re.compile(r"全\s*(\d{1,4})\s*[话話集]")
_PRE_EPISODE = re.compile(r"^\s*(\d{1,4})Pre\s*$", re.I)
_COMPOUND_EPISODE = re.compile(r"^\s*(\d{1,4})\(\d+\)\s*$")
_ATLAS_EPISODE = re.compile(r"^\s*(\d{1,4})_(?:.*[\u3040-\u30ff\u3400-\u9fff].*)$")
_BARE_EPISODE = re.compile(r"^\s*-?\s*(\d{1,4}(?:\.\d+)?)(?:v(\d+))?\s*$", re.I)
_TRAILING_EPISODE = re.compile(
r"(?:\s+-\s*|(?<=[^\d\s])-\s*|\s+)(\d{1,4}(?:\.\d+)?)(?:v(\d+))?\s*$",
re.I,
)
_SPECIAL_NUMBER = re.compile(
r"(?<![A-Za-z0-9])(OVA|OAD|SP|Special)\s*(?:[-_.]\s*)?"
r"(\d{1,4}(?:\.\d+)?)(?:v(\d+))?(?![A-Za-z0-9])",
re.I,
)
_SPECIAL_WORD = re.compile(r"(?<![A-Za-z0-9])(OVA|OAD|SP|Special)(?![A-Za-z0-9])", re.I)
_SPECIAL_CJK = re.compile(r"番外篇?|特別篇|特别篇")
_PV = re.compile(
r"(?<![A-Za-z0-9])(PV|CM)(?:\s*[-_.]?\s*(\d{1,3}))?(?![A-Za-z0-9=])",
re.I,
)
_NCOP = re.compile(r"(?<![A-Za-z0-9])NCOP(?:v?(\d+))?(?![A-Za-z0-9])", re.I)
_NCED = re.compile(r"(?<![A-Za-z0-9])NCED(?:v?(\d+))?(?![A-Za-z0-9])", re.I)
_MOVIE_CJK = re.compile(r"劇場版?|剧场版?|電影版|电影版")
_MOVIE_ROMAJI = re.compile(r"(?<!\w)Gekijou-?ban(?!\w)", re.I)
_MOVIE_EN = re.compile(r"(?<!\w)(?:The\s+)?Movie(?!\w)", re.I)
_BATCH = re.compile(
r"(?<!\w)(?:Complete(?:\s+(?:Series|Season))?(?:\s+Batch)?|Batch)(?!\w)",
re.I,
)
_COLLECTION = re.compile(r"合集|全集|全[话話集]|总集篇|總集篇|特典")
_RESOLUTION = re.compile(
r"(?<![A-Za-z0-9])(?:\d{3,4}[xX]\d{3,4}|(?:720|1080|2160)[pP](?:60)?|4[Kk])(?![A-Za-z0-9])"
)
_SOURCE = re.compile(
r"(?<!\w)(?:WEB[-_. ]?DL(?:\s+Remux)?|WebRip|BDRip|BluRay|Blu-Ray|"
r"B-Global|Baha|Bilibili|AT-X|TTFC|CR|ADN|iQIYI|Abema|AMZN|NF|"
r"HDTV|DVD|Remux|WEB)(?!\w)",
re.I,
)
_CODEC = re.compile(
r"(?<![A-Za-z0-9])(?:HEVC(?:[-_. ]?10[- ]?bit)?|AVC|x26[45]|H\.?26[45]|"
r"(?:8|10|12)bit|Ma10p)(?![A-Za-z0-9])",
re.I,
)
_AUDIO = re.compile(
r"(?<![A-Za-z0-9])(?:AAC|FLAC|MP3|AC-?3|E-?AC-?3|DDP(?:\d(?:\.\d)?)?|"
r"DTS|Opus|Dual[-_ ]?Audio)(?![A-Za-z0-9])",
re.I,
)
_CONTAINER = re.compile(r"(?<![A-Za-z0-9])(MP4|MKV|AVI)(?![A-Za-z0-9])", re.I)
_SUBTITLE = re.compile(
r"(?<![A-Za-z0-9])(?:CHS(?:_JP)?|CHT(?:_JP)?|GB(?:_(?:JP|MP4|MKV))?|Big5|"
r"VOSTFR|ENG|Multi[-_ ]?Subs?|Multiple[-_ ]?Subtitles?|ASSx2|PGS)"
r"(?![A-Za-z0-9])",
re.I,
)
_CJK_SUBTITLE_TAG = re.compile(
r"[简簡繁日英中字幕体體双雙语語内內封嵌外挂掛多国國]+(?:PGS)?", re.I
)
_VERSION = re.compile(r"(?<!\w)[vV](\d+)(?!\w)")
_PREFIX = re.compile(r"^\s*(?:★?\s*\d{1,2}月新番\s*★?|★?\s*新番\s*★?)")
_RECRUIT = re.compile(r"[((]?(?:字幕社?)?招[募人].*[))]?$|急招翻校轴|搜索用[::].*$")
_REGION = re.compile(r"[((](?:仅限)?港澳台地区[))]")
_DUB = re.compile(r"(?:配音版|Dub)", re.I)
_TRAILING_RELEASE_GROUP = re.compile(r"\s*-[\w·&.-]+(?:Raws?|Subs?)\s*$", re.I)
@dataclass(frozen=True, slots=True)
class _Segment:
text: str
start: int
end: int
enclosure: str | None = None
class _WorkingSegment:
def __init__(self, segment: _Segment):
self.segment = segment
self.mask = [False] * len(segment.text)
def available(self, match: re.Match[str]) -> bool:
return not any(self.mask[match.start() : match.end()])
def consume(self, match: re.Match[str]) -> None:
self.mask[match.start() : match.end()] = [True] * (match.end() - match.start())
def consume_span(self, start: int, end: int) -> None:
self.mask[start:end] = [True] * (end - start)
def residual(self) -> str:
return "".join(
" " if used else char for char, used in zip(self.segment.text, self.mask)
)
@dataclass(slots=True)
class _State:
raw: str
group: str | None = None
season: int | None = None
season_raw: str | None = None
episode: int | float | None = None
episode_end: int | float | None = None
episode_title: str | None = None
episode_priority: int = -1
media_type: MediaType = MediaType.UNKNOWN
media_priority: int = -1
release_kind: ReleaseKind = ReleaseKind.SINGLE
resolution: str | None = None
source: str | None = None
subtitle: str | None = None
codecs: list[str] = field(default_factory=list)
audio: list[str] = field(default_factory=list)
container: str | None = None
version: int | None = None
year: int | None = None
tags: list[str] = field(default_factory=list)
evidence: list[str] = field(default_factory=list)
def set_episode(
self,
episode: int | float,
*,
priority: int,
end: int | float | None = None,
version: int | None = None,
) -> None:
if priority >= self.episode_priority:
self.episode = episode
self.episode_end = end
self.episode_priority = priority
if version is not None:
self.version = version
def set_media(self, media_type: MediaType, priority: int) -> None:
if priority > self.media_priority:
self.media_type = media_type
self.media_priority = priority
def parse_release_title(raw: str) -> ParsedRelease | None:
"""Parse a resource name without requiring an episode number."""
if not raw or not raw.strip():
return None
normalized = normalize(raw)
segments = _scan_segments(normalized)
if not segments:
return None
state = _State(raw=raw)
group_indices = _find_group_segments(segments)
if group_indices:
state.group = "&".join(segments[index].text.strip() for index in group_indices)
state.evidence.append("group")
working = [_WorkingSegment(segment) for segment in segments]
for index in group_indices:
working[index].mask[:] = [True] * len(working[index].mask)
_preclassify_movie(working, state)
_extract_numbers(working, state)
_extract_media_and_cardinality(working, state)
_extract_technical_metadata(working, state)
title_en, title_zh, title_jp = _reconstruct_titles(working)
if not any((title_en, title_zh, title_jp)):
return None
if state.media_type is MediaType.UNKNOWN:
if state.episode is not None or state.release_kind is not ReleaseKind.SINGLE:
state.set_media(MediaType.EPISODE, 0)
return ParsedRelease(
raw=raw,
title_en=title_en,
title_zh=title_zh,
title_jp=title_jp,
group=state.group,
season=state.season,
season_raw=state.season_raw,
episode=state.episode,
episode_end=state.episode_end,
episode_title=state.episode_title,
media_type=state.media_type,
release_kind=state.release_kind,
resolution=state.resolution,
source=state.source,
subtitle=state.subtitle,
codecs=tuple(state.codecs),
audio=tuple(state.audio),
container=state.container,
version=state.version,
year=state.year,
tags=tuple(state.tags),
evidence=tuple(state.evidence),
)
def _scan_segments(text: str) -> list[_Segment]:
segments: list[_Segment] = []
free_start = 0
index = 0
bracket_pairs = {"[": "]", "(": ")"}
def add_free(start: int, end: int) -> None:
content = text[start:end]
if content.strip():
segments.append(_Segment(content, start, end))
while index < len(text):
opener = text[index]
if opener not in bracket_pairs:
index += 1
continue
add_free(free_start, index)
closer = bracket_pairs[opener]
depth = 1
cursor = index + 1
while cursor < len(text) and depth:
if text[cursor] == opener:
depth += 1
elif text[cursor] == closer:
depth -= 1
cursor += 1
content_end = cursor - 1 if depth == 0 else len(text)
content = text[index + 1 : content_end]
if content.strip():
enclosure = "square" if opener == "[" else "round"
segments.append(_Segment(content, index, cursor, enclosure))
index = cursor
free_start = cursor
add_free(free_start, len(text))
return segments
def _find_group_segments(segments: list[_Segment]) -> set[int]:
if not segments or segments[0].enclosure != "square":
return set()
first = segments[0].text.strip()
if _looks_like_metadata(first):
return set()
has_free_title = any(
segment.enclosure is None
and _contains_title(segment.text)
and not _looks_like_metadata(segment.text)
for segment in segments[1:]
)
if not (_looks_like_group(first) or has_free_title):
return set()
indices = {0}
for index, segment in enumerate(segments[1:], 1):
if segment.enclosure != "square" or not _looks_like_group(segment.text):
break
indices.add(index)
return indices
def _looks_like_group(text: str) -> bool:
value = text.strip()
lower = value.lower()
if re.search(r"字幕[组組社]?|搬运|搬運|压制|壓制|发布|發佈", value):
return True
if re.search(r"raws?|fansub|subs?|house|studio|group|team|rip", lower):
return True
if lower in {"ani", "official", "magicstar", "doomdos", "vcb-studio"}:
return True
if "&" in value and len(value) <= 48:
return True
return bool(
len(value) <= 16 and " " not in value and re.fullmatch(r"[A-Z0-9.-]+", value)
)
def _looks_like_metadata(text: str) -> bool:
value = text.strip()
if not value:
return True
structural_patterns = (
_RANGE,
_SEASON_EPISODE,
_SEASON_EPISODE_WORDS,
_SEASON,
_CHINESE_SEASON,
_FULL_COLLECTION,
_CHINESE_EPISODE,
_BARE_EPISODE,
_SPECIAL_NUMBER,
_SPECIAL_WORD,
_SPECIAL_CJK,
_PV,
_NCOP,
_NCED,
_MOVIE_CJK,
_MOVIE_ROMAJI,
_MOVIE_EN,
_BATCH,
_COLLECTION,
_DATE,
_YEAR,
_FILE_SIZE,
)
if any(pattern.fullmatch(value) for pattern in structural_patterns):
return True
if _CJK_SUBTITLE_TAG.fullmatch(value) or value.lower() in {"end", "生"}:
return True
residual = value
technical_patterns = (
_RESOLUTION,
_SOURCE,
_CODEC,
_AUDIO,
_CONTAINER,
_SUBTITLE,
)
for pattern in technical_patterns:
residual = pattern.sub(" ", residual)
residual = re.sub(r"[\s_+.,-]+", "", residual)
return not _contains_title(residual)
def _contains_title(text: str) -> bool:
return bool(_TITLE_CHAR.search(text))
def _contains_title_fragment(text: str) -> bool:
return _contains_title(text) or bool(_NUMERIC_TITLE.fullmatch(text.strip()))
def _number(value: str) -> int | float:
number = float(value)
return int(number) if number.is_integer() else number
def _is_year_number(value: str) -> bool:
return value.isdigit() and len(value) == 4 and 1800 <= int(value) <= 2199
def _consume_all(work: _WorkingSegment, pattern: re.Pattern[str]):
for match in pattern.finditer(work.segment.text):
if work.available(match):
yield match
def _preclassify_movie(working: list[_WorkingSegment], state: _State) -> None:
"""Record strong movie evidence before resolving ambiguous trailing numbers."""
for work in working:
for pattern in (_MOVIE_CJK, _MOVIE_ROMAJI):
if any(True for _ in _consume_all(work, pattern)):
state.set_media(MediaType.MOVIE, 100)
return
for match in _consume_all(work, _MOVIE_EN):
after = work.segment.text[match.end() :].lstrip()
if not re.match(r"[A-Za-z]", after):
state.set_media(MediaType.MOVIE, 100)
return
def _is_metadata_tail(text: str) -> bool:
residual = text
for pattern in (
_RESOLUTION,
_SOURCE,
_CODEC,
_AUDIO,
_CONTAINER,
_SUBTITLE,
_TRAILING_EPISODE,
):
residual = pattern.sub(" ", residual)
residual = re.sub(r"[\s_+.,\-–—]+", "", residual)
return not _contains_title(residual)
def _extract_episode_title(
work: _WorkingSegment, match: re.Match[str], state: _State
) -> None:
"""Separate Emby/Plex ``Series - SxxExx - Episode title`` layouts."""
before = work.segment.text[: match.start()]
separator = re.search(r"\s+-\s*$", before)
if separator:
work.consume_span(separator.start(), match.start())
after = work.segment.text[match.end() :]
episode_title = re.fullmatch(r"\s*-\s*(.+?)\s*", after)
if episode_title and _contains_title_fragment(episode_title.group(1)):
state.episode_title = _clean_title(episode_title.group(1))
work.consume_span(match.end(), len(work.segment.text))
state.evidence.append("episode-title")
def _extract_numbers(working: list[_WorkingSegment], state: _State) -> None:
for work in working:
text = work.segment.text
for match in _consume_all(work, _DATE):
work.consume(match)
state.tags.append(match.group())
for match in _consume_all(work, _FILE_SIZE):
work.consume(match)
state.tags.append(match.group())
for match in _consume_all(work, _YEAR):
if match.group() == text.strip() or _is_metadata_tail(text[match.end() :]):
work.consume(match)
state.year = int(match.group(1))
state.evidence.append("year")
for match in _consume_all(work, _RANGE):
start, end = _number(match.group(1)), _number(match.group(2))
prefix = text[: match.start()]
suffix = text[match.end() :]
compact = bool(
re.search(r"\d(?:-|~|~|—)(?:E(?:P)?\.?)?\d", match.group(), re.I)
)
isolated = not _contains_title(prefix + suffix)
labelled = bool(re.search(r"(?:Episodes?|Eps?)\s*$", prefix, re.I))
if (
start >= 1800
or end >= 1800
or start > end
or re.search(r"\d+\s*/\s*$", prefix)
or not (compact or isolated or labelled)
):
continue
work.consume(match)
state.set_episode(start, priority=85, end=end, version=_int(match.group(3)))
state.release_kind = ReleaseKind.RANGE
state.evidence.append("episode-range")
for pattern in (_SEASON_EPISODE, _SEASON_EPISODE_WORDS):
for match in _consume_all(work, pattern):
work.consume(match)
_extract_episode_title(work, match, state)
state.season = int(match.group(1))
state.season_raw = match.group(0)
state.set_episode(
_number(match.group(2)),
priority=100,
version=_int(match.group(3)),
)
state.evidence.extend(("season", "episode"))
for match in _consume_all(work, _SPECIAL_NUMBER):
work.consume(match)
media_type = _special_media(match.group(1))
state.set_media(media_type, 80)
state.set_episode(
_number(match.group(2)),
priority=80,
version=_int(match.group(3)),
)
state.evidence.extend((media_type.value, "episode"))
for match in _consume_all(work, _PV):
if match.group(2) is None:
continue
work.consume(match)
state.set_media(MediaType.PV, 60)
state.set_episode(_number(match.group(2)), priority=75)
state.evidence.extend(("pv", "episode"))
for match in _consume_all(work, _SEASON):
work.consume(match)
raw_season = match.group(1)
number_match = re.search(r"\d+", raw_season)
if number_match:
state.season = int(number_match.group())
state.season_raw = raw_season
state.evidence.append("season")
for match in _consume_all(work, _CHINESE_SEASON):
work.consume(match)
state.season = _chinese_number(match.group(1))
state.season_raw = match.group(0)
state.evidence.append("season")
for match in _consume_all(work, _ORDINAL_SEASON):
if state.season is None:
work.consume(match)
state.season = int(match.group(1))
state.season_raw = match.group(0)
state.evidence.append("season")
for match in _consume_all(work, _FULL_COLLECTION):
work.consume(match)
state.set_episode(1, priority=95, end=int(match.group(1)))
state.release_kind = ReleaseKind.COLLECTION
state.evidence.extend(("collection", "episode-range"))
for match in _consume_all(work, _EXPLICIT_EPISODE):
work.consume(match)
state.set_episode(
_number(match.group(1)), priority=90, version=_int(match.group(2))
)
state.evidence.append("episode")
for match in _consume_all(work, _CHINESE_EPISODE):
work.consume(match)
state.set_episode(_number(match.group(1)), priority=90)
state.evidence.append("episode")
for pattern in (_PRE_EPISODE, _COMPOUND_EPISODE, _ATLAS_EPISODE):
match = pattern.match(text)
if match and work.available(match):
work.consume(match)
state.set_episode(_number(match.group(1)), priority=75)
state.evidence.append("episode")
trailing = _TRAILING_EPISODE.search(text)
if (
trailing
and work.available(trailing)
and not _is_year_number(trailing.group(1))
and state.media_type is not MediaType.MOVIE
):
work.consume(trailing)
state.set_episode(
_number(trailing.group(1)),
priority=70,
version=_int(trailing.group(2)),
)
state.evidence.append("episode")
bare = _BARE_EPISODE.match(text)
if (
bare
and work.available(bare)
and not _is_year_number(bare.group(1))
and state.media_type is not MediaType.MOVIE
):
work.consume(bare)
state.set_episode(
_number(bare.group(1)), priority=70, version=_int(bare.group(2))
)
state.evidence.append("episode")
def _extract_media_and_cardinality(
working: list[_WorkingSegment], state: _State
) -> None:
for work in working:
text = work.segment.text
for match in _consume_all(work, _MOVIE_CJK):
work.consume(match)
state.set_media(MediaType.MOVIE, 100)
state.evidence.append("movie")
for match in _consume_all(work, _MOVIE_ROMAJI):
work.consume(match)
state.set_media(MediaType.MOVIE, 100)
state.evidence.append("movie")
for match in _consume_all(work, _MOVIE_EN):
before = text[: match.start()].rstrip()
after = text[match.end() :].lstrip()
marker_is_edge = (
not before or not after or work.segment.enclosure == "square"
)
followed_by_title_word = bool(re.match(r"[A-Za-z]", after))
if followed_by_title_word:
continue
if not marker_is_edge and state.episode is not None:
continue
work.consume(match)
state.set_media(MediaType.MOVIE, 100)
state.evidence.append("movie")
for match in _consume_all(work, _SPECIAL_WORD):
after = text[match.end() :].lstrip()
if re.match(r"[A-Za-z]", after):
continue
media_type = _special_media(match.group(1))
work.consume(match)
state.set_media(media_type, 80)
state.evidence.append(media_type.value)
for match in _consume_all(work, _SPECIAL_CJK):
work.consume(match)
state.set_media(MediaType.SPECIAL, 80)
state.evidence.append("special")
for match in _consume_all(work, _PV):
work.consume(match)
state.set_media(MediaType.PV, 60)
state.evidence.append("pv")
for match in _consume_all(work, _NCOP):
work.consume(match)
state.set_media(MediaType.OPENING, 60)
state.version = state.version or _int(match.group(1))
state.evidence.append("opening")
for match in _consume_all(work, _NCED):
work.consume(match)
state.set_media(MediaType.ENDING, 60)
state.version = state.version or _int(match.group(1))
state.evidence.append("ending")
for match in _consume_all(work, _BATCH):
work.consume(match)
state.release_kind = ReleaseKind.BATCH
state.evidence.append("batch")
for match in _consume_all(work, _COLLECTION):
work.consume(match)
state.release_kind = ReleaseKind.COLLECTION
state.evidence.append("collection")
def _extract_technical_metadata(working: list[_WorkingSegment], state: _State) -> None:
for work in working:
text = work.segment.text
for match in _consume_all(work, _RESOLUTION):
work.consume(match)
state.resolution = state.resolution or match.group()
state.evidence.append("resolution")
for match in _consume_all(work, _SOURCE):
work.consume(match)
state.source = state.source or match.group()
state.evidence.append("source")
for match in _consume_all(work, _CODEC):
work.consume(match)
_append_unique(state.codecs, match.group())
for match in _consume_all(work, _AUDIO):
work.consume(match)
_append_unique(state.audio, match.group())
for match in _consume_all(work, _SUBTITLE):
work.consume(match)
state.subtitle = _prefer_subtitle(state.subtitle, match.group())
container = re.search(r"_(MP4|MKV)$", match.group(), re.I)
if container:
state.container = state.container or container.group(1)
for match in _consume_all(work, _CONTAINER):
work.consume(match)
state.container = state.container or match.group(1)
if work.segment.enclosure == "square":
subtitle_tag = _clean_fragment(work.residual())
if _CJK_SUBTITLE_TAG.fullmatch(subtitle_tag):
work.mask[:] = [True] * len(work.mask)
state.subtitle = _prefer_subtitle(state.subtitle, subtitle_tag)
if _DUB.search(text):
work.mask[:] = [True] * len(work.mask)
state.tags.append(text.strip())
for match in _consume_all(work, _VERSION):
work.consume(match)
state.version = state.version or int(match.group(1))
for pattern in (_PREFIX, _RECRUIT, _REGION, _TRAILING_RELEASE_GROUP):
for match in _consume_all(work, pattern):
work.consume(match)
state.tags.append(match.group().strip())
for match in _consume_all(work, _HASH):
work.consume(match)
state.tags.append(match.group())
_extract_episode_after_metadata(work, state)
residual = _clean_fragment(work.residual())
if work.segment.enclosure in {"square", "round"} and residual.lower() in {
"end",
"生",
}:
work.mask[:] = [True] * len(work.mask)
state.tags.append(residual)
def _extract_episode_after_metadata(work: _WorkingSegment, state: _State) -> None:
"""Resolve episode numbers mixed into an enclosed technical tag."""
if (
state.episode is not None
or state.media_type is MediaType.MOVIE
or work.segment.enclosure != "square"
):
return
residual = work.residual()
match = _BARE_EPISODE.match(residual) or _TRAILING_EPISODE.search(residual)
if not match or _is_year_number(match.group(1)):
return
work.consume_span(match.start(), match.end())
state.set_episode(
_number(match.group(1)), priority=65, version=_int(match.group(2))
)
state.evidence.append("episode")
def _reconstruct_titles(
working: list[_WorkingSegment],
) -> tuple[str | None, str | None, str | None]:
fragments: list[str] = []
for work in working:
residual = _clean_fragment(work.residual())
if not residual or not _contains_title_fragment(residual):
continue
if work.segment.enclosure == "round":
residual = f"({residual})"
fragments.extend(_split_explicit_titles(residual))
titles: dict[MediaType | str, list[str]] = {"en": [], "zh": [], "jp": []}
for fragment in fragments:
for part in _split_mixed_title(fragment):
cleaned = _clean_title(part)
if not cleaned or not _contains_title_fragment(cleaned):
continue
language = _title_language(cleaned)
if cleaned not in titles[language]:
titles[language].append(cleaned)
return (
_join_title_parts(titles["en"]),
_join_title_parts(titles["zh"]),
_join_title_parts(titles["jp"]),
)
def _split_explicit_titles(text: str) -> list[str]:
spaced = [part.strip() for part in re.split(r"\s+[/|]\s+", text) if part.strip()]
if len(spaced) > 1:
return spaced
for match in re.finditer(r"/", text):
left, right = text[: match.start()].strip(), text[match.end() :].strip()
if not left or not right:
continue
if left[-1].isdigit() or right[0].isdigit():
continue
if _title_language(left) != _title_language(right):
return [left, right]
return [text]
def _split_mixed_title(text: str) -> list[str]:
underscore = re.split(r"_(?=[A-Za-z\u3040-\u30ff\u3400-\u9fff])", text, maxsplit=1)
if len(underscore) == 2 and _title_language(underscore[0]) != _title_language(
underscore[1]
):
return underscore
cjk_to_latin = re.search(
r"(?<=[\u3040-\u30ff\u3400-\u9fff])\s*[-–—]\s*(?=[A-Z][A-Za-z]+\s+[A-Za-z])",
text,
)
if cjk_to_latin:
return [text[: cjk_to_latin.start()], text[cjk_to_latin.end() :]]
words = text.split()
if len(words) < 2:
runs: list[tuple[str, int]] = []
for index, char in enumerate(text):
language = (
"en"
if _LATIN.match(char)
else "cjk" if (_HAN.match(char) or _KANA.match(char)) else None
)
if language is not None and (not runs or runs[-1][0] != language):
runs.append((language, index))
if len(runs) == 2 and runs[0][0] == "en":
boundary = runs[1][1]
return [text[:boundary], text[boundary:]]
return [text]
classes = [_word_language(word) for word in words]
if words[0].isdigit() and classes[1] in {"zh", "jp"}:
return [text]
if words[-1].isdigit() and classes[0] in {"zh", "jp"}:
return [text]
transitions = [
index
for index in range(1, len(classes))
if classes[index] != classes[index - 1]
]
if len(transitions) == 1:
index = transitions[0]
return [" ".join(words[:index]), " ".join(words[index:])]
if _PURE_HAN.fullmatch(words[0]) and "jp" in classes[1:] and classes[-1] == "en":
jp_index = classes.index("jp")
en_index = next(
(
index
for index in range(jp_index + 1, len(classes))
if classes[index] == "en"
),
-1,
)
if en_index > jp_index:
return [
words[0],
" ".join(words[1:en_index]),
" ".join(words[en_index:]),
]
return [text]
def _word_language(text: str) -> str:
if _KANA.search(text):
return "jp"
if _HAN.search(text):
return "zh"
return "en"
def _title_language(text: str) -> str:
if _KANA.search(text):
return "jp"
if _HAN.search(text):
return "zh"
return "en"
def _clean_fragment(text: str) -> str:
text = re.sub(r"\s+", " ", text).strip()
text = re.sub(r"^(?:[-–—]\s+)+|(?:\s+[-–—])+$", "", text).strip()
text = re.sub(r"^[\s_|]+|[\s_|]+$", "", text).strip()
text = re.sub(r"^/\s*|\s*/$", "", text).strip()
return re.sub(r"\s+", " ", text).strip()
def _clean_title(text: str) -> str:
text = _clean_fragment(text)
return re.sub(r"_+", " ", text).strip()
def _join_title_parts(parts: list[str]) -> str | None:
return " ".join(parts).strip() if parts else None
def _mostly_metadata(text: str) -> bool:
return bool(re.fullmatch(r"[\w\s+_.-]{1,48}", text))
def _prefer_subtitle(current: str | None, candidate: str) -> str:
if current is None or _HAN.search(candidate):
return re.sub(r"_(?:MP4|MKV)$", "", candidate, flags=re.I)
return current
def _append_unique(values: list[str], value: str) -> None:
if value.lower() not in {item.lower() for item in values}:
values.append(value)
def _special_media(marker: str) -> MediaType:
marker = marker.lower()
if marker == "ova":
return MediaType.OVA
if marker == "oad":
return MediaType.OAD
return MediaType.SPECIAL
def _int(value: str | None) -> int | None:
return int(value) if value is not None else None
def _chinese_number(value: str) -> int:
if value.isdigit():
return int(value)
digits = {
"零": 0,
"〇": 0,
"一": 1,
"二": 2,
"两": 2,
"三": 3,
"四": 4,
"五": 5,
"六": 6,
"七": 7,
"八": 8,
"九": 9,
}
if "百" in value:
left, _, right = value.partition("百")
return digits.get(left, 1) * 100 + _chinese_number(right or "零")
if "十" in value:
left, _, right = value.partition("十")
return digits.get(left, 1) * 10 + digits.get(right, 0)
return digits.get(value, 1)

View File

@@ -0,0 +1,70 @@
"""Domain result returned by the release-title tokenizer.
The tokenizer deliberately keeps missing values as ``None``. Defaults such as
season 1 and episode 0 belong to the legacy adapter, not to the parser itself.
"""
from __future__ import annotations
from dataclasses import dataclass
from enum import StrEnum
class MediaType(StrEnum):
"""Kind of media represented by a release."""
UNKNOWN = "unknown"
EPISODE = "episode"
MOVIE = "movie"
OVA = "ova"
OAD = "oad"
SPECIAL = "special"
PV = "pv"
OPENING = "opening"
ENDING = "ending"
class ReleaseKind(StrEnum):
"""Cardinality of a release, independent from its media type."""
SINGLE = "single"
RANGE = "range"
BATCH = "batch"
COLLECTION = "collection"
@dataclass(frozen=True, slots=True)
class ParsedRelease:
"""Structured, loss-aware representation of a resource name."""
raw: str
title_en: str | None = None
title_zh: str | None = None
title_jp: str | None = None
group: str | None = None
season: int | None = None
season_raw: str | None = None
episode: int | float | None = None
episode_end: int | float | None = None
episode_title: str | None = None
media_type: MediaType = MediaType.UNKNOWN
release_kind: ReleaseKind = ReleaseKind.SINGLE
resolution: str | None = None
source: str | None = None
subtitle: str | None = None
codecs: tuple[str, ...] = ()
audio: tuple[str, ...] = ()
container: str | None = None
version: int | None = None
year: int | None = None
tags: tuple[str, ...] = ()
evidence: tuple[str, ...] = ()
@property
def primary_title(self) -> str | None:
"""Return the best available title without applying locale policy."""
return self.title_en or self.title_zh or self.title_jp
@property
def is_movie(self) -> bool:
return self.media_type is MediaType.MOVIE

View File

@@ -1,114 +0,0 @@
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]

View File

@@ -1,119 +0,0 @@
import re
from module.models import Episode
from .keywords import CHINESE_NUMBER_MAP
from .token import Token, TokenKind
# Only strip dashes that have surrounding whitespace (separators),
# not dashes attached to characters (part of titles like 地。-关于地球的运动-)
_DASH_STRIP = re.compile(r"^\s*-\s+|\s+-\s*$")
_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:
title_en: str | None = None
title_zh: str | None = None
title_jp: str | None = None
season = 1
season_raw = ""
episode = 0
sub: str | None = None
group = ""
resolution: str | None = None
source: str | None = None
episode_type = "episode"
has_episode_marker = False
has_non_episodic_extra = False
# Collect all subtitles, prefer CJK ones
all_subs: list[str] = []
for token in tokens:
kind = token.kind
text = token.text
if kind == TokenKind.GROUP:
group = text
elif kind == TokenKind.TITLE_ZH:
text = _DASH_STRIP.sub("", text).strip()
if text:
title_zh = text
elif kind == TokenKind.TITLE_EN:
title_en = text.strip()
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())
elif kind == TokenKind.SEASON:
m = _SEASON_NUM.match(text)
if m:
s_num = int(m.group(1) or m.group(2))
else:
cm = _CHINESE_SEASON.match(text)
if cm:
s_val = cm.group(1)
try:
s_num = int(s_val)
except ValueError:
s_num = CHINESE_NUMBER_MAP.get(s_val, 1)
else:
s_num = season
# Prefer first season found (Chinese season over English)
if season == 1 or not season_raw:
season = s_num
season_raw = text
elif kind == TokenKind.RESOLUTION:
if resolution is None:
resolution = text
elif kind == TokenKind.SOURCE:
if source is None:
source = text
elif kind == TokenKind.MOVIE:
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:
all_subs.append(cleaned)
# Prefer CJK subtitle markers over technical ones
if all_subs:
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,
title_jp=title_jp,
season=season,
season_raw=season_raw,
episode=episode,
sub=sub,
group=group,
resolution=resolution,
source=source,
episode_type=episode_type,
)

View File

@@ -1,10 +0,0 @@
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

View File

@@ -1,450 +0,0 @@
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

View File

@@ -1,64 +0,0 @@
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

View File

@@ -1,33 +0,0 @@
from dataclasses import dataclass
from enum import Enum, auto
class TokenKind(Enum):
UNKNOWN = auto()
GROUP = auto()
TITLE_ZH = auto()
TITLE_EN = auto()
TITLE_JP = auto()
EPISODE = auto()
SEASON = auto()
RESOLUTION = auto()
SOURCE = auto()
SUBTITLE = auto()
CODEC = auto()
AUDIO = auto()
CONTAINER = auto()
PREFIX_TAG = auto()
VERSION = auto()
END_MARKER = auto()
HASH = auto()
MOVIE = auto() # movie/gekijouban marker
SPECIAL = auto() # OVA/OAD/SP/Special marker
EXTRA = auto()
@dataclass(slots=True)
class Token:
text: str
kind: TokenKind
position: int
enclosed: bool

View File

@@ -0,0 +1,475 @@
"""Contract tests for the generic release-title parser.
The generic API intentionally keeps an absent episode as ``None``. Legacy
``raw_parser`` compatibility (including its zero sentinel) is covered by the
older parser tests and is not part of this suite.
"""
from __future__ import annotations
from typing import Any
import pytest
from module.parser.analyser.raw_parser import raw_parser
from module.parser.analyser.tokenizer import parse_release_title
from module.parser.analyser.tokenizer.result import MediaType, ReleaseKind
REQUIRED_FIELDS = (
"raw",
"title_en",
"title_zh",
"title_jp",
"group",
"season",
"episode",
"episode_end",
"episode_title",
"media_type",
"release_kind",
"resolution",
"source",
"subtitle",
"codecs",
"audio",
"container",
"version",
)
def _title_text(parsed: Any) -> str:
"""Join title slots so tests do not prescribe a language heuristic."""
return " | ".join(
title for title in (parsed.title_en, parsed.title_zh, parsed.title_jp) if title
)
def test_generic_result_contract_and_standard_episode() -> None:
raw = (
"[LoliHouse] 葬送的芙莉莲 / Sousou no Frieren - 03 "
"[WebRip 1080p HEVC-10bit AAC][简繁内封字幕][MKV]"
)
parsed = parse_release_title(raw)
assert parsed is not None
for field in REQUIRED_FIELDS:
assert hasattr(parsed, field), field
assert parsed.raw == raw
assert parsed.group == "LoliHouse"
assert parsed.episode == 3
assert parsed.episode_end is None
assert parsed.resolution is not None
assert parsed.resolution.lower() == "1080p"
assert "葬送的芙莉莲" in _title_text(parsed)
assert "Sousou no Frieren" in _title_text(parsed)
assert parsed.media_type is MediaType.EPISODE
assert parsed.release_kind is ReleaseKind.SINGLE
@pytest.mark.parametrize(
("raw", "season", "episode", "version"),
[
("[Group] Anime Title S02E05 [WEB-DL 1080p]", 2, 5, None),
("[Group] Anime Title Season 2 - 01 [1080p]", 2, 1, None),
("[Group] Anime Title - 12.5 [1080p]", None, 12.5, None),
("[Group] Anime Title - 12v2 [1080p]", None, 12, 2),
],
)
def test_numbered_release_forms(
raw: str,
season: int | None,
episode: int | float,
version: int | None,
) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert "Anime Title" in _title_text(parsed)
assert parsed.episode == episode
assert parsed.episode_end is None
assert parsed.version == version
if season is not None:
assert parsed.season == season
assert parsed.media_type is MediaType.EPISODE
assert parsed.release_kind is ReleaseKind.SINGLE
@pytest.mark.parametrize(
"raw",
[
"[Group] Anime Title [01-12] [BDRip 1080p]",
"[Group] Anime Title [E01-E12] [BDRip 1080p]",
],
)
def test_episode_range_is_a_range_release(raw: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert "Anime Title" in _title_text(parsed)
assert parsed.episode == 1
assert parsed.episode_end == 12
assert parsed.media_type is MediaType.EPISODE
assert parsed.release_kind is ReleaseKind.RANGE
def test_complete_batch_without_episode_number() -> None:
parsed = parse_release_title(
"[VCB-Studio] Violet Evergarden Complete Batch [BDRip 1080p]"
)
assert parsed is not None
assert "Violet Evergarden" in _title_text(parsed)
assert parsed.episode is None
assert parsed.episode_end is None
assert parsed.media_type is MediaType.EPISODE
assert parsed.release_kind is ReleaseKind.BATCH
@pytest.mark.parametrize(
("raw", "title_fragment"),
[
("[Group] 葬送的芙莉莲合集 [BDRip 1080p]", "葬送的芙莉莲"),
("[Group] 葬送的芙莉莲全集 [BDRip 1080p]", "葬送的芙莉莲"),
("[Group] 进击的巨人总集篇 [BDRip 1080p]", "进击的巨人"),
("[Group] Anime Title 特典 [BDRip 1080p]", "Anime Title"),
],
)
def test_chinese_collection_without_episode(raw: str, title_fragment: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert title_fragment in _title_text(parsed)
assert parsed.episode is None
assert parsed.episode_end is None
assert parsed.release_kind is ReleaseKind.COLLECTION
def test_movie_collection_keeps_both_dimensions() -> None:
parsed = parse_release_title("[Group] 剧场版合集 - 进击的巨人 [1080p]")
assert parsed is not None
assert parsed.title_zh == "进击的巨人"
assert parsed.media_type is MediaType.MOVIE
assert parsed.release_kind is ReleaseKind.COLLECTION
@pytest.mark.parametrize(
("raw", "media_type", "episode"),
[
("[Group] Anime Title OVA01 [BDRip 1080p]", MediaType.OVA, 1),
("[Group] Anime Title OAD 02 [BDRip 1080p]", MediaType.OAD, 2),
("[Group] Anime Title SP03 [BDRip 1080p]", MediaType.SPECIAL, 3),
],
)
def test_numbered_special_media(raw: str, media_type: MediaType, episode: int) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert "Anime Title" in _title_text(parsed)
assert parsed.episode == episode
assert parsed.media_type is media_type
assert parsed.release_kind is ReleaseKind.SINGLE
@pytest.mark.parametrize(
("raw", "title_fragment"),
[
(
"[VCB-Studio]电影版哆啦A梦:大雄的地球交响乐[BDRip 1080p]",
"哆啦A梦",
),
(
"[SubGroup]劇場版SPY×FAMILY CODE: White[WEB-DL 1080p]",
"SPY×FAMILY",
),
],
)
def test_attached_movie_marker_preserves_title(raw: str, title_fragment: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert title_fragment in _title_text(parsed)
assert parsed.episode is None
assert parsed.media_type is MediaType.MOVIE
assert parsed.release_kind is ReleaseKind.SINGLE
@pytest.mark.parametrize(
("raw", "media_type", "episode"),
[
("[Official] Anime Title PV2 [1080p]", MediaType.PV, 2),
(
"[VCB-Studio] Anime Title NCOP [BDRip 1080p]",
MediaType.OPENING,
None,
),
(
"[VCB-Studio] Anime Title NCED [BDRip 1080p]",
MediaType.ENDING,
None,
),
],
)
def test_non_episode_video_kinds(
raw: str, media_type: MediaType, episode: int | None
) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert "Anime Title" in _title_text(parsed)
assert parsed.episode == episode
assert parsed.media_type is media_type
assert parsed.release_kind is ReleaseKind.SINGLE
@pytest.mark.parametrize(
("raw", "title_fragment"),
[
("[LoliHouse] Sousou no Frieren [WEB-DL 1080p]", "Sousou no Frieren"),
("[VCB-Studio] Fate/stay night [BDRip 1080p]", "Fate/stay night"),
],
)
def test_title_only_release_is_valid(raw: str, title_fragment: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert title_fragment in _title_text(parsed)
assert parsed.episode is None
assert parsed.episode_end is None
assert parsed.media_type is MediaType.UNKNOWN
assert parsed.release_kind is ReleaseKind.SINGLE
def test_year_is_not_an_episode_number() -> None:
parsed = parse_release_title("[VCB-Studio] Kimi no Na wa. (2016) [BDRip 1080p]")
assert parsed is not None
assert "Kimi no Na wa." in _title_text(parsed)
assert parsed.episode is None
def test_only_technical_tags_is_not_a_release() -> None:
assert parse_release_title("[1080p][HEVC][AAC][MKV]") is None
def test_movie_word_in_group_does_not_mark_episode_as_movie() -> None:
parsed = parse_release_title("[Movie-Fan] Ordinary Anime - 01 [WEB-DL 1080p]")
assert parsed is not None
assert parsed.group == "Movie-Fan"
assert parsed.episode == 1
assert parsed.media_type is MediaType.EPISODE
def test_movie_word_in_regular_series_title_does_not_override_episode() -> None:
parsed = parse_release_title(
"[SubGroup] The Movie Star Next Door - 03 [WEB-DL 1080p]"
)
assert parsed is not None
assert "Movie Star" in _title_text(parsed)
assert parsed.episode == 3
assert parsed.media_type is MediaType.EPISODE
assert parsed.release_kind is ReleaseKind.SINGLE
def test_fraction_in_title_is_not_an_episode_range() -> None:
parsed = parse_release_title("[SubGroup] Ranma 1/2 - 03 [WEB-DL 1080p]")
assert parsed is not None
assert "Ranma 1/2" in _title_text(parsed)
assert parsed.episode == 3
assert parsed.episode_end is None
assert parsed.release_kind is ReleaseKind.SINGLE
@pytest.mark.parametrize(
("raw", "media_type", "season", "episode", "episode_end"),
[
("[MOVIE] Anime Title [1080p]", MediaType.MOVIE, None, None, None),
("[劇場版] Anime Title [1080p]", MediaType.MOVIE, None, None, None),
("[S02E05] Anime Title [1080p]", MediaType.EPISODE, 2, 5, None),
("[Season 2] Anime Title - 01 [1080p]", MediaType.EPISODE, 2, 1, None),
("[E01-E12] Anime Title [1080p]", MediaType.EPISODE, None, 1, 12),
],
)
def test_leading_metadata_is_not_a_group(
raw: str,
media_type: MediaType,
season: int | None,
episode: int | None,
episode_end: int | None,
) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.group is None
assert "Anime Title" in _title_text(parsed)
assert parsed.media_type is media_type
assert parsed.season == season
assert parsed.episode == episode
assert parsed.episode_end == episode_end
def test_compound_technical_prefix_is_not_a_group() -> None:
parsed = parse_release_title("[BDRip 1080p] Anime Title - 01")
assert parsed is not None
assert parsed.group is None
assert parsed.source == "BDRip"
assert parsed.resolution == "1080p"
assert parsed.episode == 1
@pytest.mark.parametrize(
("raw", "media_type"),
[
("[Group] 某作品OVA [1080p]", MediaType.OVA),
("[Group] 某作品番外篇 [1080p]", MediaType.SPECIAL),
("[Group] 某作品特别篇 [1080p]", MediaType.SPECIAL),
],
)
def test_attached_special_markers_preserve_title(
raw: str, media_type: MediaType
) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.title_zh == "某作品"
assert parsed.media_type is media_type
def test_numbered_complete_collection() -> None:
parsed = parse_release_title("[Group] 迷宫饭 全24话 [1080p]")
assert parsed is not None
assert parsed.title_zh == "迷宫饭"
assert parsed.episode == 1
assert parsed.episode_end == 24
assert parsed.release_kind is ReleaseKind.COLLECTION
def test_provider_technical_labels_do_not_pollute_title() -> None:
parsed = parse_release_title(
"[EMBER] Frieren - 12.5 [BDRip 1080p AVC 8bit Dual Audio Multiple Subtitle MKV]"
)
assert parsed is not None
assert parsed.title_en == "Frieren"
assert parsed.episode == 12.5
assert "8bit" in parsed.codecs
assert "Dual Audio" in parsed.audio
assert parsed.subtitle == "Multiple Subtitle"
def test_plex_episode_title_is_separate_from_series_title() -> None:
parsed = parse_release_title(
"Mob Psycho 100 - S03E05 - Divine Tree 2 ~Peace~ [1080p]"
)
assert parsed is not None
assert parsed.title_en == "Mob Psycho 100"
assert parsed.episode_title == "Divine Tree 2 ~Peace~"
assert parsed.season == 3
assert parsed.episode == 5
def test_inline_year_is_metadata() -> None:
parsed = parse_release_title("The Boy and the Heron 2023 1080p BluRay")
assert parsed is not None
assert parsed.title_en == "The Boy and the Heron"
assert parsed.year == 2023
def test_pv_prefix_in_real_title_is_not_a_marker() -> None:
parsed = parse_release_title("[Group] PV=nRT - 01 [1080p]")
assert parsed is not None
assert parsed.title_en == "PV=nRT"
assert parsed.media_type is MediaType.EPISODE
@pytest.mark.parametrize(
("raw", "title"),
[
("[Group] The Movie Star Next Door [1080p]", "The Movie Star Next Door"),
("[Group] Special A [1080p]", "Special A"),
],
)
def test_marker_words_inside_title_only_release_are_preserved(
raw: str, title: str
) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.title_en == title
assert parsed.media_type is MediaType.UNKNOWN
@pytest.mark.parametrize(
("raw", "title"),
[
("[Group] 劇場版 呪術廻戦 0 [1080p]", "呪術廻戦 0"),
("[Group] Gekijouban Naruto 3 [1080p]", "Naruto 3"),
],
)
def test_movie_title_number_is_not_an_episode(raw: str, title: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert title in _title_text(parsed)
assert parsed.media_type is MediaType.MOVIE
assert parsed.episode is None
@pytest.mark.parametrize("title", ["86", "22/7"])
def test_numeric_anime_titles(title: str) -> None:
parsed = parse_release_title(f"[Group] {title} - 01 [1080p]")
assert parsed is not None
assert parsed.title_en == title
assert parsed.episode == 1
def test_ordinal_season_is_recognized() -> None:
parsed = parse_release_title("[Group] Dandadan 2nd Season - 01 [1080p]")
assert parsed is not None
assert parsed.title_en == "Dandadan"
assert parsed.season == 2
assert parsed.episode == 1
def test_episode_mixed_with_technical_tags() -> None:
parsed = parse_release_title("[Group] Anime Title [01 1080p HEVC]")
assert parsed is not None
assert parsed.title_en == "Anime Title"
assert parsed.episode == 1
assert parsed.resolution == "1080p"
def test_surround_audio_metadata_is_not_a_title() -> None:
parsed = parse_release_title("[Group] Anime Title - 01 [Dual Audio][DDP5.1][1080p]")
assert parsed is not None
assert parsed.title_en == "Anime Title"
assert "Dual Audio" in parsed.audio
assert "DDP5.1" in parsed.audio
@pytest.mark.parametrize(
"raw",
[
"[Group] Anime [01-12] [1080p]",
"[Group] Anime Complete Batch [1080p]",
"[Group] Anime PV2 [1080p]",
"[Group] Anime NCOP [1080p]",
"[Group] Anime - 12.5 [1080p]",
],
)
def test_generic_only_kinds_do_not_leak_into_legacy_parser(raw: str) -> None:
assert parse_release_title(raw) is not None
assert raw_parser(raw) is None

View File

@@ -1,110 +0,0 @@
"""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