diff --git a/backend/src/module/parser/analyser/tokenizer/__init__.py b/backend/src/module/parser/analyser/tokenizer/__init__.py index 888c78e3..6446b4d6 100644 --- a/backend/src/module/parser/analyser/tokenizer/__init__.py +++ b/backend/src/module/parser/analyser/tokenizer/__init__.py @@ -1,13 +1,17 @@ """Public APIs for generic and legacy resource-title parsing.""" from .compat import tokenize_title -from .parser import parse_release_title +from .parser import parse_release_title, parse_release_title_with_trace from .result import MediaType, ParsedRelease, ReleaseKind +from .trace import ParseOutcome, ParseTrace __all__ = [ "MediaType", + "ParseOutcome", + "ParseTrace", "ParsedRelease", "ReleaseKind", "parse_release_title", + "parse_release_title_with_trace", "tokenize_title", ] diff --git a/backend/src/module/parser/analyser/tokenizer/candidate.py b/backend/src/module/parser/analyser/tokenizer/candidate.py new file mode 100644 index 00000000..495fca48 --- /dev/null +++ b/backend/src/module/parser/analyser/tokenizer/candidate.py @@ -0,0 +1,240 @@ +"""Immutable evidence and candidate models used by the title resolver. + +Regex rules should only describe what they observed. They must not mutate the +parse result or remove text from the title. Semantic candidates group one or +more observations into an interpretation that can be resolved deterministically. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum + +from .result import MediaType, ReleaseKind + + +class ClaimField(StrEnum): + """Scalar and repeatable fields a candidate can contribute.""" + + GROUP = "group" + SEASON = "season" + SEASON_RAW = "season_raw" + EPISODE = "episode" + EPISODE_END = "episode_end" + EPISODE_TITLE = "episode_title" + MEDIA_TYPE = "media_type" + RELEASE_KIND = "release_kind" + RESOLUTION = "resolution" + SOURCE = "source" + SUBTITLE = "subtitle" + CODECS = "codecs" + AUDIO = "audio" + CONTAINER = "container" + VERSION = "version" + YEAR = "year" + TAGS = "tags" + + +class ShadowedSpanPolicy(StrEnum): + """Whether valid-but-shadowed metadata remains excluded from the title.""" + + EXCLUDE = "exclude" + KEEP = "keep" + + +class OverlapPolicy(StrEnum): + """Whether a candidate may share source characters with another winner.""" + + EXCLUSIVE = "exclusive" + SHARED = "shared" + + +class DecisionStatus(StrEnum): + """Outcome of resolving one semantic candidate.""" + + SELECTED = "selected" + SHADOWED = "shadowed" + REJECTED_AS_TITLE = "rejected_as_title" + REJECTED_CONFLICT = "rejected_conflict" + + +@dataclass(frozen=True, slots=True, order=True) +class Span: + """Half-open character range using coordinates local to one segment.""" + + segment: int + start: int + end: int + + def __post_init__(self) -> None: + if self.segment < 0: + raise ValueError("segment must be non-negative") + if self.start < 0: + raise ValueError("span start must be non-negative") + if self.end <= self.start: + raise ValueError("span end must be greater than start") + + @property + def length(self) -> int: + return self.end - self.start + + def overlaps(self, other: Span) -> bool: + return ( + self.segment == other.segment + and self.start < other.end + and other.start < self.end + ) + + def contains(self, other: Span) -> bool: + return ( + self.segment == other.segment + and self.start <= other.start + and self.end >= other.end + ) + + +@dataclass(frozen=True, slots=True) +class Observation: + """A direct, interpretation-free match produced by a tokenizer rule.""" + + id: str + rule_id: str + kind: str + span: Span + text: str + captures: tuple[str | None, ...] = () + + def __post_init__(self) -> None: + if not self.id: + raise ValueError("observation id must not be empty") + if not self.rule_id: + raise ValueError("observation rule_id must not be empty") + if not self.kind: + raise ValueError("observation kind must not be empty") + + @property + def sort_key(self) -> tuple[int, int, int, str, str]: + return ( + self.span.segment, + self.span.start, + self.span.end, + self.rule_id, + self.id, + ) + + +@dataclass(frozen=True, slots=True) +class Claims: + """Typed semantic fields proposed by a candidate. + + ``None`` means that the candidate makes no claim for a scalar field. Empty + tuples likewise mean no contribution to repeatable fields. + """ + + 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 | None = None + release_kind: ReleaseKind | None = None + 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, ...] = () + + def scalar_items(self) -> tuple[tuple[ClaimField, object], ...]: + values: tuple[tuple[ClaimField, object | None], ...] = ( + (ClaimField.GROUP, self.group), + (ClaimField.SEASON, self.season), + (ClaimField.SEASON_RAW, self.season_raw), + (ClaimField.EPISODE, self.episode), + (ClaimField.EPISODE_END, self.episode_end), + (ClaimField.EPISODE_TITLE, self.episode_title), + (ClaimField.MEDIA_TYPE, self.media_type), + (ClaimField.RELEASE_KIND, self.release_kind), + (ClaimField.RESOLUTION, self.resolution), + (ClaimField.SOURCE, self.source), + (ClaimField.SUBTITLE, self.subtitle), + (ClaimField.CONTAINER, self.container), + (ClaimField.VERSION, self.version), + (ClaimField.YEAR, self.year), + ) + return tuple((field, value) for field, value in values if value is not None) + + def repeatable_items(self) -> tuple[tuple[ClaimField, tuple[str, ...]], ...]: + return ( + (ClaimField.CODECS, self.codecs), + (ClaimField.AUDIO, self.audio), + (ClaimField.TAGS, self.tags), + ) + + @property + def is_empty(self) -> bool: + return not self.scalar_items() and not any( + values for _, values in self.repeatable_items() + ) + + +@dataclass(frozen=True, slots=True) +class Candidate: + """One semantic interpretation offered to the resolver. + + ``conflict_tags`` and ``blocks`` express cross-field incompatibilities. A + strong movie marker, for example, can block the ``ambiguous_episode`` tag + carried by a trailing bare number without blocking an explicit SxxExx. + """ + + id: str + rule_id: str + spans: tuple[Span, ...] + claims: Claims = Claims() + priority: int = 0 + specificity: int = 0 + observation_ids: tuple[str, ...] = () + evidence: tuple[str, ...] = () + conflict_tags: frozenset[str] = frozenset() + blocks: frozenset[str] = frozenset() + preserve_as_title_on_conflict: bool = False + shadowed_span_policy: ShadowedSpanPolicy = ShadowedSpanPolicy.EXCLUDE + shadowed_spans: tuple[Span, ...] | None = None + overlap_policy: OverlapPolicy = OverlapPolicy.EXCLUSIVE + + def __post_init__(self) -> None: + if not self.id: + raise ValueError("candidate id must not be empty") + if not self.rule_id: + raise ValueError("candidate rule_id must not be empty") + canonical_spans = tuple(sorted(set(self.spans))) + object.__setattr__(self, "spans", canonical_spans) + if self.shadowed_spans is not None: + canonical_shadowed = tuple(sorted(set(self.shadowed_spans))) + if not set(canonical_shadowed).issubset(canonical_spans): + raise ValueError("shadowed_spans must be a subset of spans") + object.__setattr__(self, "shadowed_spans", canonical_shadowed) + + @property + def sort_key(self) -> tuple[int, int, tuple[Span, ...], str, str]: + """Stable ordering key; it never depends on collection insertion order.""" + return (-self.priority, -self.specificity, self.spans, self.rule_id, self.id) + + +@dataclass(frozen=True, slots=True) +class Decision: + """Auditable outcome for one candidate.""" + + candidate_id: str + status: DecisionStatus + reason: str + won_fields: tuple[ClaimField, ...] = () + excluded_spans: tuple[Span, ...] = () + + @property + def exclude_from_title(self) -> bool: + return bool(self.excluded_spans) diff --git a/backend/src/module/parser/analyser/tokenizer/parser.py b/backend/src/module/parser/analyser/tokenizer/parser.py index 820ea387..2c3a75f4 100644 --- a/backend/src/module/parser/analyser/tokenizer/parser.py +++ b/backend/src/module/parser/analyser/tokenizer/parser.py @@ -11,8 +11,18 @@ from __future__ import annotations import re from dataclasses import dataclass, field +from .candidate import ( + Candidate, + Claims, + Observation, + OverlapPolicy, + ShadowedSpanPolicy, + Span, +) from .normalization import normalize +from .resolver import Resolution, resolve_candidates from .result import MediaType, ParsedRelease, ReleaseKind +from .trace import ParseOutcome, ParseTrace, TraceSegment _TITLE_CHAR = re.compile(r"[A-Za-z\u3040-\u30ff\u3400-\u9fff\uf900-\ufaff]") _HAN = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff\uf900-\ufaff]") @@ -36,10 +46,22 @@ _SEASON_EPISODE = 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: + def mask_span(self, start: int, end: int) -> None: self.mask[start:end] = [True] * (end - start) def residual(self) -> str: @@ -147,109 +169,248 @@ class _WorkingSegment: @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) +class _CandidateCollector: + segments: list[_Segment] + capture_observations: bool = False + observations: list[Observation] = field(default_factory=list) + candidates: list[Candidate] = field(default_factory=list) - def set_episode( + def add_span( 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 + rule_id: str, + kind: str, + segment_index: int, + start: int, + end: int, + claims: Claims = Claims(), + priority: int = 0, + specificity: int = 0, + evidence: tuple[str, ...] = (), + extra_spans: tuple[Span, ...] = (), + conflict_tags: frozenset[str] = frozenset(), + blocks: frozenset[str] = frozenset(), + preserve_as_title_on_conflict: bool = False, + shadowed_span_policy: ShadowedSpanPolicy = ShadowedSpanPolicy.EXCLUDE, + overlap_policy: OverlapPolicy = OverlapPolicy.EXCLUSIVE, + captures: tuple[str | None, ...] = (), + ) -> Candidate: + span = Span(segment_index, start, end) + suffix = f"{segment_index}:{start}:{end}" + candidate_id = f"{rule_id}:{suffix}" + observation_ids: tuple[str, ...] = () + if self.capture_observations: + observation_id = f"observation:{candidate_id}" + observation_ids = (observation_id,) + self.observations.append( + Observation( + id=observation_id, + rule_id=rule_id, + kind=kind, + span=span, + text=self.segments[segment_index].text[start:end], + captures=captures, + ) + ) + candidate = Candidate( + id=candidate_id, + rule_id=rule_id, + spans=(span, *extra_spans), + claims=claims, + priority=priority, + specificity=specificity, + observation_ids=observation_ids, + evidence=evidence, + conflict_tags=conflict_tags, + blocks=blocks, + preserve_as_title_on_conflict=preserve_as_title_on_conflict, + shadowed_span_policy=shadowed_span_policy, + shadowed_spans=(span,) if extra_spans else None, + overlap_policy=overlap_policy, + ) + self.candidates.append(candidate) + return candidate - 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 add_match( + self, + *, + rule_id: str, + kind: str, + segment_index: int, + match: re.Match[str], + **kwargs: object, + ) -> Candidate: + return self.add_span( + rule_id=rule_id, + kind=kind, + segment_index=segment_index, + start=match.start(), + end=match.end(), + captures=match.groups(), + **kwargs, # type: ignore[arg-type] + ) + + def add_segments( + self, + *, + rule_id: str, + kind: str, + segment_indices: tuple[int, ...], + claims: Claims, + priority: int, + evidence: tuple[str, ...] = (), + ) -> Candidate: + spans = tuple( + Span(index, 0, len(self.segments[index].text)) for index in segment_indices + ) + observation_ids: list[str] = [] + if self.capture_observations: + for span in spans: + observation_id = ( + f"observation:{rule_id}:{span.segment}:{span.start}:{span.end}" + ) + observation_ids.append(observation_id) + self.observations.append( + Observation( + id=observation_id, + rule_id=rule_id, + kind=kind, + span=span, + text=self.segments[span.segment].text, + ) + ) + candidate = Candidate( + id=f"{rule_id}:" + ",".join(str(index) for index in segment_indices), + rule_id=rule_id, + spans=spans, + claims=claims, + priority=priority, + observation_ids=tuple(observation_ids), + evidence=evidence, + ) + self.candidates.append(candidate) + return candidate 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 + result, _ = _parse_release_title_with_candidates(raw, capture_trace=False) + return result - normalized = normalize(raw) - segments = _scan_segments(normalized) + +def parse_release_title_with_trace(raw: str) -> ParseOutcome: + """Parse a resource name together with immutable resolver diagnostics.""" + result, trace = _parse_release_title_with_candidates(raw, capture_trace=True) + assert trace is not None + return ParseOutcome(result=result, trace=trace) + + +def _parse_release_title_with_candidates( + raw: str, *, capture_trace: bool +) -> tuple[ParsedRelease | None, ParseTrace | None]: + normalized = normalize(raw) if raw and raw.strip() else "" + segments = _scan_segments(normalized) if normalized else [] if not segments: - return None + trace = ParseTrace(raw=raw, normalized=normalized) if capture_trace else None + return None, trace - 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) + collector = _CandidateCollector(segments, capture_observations=capture_trace) + _collect_group_candidates(collector) + _collect_structural_candidates(collector) + _collect_media_candidates(collector) + _collect_technical_candidates(collector) + _collect_metadata_episode_candidates(collector) + resolution = resolve_candidates( + collector.candidates, collect_warnings=capture_trace + ) + working = _working_from_spans(segments, resolution.excluded_spans) title_en, title_zh, title_jp = _reconstruct_titles(working) - if not any((title_en, title_zh, title_jp)): - return None + result = None + if any((title_en, title_zh, title_jp)): + result = _result_from_resolution(raw, title_en, title_zh, title_jp, resolution) - 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) + trace = None + if capture_trace: + trace = ParseTrace.from_resolution( + raw=raw, + normalized=normalized, + resolution=resolution, + segments=_trace_segments(segments), + observations=tuple(collector.observations), + candidates=tuple(collector.candidates), + residuals=tuple(work.residual() for work in working), + ) + return result, trace + +def _result_from_resolution( + raw: str, + title_en: str | None, + title_zh: str | None, + title_jp: str | None, + resolution: Resolution, +) -> ParsedRelease: + claims = resolution.claims + media_type = claims.media_type + if media_type is None and ( + claims.episode is not None or claims.release_kind is not None + ): + media_type = MediaType.EPISODE 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), + group=claims.group, + season=claims.season, + season_raw=claims.season_raw, + episode=claims.episode, + episode_end=claims.episode_end, + episode_title=claims.episode_title, + media_type=media_type or MediaType.UNKNOWN, + release_kind=claims.release_kind or ReleaseKind.SINGLE, + resolution=claims.resolution, + source=claims.source, + subtitle=claims.subtitle, + codecs=claims.codecs, + audio=claims.audio, + container=claims.container, + version=claims.version, + year=claims.year, + tags=claims.tags, + evidence=resolution.evidence, ) +def _working_from_spans( + segments: list[_Segment], excluded_spans: tuple[Span, ...] +) -> list[_ResidualSegment]: + working = [_ResidualSegment(segment) for segment in segments] + for span in excluded_spans: + working[span.segment].mask_span(span.start, span.end) + return working + + +def _trace_segments(segments: list[_Segment]) -> tuple[TraceSegment, ...]: + traced: list[TraceSegment] = [] + for index, segment in enumerate(segments): + enclosed = segment.enclosure is not None + content_start = segment.start + 1 if enclosed else segment.start + traced.append( + TraceSegment( + index=index, + text=segment.text, + outer_start=segment.start, + outer_end=segment.end, + content_start=content_start, + content_end=content_start + len(segment.text), + enclosure=segment.enclosure, + ) + ) + return tuple(traced) + + def _scan_segments(text: str) -> list[_Segment]: segments: list[_Segment] = [] free_start = 0 @@ -289,24 +450,42 @@ def _scan_segments(text: str) -> list[_Segment]: 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): + if not segments: 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:] + first_free_title = next( + ( + index + for index, segment in enumerate(segments) + if segment.enclosure is None + and _contains_title_fragment(segment.text) + and not _looks_like_metadata(segment.text) + ), + None, ) - if not (_looks_like_group(first) or has_free_title): + has_free_title = first_free_title is not None + + group_index = next( + ( + index + for index, segment in enumerate(segments) + if segment.enclosure == "square" + and (first_free_title is None or index < first_free_title) + and not _looks_like_metadata(segment.text) + and (_looks_like_group(segment.text) or has_free_title) + ), + None, + ) + if group_index is None: return set() - indices = {0} - for index, segment in enumerate(segments[1:], 1): - if segment.enclosure != "square" or not _looks_like_group(segment.text): + indices = {group_index} + for index, segment in enumerate(segments[group_index + 1 :], group_index + 1): + if ( + segment.enclosure != "square" + or _looks_like_metadata(segment.text) + or not _looks_like_group(segment.text) + ): break indices.add(index) return indices @@ -319,6 +498,8 @@ def _looks_like_group(text: str) -> bool: return True if re.search(r"raws?|fansub|subs?|house|studio|group|team|rip", lower): return True + if re.search(r"(?:^|[-_. ])fan(?:$|[-_. ])", lower): + return True if lower in {"ani", "official", "magicstar", "doomdos", "vcb-studio"}: return True if "&" in value and len(value) <= 48: @@ -334,6 +515,8 @@ def _looks_like_metadata(text: str) -> bool: return True structural_patterns = ( _RANGE, + _SEASON_EPISODE_RANGE, + _SEASON_EPISODE_WORDS_RANGE, _SEASON_EPISODE, _SEASON_EPISODE_WORDS, _SEASON, @@ -342,6 +525,7 @@ def _looks_like_metadata(text: str) -> bool: _CHINESE_EPISODE, _BARE_EPISODE, _SPECIAL_NUMBER, + _SPECIAL_RANGE, _SPECIAL_WORD, _SPECIAL_CJK, _PV, @@ -393,24 +577,667 @@ 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 _collect_group_candidates(collector: _CandidateCollector) -> set[int]: + group_indices = _find_group_segments(collector.segments) + if not group_indices: + return set() + ordered = tuple(sorted(group_indices)) + collector.add_segments( + rule_id="group.square-prefix", + kind="group", + segment_indices=ordered, + claims=Claims( + group="&".join(collector.segments[index].text.strip() for index in ordered) + ), + priority=1000, + evidence=("group",), + ) + return group_indices -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 _episode_title_parts( + segment_index: int, text: str, match: re.Match[str] +) -> tuple[str | None, tuple[Span, ...]]: + spans: list[Span] = [] + before = text[: match.start()] + separator = re.search(r"\s+-\s*$", before) + if separator: + spans.append(Span(segment_index, separator.start(), match.start())) + + after = text[match.end() :] + episode_title = re.fullmatch(r"\s*-\s*(.+?)\s*", after) + if not episode_title or not _contains_title_fragment(episode_title.group(1)): + return None, tuple(spans) + spans.append(Span(segment_index, match.end(), len(text))) + return _clean_title(episode_title.group(1)), tuple(spans) + + +def _collect_structural_candidates(collector: _CandidateCollector) -> None: + for index, segment in enumerate(collector.segments): + text = segment.text + + for match in _DATE.finditer(text): + collector.add_match( + rule_id="metadata.date", + kind="tag", + segment_index=index, + match=match, + claims=Claims(tags=(match.group(),)), + priority=110, + ) + + for match in _FILE_SIZE.finditer(text): + collector.add_match( + rule_id="metadata.file-size", + kind="tag", + segment_index=index, + match=match, + claims=Claims(tags=(match.group(),)), + priority=110, + ) + + for match in _YEAR.finditer(text): + if match.group() != text.strip() and not _is_metadata_tail( + text[match.end() :] + ): + continue + collector.add_match( + rule_id="metadata.year", + kind="year", + segment_index=index, + match=match, + claims=Claims(year=int(match.group(1))), + priority=100, + evidence=("year",), + ) + + for rule_id, pattern in ( + ("episode.season-range-compact", _SEASON_EPISODE_RANGE), + ("episode.season-range-words", _SEASON_EPISODE_WORDS_RANGE), + ): + for match in pattern.finditer(text): + start, end = _number(match.group(2)), _number(match.group(3)) + if start >= 1800 or end >= 1800 or start > end: + continue + collector.add_match( + rule_id=rule_id, + kind="season-episode-range", + segment_index=index, + match=match, + claims=Claims( + season=int(match.group(1)), + season_raw=match.group(0), + episode=start, + episode_end=end, + release_kind=ReleaseKind.RANGE, + version=_int(match.group(4)), + ), + priority=105, + specificity=5, + evidence=("season", "episode-range"), + ) + + for match in _SPECIAL_RANGE.finditer(text): + start, end = _number(match.group(2)), _number(match.group(3)) + if start >= 1800 or end >= 1800 or start > end: + continue + media_type = _special_media(match.group(1)) + collector.add_match( + rule_id=f"episode.{media_type.value}-range", + kind="special-episode-range", + segment_index=index, + match=match, + claims=Claims( + episode=start, + episode_end=end, + media_type=media_type, + release_kind=ReleaseKind.RANGE, + version=_int(match.group(4)), + ), + priority=105, + specificity=5, + evidence=(media_type.value, "episode-range"), + ) + + for match in _RANGE.finditer(text): + 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)?\.?\s*)?\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 + collector.add_match( + rule_id="episode.range", + kind="episode-range", + segment_index=index, + match=match, + claims=Claims( + episode=start, + episode_end=end, + release_kind=ReleaseKind.RANGE, + version=_int(match.group(3)), + ), + priority=95, + specificity=3, + evidence=("episode-range",), + ) + + for rule_id, pattern in ( + ("episode.season-compact", _SEASON_EPISODE), + ("episode.season-words", _SEASON_EPISODE_WORDS), + ): + for match in pattern.finditer(text): + episode_title, extra_spans = _episode_title_parts(index, text, match) + evidence: tuple[str, ...] = ("season", "episode") + if episode_title: + evidence += ("episode-title",) + collector.add_match( + rule_id=rule_id, + kind="season-episode", + segment_index=index, + match=match, + claims=Claims( + season=int(match.group(1)), + season_raw=match.group(0), + episode=_number(match.group(2)), + episode_title=episode_title, + version=_int(match.group(3)), + ), + priority=100, + specificity=4, + evidence=evidence, + extra_spans=extra_spans, + ) + + for match in _SPECIAL_NUMBER.finditer(text): + media_type = _special_media(match.group(1)) + collector.add_match( + rule_id=f"episode.{media_type.value}-numbered", + kind="special-episode", + segment_index=index, + match=match, + claims=Claims( + episode=_number(match.group(2)), + media_type=media_type, + version=_int(match.group(3)), + ), + priority=80, + specificity=3, + evidence=(media_type.value, "episode"), + ) + + for match in _PV.finditer(text): + if match.group(2) is None: + continue + collector.add_match( + rule_id="media.pv-numbered", + kind="non-episode-video", + segment_index=index, + match=match, + claims=Claims(episode=_number(match.group(2)), media_type=MediaType.PV), + priority=75, + specificity=2, + evidence=("pv", "episode"), + ) + + for match in _SEASON.finditer(text): + number_match = re.search(r"\d+", match.group(1)) + if number_match: + collector.add_match( + rule_id="season.marker", + kind="season", + segment_index=index, + match=match, + claims=Claims( + season=int(number_match.group()), season_raw=match.group(1) + ), + priority=80, + evidence=("season",), + ) + + for match in _CHINESE_SEASON.finditer(text): + collector.add_match( + rule_id="season.chinese", + kind="season", + segment_index=index, + match=match, + claims=Claims( + season=_chinese_number(match.group(1)), season_raw=match.group(0) + ), + priority=80, + evidence=("season",), + ) + + for match in _ORDINAL_SEASON.finditer(text): + collector.add_match( + rule_id="season.ordinal", + kind="season", + segment_index=index, + match=match, + claims=Claims(season=int(match.group(1)), season_raw=match.group(0)), + priority=50, + evidence=("season",), + shadowed_span_policy=ShadowedSpanPolicy.KEEP, + ) + + for match in _FULL_COLLECTION.finditer(text): + collector.add_match( + rule_id="episode.full-collection", + kind="collection", + segment_index=index, + match=match, + claims=Claims( + episode=1, + episode_end=int(match.group(1)), + release_kind=ReleaseKind.COLLECTION, + ), + priority=95, + specificity=3, + evidence=("collection", "episode-range"), + ) + + for match in _EXPLICIT_EPISODE.finditer(text): + collector.add_match( + rule_id="episode.explicit", + kind="episode", + segment_index=index, + match=match, + claims=Claims( + episode=_number(match.group(1)), version=_int(match.group(2)) + ), + priority=90, + specificity=2, + evidence=("episode",), + ) + + for match in _CHINESE_EPISODE.finditer(text): + collector.add_match( + rule_id="episode.chinese", + kind="episode", + segment_index=index, + match=match, + claims=Claims(episode=_number(match.group(1))), + priority=90, + specificity=2, + evidence=("episode",), + ) + + for rule_id, pattern in ( + ("episode.pre", _PRE_EPISODE), + ("episode.compound", _COMPOUND_EPISODE), + ("episode.atlas", _ATLAS_EPISODE), + ): + special_match = pattern.match(text) + if special_match: + collector.add_match( + rule_id=rule_id, + kind="episode", + segment_index=index, + match=special_match, + claims=Claims(episode=_number(special_match.group(1))), + priority=75, + evidence=("episode",), + ) + + trailing = _TRAILING_EPISODE.search(text) + if trailing and not _is_year_number(trailing.group(1)): + collector.add_match( + rule_id="episode.trailing", + kind="episode", + segment_index=index, + match=trailing, + claims=Claims( + episode=_number(trailing.group(1)), + version=_int(trailing.group(2)), + ), + priority=70, + evidence=("episode",), + conflict_tags=frozenset({"ambiguous-episode"}), + preserve_as_title_on_conflict=True, + ) + + bare = _BARE_EPISODE.match(text) + if bare and not _is_year_number(bare.group(1)): + collector.add_match( + rule_id="episode.bare", + kind="episode", + segment_index=index, + match=bare, + claims=Claims( + episode=_number(bare.group(1)), version=_int(bare.group(2)) + ), + priority=70, + evidence=("episode",), + conflict_tags=frozenset({"ambiguous-episode"}), + preserve_as_title_on_conflict=True, + ) + + +def _collect_media_candidates(collector: _CandidateCollector) -> None: + for index, segment in enumerate(collector.segments): + text = segment.text + + for rule_id, pattern in ( + ("media.movie-cjk", _MOVIE_CJK), + ("media.movie-romaji", _MOVIE_ROMAJI), + ): + for match in pattern.finditer(text): + collector.add_match( + rule_id=rule_id, + kind="movie", + segment_index=index, + match=match, + claims=Claims(media_type=MediaType.MOVIE), + priority=100, + specificity=2, + evidence=("movie",), + blocks=frozenset({"ambiguous-episode"}), + ) + + for match in _MOVIE_EN.finditer(text): + before = text[: match.start()].rstrip() + after = text[match.end() :].lstrip() + marker_is_edge = not before or not after or segment.enclosure == "square" + if re.match(r"[A-Za-z]", after): + continue + collector.add_match( + rule_id="media.movie-english", + kind="movie", + segment_index=index, + match=match, + claims=Claims(media_type=MediaType.MOVIE), + priority=100 if marker_is_edge else 99, + specificity=2, + evidence=("movie",), + blocks=frozenset({"ambiguous-episode"}), + ) + + for match in _SPECIAL_WORD.finditer(text): + after = text[match.end() :].lstrip() + if re.match(r"[A-Za-z]", after): + continue + media_type = _special_media(match.group(1)) + collector.add_match( + rule_id=f"media.{media_type.value}", + kind="special", + segment_index=index, + match=match, + claims=Claims(media_type=media_type), + priority=80, + specificity=1, + evidence=(media_type.value,), + ) + + for match in _SPECIAL_CJK.finditer(text): + collector.add_match( + rule_id="media.special-cjk", + kind="special", + segment_index=index, + match=match, + claims=Claims(media_type=MediaType.SPECIAL), + priority=80, + evidence=("special",), + ) + + for match in _PV.finditer(text): + if match.group(2) is not None: + continue + collector.add_match( + rule_id="media.pv", + kind="non-episode-video", + segment_index=index, + match=match, + claims=Claims(media_type=MediaType.PV), + priority=60, + evidence=("pv",), + ) + + for rule_id, pattern, media_type, evidence in ( + ("media.opening", _NCOP, MediaType.OPENING, "opening"), + ("media.ending", _NCED, MediaType.ENDING, "ending"), + ): + for match in pattern.finditer(text): + collector.add_match( + rule_id=rule_id, + kind="non-episode-video", + segment_index=index, + match=match, + claims=Claims(media_type=media_type, version=_int(match.group(1))), + priority=60, + evidence=(evidence,), + ) + + for match in _BATCH.finditer(text): + collector.add_match( + rule_id="cardinality.batch", + kind="batch", + segment_index=index, + match=match, + claims=Claims(release_kind=ReleaseKind.BATCH), + priority=80, + evidence=("batch",), + ) + + for match in _COLLECTION.finditer(text): + collector.add_match( + rule_id="cardinality.collection", + kind="collection", + segment_index=index, + match=match, + claims=Claims(release_kind=ReleaseKind.COLLECTION), + priority=80, + evidence=("collection",), + ) + + +_TECHNICAL_PATTERNS = ( + _RESOLUTION, + _SOURCE, + _CODEC, + _AUDIO, + _SUBTITLE, + _CONTAINER, +) + + +def _pattern_residual(text: str, patterns: tuple[re.Pattern[str], ...]) -> str: + mask = [False] * len(text) + for pattern in patterns: + for match in pattern.finditer(text): + mask[match.start() : match.end()] = [True] * (match.end() - match.start()) + return "".join(" " if used else char for char, used in zip(text, mask)) + + +def _collect_technical_candidates(collector: _CandidateCollector) -> None: + for index, segment in enumerate(collector.segments): + text = segment.text + + for match in _RESOLUTION.finditer(text): + collector.add_match( + rule_id="technical.resolution", + kind="resolution", + segment_index=index, + match=match, + claims=Claims(resolution=match.group()), + priority=30, + evidence=("resolution",), + ) + + for match in _SOURCE.finditer(text): + collector.add_match( + rule_id="technical.source", + kind="source", + segment_index=index, + match=match, + claims=Claims(source=match.group()), + priority=30, + evidence=("source",), + ) + + for match in _CODEC.finditer(text): + collector.add_match( + rule_id="technical.codec", + kind="codec", + segment_index=index, + match=match, + claims=Claims(codecs=(match.group(),)), + priority=30, + ) + + for match in _AUDIO.finditer(text): + collector.add_match( + rule_id="technical.audio", + kind="audio", + segment_index=index, + match=match, + claims=Claims(audio=(match.group(),)), + priority=30, + ) + + for match in _SUBTITLE.finditer(text): + container_match = re.search(r"_(MP4|MKV)$", match.group(), re.I) + collector.add_match( + rule_id="technical.subtitle", + kind="subtitle", + segment_index=index, + match=match, + claims=Claims( + subtitle=re.sub(r"_(?:MP4|MKV)$", "", match.group(), flags=re.I), + container=(container_match.group(1) if container_match else None), + ), + priority=35 if container_match else 30, + specificity=2 if container_match else 1, + ) + + for match in _CONTAINER.finditer(text): + collector.add_match( + rule_id="technical.container", + kind="container", + segment_index=index, + match=match, + claims=Claims(container=match.group(1)), + priority=30, + ) + + if segment.enclosure == "square": + residual = _clean_fragment(_pattern_residual(text, _TECHNICAL_PATTERNS)) + if residual and _CJK_SUBTITLE_TAG.fullmatch(residual): + collector.add_span( + rule_id="technical.subtitle-cjk", + kind="subtitle", + segment_index=index, + start=0, + end=len(text), + claims=Claims(subtitle=residual), + priority=45, + evidence=(), + overlap_policy=OverlapPolicy.SHARED, + ) + if _DUB.search(text): + collector.add_span( + rule_id="metadata.dub", + kind="tag", + segment_index=index, + start=0, + end=len(text), + claims=Claims(tags=(text.strip(),)), + priority=45, + overlap_policy=OverlapPolicy.SHARED, + ) + + for match in _VERSION.finditer(text): + collector.add_match( + rule_id="metadata.version", + kind="version", + segment_index=index, + match=match, + claims=Claims(version=int(match.group(1))), + priority=20, + ) + + for rule_id, pattern in ( + ("metadata.prefix", _PREFIX), + ("metadata.recruitment", _RECRUIT), + ("metadata.region", _REGION), + ("metadata.trailing-group", _TRAILING_RELEASE_GROUP), + ): + for match in pattern.finditer(text): + collector.add_match( + rule_id=rule_id, + kind="tag", + segment_index=index, + match=match, + claims=Claims(tags=(match.group().strip(),)), + priority=25, + ) + + for match in _HASH.finditer(text): + collector.add_match( + rule_id="metadata.hash", + kind="tag", + segment_index=index, + match=match, + claims=Claims(tags=(match.group(),)), + priority=25, + ) + + if segment.enclosure in {"square", "round"}: + residual = _clean_fragment(_pattern_residual(text, _TECHNICAL_PATTERNS)) + if residual.lower() in {"end", "生"}: + collector.add_span( + rule_id="metadata.end-marker", + kind="tag", + segment_index=index, + start=0, + end=len(text), + claims=Claims(tags=(residual,)), + priority=25, + overlap_policy=OverlapPolicy.SHARED, + ) + + +def _collect_metadata_episode_candidates(collector: _CandidateCollector) -> None: + for index, segment in enumerate(collector.segments): + if segment.enclosure != "square": + continue + residual = _pattern_residual(segment.text, _TECHNICAL_PATTERNS) + episode_match = _BARE_EPISODE.match(residual) or _TRAILING_EPISODE.search( + residual + ) + if not episode_match or _is_year_number(episode_match.group(1)): + continue + start = episode_match.start(1) + end = ( + episode_match.end(2) + if episode_match.group(2) is not None + else episode_match.end(1) + ) + collector.add_span( + rule_id="episode.metadata-mixed", + kind="episode", + segment_index=index, + start=start, + end=end, + claims=Claims( + episode=_number(episode_match.group(1)), + version=_int(episode_match.group(2)), + ), + priority=65, + evidence=("episode",), + conflict_tags=frozenset({"ambiguous-episode"}), + preserve_as_title_on_conflict=True, + captures=episode_match.groups(), + ) def _is_metadata_tail(text: str) -> bool: @@ -429,319 +1256,8 @@ def _is_metadata_tail(text: str) -> bool: 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], + working: list[_ResidualSegment], ) -> tuple[str | None, str | None, str | None]: fragments: list[str] = [] for work in working: @@ -880,21 +1396,6 @@ 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": diff --git a/backend/src/module/parser/analyser/tokenizer/resolver.py b/backend/src/module/parser/analyser/tokenizer/resolver.py new file mode 100644 index 00000000..c3f45692 --- /dev/null +++ b/backend/src/module/parser/analyser/tokenizer/resolver.py @@ -0,0 +1,339 @@ +"""Deterministic conflict resolution for tokenizer candidates.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from itertools import combinations +from typing import Iterable + +from .candidate import ( + Candidate, + ClaimField, + Claims, + Decision, + DecisionStatus, + OverlapPolicy, + ShadowedSpanPolicy, + Span, +) +from .result import MediaType, ReleaseKind + + +@dataclass(frozen=True, slots=True) +class Resolution: + """Resolved semantic values plus the decisions that produced them.""" + + claims: Claims + decisions: tuple[Decision, ...] + excluded_spans: tuple[Span, ...] + evidence: tuple[str, ...] + warnings: tuple[ResolutionWarning, ...] = () + + @property + def selected_candidate_ids(self) -> tuple[str, ...]: + return tuple( + decision.candidate_id + for decision in self.decisions + if decision.status is DecisionStatus.SELECTED + ) + + def decision_for(self, candidate_id: str) -> Decision: + for decision in self.decisions: + if decision.candidate_id == candidate_id: + return decision + raise KeyError(candidate_id) + + +@dataclass(frozen=True, slots=True, order=True) +class ResolutionWarning: + """Non-fatal ambiguity surfaced instead of hidden by a stable tie-break.""" + + field: ClaimField + candidate_ids: tuple[str, str] + reason: str = "equal-rank-conflicting-claims" + + +def resolve_candidates( + candidates: Iterable[Candidate], *, collect_warnings: bool = True +) -> Resolution: + """Resolve candidates without depending on their input iteration order. + + Higher priority and specificity win. Remaining ties are settled by source + coordinates, rule id, and candidate id. Rules should encode intentional + precedence in ``priority`` rather than relying on declaration order. + """ + + ordered = tuple(sorted(candidates, key=lambda candidate: candidate.sort_key)) + _ensure_unique_ids(ordered) + + selected: list[Candidate] = [] + scalar_values: dict[ClaimField, object] = {} + repeatable_values: dict[ClaimField, list[str]] = { + ClaimField.CODECS: [], + ClaimField.AUDIO: [], + ClaimField.TAGS: [], + } + decisions: list[Decision] = [] + + for candidate in ordered: + explicit_blocker = _find_explicit_blocker(candidate, selected) + if explicit_blocker is not None: + as_title = candidate.preserve_as_title_on_conflict + decisions.append( + Decision( + candidate_id=candidate.id, + status=( + DecisionStatus.REJECTED_AS_TITLE + if as_title + else DecisionStatus.REJECTED_CONFLICT + ), + reason=f"blocked-by:{explicit_blocker.id}", + ) + ) + continue + + overlap_blocker = _find_overlap_blocker(candidate, selected) + if overlap_blocker is not None: + as_title = candidate.preserve_as_title_on_conflict + decisions.append( + Decision( + candidate_id=candidate.id, + status=( + DecisionStatus.REJECTED_AS_TITLE + if as_title + else DecisionStatus.REJECTED_CONFLICT + ), + reason=f"overlaps:{overlap_blocker.id}", + ) + ) + continue + + scalar_items = candidate.claims.scalar_items() + occupied = tuple(field for field, _ in scalar_items if field in scalar_values) + if occupied: + shadowed_excluded_spans: tuple[Span, ...] = () + if candidate.shadowed_span_policy is ShadowedSpanPolicy.EXCLUDE: + shadowed_excluded_spans = ( + candidate.shadowed_spans + if candidate.shadowed_spans is not None + else candidate.spans + ) + decisions.append( + Decision( + candidate_id=candidate.id, + status=DecisionStatus.SHADOWED, + reason="shadowed-fields:" + + ",".join(field.value for field in occupied), + excluded_spans=shadowed_excluded_spans, + ) + ) + continue + + won_fields: list[ClaimField] = [] + for field, value in scalar_items: + if field not in scalar_values: + scalar_values[field] = value + won_fields.append(field) + + for field, values in candidate.claims.repeatable_items(): + target = repeatable_values[field] + before = len(target) + for value in values: + _append_unique_casefold(target, value) + if len(target) > before: + won_fields.append(field) + + selected.append(candidate) + decisions.append( + Decision( + candidate_id=candidate.id, + status=DecisionStatus.SELECTED, + reason="selected", + won_fields=tuple(won_fields), + excluded_spans=candidate.spans, + ) + ) + + decisions_tuple = tuple(decisions) + warnings = ( + _find_equal_rank_conflicts(ordered, decisions_tuple) if collect_warnings else () + ) + excluded_spans = _merge_spans( + span for decision in decisions_tuple for span in decision.excluded_spans + ) + evidence = _collect_evidence(ordered, decisions_tuple) + + return Resolution( + claims=_build_claims(scalar_values, repeatable_values), + decisions=decisions_tuple, + excluded_spans=excluded_spans, + evidence=evidence, + warnings=warnings, + ) + + +def _ensure_unique_ids(candidates: tuple[Candidate, ...]) -> None: + seen: set[str] = set() + for candidate in candidates: + if candidate.id in seen: + raise ValueError(f"duplicate candidate id: {candidate.id}") + seen.add(candidate.id) + + +def _find_equal_rank_conflicts( + candidates: tuple[Candidate, ...], + decisions: tuple[Decision, ...], +) -> tuple[ResolutionWarning, ...]: + selected_ids = { + decision.candidate_id + for decision in decisions + if decision.status is DecisionStatus.SELECTED + } + warnings: set[ResolutionWarning] = set() + for left, right in combinations(candidates, 2): + if ( + left.priority != right.priority + or left.specificity != right.specificity + or not ({left.id, right.id} & selected_ids) + ): + continue + left_values = dict(left.claims.scalar_items()) + right_values = dict(right.claims.scalar_items()) + for field in left_values.keys() & right_values.keys(): + if _warning_values_equal(field, left_values[field], right_values[field]): + continue + first_id, second_id = sorted((left.id, right.id)) + warnings.add( + ResolutionWarning( + field=field, + candidate_ids=(first_id, second_id), + ) + ) + return tuple(sorted(warnings)) + + +def _warning_values_equal(field: ClaimField, left: object, right: object) -> bool: + # ``season_raw`` records the spelling that supplied the semantic season; + # different spellings are provenance, not contradictory values. + if field is ClaimField.SEASON_RAW: + return True + if isinstance(left, str) and isinstance(right, str): + left_normalized = left.casefold() + right_normalized = right.casefold() + if field is ClaimField.SOURCE: + left_normalized = re.sub(r"[-_.\s]+", "", left_normalized) + right_normalized = re.sub(r"[-_.\s]+", "", right_normalized) + return left_normalized == right_normalized + return left == right + + +def _find_explicit_blocker( + candidate: Candidate, selected: list[Candidate] +) -> Candidate | None: + for winner in selected: + if (winner.blocks & candidate.conflict_tags) or ( + candidate.blocks & winner.conflict_tags + ): + return winner + return None + + +def _find_overlap_blocker( + candidate: Candidate, selected: list[Candidate] +) -> Candidate | None: + if candidate.overlap_policy is OverlapPolicy.SHARED: + return None + for winner in selected: + if winner.overlap_policy is OverlapPolicy.SHARED: + continue + if any( + span.overlaps(winner_span) + for span in candidate.spans + for winner_span in winner.spans + ): + return winner + return None + + +def _append_unique_casefold(values: list[str], value: str) -> None: + folded = value.casefold() + if folded not in {existing.casefold() for existing in values}: + values.append(value) + + +def _build_claims( + scalar_values: dict[ClaimField, object], + repeatable_values: dict[ClaimField, list[str]], +) -> Claims: + return Claims( + group=_scalar(scalar_values, ClaimField.GROUP, str), + season=_scalar(scalar_values, ClaimField.SEASON, int), + season_raw=_scalar(scalar_values, ClaimField.SEASON_RAW, str), + episode=_number_scalar(scalar_values, ClaimField.EPISODE), + episode_end=_number_scalar(scalar_values, ClaimField.EPISODE_END), + episode_title=_scalar(scalar_values, ClaimField.EPISODE_TITLE, str), + media_type=_scalar(scalar_values, ClaimField.MEDIA_TYPE, MediaType), + release_kind=_scalar(scalar_values, ClaimField.RELEASE_KIND, ReleaseKind), + resolution=_scalar(scalar_values, ClaimField.RESOLUTION, str), + source=_scalar(scalar_values, ClaimField.SOURCE, str), + subtitle=_scalar(scalar_values, ClaimField.SUBTITLE, str), + codecs=tuple(repeatable_values[ClaimField.CODECS]), + audio=tuple(repeatable_values[ClaimField.AUDIO]), + container=_scalar(scalar_values, ClaimField.CONTAINER, str), + version=_scalar(scalar_values, ClaimField.VERSION, int), + year=_scalar(scalar_values, ClaimField.YEAR, int), + tags=tuple(repeatable_values[ClaimField.TAGS]), + ) + + +def _scalar[T]( + values: dict[ClaimField, object], field: ClaimField, expected: type[T] +) -> T | None: + value = values.get(field) + if value is None: + return None + if expected is not object and not isinstance(value, expected): + raise TypeError(f"invalid value for {field.value}: {value!r}") + return value # type: ignore[return-value] + + +def _number_scalar( + values: dict[ClaimField, object], field: ClaimField +) -> int | float | None: + value = values.get(field) + if value is None: + return None + if not isinstance(value, (int, float)): + raise TypeError(f"invalid value for {field.value}: {value!r}") + return value + + +def _merge_spans(spans: Iterable[Span]) -> tuple[Span, ...]: + ordered = sorted(set(spans)) + if not ordered: + return () + + merged: list[Span] = [ordered[0]] + for span in ordered[1:]: + previous = merged[-1] + if span.segment == previous.segment and span.start <= previous.end: + merged[-1] = Span( + previous.segment, previous.start, max(previous.end, span.end) + ) + else: + merged.append(span) + return tuple(merged) + + +def _collect_evidence( + candidates: tuple[Candidate, ...], decisions: tuple[Decision, ...] +) -> tuple[str, ...]: + evidence: list[str] = [] + for candidate, decision in zip(candidates, decisions, strict=True): + if decision.status is not DecisionStatus.SELECTED: + continue + for item in candidate.evidence: + if item not in evidence: + evidence.append(item) + return tuple(evidence) diff --git a/backend/src/module/parser/analyser/tokenizer/trace.py b/backend/src/module/parser/analyser/tokenizer/trace.py new file mode 100644 index 00000000..a0100823 --- /dev/null +++ b/backend/src/module/parser/analyser/tokenizer/trace.py @@ -0,0 +1,108 @@ +"""Immutable diagnostics for explaining release-title parser decisions.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .candidate import Candidate, Claims, Decision, DecisionStatus, Observation, Span +from .resolver import Resolution, ResolutionWarning +from .result import ParsedRelease + + +@dataclass(frozen=True, slots=True) +class TraceSegment: + """Segment snapshot with explicit normalized-text coordinates.""" + + index: int + text: str + outer_start: int + outer_end: int + content_start: int + content_end: int + enclosure: str | None = None + + def __post_init__(self) -> None: + if self.index < 0: + raise ValueError("segment index must be non-negative") + if not (0 <= self.outer_start <= self.content_start): + raise ValueError("invalid segment start coordinates") + if not (self.content_start <= self.content_end <= self.outer_end): + raise ValueError("invalid segment end coordinates") + + +@dataclass(frozen=True, slots=True) +class ParseTrace: + """All observations and resolver decisions for one parser invocation.""" + + raw: str + normalized: str + segments: tuple[TraceSegment, ...] = () + observations: tuple[Observation, ...] = () + candidates: tuple[Candidate, ...] = () + claims: Claims = Claims() + decisions: tuple[Decision, ...] = () + excluded_spans: tuple[Span, ...] = () + residuals: tuple[str, ...] = () + warnings: tuple[ResolutionWarning, ...] = () + + def __post_init__(self) -> None: + object.__setattr__( + self, "segments", tuple(sorted(self.segments, key=lambda item: item.index)) + ) + object.__setattr__( + self, + "observations", + tuple(sorted(self.observations, key=lambda item: item.sort_key)), + ) + object.__setattr__( + self, + "candidates", + tuple(sorted(self.candidates, key=lambda item: item.sort_key)), + ) + + @classmethod + def from_resolution( + cls, + *, + raw: str, + normalized: str, + resolution: Resolution, + segments: tuple[TraceSegment, ...] = (), + observations: tuple[Observation, ...] = (), + candidates: tuple[Candidate, ...] = (), + residuals: tuple[str, ...] = (), + ) -> ParseTrace: + return cls( + raw=raw, + normalized=normalized, + segments=segments, + observations=observations, + candidates=candidates, + claims=resolution.claims, + decisions=resolution.decisions, + excluded_spans=resolution.excluded_spans, + residuals=residuals, + warnings=resolution.warnings, + ) + + def decision_for(self, candidate_id: str) -> Decision: + for decision in self.decisions: + if decision.candidate_id == candidate_id: + return decision + raise KeyError(candidate_id) + + @property + def rejected_as_title(self) -> tuple[str, ...]: + return tuple( + decision.candidate_id + for decision in self.decisions + if decision.status is DecisionStatus.REJECTED_AS_TITLE + ) + + +@dataclass(frozen=True, slots=True) +class ParseOutcome: + """Optional parse result paired with its complete diagnostic trace.""" + + result: ParsedRelease | None + trace: ParseTrace diff --git a/backend/src/test/fixtures/release_title_golden.json b/backend/src/test/fixtures/release_title_golden.json new file mode 100644 index 00000000..2a7c5f0a --- /dev/null +++ b/backend/src/test/fixtures/release_title_golden.json @@ -0,0 +1,1603 @@ +{ + "schema_version": 1, + "cases": [ + { + "id": "episode_multilingual_lolihouse", + "category": "episode", + "raw": "[LoliHouse] 葬送的芙莉莲 / Sousou no Frieren - 03 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕][MKV]", + "expected": { + "title_en": "Sousou no Frieren", + "title_zh": "葬送的芙莉莲", + "group": "LoliHouse", + "episode": 3, + "media_type": "episode", + "resolution": "1080p", + "source": "WebRip", + "subtitle": "简繁内封字幕", + "codecs": [ + "HEVC-10bit" + ], + "audio": [ + "AAC" + ], + "container": "MKV" + }, + "legacy": { + "title_en": "Sousou no Frieren", + "title_zh": "葬送的芙莉莲", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 3, + "sub": "简繁内封字幕", + "group": "LoliHouse", + "resolution": "1080p", + "source": "WebRip", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_sxxexx", + "category": "episode", + "raw": "[Group] Anime Title S02E05 [WEB-DL 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "season": 2, + "season_raw": "S02E05", + "episode": 5, + "media_type": "episode", + "resolution": "1080p", + "source": "WEB-DL" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 2, + "season_raw": "S02E05", + "episode": 5, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "WEB-DL", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_season_words", + "category": "episode", + "raw": "[Group] Anime Title Season 2 - 01 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "season": 2, + "season_raw": "Season 2", + "episode": 1, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 2, + "season_raw": "Season 2", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_fractional", + "category": "episode", + "raw": "[Group] Anime Title - 12.5 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 12.5, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "episode_versioned", + "category": "episode", + "raw": "[Group] Anime Title - 12v2 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 12, + "media_type": "episode", + "resolution": "1080p", + "version": 2 + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 12, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "range_bare", + "category": "range", + "raw": "[Group] Anime Title [01-12] [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 1, + "episode_end": 12, + "media_type": "episode", + "release_kind": "range", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": null + }, + { + "id": "range_e_prefix", + "category": "range", + "raw": "[Group] Anime Title [E01-E12] [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 1, + "episode_end": 12, + "media_type": "episode", + "release_kind": "range", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": null + }, + { + "id": "batch_complete", + "category": "batch", + "raw": "[VCB-Studio] Violet Evergarden Complete Batch [BDRip 1080p]", + "expected": { + "title_en": "Violet Evergarden", + "group": "VCB-Studio", + "media_type": "episode", + "release_kind": "batch", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": null + }, + { + "id": "collection_chinese_heji", + "category": "collection", + "raw": "[Group] 葬送的芙莉莲合集 [BDRip 1080p]", + "expected": { + "title_zh": "葬送的芙莉莲", + "group": "Group", + "media_type": "episode", + "release_kind": "collection", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": null, + "title_zh": "葬送的芙莉莲", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "collection_chinese_quanji", + "category": "collection", + "raw": "[Group] 葬送的芙莉莲全集 [BDRip 1080p]", + "expected": { + "title_zh": "葬送的芙莉莲", + "group": "Group", + "media_type": "episode", + "release_kind": "collection", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": null, + "title_zh": "葬送的芙莉莲", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "collection_chinese_summary", + "category": "collection", + "raw": "[Group] 进击的巨人总集篇 [BDRip 1080p]", + "expected": { + "title_zh": "进击的巨人", + "group": "Group", + "media_type": "episode", + "release_kind": "collection", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": null, + "title_zh": "进击的巨人", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "collection_bonus", + "category": "collection", + "raw": "[Group] Anime Title 特典 [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "media_type": "episode", + "release_kind": "collection", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "collection_movie", + "category": "collection", + "raw": "[Group] 剧场版合集 - 进击的巨人 [1080p]", + "expected": { + "title_zh": "进击的巨人", + "group": "Group", + "media_type": "movie", + "release_kind": "collection", + "resolution": "1080p" + }, + "legacy": { + "title_en": null, + "title_zh": "进击的巨人", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "special_ova_numbered", + "category": "special", + "raw": "[Group] Anime Title OVA01 [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 1, + "media_type": "ova", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 0, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "special", + "is_movie": false + } + }, + { + "id": "special_oad_numbered", + "category": "special", + "raw": "[Group] Anime Title OAD 02 [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 2, + "media_type": "oad", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 0, + "season_raw": "", + "episode": 2, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "special", + "is_movie": false + } + }, + { + "id": "special_sp_numbered", + "category": "special", + "raw": "[Group] Anime Title SP03 [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 3, + "media_type": "special", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 0, + "season_raw": "", + "episode": 3, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "special", + "is_movie": false + } + }, + { + "id": "movie_attached_chinese", + "category": "movie", + "raw": "[VCB-Studio]电影版哆啦A梦：大雄的地球交响乐[BDRip 1080p]", + "expected": { + "title_zh": "哆啦A梦：大雄的地球交响乐", + "group": "VCB-Studio", + "media_type": "movie", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": null, + "title_zh": "哆啦A梦：大雄的地球交响乐", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "VCB-Studio", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "movie_attached_japanese", + "category": "movie", + "raw": "[SubGroup]劇場版SPY×FAMILY CODE: White[WEB-DL 1080p]", + "expected": { + "title_en": "SPY×FAMILY CODE: White", + "group": "SubGroup", + "media_type": "movie", + "resolution": "1080p", + "source": "WEB-DL" + }, + "legacy": { + "title_en": "SPY×FAMILY CODE: White", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "SubGroup", + "resolution": "1080p", + "source": "WEB-DL", + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "auxiliary_pv_numbered", + "category": "auxiliary", + "raw": "[Official] Anime Title PV2 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Official", + "episode": 2, + "media_type": "pv", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "auxiliary_ncop", + "category": "auxiliary", + "raw": "[VCB-Studio] Anime Title NCOP [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "VCB-Studio", + "media_type": "opening", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": null + }, + { + "id": "auxiliary_nced", + "category": "auxiliary", + "raw": "[VCB-Studio] Anime Title NCED [BDRip 1080p]", + "expected": { + "title_en": "Anime Title", + "group": "VCB-Studio", + "media_type": "ending", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": null + }, + { + "id": "title_only_english", + "category": "title-only", + "raw": "[LoliHouse] Sousou no Frieren [WEB-DL 1080p]", + "expected": { + "title_en": "Sousou no Frieren", + "group": "LoliHouse", + "resolution": "1080p", + "source": "WEB-DL" + }, + "legacy": null + }, + { + "id": "title_only_slash", + "category": "title-only", + "raw": "[VCB-Studio] Fate/stay night [BDRip 1080p]", + "expected": { + "title_en": "Fate/stay night", + "group": "VCB-Studio", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": null + }, + { + "id": "movie_year_parenthesized", + "category": "movie", + "raw": "[VCB-Studio] Kimi no Na wa. (2016) [BDRip 1080p]", + "expected": { + "title_en": "Kimi no Na wa.", + "group": "VCB-Studio", + "resolution": "1080p", + "source": "BDRip", + "year": 2016 + }, + "legacy": null + }, + { + "id": "negative_only_technical", + "category": "negative", + "raw": "[1080p][HEVC][AAC][MKV]", + "expected": null, + "legacy": null + }, + { + "id": "ambiguity_movie_in_group", + "category": "ambiguity", + "raw": "[Movie-Fan] Ordinary Anime - 01 [WEB-DL 1080p]", + "expected": { + "title_en": "Ordinary Anime", + "group": "Movie-Fan", + "episode": 1, + "media_type": "episode", + "resolution": "1080p", + "source": "WEB-DL" + }, + "legacy": { + "title_en": "Ordinary Anime", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Movie-Fan", + "resolution": "1080p", + "source": "WEB-DL", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "ambiguity_movie_in_episode_title", + "category": "ambiguity", + "raw": "[SubGroup] The Movie Star Next Door - 03 [WEB-DL 1080p]", + "expected": { + "title_en": "The Movie Star Next Door", + "group": "SubGroup", + "episode": 3, + "media_type": "episode", + "resolution": "1080p", + "source": "WEB-DL" + }, + "legacy": { + "title_en": "The Movie Star Next Door", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 3, + "sub": null, + "group": "SubGroup", + "resolution": "1080p", + "source": "WEB-DL", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "ambiguity_fraction_in_title", + "category": "ambiguity", + "raw": "[SubGroup] Ranma 1/2 - 03 [WEB-DL 1080p]", + "expected": { + "title_en": "Ranma 1/2", + "group": "SubGroup", + "episode": 3, + "media_type": "episode", + "resolution": "1080p", + "source": "WEB-DL" + }, + "legacy": { + "title_en": "Ranma 1/2", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 3, + "sub": null, + "group": "SubGroup", + "resolution": "1080p", + "source": "WEB-DL", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "metadata_leading_movie", + "category": "movie", + "raw": "[MOVIE] Anime Title [1080p]", + "expected": { + "title_en": "Anime Title", + "media_type": "movie", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "", + "resolution": "1080p", + "source": null, + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "metadata_leading_movie_cjk", + "category": "movie", + "raw": "[劇場版] Anime Title [1080p]", + "expected": { + "title_en": "Anime Title", + "media_type": "movie", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "", + "resolution": "1080p", + "source": null, + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "metadata_leading_sxxexx", + "category": "episode", + "raw": "[S02E05] Anime Title [1080p]", + "expected": { + "title_en": "Anime Title", + "season": 2, + "season_raw": "S02E05", + "episode": 5, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 2, + "season_raw": "S02E05", + "episode": 5, + "sub": null, + "group": "", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "metadata_leading_season", + "category": "episode", + "raw": "[Season 2] Anime Title - 01 [1080p]", + "expected": { + "title_en": "Anime Title", + "season": 2, + "season_raw": "Season 2", + "episode": 1, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 2, + "season_raw": "Season 2", + "episode": 1, + "sub": null, + "group": "", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "metadata_leading_range", + "category": "range", + "raw": "[E01-E12] Anime Title [1080p]", + "expected": { + "title_en": "Anime Title", + "episode": 1, + "episode_end": 12, + "media_type": "episode", + "release_kind": "range", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "metadata_leading_technical", + "category": "technical", + "raw": "[BDRip 1080p] Anime Title - 01", + "expected": { + "title_en": "Anime Title", + "episode": 1, + "media_type": "episode", + "resolution": "1080p", + "source": "BDRip" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "", + "resolution": "1080p", + "source": "BDRip", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "special_attached_ova", + "category": "special", + "raw": "[Group] 某作品OVA [1080p]", + "expected": { + "title_zh": "某作品", + "group": "Group", + "media_type": "ova", + "resolution": "1080p" + }, + "legacy": { + "title_en": null, + "title_zh": "某作品", + "title_jp": null, + "season": 0, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "special", + "is_movie": false + } + }, + { + "id": "special_attached_side_story", + "category": "special", + "raw": "[Group] 某作品番外篇 [1080p]", + "expected": { + "title_zh": "某作品", + "group": "Group", + "media_type": "special", + "resolution": "1080p" + }, + "legacy": { + "title_en": null, + "title_zh": "某作品", + "title_jp": null, + "season": 0, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "special", + "is_movie": false + } + }, + { + "id": "special_attached_special_chinese", + "category": "special", + "raw": "[Group] 某作品特别篇 [1080p]", + "expected": { + "title_zh": "某作品", + "group": "Group", + "media_type": "special", + "resolution": "1080p" + }, + "legacy": { + "title_en": null, + "title_zh": "某作品", + "title_jp": null, + "season": 0, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "special", + "is_movie": false + } + }, + { + "id": "collection_numbered_complete", + "category": "collection", + "raw": "[Group] 迷宫饭 全24话 [1080p]", + "expected": { + "title_zh": "迷宫饭", + "group": "Group", + "episode": 1, + "episode_end": 24, + "media_type": "episode", + "release_kind": "collection", + "resolution": "1080p" + }, + "legacy": { + "title_en": null, + "title_zh": "迷宫饭", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "technical_provider_labels", + "category": "technical", + "raw": "[EMBER] Frieren - 12.5 [BDRip 1080p AVC 8bit Dual Audio Multiple Subtitle MKV]", + "expected": { + "title_en": "Frieren", + "group": "EMBER", + "episode": 12.5, + "media_type": "episode", + "resolution": "1080p", + "source": "BDRip", + "subtitle": "Multiple Subtitle", + "codecs": [ + "AVC", + "8bit" + ], + "audio": [ + "Dual Audio" + ], + "container": "MKV" + }, + "legacy": null + }, + { + "id": "episode_plex_title", + "category": "episode", + "raw": "Mob Psycho 100 - S03E05 - Divine Tree 2 ~Peace~ [1080p]", + "expected": { + "title_en": "Mob Psycho 100", + "season": 3, + "season_raw": "S03E05", + "episode": 5, + "episode_title": "Divine Tree 2 ~Peace~", + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Mob Psycho 100", + "title_zh": null, + "title_jp": null, + "season": 3, + "season_raw": "S03E05", + "episode": 5, + "sub": null, + "group": "", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "movie_inline_year", + "category": "movie", + "raw": "The Boy and the Heron 2023 1080p BluRay", + "expected": { + "title_en": "The Boy and the Heron", + "resolution": "1080p", + "source": "BluRay", + "year": 2023 + }, + "legacy": null + }, + { + "id": "ambiguity_pv_in_title", + "category": "ambiguity", + "raw": "[Group] PV=nRT - 01 [1080p]", + "expected": { + "title_en": "PV=nRT", + "group": "Group", + "episode": 1, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "PV=nRT", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "ambiguity_movie_title_only", + "category": "ambiguity", + "raw": "[Group] The Movie Star Next Door [1080p]", + "expected": { + "title_en": "The Movie Star Next Door", + "group": "Group", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "ambiguity_special_title_only", + "category": "ambiguity", + "raw": "[Group] Special A [1080p]", + "expected": { + "title_en": "Special A", + "group": "Group", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "movie_numeric_title_cjk", + "category": "movie", + "raw": "[Group] 劇場版 呪術廻戦 0 [1080p]", + "expected": { + "title_zh": "呪術廻戦 0", + "group": "Group", + "media_type": "movie", + "resolution": "1080p" + }, + "legacy": { + "title_en": null, + "title_zh": "呪術廻戦 0", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "movie_numeric_title_romaji", + "category": "movie", + "raw": "[Group] Gekijouban Naruto 3 [1080p]", + "expected": { + "title_en": "Naruto 3", + "group": "Group", + "media_type": "movie", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Naruto 3", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "episode_numeric_title_86", + "category": "episode", + "raw": "[Group] 86 - 01 [1080p]", + "expected": { + "title_en": "86", + "group": "Group", + "episode": 1, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "86", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_numeric_title_22_7", + "category": "episode", + "raw": "[Group] 22/7 - 01 [1080p]", + "expected": { + "title_en": "22/7", + "group": "Group", + "episode": 1, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "22/7", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_ordinal_season", + "category": "episode", + "raw": "[Group] Dandadan 2nd Season - 01 [1080p]", + "expected": { + "title_en": "Dandadan", + "group": "Group", + "season": 2, + "season_raw": "2nd Season", + "episode": 1, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Dandadan", + "title_zh": null, + "title_jp": null, + "season": 2, + "season_raw": "2nd Season", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_inside_technical_tag", + "category": "episode", + "raw": "[Group] Anime Title [01 1080p HEVC]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 1, + "media_type": "episode", + "resolution": "1080p", + "codecs": [ + "HEVC" + ] + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "technical_surround_audio", + "category": "technical", + "raw": "[Group] Anime Title - 01 [Dual Audio][DDP5.1][1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 1, + "media_type": "episode", + "resolution": "1080p", + "audio": [ + "Dual Audio", + "DDP5.1" + ] + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "legacy_reject_range", + "category": "legacy-boundary", + "raw": "[Group] Anime [01-12] [1080p]", + "expected": { + "title_en": "Anime", + "group": "Group", + "episode": 1, + "episode_end": 12, + "media_type": "episode", + "release_kind": "range", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "legacy_reject_batch", + "category": "legacy-boundary", + "raw": "[Group] Anime Complete Batch [1080p]", + "expected": { + "title_en": "Anime", + "group": "Group", + "media_type": "episode", + "release_kind": "batch", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "legacy_reject_pv", + "category": "legacy-boundary", + "raw": "[Group] Anime PV2 [1080p]", + "expected": { + "title_en": "Anime", + "group": "Group", + "episode": 2, + "media_type": "pv", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "legacy_reject_ncop", + "category": "legacy-boundary", + "raw": "[Group] Anime NCOP [1080p]", + "expected": { + "title_en": "Anime", + "group": "Group", + "media_type": "opening", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "legacy_reject_fractional", + "category": "legacy-boundary", + "raw": "[Group] Anime - 12.5 [1080p]", + "expected": { + "title_en": "Anime", + "group": "Group", + "episode": 12.5, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "episode_mikan_multiline", + "category": "episode", + "raw": "[喵萌奶茶屋&LoliHouse] 鹿乃子乃子乃子虎视眈眈 / Shikanoko Nokonoko Koshitantan\n- 01 [WebRip 1080p HEVC-10bit AAC][简繁内封字幕]", + "expected": { + "title_en": "Shikanoko Nokonoko Koshitantan", + "title_zh": "鹿乃子乃子乃子虎视眈眈", + "group": "喵萌奶茶屋&LoliHouse", + "episode": 1, + "media_type": "episode", + "resolution": "1080p", + "source": "WebRip", + "subtitle": "简繁内封字幕", + "codecs": [ + "HEVC-10bit" + ], + "audio": [ + "AAC" + ] + }, + "legacy": { + "title_en": "Shikanoko Nokonoko Koshitantan", + "title_zh": "鹿乃子乃子乃子虎视眈眈", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": "简繁内封字幕", + "group": "喵萌奶茶屋&LoliHouse", + "resolution": "1080p", + "source": "WebRip", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_fullwidth_brackets", + "category": "episode", + "raw": "【幻樱字幕组】【4月新番】【古见同学有交流障碍症 第二季 Komi-san wa, Komyushou Desu. S02】【22】【GB_MP4】【1920X1080】", + "expected": { + "title_en": "Komi-san wa, Komyushou Desu.", + "title_zh": "古见同学有交流障碍症", + "group": "幻樱字幕组", + "season": 2, + "season_raw": "第二季", + "episode": 22, + "media_type": "episode", + "resolution": "1920X1080", + "subtitle": "GB", + "container": "MP4", + "tags": [ + "4月新番" + ] + }, + "legacy": { + "title_en": "Komi-san wa, Komyushou Desu.", + "title_zh": "古见同学有交流障碍症", + "title_jp": null, + "season": 2, + "season_raw": "第二季", + "episode": 22, + "sub": "GB", + "group": "幻樱字幕组", + "resolution": "1920X1080", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_chinese_marker", + "category": "episode", + "raw": "【极影字幕社】★4月新番 天国大魔境 Tengoku Daimakyou 第05话 GB 720P MP4（字幕社招人内详）", + "expected": { + "title_en": "Tengoku Daimakyou", + "title_zh": "天国大魔境", + "group": "极影字幕社", + "episode": 5, + "media_type": "episode", + "resolution": "720P", + "subtitle": "GB", + "container": "MP4", + "tags": [ + "★4月新番", + "（字幕社招人内详）" + ] + }, + "legacy": { + "title_en": "Tengoku Daimakyou", + "title_zh": "天国大魔境", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 5, + "sub": "GB", + "group": "极影字幕社", + "resolution": "720P", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_date_and_size_tags", + "category": "episode", + "raw": "[梦蓝字幕组]New Doraemon 哆啦A梦新番[747][2023.02.25][AVC][1080P][GB_JP][MP4][101.69 MB]", + "expected": { + "title_en": "New Doraemon", + "title_zh": "哆啦A梦新番", + "group": "梦蓝字幕组", + "episode": 747, + "media_type": "episode", + "resolution": "1080P", + "subtitle": "GB_JP", + "codecs": [ + "AVC" + ], + "container": "MP4", + "tags": [ + "2023.02.25", + "101.69 MB" + ] + }, + "legacy": { + "title_en": "New Doraemon", + "title_zh": "哆啦A梦新番", + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 747, + "sub": "GB_JP", + "group": "梦蓝字幕组", + "resolution": "1080P", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_file_extension", + "category": "episode", + "raw": "[Lilith-Raws] Otonari no Tenshi-sama - 09 [Baha][WEB-DL][1080p][AVC AAC][CHT][MP4].mkv", + "expected": { + "title_en": "Otonari no Tenshi-sama", + "group": "Lilith-Raws", + "episode": 9, + "media_type": "episode", + "resolution": "1080p", + "source": "Baha", + "subtitle": "CHT", + "codecs": [ + "AVC" + ], + "audio": [ + "AAC" + ], + "container": "MP4" + }, + "legacy": { + "title_en": "Otonari no Tenshi-sama", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 9, + "sub": "CHT", + "group": "Lilith-Raws", + "resolution": "1080p", + "source": "Baha", + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_pre_release", + "category": "episode", + "raw": "[Group] Anime Title [01Pre] [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 1, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 1, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_atlas_style", + "category": "episode", + "raw": "[Group] Anime Title [04_アニメサブタイトル] [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 4, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 4, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_chinese_season", + "category": "episode", + "raw": "[Group] 某作品 第二季 - 08 [1080p]", + "expected": { + "title_zh": "某作品", + "group": "Group", + "season": 2, + "season_raw": "第二季", + "episode": 8, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": null, + "title_zh": "某作品", + "title_jp": null, + "season": 2, + "season_raw": "第二季", + "episode": 8, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "episode_season_episode_words", + "category": "episode", + "raw": "[Group] Anime Title Season 2 Episode 08 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "season": 2, + "season_raw": "Season 2 Episode 08", + "episode": 8, + "media_type": "episode", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 2, + "season_raw": "Season 2 Episode 08", + "episode": 8, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "episode", + "is_movie": false + } + }, + { + "id": "auxiliary_ncop_versioned", + "category": "auxiliary", + "raw": "[Group] Anime Title NCOPv2 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "media_type": "opening", + "resolution": "1080p", + "version": 2 + }, + "legacy": null + }, + { + "id": "auxiliary_nced_versioned", + "category": "auxiliary", + "raw": "[Group] Anime Title NCEDv3 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "media_type": "ending", + "resolution": "1080p", + "version": 3 + }, + "legacy": null + }, + { + "id": "auxiliary_cm_numbered", + "category": "auxiliary", + "raw": "[Group] Anime Title CM1 [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "episode": 1, + "media_type": "pv", + "resolution": "1080p" + }, + "legacy": null + }, + { + "id": "movie_english_marker", + "category": "movie", + "raw": "[Group] Anime Title Movie [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "media_type": "movie", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "movie_romaji_marker", + "category": "movie", + "raw": "[Group] Gekijouban Anime Title [1080p]", + "expected": { + "title_en": "Anime Title", + "group": "Group", + "media_type": "movie", + "resolution": "1080p" + }, + "legacy": { + "title_en": "Anime Title", + "title_zh": null, + "title_jp": null, + "season": 1, + "season_raw": "", + "episode": 0, + "sub": null, + "group": "Group", + "resolution": "1080p", + "source": null, + "episode_type": "movie", + "is_movie": true + } + }, + { + "id": "negative_date_hash_size", + "category": "negative", + "raw": "[A1B2C3D4][1.2 GiB][2023-01-12]", + "expected": null, + "legacy": null + } + ] +} diff --git a/backend/src/test/test_release_parser.py b/backend/src/test/test_release_parser.py index 9c9c2ddc..43ff57c3 100644 --- a/backend/src/test/test_release_parser.py +++ b/backend/src/test/test_release_parser.py @@ -113,6 +113,52 @@ def test_episode_range_is_a_range_release(raw: str) -> None: assert parsed.release_kind is ReleaseKind.RANGE +@pytest.mark.parametrize( + ("raw", "media_type", "season", "episode_end"), + ( + ( + "[Group] Anime Title S02E01-E12 [1080p]", + MediaType.EPISODE, + 2, + 12, + ), + ( + "[Group] Anime Title Season 2 Episodes 01-12 [1080p]", + MediaType.EPISODE, + 2, + 12, + ), + ( + "[Group] Anime Title OVA01-02 [1080p]", + MediaType.OVA, + None, + 2, + ), + ( + "[Group] Anime Title OAD01-OAD03 [1080p]", + MediaType.OAD, + None, + 3, + ), + ), +) +def test_compound_range_forms_do_not_degrade_to_single_episode( + raw: str, + media_type: MediaType, + season: int | None, + episode_end: int, +) -> None: + parsed = parse_release_title(raw) + + assert parsed is not None + assert parsed.media_type is media_type + assert parsed.release_kind is ReleaseKind.RANGE + assert parsed.season == season + assert parsed.episode == 1 + assert parsed.episode_end == episode_end + assert raw_parser(raw) is None + + def test_complete_batch_without_episode_number() -> None: parsed = parse_release_title( "[VCB-Studio] Violet Evergarden Complete Batch [BDRip 1080p]" @@ -262,6 +308,36 @@ def test_movie_word_in_group_does_not_mark_episode_as_movie() -> None: assert parsed.media_type is MediaType.EPISODE +def test_rejected_structural_marker_inside_group_does_not_infer_media() -> None: + parsed = parse_release_title("[S02E05-Fansub] Plain Anime Title [1080p]") + + assert parsed is not None + assert parsed.group == "S02E05-Fansub" + assert parsed.episode is None + assert parsed.media_type is MediaType.UNKNOWN + + +def test_shadowed_episode_title_candidate_does_not_remove_title_text() -> None: + raw = "[Group] Anime S01E01 Season 2 Episode 3 - Another Title " "[1080p]" + + parsed = parse_release_title(raw) + + assert parsed is not None + assert parsed.season == 1 + assert parsed.episode == 1 + assert "Another Title" in _title_text(parsed) + assert "episode-title" not in parsed.evidence + + +def test_resolved_evidence_excludes_shadowed_special_candidate() -> None: + parsed = parse_release_title("[Group] Anime OVA01 OAD02 [1080p]") + + assert parsed is not None + assert parsed.media_type is MediaType.OVA + assert "ova" in parsed.evidence + assert "oad" not in parsed.evidence + + 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]" diff --git a/backend/src/test/test_release_parser_golden.py b/backend/src/test/test_release_parser_golden.py new file mode 100644 index 00000000..1e57c585 --- /dev/null +++ b/backend/src/test/test_release_parser_golden.py @@ -0,0 +1,160 @@ +"""Golden-corpus regression tests for release-title parsing. + +The fixture snapshots only the public generic result and the legacy adapter. +Parser evidence and trace data are deliberately excluded so internal rule and +resolver changes do not create meaningless snapshot churn. +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, fields +from enum import Enum +from pathlib import Path +from typing import Any + +import pytest + +from module.parser.analyser.raw_parser import raw_parser +from module.parser.analyser.tokenizer import ( + parse_release_title, + parse_release_title_with_trace, +) + +_CORPUS_PATH = Path(__file__).parent / "fixtures" / "release_title_golden.json" +_CORPUS_SIZE = 71 + +_GENERIC_DEFAULTS: dict[str, Any] = { + "title_en": None, + "title_zh": None, + "title_jp": None, + "group": None, + "season": None, + "season_raw": None, + "episode": None, + "episode_end": None, + "episode_title": None, + "media_type": "unknown", + "release_kind": "single", + "resolution": None, + "source": None, + "subtitle": None, + "codecs": [], + "audio": [], + "container": None, + "version": None, + "year": None, + "tags": [], +} + + +def _load_corpus() -> dict[str, Any]: + with _CORPUS_PATH.open(encoding="utf-8") as corpus_file: + return json.load(corpus_file) + + +_CORPUS = _load_corpus() +_CASES = _CORPUS["cases"] + + +def _case_params() -> list[Any]: + return [pytest.param(case, id=case["id"]) for case in _CASES] + + +def _json_value(value: Any) -> Any: + if isinstance(value, Enum): + return value.value + if isinstance(value, tuple): + return list(value) + return value + + +def _generic_snapshot(parsed: Any) -> dict[str, Any]: + return { + field_name: _json_value(getattr(parsed, field_name)) + for field_name in _GENERIC_DEFAULTS + } + + +def _expected_generic(case: dict[str, Any]) -> dict[str, Any] | None: + overrides = case["expected"] + if overrides is None: + return None + return {**_GENERIC_DEFAULTS, **overrides} + + +def test_golden_corpus_integrity() -> None: + assert _CORPUS["schema_version"] == 1 + assert len(_CASES) == _CORPUS_SIZE + + ids = [case["id"] for case in _CASES] + raw_titles = [case["raw"] for case in _CASES] + identities = [(case["category"], case["id"], case["raw"]) for case in _CASES] + + assert len(ids) == len(set(ids)) + assert len(raw_titles) == len(set(raw_titles)) + assert len(identities) == len(set(identities)) + assert all(case["category"] for case in _CASES) + assert all("evidence" not in case for case in _CASES) + assert all( + case["expected"] is None or "evidence" not in case["expected"] + for case in _CASES + ) + assert all("trace" not in case for case in _CASES) + + +@pytest.mark.parametrize("case", _case_params()) +def test_generic_parser_matches_golden_corpus(case: dict[str, Any]) -> None: + parsed = parse_release_title(case["raw"]) + expected = _expected_generic(case) + + if expected is None: + assert parsed is None + return + + assert parsed is not None + assert parsed.raw == case["raw"] + public_fields = { + field.name for field in fields(parsed) if field.name not in {"raw", "evidence"} + } + assert public_fields == set(_GENERIC_DEFAULTS) + assert _generic_snapshot(parsed) == expected + + +@pytest.mark.parametrize("case", _case_params()) +def test_legacy_parser_matches_golden_corpus(case: dict[str, Any]) -> None: + parsed = raw_parser(case["raw"]) + expected = case["legacy"] + + if expected is None: + assert parsed is None + return + + assert parsed is not None + assert asdict(parsed) == expected + + +@pytest.mark.parametrize("case", _case_params()) +def test_traced_parser_matches_plain_result_and_has_valid_references( + case: dict[str, Any], +) -> None: + outcome = parse_release_title_with_trace(case["raw"]) + + assert outcome.result == parse_release_title(case["raw"]) + candidate_ids = {candidate.id for candidate in outcome.trace.candidates} + observation_ids = {observation.id for observation in outcome.trace.observations} + assert {decision.candidate_id for decision in outcome.trace.decisions} == ( + candidate_ids + ) + assert { + observation_id + for candidate in outcome.trace.candidates + for observation_id in candidate.observation_ids + } == observation_ids + for span in ( + span for candidate in outcome.trace.candidates for span in candidate.spans + ): + assert 0 <= span.segment < len(outcome.trace.segments) + assert ( + 0 <= span.start < span.end <= len(outcome.trace.segments[span.segment].text) + ) diff --git a/backend/src/test/test_release_parser_properties.py b/backend/src/test/test_release_parser_properties.py new file mode 100644 index 00000000..50c78eef --- /dev/null +++ b/backend/src/test/test_release_parser_properties.py @@ -0,0 +1,331 @@ +"""Metamorphic and invariant checks for the public release parser. + +The suite intentionally avoids depending on tokenizer implementation details. +It uses a fixed pseudo-random seed and keeps the combined generated corpus below +500 cases so failures stay reproducible and the test remains cheap to run. +""" + +from __future__ import annotations + +import math +import random +import re +import string +from collections.abc import Iterable + +from module.parser.analyser.tokenizer import ( + MediaType, + ParsedRelease, + ReleaseKind, + parse_release_title, +) + +SEED = 0xAB330 +STRUCTURED_EPISODE_CASES = 4 * 3 * 3 * 3 +STRUCTURED_MEDIA_CASES = 15 * 4 +METAMORPHIC_CASES = 60 +FUZZ_CASES = 220 +GENERATED_CASES = ( + STRUCTURED_EPISODE_CASES + STRUCTURED_MEDIA_CASES + METAMORPHIC_CASES + FUZZ_CASES +) + + +def _normalized_titles(parsed: ParsedRelease) -> tuple[str, ...]: + """Compare title content without prescribing language-slot heuristics.""" + return tuple( + sorted( + re.sub(r"\s+", " ", title).strip().casefold() + for title in (parsed.title_en, parsed.title_zh, parsed.title_jp) + if title + ) + ) + + +def _core_identity(parsed: ParsedRelease) -> tuple[object, ...]: + """Fields that technical adornments must not change.""" + return ( + _normalized_titles(parsed), + parsed.group, + parsed.season, + parsed.episode, + parsed.episode_end, + parsed.episode_title, + parsed.media_type, + parsed.release_kind, + parsed.version, + parsed.year, + ) + + +def _title_contains(parsed: ParsedRelease, expected: str) -> bool: + expected = re.sub(r"\s+", " ", expected).strip().casefold() + return any(expected in title for title in _normalized_titles(parsed)) + + +def _assert_public_invariants(raw: str, parsed: ParsedRelease | None) -> None: + """Check only guarantees implied by the public result contract.""" + assert parsed == parse_release_title(raw), f"non-deterministic parse for {raw!r}" + if parsed is None: + return + + assert parsed.raw == raw + assert isinstance(parsed.media_type, MediaType) + assert isinstance(parsed.release_kind, ReleaseKind) + assert parsed.primary_title is not None + assert parsed.primary_title.strip() + assert _normalized_titles(parsed) + + for title in (parsed.title_en, parsed.title_zh, parsed.title_jp): + if title is not None: + assert title == title.strip() + + for number in (parsed.episode, parsed.episode_end): + if number is not None: + assert isinstance(number, (int, float)) and not isinstance(number, bool) + assert math.isfinite(float(number)) + assert number >= 0 + + if parsed.episode_end is not None: + assert parsed.episode is not None + if parsed.season is not None: + assert isinstance(parsed.season, int) and not isinstance(parsed.season, bool) + assert parsed.season >= 0 + if parsed.version is not None: + assert isinstance(parsed.version, int) and parsed.version >= 0 + + assert isinstance(parsed.codecs, tuple) + assert isinstance(parsed.audio, tuple) + assert isinstance(parsed.tags, tuple) + assert isinstance(parsed.evidence, tuple) + + +def test_generated_corpus_stays_within_budget() -> None: + assert GENERATED_CASES == 448 + assert GENERATED_CASES <= 500 + + +def test_technical_adornments_preserve_numbered_episode_identity() -> None: + titles = ("Sousou no Frieren", "葬送的芙莉莲", "Fate stay night", "86") + groups = ("LoliHouse", "字幕组", "Movie-Fan") + episodes = ((1, "01"), (7, "07"), (12.5, "12.5")) + adornments = ( + " [1080p WEB-DL HEVC AAC MKV]", + " [720p BDRip AVC FLAC].mkv", + " 【2160p BluRay x265 Dual Audio】", + ) + + seen = 0 + for title in titles: + for group in groups: + for expected_episode, episode_text in episodes: + baseline_raw = f"[{group}] {title} - {episode_text}" + baseline = parse_release_title(baseline_raw) + assert baseline is not None, baseline_raw + _assert_public_invariants(baseline_raw, baseline) + + for adornment in adornments: + raw = baseline_raw + adornment + parsed = parse_release_title(raw) + assert parsed is not None, raw + _assert_public_invariants(raw, parsed) + assert _core_identity(parsed) == _core_identity(baseline), raw + assert parsed.episode == expected_episode, raw + assert parsed.media_type is MediaType.EPISODE, raw + assert parsed.release_kind is ReleaseKind.SINGLE, raw + seen += 1 + + assert seen == STRUCTURED_EPISODE_CASES + + +def test_structured_media_forms_survive_safe_wrapping_variants() -> None: + forms: tuple[ + tuple[str, str, MediaType, ReleaseKind, int | float | None, int | float | None], + ..., + ] = ( + ( + "Meta Anime - 01", + "Meta Anime", + MediaType.EPISODE, + ReleaseKind.SINGLE, + 1, + None, + ), + ( + "Meta Anime - 12.5", + "Meta Anime", + MediaType.EPISODE, + ReleaseKind.SINGLE, + 12.5, + None, + ), + ( + "Meta Anime - 12v2", + "Meta Anime", + MediaType.EPISODE, + ReleaseKind.SINGLE, + 12, + None, + ), + ( + "Meta Anime S02E05", + "Meta Anime", + MediaType.EPISODE, + ReleaseKind.SINGLE, + 5, + None, + ), + ( + "Meta Anime [01-12]", + "Meta Anime", + MediaType.EPISODE, + ReleaseKind.RANGE, + 1, + 12, + ), + ( + "Meta Anime Complete Batch", + "Meta Anime", + MediaType.EPISODE, + ReleaseKind.BATCH, + None, + None, + ), + ("迷宫饭 全24话", "迷宫饭", MediaType.EPISODE, ReleaseKind.COLLECTION, 1, 24), + ("Meta Anime OVA01", "Meta Anime", MediaType.OVA, ReleaseKind.SINGLE, 1, None), + ("Meta Anime OAD02", "Meta Anime", MediaType.OAD, ReleaseKind.SINGLE, 2, None), + ( + "Meta Anime SP03", + "Meta Anime", + MediaType.SPECIAL, + ReleaseKind.SINGLE, + 3, + None, + ), + ( + "劇場版 Meta Anime", + "Meta Anime", + MediaType.MOVIE, + ReleaseKind.SINGLE, + None, + None, + ), + ("Meta Anime PV2", "Meta Anime", MediaType.PV, ReleaseKind.SINGLE, 2, None), + ( + "Meta Anime NCOP", + "Meta Anime", + MediaType.OPENING, + ReleaseKind.SINGLE, + None, + None, + ), + ( + "Meta Anime NCED", + "Meta Anime", + MediaType.ENDING, + ReleaseKind.SINGLE, + None, + None, + ), + ("Meta Anime", "Meta Anime", MediaType.UNKNOWN, ReleaseKind.SINGLE, None, None), + ) + + wrappers = ( + lambda value: value, + lambda value: f"[MetaGroup] {value} [1080p WEB-DL]", + lambda value: f"【MetaGroup】 {value} 【720p BDRip AVC AAC】.mkv", + lambda value: f"\n [MetaGroup] {value} [HEVC][1080P] \n", + ) + + seen = 0 + for core, title, media_type, release_kind, episode, episode_end in forms: + for wrap in wrappers: + raw = wrap(core) + parsed = parse_release_title(raw) + assert parsed is not None, raw + _assert_public_invariants(raw, parsed) + assert _title_contains(parsed, title), raw + assert parsed.media_type is media_type, raw + assert parsed.release_kind is release_kind, raw + assert parsed.episode == episode, raw + assert parsed.episode_end == episode_end, raw + seen += 1 + + assert seen == STRUCTURED_MEDIA_CASES + + +def test_whitespace_bracket_and_metadata_variants_preserve_core_identity() -> None: + rng = random.Random(SEED) + titles = ("Mob Psycho 100", "Dungeon Meshi", "葬送的芙莉莲", "Fate stay night") + groups = ("Group-A", "Lilith-Raws", "喵萌奶茶屋") + + for _ in range(METAMORPHIC_CASES): + title = rng.choice(titles) + group = rng.choice(groups) + season = rng.randint(1, 12) + episode = rng.randint(1, 99) + version = rng.randint(1, 3) + marker = f"S{season:02d}E{episode:02d}v{version}" + baseline_raw = f"[{group}] {title} {marker}" + variant_raw = ( + f"\n 【{group}】 {title} {marker.lower()} " + "【WEB-DL 1080p HEVC AAC MKV】.mkv \n" + ) + + baseline = parse_release_title(baseline_raw) + variant = parse_release_title(variant_raw) + assert baseline is not None, baseline_raw + assert variant is not None, variant_raw + _assert_public_invariants(baseline_raw, baseline) + _assert_public_invariants(variant_raw, variant) + assert _core_identity(variant) == _core_identity(baseline), variant_raw + + +def _random_inputs(rng: random.Random, count: int) -> Iterable[str]: + atoms = ( + "Anime", + "Title", + "葬送的芙莉莲", + "アニメ", + "OVA", + "OAD02", + "SP", + "PV2", + "NCOP", + "NCED", + "S02E05", + "Season 3", + "01-12", + "12.5", + "Complete Batch", + "劇場版", + "WEB-DL", + "BDRip", + "1080p", + "HEVC", + "AAC", + "MKV", + "[", + "]", + "(", + ")", + "【", + "】", + ) + separators = ("", " ", " ", " - ", "_", "/", ".", "\n") + alphabet = string.ascii_letters + string.digits + " []()_-./~中日語話季\n" + + for index in range(count): + if index % 2: + parts = [rng.choice(atoms) for _ in range(rng.randint(0, 12))] + yield rng.choice(separators).join(parts) + else: + yield "".join(rng.choice(alphabet) for _ in range(rng.randint(0, 120))) + + +def test_seeded_fuzz_inputs_are_total_and_deterministic() -> None: + rng = random.Random(SEED) + corpus = list(_random_inputs(rng, FUZZ_CASES)) + assert len(corpus) == FUZZ_CASES + + for raw in corpus: + _assert_public_invariants(raw, parse_release_title(raw)) diff --git a/backend/src/test/test_tokenizer_resolver.py b/backend/src/test/test_tokenizer_resolver.py new file mode 100644 index 00000000..a323d599 --- /dev/null +++ b/backend/src/test/test_tokenizer_resolver.py @@ -0,0 +1,331 @@ +from __future__ import annotations + +from itertools import permutations + +import pytest + +from module.parser.analyser.tokenizer.candidate import ( + Candidate, + ClaimField, + Claims, + DecisionStatus, + OverlapPolicy, + ShadowedSpanPolicy, + Span, +) +from module.parser.analyser.tokenizer.resolver import resolve_candidates +from module.parser.analyser.tokenizer.result import MediaType + + +def _candidate( + candidate_id: str, + *, + start: int, + end: int, + claims: Claims, + priority: int, + **kwargs: object, +) -> Candidate: + return Candidate( + id=candidate_id, + rule_id=candidate_id, + spans=(Span(0, start, end),), + claims=claims, + priority=priority, + **kwargs, # type: ignore[arg-type] + ) + + +def test_span_is_half_open_and_validated() -> None: + assert Span(0, 1, 4).length == 3 + assert Span(0, 1, 4).overlaps(Span(0, 3, 5)) + assert not Span(0, 1, 4).overlaps(Span(0, 4, 5)) + assert not Span(0, 1, 4).overlaps(Span(1, 1, 4)) + assert Span(0, 1, 4).contains(Span(0, 2, 3)) + + with pytest.raises(ValueError): + Span(0, 3, 3) + + +def test_resolution_is_independent_of_input_order() -> None: + explicit = _candidate( + "explicit", + start=10, + end=16, + claims=Claims(season=2, episode=5), + priority=100, + evidence=("season", "episode"), + ) + trailing = _candidate( + "trailing", + start=20, + end=22, + claims=Claims(episode=99), + priority=70, + evidence=("episode",), + ) + + resolutions = [ + resolve_candidates(order) for order in permutations((trailing, explicit)) + ] + + assert resolutions[0] == resolutions[1] + result = resolutions[0] + assert result.claims.season == 2 + assert result.claims.episode == 5 + assert result.decision_for("explicit").status is DecisionStatus.SELECTED + shadowed = result.decision_for("trailing") + assert shadowed.status is DecisionStatus.SHADOWED + assert shadowed.exclude_from_title + assert result.excluded_spans == (Span(0, 10, 16), Span(0, 20, 22)) + + +def test_movie_rejects_ambiguous_number_as_title_without_masking_it() -> None: + movie = _candidate( + "movie", + start=0, + end=4, + claims=Claims(media_type=MediaType.MOVIE), + priority=100, + blocks=frozenset({"ambiguous-episode"}), + evidence=("movie",), + ) + title_number = _candidate( + "trailing-number", + start=12, + end=13, + claims=Claims(episode=0), + priority=70, + conflict_tags=frozenset({"ambiguous-episode"}), + preserve_as_title_on_conflict=True, + ) + + result = resolve_candidates((title_number, movie)) + + assert result.claims.media_type is MediaType.MOVIE + assert result.claims.episode is None + decision = result.decision_for("trailing-number") + assert decision.status is DecisionStatus.REJECTED_AS_TITLE + assert not decision.exclude_from_title + assert result.excluded_spans == (Span(0, 0, 4),) + + +def test_overlapping_shadowed_field_reports_conflict_without_extra_mask() -> None: + combined = _candidate( + "combined", + start=2, + end=10, + claims=Claims(subtitle="GB", container="MKV"), + priority=80, + specificity=2, + ) + container = _candidate( + "container", + start=5, + end=10, + claims=Claims(container="MKV"), + priority=70, + ) + + result = resolve_candidates((container, combined)) + + assert result.claims.subtitle == "GB" + assert result.claims.container == "MKV" + rejected = result.decision_for("container") + assert rejected.status is DecisionStatus.REJECTED_CONFLICT + assert result.excluded_spans == (Span(0, 2, 10),) + + +def test_partially_overlapping_same_field_does_not_mask_loser_only_text() -> None: + winner = _candidate( + "winner", start=2, end=10, claims=Claims(episode=1), priority=80 + ) + partial_loser = _candidate( + "partial-loser", start=8, end=16, claims=Claims(episode=2), priority=70 + ) + + result = resolve_candidates((partial_loser, winner)) + + assert result.claims.episode == 1 + assert ( + result.decision_for("partial-loser").status is DecisionStatus.REJECTED_CONFLICT + ) + assert result.excluded_spans == (Span(0, 2, 10),) + + +def test_different_fields_on_overlapping_exclusive_spans_are_rejected() -> None: + source = _candidate( + "source", + start=0, + end=6, + claims=Claims(source="WEB-DL"), + priority=80, + ) + codec = _candidate( + "codec", + start=4, + end=8, + claims=Claims(codecs=("HEVC",)), + priority=70, + ) + + result = resolve_candidates((codec, source)) + + assert result.claims.source == "WEB-DL" + assert result.claims.codecs == () + assert result.decision_for("codec").status is DecisionStatus.REJECTED_CONFLICT + + +def test_shared_overlay_allows_explicit_transparent_overlap() -> None: + overlay = _candidate( + "overlay", + start=0, + end=12, + claims=Claims(tags=("dub",)), + priority=80, + overlap_policy=OverlapPolicy.SHARED, + ) + resolution = _candidate( + "resolution", + start=2, + end=7, + claims=Claims(resolution="1080p"), + priority=70, + ) + + result = resolve_candidates((resolution, overlay)) + + assert result.claims.tags == ("dub",) + assert result.claims.resolution == "1080p" + assert result.decision_for("overlay").status is DecisionStatus.SELECTED + assert result.decision_for("resolution").status is DecisionStatus.SELECTED + + +def test_shadowed_candidate_can_keep_its_span_in_title() -> None: + strong = _candidate( + "strong", + start=0, + end=4, + claims=Claims(episode=1), + priority=100, + ) + ambiguous = _candidate( + "ambiguous", + start=8, + end=9, + claims=Claims(episode=2), + priority=70, + shadowed_span_policy=ShadowedSpanPolicy.KEEP, + ) + + result = resolve_candidates((ambiguous, strong)) + + assert result.decision_for("ambiguous").status is DecisionStatus.SHADOWED + assert result.excluded_spans == (Span(0, 0, 4),) + + +def test_shadowed_candidate_excludes_marker_but_keeps_dependent_span() -> None: + winner = _candidate( + "winner", start=0, end=4, claims=Claims(episode=1), priority=100 + ) + marker = Span(0, 8, 12) + dependent_title = Span(0, 12, 24) + loser = Candidate( + id="loser", + rule_id="loser", + spans=(marker, dependent_title), + shadowed_spans=(marker,), + claims=Claims(episode=2, episode_title="Dependent title"), + priority=70, + ) + + result = resolve_candidates((loser, winner)) + + decision = result.decision_for("loser") + assert decision.status is DecisionStatus.SHADOWED + assert decision.excluded_spans == (marker,) + assert result.excluded_spans == (Span(0, 0, 4), marker) + + +def test_repeatable_claims_are_case_insensitively_deduplicated() -> None: + first = _candidate( + "codec-a", + start=0, + end=4, + claims=Claims(codecs=("HEVC",), tags=("tag-a",)), + priority=10, + ) + second = _candidate( + "codec-b", + start=5, + end=9, + claims=Claims(codecs=("hevc", "AVC"), tags=("TAG-A",)), + priority=10, + ) + + result = resolve_candidates((second, first)) + + assert result.claims.codecs == ("HEVC", "AVC") + assert result.claims.tags == ("tag-a",) + assert ClaimField.CODECS in result.decision_for("codec-a").won_fields + + +def test_duplicate_candidate_ids_are_rejected() -> None: + first = _candidate( + "duplicate", start=0, end=1, claims=Claims(episode=1), priority=1 + ) + second = _candidate( + "duplicate", start=2, end=3, claims=Claims(episode=2), priority=2 + ) + + with pytest.raises(ValueError, match="duplicate candidate id"): + resolve_candidates((first, second)) + + +def test_equal_priority_conflicting_claims_emit_stable_warning() -> None: + first = _candidate( + "episode-a", start=0, end=2, claims=Claims(episode=1), priority=70 + ) + second = _candidate( + "episode-b", start=4, end=6, claims=Claims(episode=2), priority=70 + ) + + forward = resolve_candidates((first, second)) + reverse = resolve_candidates((second, first)) + + assert forward == reverse + assert len(forward.warnings) == 1 + assert forward.warnings[0].field is ClaimField.EPISODE + assert forward.warnings[0].candidate_ids == ("episode-a", "episode-b") + + +def test_specificity_resolves_same_priority_without_ambiguity_warning() -> None: + explicit = _candidate( + "explicit", + start=0, + end=2, + claims=Claims(episode=1), + priority=70, + specificity=2, + ) + generic = _candidate( + "generic", start=4, end=6, claims=Claims(episode=2), priority=70 + ) + + result = resolve_candidates((generic, explicit)) + + assert result.claims.episode == 1 + assert result.warnings == () + + +def test_fully_shadowed_low_priority_tie_does_not_warn() -> None: + winner = _candidate( + "winner", start=0, end=2, claims=Claims(episode=1), priority=100 + ) + low_a = _candidate("low-a", start=4, end=6, claims=Claims(episode=2), priority=70) + low_b = _candidate("low-b", start=8, end=10, claims=Claims(episode=3), priority=70) + + result = resolve_candidates((low_b, winner, low_a)) + + assert result.claims.episode == 1 + assert result.warnings == () diff --git a/backend/src/test/test_tokenizer_trace.py b/backend/src/test/test_tokenizer_trace.py new file mode 100644 index 00000000..a7d53dcd --- /dev/null +++ b/backend/src/test/test_tokenizer_trace.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +from dataclasses import FrozenInstanceError + +import pytest + +from module.parser.analyser.tokenizer.candidate import ( + Candidate, + Claims, + DecisionStatus, + Observation, + Span, +) +from module.parser.analyser.tokenizer.parser import ( + parse_release_title, + parse_release_title_with_trace, +) +from module.parser.analyser.tokenizer.resolver import resolve_candidates +from module.parser.analyser.tokenizer.result import MediaType +from module.parser.analyser.tokenizer.trace import ParseTrace, TraceSegment + + +def test_trace_canonicalizes_diagnostic_input_order() -> None: + early_observation = Observation( + id="episode-observation", + rule_id="episode", + kind="episode", + span=Span(0, 8, 10), + text="01", + captures=("01",), + ) + late_observation = Observation( + id="resolution-observation", + rule_id="resolution", + kind="resolution", + span=Span(1, 0, 5), + text="1080p", + ) + episode = Candidate( + id="episode", + rule_id="episode", + spans=(early_observation.span,), + claims=Claims(episode=1), + priority=90, + observation_ids=(early_observation.id,), + ) + resolution_candidate = Candidate( + id="resolution", + rule_id="resolution", + spans=(late_observation.span,), + claims=Claims(resolution="1080p"), + priority=10, + observation_ids=(late_observation.id,), + ) + candidates = (resolution_candidate, episode) + resolution = resolve_candidates(candidates) + + trace = ParseTrace.from_resolution( + raw="Anime - 01 [1080p]", + normalized="Anime - 01 [1080p]", + resolution=resolution, + segments=( + TraceSegment(1, "1080p", 11, 18, 12, 17, "square"), + TraceSegment(0, "Anime - 01 ", 0, 11, 0, 11), + ), + observations=(late_observation, early_observation), + candidates=candidates, + residuals=("Anime", ""), + ) + + assert tuple(segment.index for segment in trace.segments) == (0, 1) + assert tuple(item.id for item in trace.observations) == ( + "episode-observation", + "resolution-observation", + ) + assert tuple(item.id for item in trace.candidates) == ("episode", "resolution") + assert trace.decision_for("episode").status is DecisionStatus.SELECTED + + attributable_spans = { + span for decision in trace.decisions for span in decision.excluded_spans + } + for excluded in trace.excluded_spans: + cursor = excluded.start + for span in sorted(attributable_spans): + if span.segment != excluded.segment or span.end <= cursor: + continue + assert span.start <= cursor + cursor = max(cursor, span.end) + if cursor >= excluded.end: + break + assert cursor >= excluded.end + + +def test_trace_exposes_title_rejections_and_is_frozen() -> None: + winner = Candidate( + id="movie", + rule_id="movie", + spans=(Span(0, 0, 4),), + claims=Claims(), + priority=100, + blocks=frozenset({"ambiguous"}), + ) + rejected = Candidate( + id="title-number", + rule_id="bare-number", + spans=(Span(0, 8, 9),), + claims=Claims(episode=0), + priority=10, + conflict_tags=frozenset({"ambiguous"}), + preserve_as_title_on_conflict=True, + ) + resolution = resolve_candidates((rejected, winner)) + trace = ParseTrace.from_resolution( + raw="Movie X 0", + normalized="Movie X 0", + resolution=resolution, + candidates=(rejected, winner), + ) + + assert trace.rejected_as_title == ("title-number",) + with pytest.raises(FrozenInstanceError): + trace.raw = "changed" # type: ignore[misc] + + +def test_trace_missing_candidate_decision_raises_key_error() -> None: + trace = ParseTrace(raw="Anime", normalized="Anime") + + with pytest.raises(KeyError): + trace.decision_for("missing") + + +def test_real_parser_trace_reproduces_public_result_and_accounts_for_masks() -> None: + raw = "[Group] Anime - 01 [1080p WEB-DL HEVC AAC MKV]" + + outcome = parse_release_title_with_trace(raw) + + assert outcome.result == parse_release_title(raw) + assert outcome.result is not None + assert outcome.result.episode == 1 + assert outcome.trace.claims.episode == 1 + assert outcome.result.codecs == ("HEVC",) + assert outcome.result.audio == ("AAC",) + assert len(outcome.trace.candidates) == len(outcome.trace.decisions) + assert {candidate.id for candidate in outcome.trace.candidates} == { + decision.candidate_id for decision in outcome.trace.decisions + } + + excluded_source_spans = tuple( + span for decision in outcome.trace.decisions for span in decision.excluded_spans + ) + for excluded in outcome.trace.excluded_spans: + for position in range(excluded.start, excluded.end): + assert any( + source.segment == excluded.segment + and source.start <= position < source.end + for source in excluded_source_spans + ) + + +def test_real_parser_trace_preserves_movie_title_number_with_reason() -> None: + outcome = parse_release_title_with_trace("[Group] 劇場版 86 - 0 [1080p WEB-DL]") + + assert outcome.result is not None + assert outcome.result.media_type is MediaType.MOVIE + assert outcome.result.episode is None + assert outcome.result.title_en == "86 - 0" + assert any( + candidate_id.startswith("episode.trailing:") + for candidate_id in outcome.trace.rejected_as_title + ) + + +def test_empty_input_still_returns_an_explainable_outcome() -> None: + outcome = parse_release_title_with_trace(" ") + + assert outcome.result is None + assert outcome.trace.raw == " " + assert outcome.trace.normalized == "" + assert outcome.trace.candidates == () + assert outcome.trace.decisions == () + + +@pytest.mark.parametrize( + "raw", + ( + "Anime S01E01 Season 1 Episode 1 [1080p]", + "Anime - 01 [1080p][1080P]", + "Anime - 01 [WEB-DL][web_dl]", + ), +) +def test_equivalent_spellings_do_not_emit_ambiguity_warnings(raw: str) -> None: + outcome = parse_release_title_with_trace(raw) + + assert outcome.result is not None + assert outcome.trace.warnings == () + + +@pytest.mark.parametrize( + "raw", + ( + "[Group] Anime (Alt [Nested]) - 01 [1080p]", + "[Group] Anime - 01 [1080p", + ), +) +def test_nested_or_unclosed_segment_coordinates_remain_valid(raw: str) -> None: + outcome = parse_release_title_with_trace(raw) + + assert outcome.result is not None + for segment in outcome.trace.segments: + assert ( + outcome.trace.normalized[segment.content_start : segment.content_end] + == segment.text + ) + assert segment.outer_start <= segment.content_start + assert segment.content_end <= segment.outer_end