diff --git a/backend/src/module/models/bangumi.py b/backend/src/module/models/bangumi.py index 30d8f283..35834de7 100644 --- a/backend/src/module/models/bangumi.py +++ b/backend/src/module/models/bangumi.py @@ -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 diff --git a/backend/src/module/parser/analyser/raw_parser.py b/backend/src/module/parser/analyser/raw_parser.py index 07891fca..3ccce919 100644 --- a/backend/src/module/parser/analyser/raw_parser.py +++ b/backend/src/module/parser/analyser/raw_parser.py @@ -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: diff --git a/backend/src/module/parser/analyser/tokenizer/__init__.py b/backend/src/module/parser/analyser/tokenizer/__init__.py index c344ec1b..888c78e3 100644 --- a/backend/src/module/parser/analyser/tokenizer/__init__.py +++ b/backend/src/module/parser/analyser/tokenizer/__init__.py @@ -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", +] diff --git a/backend/src/module/parser/analyser/tokenizer/compat.py b/backend/src/module/parser/analyser/tokenizer/compat.py new file mode 100644 index 00000000..cf986c67 --- /dev/null +++ b/backend/src/module/parser/analyser/tokenizer/compat.py @@ -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 diff --git a/backend/src/module/parser/analyser/tokenizer/hot_reload.py b/backend/src/module/parser/analyser/tokenizer/hot_reload.py deleted file mode 100644 index 7a5787d2..00000000 --- a/backend/src/module/parser/analyser/tokenizer/hot_reload.py +++ /dev/null @@ -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") diff --git a/backend/src/module/parser/analyser/tokenizer/keywords.py b/backend/src/module/parser/analyser/tokenizer/keywords.py deleted file mode 100644 index 9012b1bf..00000000 --- a/backend/src/module/parser/analyser/tokenizer/keywords.py +++ /dev/null @@ -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) diff --git a/backend/src/module/parser/analyser/tokenizer/keywords_extra.json b/backend/src/module/parser/analyser/tokenizer/keywords_extra.json deleted file mode 100644 index 26471d0f..00000000 --- a/backend/src/module/parser/analyser/tokenizer/keywords_extra.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "version": 1, - "keywords": {}, - "cjk_subwords": {}, - "prefix_tags": [] -} diff --git a/backend/src/module/parser/analyser/tokenizer/normalization.py b/backend/src/module/parser/analyser/tokenizer/normalization.py new file mode 100644 index 00000000..e18e68d4 --- /dev/null +++ b/backend/src/module/parser/analyser/tokenizer/normalization.py @@ -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) diff --git a/backend/src/module/parser/analyser/tokenizer/parser.py b/backend/src/module/parser/analyser/tokenizer/parser.py new file mode 100644 index 00000000..820ea387 --- /dev/null +++ b/backend/src/module/parser/analyser/tokenizer/parser.py @@ -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"(? 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) diff --git a/backend/src/module/parser/analyser/tokenizer/result.py b/backend/src/module/parser/analyser/tokenizer/result.py new file mode 100644 index 00000000..b83d0edf --- /dev/null +++ b/backend/src/module/parser/analyser/tokenizer/result.py @@ -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 diff --git a/backend/src/module/parser/analyser/tokenizer/stage_classify.py b/backend/src/module/parser/analyser/tokenizer/stage_classify.py deleted file mode 100644 index 30752278..00000000 --- a/backend/src/module/parser/analyser/tokenizer/stage_classify.py +++ /dev/null @@ -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] diff --git a/backend/src/module/parser/analyser/tokenizer/stage_compose.py b/backend/src/module/parser/analyser/tokenizer/stage_compose.py deleted file mode 100644 index 9875d726..00000000 --- a/backend/src/module/parser/analyser/tokenizer/stage_compose.py +++ /dev/null @@ -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, - ) diff --git a/backend/src/module/parser/analyser/tokenizer/stage_normalize.py b/backend/src/module/parser/analyser/tokenizer/stage_normalize.py deleted file mode 100644 index 7975022f..00000000 --- a/backend/src/module/parser/analyser/tokenizer/stage_normalize.py +++ /dev/null @@ -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 diff --git a/backend/src/module/parser/analyser/tokenizer/stage_resolve.py b/backend/src/module/parser/analyser/tokenizer/stage_resolve.py deleted file mode 100644 index e931db68..00000000 --- a/backend/src/module/parser/analyser/tokenizer/stage_resolve.py +++ /dev/null @@ -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 diff --git a/backend/src/module/parser/analyser/tokenizer/stage_tokenize.py b/backend/src/module/parser/analyser/tokenizer/stage_tokenize.py deleted file mode 100644 index 54ceb04c..00000000 --- a/backend/src/module/parser/analyser/tokenizer/stage_tokenize.py +++ /dev/null @@ -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 diff --git a/backend/src/module/parser/analyser/tokenizer/token.py b/backend/src/module/parser/analyser/tokenizer/token.py deleted file mode 100644 index 986b10c0..00000000 --- a/backend/src/module/parser/analyser/tokenizer/token.py +++ /dev/null @@ -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 diff --git a/backend/src/test/test_release_parser.py b/backend/src/test/test_release_parser.py new file mode 100644 index 00000000..9c9c2ddc --- /dev/null +++ b/backend/src/test/test_release_parser.py @@ -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 diff --git a/backend/src/test/test_token_features.py b/backend/src/test/test_token_features.py deleted file mode 100644 index 93a18ec6..00000000 --- a/backend/src/test/test_token_features.py +++ /dev/null @@ -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