fix(parser): handle ranges and mixed collections safely

This commit is contained in:
Estrella Pan
2026-07-12 14:39:35 +02:00
parent 8ceec47174
commit e744b3601a
13 changed files with 708 additions and 14 deletions

View File

@@ -185,6 +185,8 @@ def match_bangumi_in_list(
def _release_matches_bangumi(release: "ParsedRelease | None", bangumi: Bangumi) -> bool:
if release is None:
return True
if release.is_mixed_collection:
return False
from module.parser.analyser.tokenizer import MediaType

View File

@@ -23,6 +23,8 @@ def to_legacy_episode(parsed: ParsedRelease) -> Episode | None:
"""
if not parsed.primary_title:
return None
if parsed.is_mixed_collection:
return None
if parsed.release_kind in {ReleaseKind.RANGE, ReleaseKind.BATCH}:
return None
if parsed.media_type in {

View File

@@ -14,6 +14,7 @@ from dataclasses import dataclass, field
from .candidate import (
Candidate,
Claims,
DecisionStatus,
Observation,
OverlapPolicy,
ShadowedSpanPolicy,
@@ -32,6 +33,9 @@ _LATIN = re.compile(r"[A-Za-z]")
_NUMERIC_TITLE = re.compile(r"\d{1,3}(?:\s*[/.-]\s*\d{1,3})?")
_DATE = re.compile(r"(?<!\d)(?:19|20)\d{2}[.\-/]\d{1,2}[.\-/]\d{1,2}(?!\d)")
_ORGANIZED_DATE_CONTEXT = re.compile(
r"^\s*整理时间\s*[::]\s*((?:19|20)\d{2}[.\-/]\d{1,2}[.\-/]\d{1,2})\s*$"
)
_YEAR = re.compile(r"(?<!\d)((?:18|19|20|21)\d{2})(?!\d)")
_FILE_SIZE = re.compile(r"(?<!\w)\d+(?:\.\d+)?\s*[KMGT]i?B(?!\w)", re.I)
_HASH = re.compile(r"(?<![0-9A-F])[0-9A-F]{8}(?![0-9A-F])", re.I)
@@ -109,6 +113,28 @@ _BATCH = re.compile(
)
_COLLECTION = re.compile(r"合集|全集|全[话話集]|总集篇|總集篇|特典")
_MIXED_CONTENT_NAME = (
r"TV(?:\s*(?:动画|動畫|Anime))?|OVA|OAD|Special|SP|Movie|"
r"劇場版?|剧场版?|Manga|漫画|漫畫|Novel|小说|小說|"
r"Music|音乐|音樂|Other|其他|Extras?"
)
_MIXED_CONTENT_NUMBERS = (
r"(?:\s*(?:E(?:P)?\.?\s*)?\d{1,4}"
r"(?:\s*(?:-|~|~|—)\s*(?:E(?:P)?\.?\s*)?\d{1,4})?)?"
)
_MIXED_CONTENT_ITEM = rf"(?:{_MIXED_CONTENT_NAME}){_MIXED_CONTENT_NUMBERS}"
_MIXED_CONTENT_MANIFEST = re.compile(
rf"(?<![A-Za-z0-9])(?P<items>{_MIXED_CONTENT_ITEM}"
rf"(?:\s*[++]\s*{_MIXED_CONTENT_ITEM})+)\s*(?=$|[;;_])",
re.I,
)
_MIXED_CONTENT_NAME_AT_START = re.compile(rf"^(?P<name>{_MIXED_CONTENT_NAME})", re.I)
_MIXED_CONTENT_TAIL_TOKEN = re.compile(
r"\b(?:dub|jpn?|chs|cht|eng|subs?|audio|tracks?|external)\b|"
r"(?:日[语語]|英[语語]|中[文语語]|简中|簡中|繁中|外挂|外掛|音[轨軌]|字幕)",
re.I,
)
_RESOLUTION = re.compile(
r"(?<![A-Za-z0-9])(?:\d{3,4}[xX]\d{3,4}|(?:720|1080|2160)[pP](?:60)?|4[Kk])(?![A-Za-z0-9])"
)
@@ -326,8 +352,15 @@ def _parse_release_title_with_candidates(
)
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)) and _is_mixed_resolution(resolution):
retry_spans = _mixed_title_retry_spans(collector.candidates, resolution)
if retry_spans != resolution.excluded_spans:
retry_working = _working_from_spans(segments, retry_spans)
retry_titles = _reconstruct_titles(retry_working)
if any(retry_titles):
title_en, title_zh, title_jp = retry_titles
result = None
if any((title_en, title_zh, title_jp)):
if any((title_en, title_zh, title_jp)) or _is_mixed_resolution(resolution):
result = _result_from_resolution(raw, title_en, title_zh, title_jp, resolution)
trace = None
@@ -344,6 +377,31 @@ def _parse_release_title_with_candidates(
return result, trace
def _is_mixed_resolution(resolution: Resolution) -> bool:
return (
resolution.claims.media_type is MediaType.UNKNOWN
and resolution.claims.release_kind is ReleaseKind.COLLECTION
)
def _mixed_title_retry_spans(
candidates: list[Candidate], resolution: Resolution
) -> tuple[Span, ...]:
"""Keep an ambiguous numeric episode only when it is the sole mixed title."""
by_id = {candidate.id: candidate for candidate in candidates}
spans: list[Span] = []
for decision in resolution.decisions:
candidate = by_id[decision.candidate_id]
ambiguous_episode = (
decision.status is DecisionStatus.SELECTED
and candidate.preserve_as_title_on_conflict
and "ambiguous-episode" in candidate.conflict_tags
)
if not ambiguous_episode:
spans.extend(decision.excluded_spans)
return tuple(spans)
def _result_from_resolution(
raw: str,
title_en: str | None,
@@ -357,19 +415,34 @@ def _result_from_resolution(
claims.episode is not None or claims.release_kind is not None
):
media_type = MediaType.EPISODE
resolved_media_type = media_type or MediaType.UNKNOWN
release_kind = claims.release_kind or ReleaseKind.SINGLE
mixed_collection = (
resolved_media_type is MediaType.UNKNOWN
and release_kind is ReleaseKind.COLLECTION
)
evidence = resolution.evidence
if mixed_collection:
# The trace keeps pre-normalization resolver claims for diagnostics, but
# the public collection must not advertise a single-episode identity.
evidence = tuple(
item
for item in evidence
if item not in {"season", "episode", "episode-title", "episode-range"}
)
return ParsedRelease(
raw=raw,
title_en=title_en,
title_zh=title_zh,
title_jp=title_jp,
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,
season=None if mixed_collection else claims.season,
season_raw=None if mixed_collection else claims.season_raw,
episode=None if mixed_collection else claims.episode,
episode_end=None if mixed_collection else claims.episode_end,
episode_title=None if mixed_collection else claims.episode_title,
media_type=resolved_media_type,
release_kind=release_kind,
resolution=claims.resolution,
source=claims.source,
subtitle=claims.subtitle,
@@ -379,7 +452,7 @@ def _result_from_resolution(
version=claims.version,
year=claims.year,
tags=claims.tags,
evidence=resolution.evidence,
evidence=evidence,
)
@@ -542,6 +615,15 @@ def _looks_like_metadata(text: str) -> bool:
)
if any(pattern.fullmatch(value) for pattern in structural_patterns):
return True
if "+" in value or "+" in value:
manifest = _MIXED_CONTENT_MANIFEST.match(value)
if manifest is not None:
categories = _mixed_content_categories(manifest.group("items"))
suffix = value[manifest.end() :]
if len(categories) >= 2 and (
not suffix or _is_mixed_content_metadata_tail(suffix)
):
return True
if _CJK_SUBTITLE_TAG.fullmatch(value) or value.lower() in {"end", "生"}:
return True
@@ -577,6 +659,13 @@ def _is_year_number(value: str) -> bool:
return value.isdigit() and len(value) == 4 and 1800 <= int(value) <= 2199
def _trailing_release_episode(text: str) -> re.Match[str] | None:
match = _TRAILING_EPISODE.search(text)
if match is None or _is_year_number(match.group(1)):
return None
return match
def _collect_group_candidates(collector: _CandidateCollector) -> set[int]:
group_indices = _find_group_segments(collector.segments)
if not group_indices:
@@ -616,6 +705,18 @@ def _collect_structural_candidates(collector: _CandidateCollector) -> None:
for index, segment in enumerate(collector.segments):
text = segment.text
organized_date = _ORGANIZED_DATE_CONTEXT.fullmatch(text)
if organized_date is not None and segment.enclosure in {"square", "round"}:
collector.add_span(
rule_id="metadata.organized-date",
kind="tag",
segment_index=index,
start=0,
end=len(text),
claims=Claims(tags=(organized_date.group(1),)),
priority=111,
)
for match in _DATE.finditer(text):
collector.add_match(
rule_id="metadata.date",
@@ -708,12 +809,39 @@ def _collect_structural_candidates(collector: _CandidateCollector) -> None:
)
isolated = not _contains_title(prefix + suffix)
labelled = bool(re.search(r"(?:Episodes?|Eps?)\s*$", prefix, re.I))
explicit_endpoints = sum(
1 for _ in _EXPLICIT_EPISODE.finditer(match.group())
)
strong_structure_patterns = (
_SEASON_EPISODE_RANGE,
_SEASON_EPISODE_WORDS_RANGE,
_SEASON_EPISODE,
_SEASON_EPISODE_WORDS,
_SPECIAL_RANGE,
_SPECIAL_NUMBER,
_EXPLICIT_EPISODE,
_CHINESE_EPISODE,
)
has_prior_structure = any(
pattern.search(prefix) for pattern in strong_structure_patterns
) or any(
pattern.search(prior.text)
for prior in collector.segments[:index]
for pattern in strong_structure_patterns
)
strong_labelled_range = (
explicit_endpoints == 2
and not has_prior_structure
and _trailing_release_episode(suffix) is None
and _is_metadata_tail(suffix)
)
if (
start >= 1800
or end >= 1800
or start > end
or re.search(r"\d+\s*/\s*$", prefix)
or not (compact or isolated or labelled)
or _trailing_release_episode(suffix) is not None
or not (compact or isolated or labelled or strong_labelled_range)
):
continue
collector.add_match(
@@ -847,6 +975,8 @@ def _collect_structural_candidates(collector: _CandidateCollector) -> None:
)
for match in _EXPLICIT_EPISODE.finditer(text):
if _inside_range_before_trailing_episode(text, match):
continue
collector.add_match(
rule_id="episode.explicit",
kind="episode",
@@ -927,6 +1057,62 @@ def _collect_media_candidates(collector: _CandidateCollector) -> None:
for index, segment in enumerate(collector.segments):
text = segment.text
mixed_matches = (
_MIXED_CONTENT_MANIFEST.finditer(text)
if "+" in text or "+" in text
else ()
)
for match in mixed_matches:
prefix = text[: match.start()]
stripped_prefix = prefix.rstrip()
free_segment = segment.enclosure is None
after_colon = free_segment and stripped_prefix.endswith((":", ":"))
after_underscore = free_segment and stripped_prefix.endswith("_")
independent_square = segment.enclosure == "square" and not stripped_prefix
if not (after_colon or after_underscore or independent_square):
continue
categories = _mixed_content_categories(match.group("items"))
if len(categories) < 2:
continue
start = len(stripped_prefix) - 1 if after_colon else match.start()
extra_spans: tuple[Span, ...] = ()
if independent_square and index > 0:
previous = collector.segments[index - 1]
trailing_colon = re.search(r"[::]\s*$", previous.text)
if previous.enclosure is None and trailing_colon is not None:
extra_spans = (
Span(index - 1, trailing_colon.start(), trailing_colon.end()),
)
collector.add_span(
rule_id="cardinality.mixed-content",
kind="mixed-collection",
segment_index=index,
start=start,
end=match.end(),
claims=Claims(
media_type=MediaType.UNKNOWN,
release_kind=ReleaseKind.COLLECTION,
),
priority=110,
specificity=len(categories),
evidence=("collection", "mixed-content"),
extra_spans=extra_spans,
captures=(match.group("items"),),
)
suffix = text[match.end() :]
if suffix and _is_mixed_content_metadata_tail(suffix):
collector.add_span(
rule_id="metadata.mixed-content-tail",
kind="tag",
segment_index=index,
start=match.end(),
end=len(text),
priority=25,
overlap_policy=OverlapPolicy.SHARED,
)
for rule_id, pattern in (
("media.movie-cjk", _MOVIE_CJK),
("media.movie-romaji", _MOVIE_ROMAJI),
@@ -1256,6 +1442,39 @@ def _is_metadata_tail(text: str) -> bool:
return not _contains_title(residual)
def _inside_range_before_trailing_episode(
text: str, explicit_match: re.Match[str]
) -> bool:
for range_match in _RANGE.finditer(text):
if not (
range_match.start() <= explicit_match.start()
and explicit_match.end() <= range_match.end()
):
continue
if sum(1 for _ in _EXPLICIT_EPISODE.finditer(range_match.group())) != 2:
continue
if _trailing_release_episode(text[range_match.end() :]) is not None:
return True
return False
def _is_mixed_content_metadata_tail(text: str) -> bool:
residual = text
for pattern in (
*_TECHNICAL_PATTERNS,
_CJK_SUBTITLE_TAG,
_DUB,
_DATE,
_YEAR,
_FILE_SIZE,
_HASH,
):
residual = pattern.sub(" ", residual)
residual = _MIXED_CONTENT_TAIL_TOKEN.sub(" ", residual)
residual = re.sub(r"[\s_;;::,,/|+.-]+", "", residual)
return not residual
def _reconstruct_titles(
working: list[_ResidualSegment],
) -> tuple[str | None, str | None, str | None]:
@@ -1405,6 +1624,33 @@ def _special_media(marker: str) -> MediaType:
return MediaType.SPECIAL
def _mixed_content_categories(manifest: str) -> frozenset[str]:
categories: set[str] = set()
for item in re.split(r"\s*[++]\s*", manifest):
match = _MIXED_CONTENT_NAME_AT_START.match(item)
if match is None:
continue
marker = match.group("name").casefold().replace(" ", "")
if marker.startswith("tv"):
category = "tv"
elif marker in {"sp", "special"}:
category = "special"
elif marker in {"movie", "劇場版", "劇場", "剧场版", "剧场"}:
category = "movie"
elif marker in {"manga", "漫画", "漫畫"}:
category = "manga"
elif marker in {"novel", "小说", "小說"}:
category = "novel"
elif marker in {"music", "音乐", "音樂"}:
category = "music"
elif marker in {"other", "其他", "extra", "extras"}:
category = "other"
else:
category = marker
categories.add(category)
return frozenset(categories)
def _int(value: str | None) -> int | None:
return int(value) if value is not None else None

View File

@@ -68,3 +68,11 @@ class ParsedRelease:
@property
def is_movie(self) -> bool:
return self.media_type is MediaType.MOVIE
@property
def is_mixed_collection(self) -> bool:
"""Whether no single media type can represent this collection."""
return (
self.media_type is MediaType.UNKNOWN
and self.release_kind is ReleaseKind.COLLECTION
)

View File

@@ -257,7 +257,10 @@ def _merge_llm_release(
release_kind = ReleaseKind.SINGLE
use_deterministic_structure = False
if deterministic is not None:
if deterministic.media_type is not MediaType.UNKNOWN:
if (
deterministic.media_type is not MediaType.UNKNOWN
or deterministic.is_mixed_collection
):
media_type = deterministic.media_type
release_kind = deterministic.release_kind
use_deterministic_structure = bool(
@@ -428,6 +431,9 @@ class TitleParser:
):
logger.debug("Structured resource has no usable title: %s", raw)
return None
if deterministic is not None and not deterministic.primary_title:
logger.debug("Structured resource has no usable title: %s", raw)
return None
release: ParsedRelease | None = None
# primary 模式保留 LLM 标题语义,但确定性结构提示仍参与合并,

View File

@@ -1,5 +1,7 @@
import json
import pytest
from module.database.bangumi import BangumiDatabase, match_bangumi_in_list
from module.database.movie import MovieDatabase
from module.database.rss import RSSDatabase
@@ -485,6 +487,68 @@ async def test_typed_bangumi_identities_can_coexist_and_match(db_session):
assert matched_episode is not None and matched_episode.episode_type == "episode"
@pytest.mark.parametrize(
"raw",
(
"[Group] Shared Anime [TV 01-12 + OVA 01-02] [1080p]",
"Shared Anime_TV+OVA+Manga+Novel+Music+Other_dub jpn sub cht",
),
)
async def test_mixed_collection_does_not_match_episode_or_special_bangumi(
db_session, raw: str
):
db = BangumiDatabase(db_session)
shared = dict(
official_title="Shared Anime",
title_raw="Shared Anime",
group_name="Group",
rss_link="rss",
)
assert await db.add(Bangumi(**shared, season=1, episode_type="episode")) is True
assert await db.add(Bangumi(**shared, season=0, episode_type="special")) is True
matched = match_bangumi_in_list(raw, await db.search_all())
assert matched is None
@pytest.mark.parametrize(
"raw",
(
"[TV+OVA] 86 [1080p]",
"86:[TV+OVA] [1080p]",
"86: [TV+OVA] [1080p]",
),
)
def test_numeric_mixed_collection_does_not_fall_through_to_title_matching(
raw: str,
) -> None:
bangumi = Bangumi(
official_title="86",
title_raw="86",
group_name="Group",
rss_link="rss",
season=1,
episode_type="episode",
)
assert match_bangumi_in_list(raw, [bangumi]) is None
def test_titleless_mixed_collection_does_not_bypass_typed_matching() -> None:
bangumi = Bangumi(
official_title="名侦探柯南",
title_raw="名侦探柯南",
group_name="CONAN",
rss_link="rss",
season=1,
episode_type="episode",
)
raw = "[APTX4869][CONAN][名侦探柯南 1045&1046 降下天罚的生日派对]" "[TV+OVA][1080p]"
assert match_bangumi_in_list(raw, [bangumi]) is None
async def test_add_same_title_new_entry_inherits_year(db_session):
"""A new entry for an already-known anime (same official_title, different
group/quality so no semantic merge) must inherit the existing year, so both

View File

@@ -94,10 +94,12 @@ class TestGetLatestParsedEpisode:
"[TestGroup] Signal Anime OAD98 [1080p].mkv",
"[TestGroup] Signal Anime SP97 [1080p].mkv",
"[TestGroup] Signal Anime [01-99] [1080p].mkv",
"[TestGroup] Signal Anime EP01 ~ EP99 [1080p].mkv",
"[TestGroup] Signal Anime S02E01-E99 [1080p].mkv",
"[TestGroup] Signal Anime OVA01-99 [1080p].mkv",
"[TestGroup] Signal Anime Complete Batch [1080p].mkv",
"[TestGroup] Signal Anime 合集 [1080p].mkv",
"[TestGroup] Signal Anime [TV 01-12 + OVA 01-02] [1080p].mkv",
"[TestGroup] Signal Anime PV99 [1080p].mkv",
"[TestGroup] Signal Anime [1080p].mkv",
],
@@ -108,10 +110,12 @@ class TestGetLatestParsedEpisode:
"oad",
"special",
"range",
"spaced-explicit-range",
"season-range",
"ova-range",
"batch",
"collection",
"mixed-collection",
"pv",
"title-only",
],
@@ -141,10 +145,12 @@ class TestGetLatestParsedEpisode:
"[TestGroup] Signal Anime OAD98 [1080p].mkv",
"[TestGroup] Signal Anime SP97 [1080p].mkv",
"[TestGroup] Signal Anime [01-100] [1080p].mkv",
"[TestGroup] Signal Anime EP01 ~ EP100 [1080p].mkv",
"[TestGroup] Signal Anime S02E01-E100 [1080p].mkv",
"[TestGroup] Signal Anime OVA01-100 [1080p].mkv",
"[TestGroup] Signal Anime Complete Batch [1080p].mkv",
"[TestGroup] Signal Anime 合集 [1080p].mkv",
"[TestGroup] Signal Anime [TV 01-12 + OVA 01-02] [1080p].mkv",
"[TestGroup] Signal Anime PV99 [1080p].mkv",
"[TestGroup] Signal Anime [1080p].mkv",
]

View File

@@ -36,6 +36,19 @@ REQUIRED_FIELDS = (
"version",
)
MIKAN_EXPLICIT_RANGE = (
"才女的侍从 - EP01 ~ EP02 [简/繁] "
"(1080p H.264 AAC SRTx2) {才女的侍从 | 才女のお世话}"
)
MIKAN_MIXED_COLLECTION = (
"\u200b\u200b[整理搬运] 不可思议的游戏/不思议游戏/梦幻游戏/幻梦游戏 "
"(ふしぎ游戯) (Fushigi Yugi):TV动画+OVA+漫画+小说+音乐+其他;"
"日语音轨; 外挂繁中字幕 (整理时间:2025.11.12)"
)
MIKAN_MIXED_ROOT = (
"Fushigi Yugi_TV+OVA+Manga+Novel+Music+Other_" "dub jpn sub cht (2025-11-12)"
)
def _title_text(parsed: Any) -> str:
"""Join title slots so tests do not prescribe a language heuristic."""
@@ -113,6 +126,28 @@ def test_episode_range_is_a_range_release(raw: str) -> None:
assert parsed.release_kind is ReleaseKind.RANGE
@pytest.mark.parametrize(
("raw", "episode_end", "version"),
(
(MIKAN_EXPLICIT_RANGE, 2, None),
("[Group] Anime Title E01 ~ E02 [1080p]", 2, None),
("[Group] Anime Title EP. 01 ~ EP. 02v2 [1080p]", 2, 2),
),
)
def test_both_explicit_endpoints_admit_spaced_episode_range(
raw: str, episode_end: int, version: int | None
) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.episode == 1
assert parsed.episode_end == episode_end
assert parsed.version == version
assert parsed.media_type is MediaType.EPISODE
assert parsed.release_kind is ReleaseKind.RANGE
assert raw_parser(raw) is None
@pytest.mark.parametrize(
("raw", "media_type", "season", "episode_end"),
(
@@ -200,6 +235,140 @@ def test_movie_collection_keeps_both_dimensions() -> None:
assert parsed.release_kind is ReleaseKind.COLLECTION
@pytest.mark.parametrize(
"raw",
(
MIKAN_MIXED_COLLECTION,
"[Group] Anime Title [TV 01-12 + OVA 01-02] [BDRip 1080p]",
"Anime Title:TV Anime+OVA+Manga+Music",
"Anime Title:TV+剧场版+OVA",
"Anime Title:[TV+OVA] [1080p]",
"[TV+OVA; dub jpn] Anime Title [1080p]",
"[TV+OVA_1080p] Anime Title",
MIKAN_MIXED_ROOT,
),
)
def test_plus_delimited_mixed_content_manifest_is_a_collection(raw: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.media_type is MediaType.UNKNOWN
assert parsed.release_kind is ReleaseKind.COLLECTION
assert parsed.episode is None
assert parsed.episode_end is None
assert parsed.is_mixed_collection
assert "mixed-content" in parsed.evidence
title = _title_text(parsed)
assert "TV 01-12" not in title
assert "TV动画" not in title
assert "+OVA" not in title
assert raw_parser(raw) is None
if raw == MIKAN_MIXED_COLLECTION:
assert "整理时间" not in title
assert "2025.11.12" in parsed.tags
if raw == MIKAN_MIXED_ROOT:
assert parsed.title_en == "Fushigi Yugi"
assert parsed.subtitle is not None
assert parsed.subtitle.casefold() == "cht"
def test_mixed_content_metadata_cleanup_preserves_real_subtitle() -> None:
parsed = parse_release_title("Anime Title:TV+OVA; The Story")
assert parsed is not None
assert parsed.is_mixed_collection
assert "The Story" in _title_text(parsed)
def test_mixed_content_date_context_does_not_consume_a_real_subtitle() -> None:
parsed = parse_release_title("Anime Title:TV+OVA (The Story 2025-11-12)")
assert parsed is not None
assert parsed.is_mixed_collection
assert "The Story" in _title_text(parsed)
@pytest.mark.parametrize(
"raw",
(
"[TV+OVA] 86 [1080p]",
"86:[TV+OVA] [1080p]",
"86: [TV+OVA] [1080p]",
),
)
def test_mixed_collection_preserves_a_numeric_work_title(raw: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.is_mixed_collection
assert parsed.primary_title == "86"
assert parsed.episode is None
assert "episode" not in parsed.evidence
@pytest.mark.parametrize(
"raw",
(
"[TV+OVA][1080p]",
"[APTX4869][CONAN][名侦探柯南 1045&1046 降下天罚的生日派对]" "[TV+OVA][1080p]",
),
)
def test_titleless_mixed_structure_is_still_materialized(raw: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.primary_title is None
assert parsed.is_mixed_collection
assert parsed.episode is None
def test_mixed_root_metadata_tail_accepts_inline_date() -> None:
raw = "Fushigi Yugi_TV+OVA+Manga_dub jpn sub cht_2025-11-12"
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.is_mixed_collection
assert parsed.title_en == "Fushigi Yugi"
assert parsed.subtitle is not None
assert parsed.subtitle.casefold() == "cht"
def test_mixed_collection_does_not_expose_external_episode_claim() -> None:
parsed = parse_release_title("[TV + OVA] Anime Title - 01 [1080p]")
assert parsed is not None
assert parsed.is_mixed_collection
assert parsed.episode is None
assert parsed.episode_end is None
assert "episode" not in parsed.evidence
@pytest.mark.parametrize(
"raw",
(
"[TV-OVA-Fansub] Anime - 01",
"[Group] Anime OVA01 [1080p]",
"[Group] Anime OVA01-02 [1080p]",
"[Group] Anime S01E03 - TV + OVA Discussion [1080p]",
"[Group] Anime EP03 - TV + OVA [1080p]",
"[Group] Anime TV 01-12 [1080p]",
"Anime Title:SP+Special",
"Anime Title:TV+TV Anime",
"Anime - 01 (Alt_TV+OVA) [1080p]",
"Anime - 01 (Alt:TV+OVA) [1080p]",
"[Group] Anime - 01 [Tag_TV+OVA] [1080p]",
"[Group] Anime - 01 [Tag:TV+OVA] [1080p]",
"[TV+OVA_Fansub] Anime - 01 [1080p]",
),
)
def test_mixed_content_manifest_requires_metadata_shaped_context(raw: str) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert not parsed.is_mixed_collection
@pytest.mark.parametrize(
("raw", "media_type", "episode"),
[
@@ -360,6 +529,37 @@ def test_fraction_in_title_is_not_an_episode_range() -> None:
assert parsed.release_kind is ReleaseKind.SINGLE
@pytest.mark.parametrize(
("raw", "episode"),
(
("[Group] Room No. 01 ~ 02 - 03 [1080p]", 3),
("[Group] Room EP01 ~ EP02 - 03 [1080p]", 3),
("[Group] Anime EP03 - Review EP01 ~ EP02 [1080p]", 3),
("[EP01 ~ EP02-Fansub] Anime Title - 03 [1080p]", 3),
("[Group][S01E03] Anime - Review EP01 ~ EP02 [1080p]", 3),
),
)
def test_spaced_numbers_in_titles_or_groups_are_not_episode_ranges(
raw: str, episode: int
) -> None:
parsed = parse_release_title(raw)
assert parsed is not None
assert parsed.episode == episode
assert parsed.episode_end is None
assert parsed.release_kind is ReleaseKind.SINGLE
def test_explicit_range_allows_a_same_segment_year_suffix() -> None:
parsed = parse_release_title("[Group] Anime EP01 ~ EP02 - 2025 [1080p]")
assert parsed is not None
assert parsed.episode == 1
assert parsed.episode_end == 2
assert parsed.release_kind is ReleaseKind.RANGE
assert parsed.year == 2025
@pytest.mark.parametrize(
("raw", "media_type", "season", "episode", "episode_end"),
[

View File

@@ -157,6 +157,19 @@ def test_non_single_or_unnumbered_release_has_no_preference_identity() -> None:
assert preference_identity(_release(media_type=MediaType.PV)) is None
def test_mixed_collection_is_rejected_by_every_single_release_policy() -> None:
mixed = _release(
episode=None,
media_type=MediaType.UNKNOWN,
release_kind=ReleaseKind.COLLECTION,
)
assert mixed.is_mixed_collection
assert persistence_target(mixed) is None
assert preference_identity(mixed) is None
assert is_offset_signal(mixed) is False
def test_preference_revision_defaults_to_one_and_uses_explicit_version() -> None:
assert preference_revision(_release()) == 1
assert preference_revision(replace(_release(), version=3)) == 3

View File

@@ -44,3 +44,17 @@ async def test_feed_keeps_episode_special_and_movie_with_same_title(
assert isinstance(movies[0], Movie)
assert {item.title_raw for item in bangumi} == {"Shared Anime"}
assert {item.title_raw for item in movies} == {"Shared Anime"}
async def test_feed_does_not_create_entry_for_mixed_collection() -> None:
torrent = Torrent(
name="[Group] Shared Anime:TV+OVA+Manga+Music [1080p WEB-DL]",
url="https://example.com/mixed",
)
rss = make_rss_item(parser="none")
with patch.object(settings, "llm", LLM(enable=False)):
bangumi, movies = await RSSAnalyser().torrents_to_data([torrent], rss)
assert bangumi == []
assert movies == []

View File

@@ -116,6 +116,26 @@ class TestMatchTorrent:
assert result is None
async def test_mixed_collection_does_not_match_episode_or_special(self, rss_engine):
episode = make_bangumi(
title_raw="Shared Anime", season=1, episode_type="episode", filter=""
)
special = make_bangumi(
title_raw="Shared Anime", season=0, episode_type="special", filter=""
)
await rss_engine.db.bangumi.add(episode)
await rss_engine.db.bangumi.add(special)
torrent = make_torrent(
name="[Group] Shared Anime [TV 01-12 + OVA 01-02] [1080p].mkv"
)
result = rss_engine.match_torrent(
torrent, await rss_engine.db.bangumi.search_all()
)
assert result is None
assert torrent.bangumi_id is None
async def test_filter_excludes_matching_torrent(self, rss_engine):
"""When torrent name matches the filter regex, returns None."""
bangumi = make_bangumi(title_raw="Mushoku Tensei", filter="720")
@@ -1089,20 +1109,24 @@ class TestPreferenceDedup:
[
"Mushoku Tensei Movie [1080p].mkv",
"Mushoku Tensei [01-12] [1080p].mkv",
"Mushoku Tensei EP01 ~ EP02 [1080p].mkv",
"Mushoku Tensei S02E01-E12 [1080p].mkv",
"Mushoku Tensei OVA01-02 [1080p].mkv",
"Mushoku Tensei Complete Batch [1080p].mkv",
"Mushoku Tensei 合集 [1080p].mkv",
"Mushoku Tensei [TV 01-12 + OVA 01-02] [1080p].mkv",
"Mushoku Tensei PV2 [1080p].mkv",
"Mushoku Tensei [1080p].mkv",
],
ids=[
"movie",
"range",
"spaced-explicit-range",
"season-range",
"ova-range",
"batch",
"collection",
"mixed-collection",
"pv",
"title-only",
],

View File

@@ -9,12 +9,19 @@ from module.models import Bangumi, Movie
from module.models.bangumi import Episode
from module.models.config import LLM, ExperimentalOpenAI
from module.parser import title_parser as title_parser_module
from module.parser.analyser.tokenizer import ParseOutcome, ParseTrace
from module.parser.analyser.tokenizer import (
MediaType,
ParseOutcome,
ParseTrace,
ReleaseKind,
parse_release_title,
)
from module.parser.title_parser import TitleParser, _llm_config
RAW_TITLE = (
"[梦蓝字幕组]New Doraemon 哆啦A梦新番[747][2023.02.25][AVC][1080P][GB_JP][MP4]"
)
MIXED_COLLECTION = "[Group] Anime Title [TV 01-12 + OVA 01-02] [1080p]"
def _make_llm_episode(episode_type: str = "episode") -> Episode:
@@ -247,6 +254,7 @@ class TestParsedReleaseAdmission:
"[Group] Anime Title [01-12] [1080p]",
"[Group] Anime Title Complete Batch [1080p]",
"[Group] Anime Title 全集 [1080p]",
MIXED_COLLECTION,
"[Group] Anime Title PV2 [1080p]",
"[Group] Anime Title NCOP [1080p]",
"[Group] Anime Title NCED [1080p]",
@@ -328,6 +336,25 @@ class TestParsedReleaseAdmission:
assert result.season == 0
assert result.episode_type == "special"
@pytest.mark.parametrize("episode_type", ("episode", "movie", "special"))
def test_mixed_collection_structure_stays_atomic_during_llm_merge(
self, episode_type: str
):
deterministic = parse_release_title(MIXED_COLLECTION)
assert deterministic is not None
merged = title_parser_module._merge_llm_release(
MIXED_COLLECTION,
_make_llm_episode(episode_type=episode_type),
deterministic,
)
assert merged.media_type is MediaType.UNKNOWN
assert merged.release_kind is ReleaseKind.COLLECTION
assert merged.episode is None
assert merged.episode_end is None
assert merged.is_mixed_collection
@pytest.mark.parametrize(
"raw",
(
@@ -335,6 +362,7 @@ class TestParsedReleaseAdmission:
"[Group] Anime Title PV2 [1080p]",
"[Group] Anime Title Complete Batch [1080p]",
"[Group] Anime Title 合集 [1080p]",
MIXED_COLLECTION,
),
)
async def test_primary_llm_cannot_bypass_deterministic_admission(self, raw: str):
@@ -350,14 +378,17 @@ class TestParsedReleaseAdmission:
assert result is None
async def test_fallback_does_not_call_llm_for_known_rejected_kind(self):
@pytest.mark.parametrize(
"raw", ("[Group] Anime Title PV2 [1080p]", MIXED_COLLECTION)
)
async def test_fallback_does_not_call_llm_for_known_rejected_kind(self, raw: str):
with patch.object(
settings, "llm", LLM(enable=True, api_key="k", mode="fallback")
):
with patch(
"module.parser.title_parser._llm_parse", new_callable=AsyncMock
) as mock_llm:
result = await TitleParser.raw_parser("[Group] Anime Title PV2 [1080p]")
result = await TitleParser.raw_parser(raw)
assert result is None
mock_llm.assert_not_awaited()
@@ -369,6 +400,7 @@ class TestParsedReleaseAdmission:
"[Group] PV [1080p]",
"[Group] [01-12] [1080p]",
"[Group] OVA [1080p]",
"[TV+OVA][1080p]",
),
)
async def test_titleless_structured_resource_never_reaches_llm(

View File

@@ -170,6 +170,83 @@ def test_real_parser_trace_preserves_movie_title_number_with_reason() -> None:
)
def test_spaced_explicit_range_wins_over_both_single_episode_candidates() -> None:
raw = "[Group] Anime Title EP01 ~ EP02 [1080p]"
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.result.episode_end == 2
range_candidate = next(
candidate
for candidate in outcome.trace.candidates
if candidate.rule_id == "episode.range"
)
explicit_candidates = tuple(
candidate
for candidate in outcome.trace.candidates
if candidate.rule_id == "episode.explicit"
)
assert (
outcome.trace.decision_for(range_candidate.id).status is DecisionStatus.SELECTED
)
assert len(explicit_candidates) == 2
assert all(
outcome.trace.decision_for(candidate.id).status
is DecisionStatus.REJECTED_CONFLICT
for candidate in explicit_candidates
)
assert outcome.trace.warnings == ()
def test_mixed_collection_candidate_shadows_embedded_ova_and_range() -> None:
raw = "[Group] Anime Title [TV 01-12 + OVA 01-02] [1080p]"
outcome = parse_release_title_with_trace(raw)
assert outcome.result is not None
assert outcome.result.is_mixed_collection
mixed = next(
candidate
for candidate in outcome.trace.candidates
if candidate.rule_id == "cardinality.mixed-content"
)
embedded = tuple(
candidate
for candidate in outcome.trace.candidates
if candidate.rule_id in {"episode.range", "episode.ova-range", "media.ova"}
)
assert outcome.trace.decision_for(mixed.id).status is DecisionStatus.SELECTED
assert embedded
assert all(
outcome.trace.decision_for(candidate.id).status
is DecisionStatus.REJECTED_CONFLICT
for candidate in embedded
)
def test_mixed_public_result_normalizes_external_episode_resolution() -> None:
outcome = parse_release_title_with_trace("[TV + OVA] Anime Title - 01 [1080p]")
assert outcome.result is not None
assert outcome.result.is_mixed_collection
assert outcome.result.episode is None
assert "episode" not in outcome.result.evidence
assert outcome.trace.claims.episode == 1
def test_mixed_numeric_title_retry_keeps_resolver_residuals_consistent() -> None:
outcome = parse_release_title_with_trace("[TV+OVA] 86 [1080p]")
assert outcome.result is not None
assert outcome.result.primary_title == "86"
for span in outcome.trace.excluded_spans:
residual = outcome.trace.residuals[span.segment]
assert not residual[span.start : span.end].strip()
def test_empty_input_still_returns_an_explainable_outcome() -> None:
outcome = parse_release_title_with_trace(" ")