refactor(parser): migrate consumers to parsed releases

backend/src/module/core/offset_scanner.py

@@ -7,8 +7,9 @@ from module.database import Database
 from module.models import Bangumi
 from module.notification.events import OffsetReviewEvent
 from module.parser.analyser.offset_detector import detect_offset_mismatch
-from module.parser.analyser.raw_parser import raw_parser
 from module.parser.analyser.tmdb_parser import tmdb_parser
+from module.parser.analyser.tokenizer import parse_release_title
+from module.parser.release_policy import is_offset_signal
 
 logger = logging.getLogger(__name__)
 
@@ -58,6 +59,16 @@ class OffsetScanner:
             logger.debug(f"Skipping {bangumi.official_title}: already needs review")
             return None
 
+        # OVA/OAD/SP are persisted as Bangumi for library compatibility, but
+        # they are not weekly episode streams and must never drive offsets.
+        if bangumi.episode_type != "episode":
+            logger.debug(
+                "Skipping %s: episode_type=%s",
+                bangumi.official_title,
+                bangumi.episode_type,
+            )
+            return None
+
         # Skip if user has already configured offsets
         if bangumi.season_offset != 0 or bangumi.episode_offset != 0:
             logger.debug(f"Skipping {bangumi.official_title}: has configured offsets")
@@ -114,11 +125,16 @@ class OffsetScanner:
 
         latest: int | None = None
         for torrent in torrents:
-            episode = raw_parser(torrent.name)
-            if episode is None or not episode.episode:
+            release = parse_release_title(torrent.name)
+            if release is None or not is_offset_signal(release):
                 continue
-            if latest is None or episode.episode > latest:
-                latest = episode.episode
+            episode = release.episode
+            # ``is_offset_signal`` guarantees this at runtime; retain the local
+            # guard so static type checking can narrow ``int | float | None``.
+            if type(episode) is not int:
+                continue
+            if latest is None or episode > latest:
+                latest = episode
         return latest
 
     async def check_single(self, bangumi_id: int) -> bool:

backend/src/module/parser/release_policy.py

@@ -0,0 +1,107 @@
+"""Business admission policies for generic parsed releases.
+
+Parsing answers what a resource name contains.  These helpers answer whether a
+particular consumer can safely use that result.  Keeping the policy outside the
+tokenizer prevents download/database constraints from leaking into parsing.
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+from module.parser.analyser.tokenizer.result import (
+    MediaType,
+    ParsedRelease,
+    ReleaseKind,
+)
+
+
+class PersistenceTarget(StrEnum):
+    BANGUMI = "bangumi"
+    MOVIE = "movie"
+
+
+_SPECIAL_MEDIA = {MediaType.OVA, MediaType.OAD, MediaType.SPECIAL}
+_NUMBERED_MEDIA = {MediaType.EPISODE, *_SPECIAL_MEDIA}
+_NON_PERSISTED_MEDIA = {MediaType.PV, MediaType.OPENING, MediaType.ENDING}
+
+
+def has_release_evidence(release: ParsedRelease) -> bool:
+    """Whether a title-only result still looks like an actual release name."""
+    return bool(
+        release.group
+        or release.resolution
+        or release.source
+        or release.subtitle
+        or release.codecs
+        or release.audio
+        or release.container
+    )
+
+
+def is_weak_title_only(release: ParsedRelease) -> bool:
+    """Whether parsing found only a title, with no release-shaped evidence."""
+    return bool(
+        release.primary_title
+        and release.media_type is MediaType.UNKNOWN
+        and release.release_kind is ReleaseKind.SINGLE
+        and release.episode is None
+        and not has_release_evidence(release)
+    )
+
+
+def persistence_target(release: ParsedRelease) -> PersistenceTarget | None:
+    """Return the database projection currently supported for a release."""
+    if (
+        not release.primary_title
+        or release.release_kind is not ReleaseKind.SINGLE
+        or release.media_type in _NON_PERSISTED_MEDIA
+    ):
+        return None
+    if release.media_type is MediaType.MOVIE:
+        return PersistenceTarget.MOVIE
+    if is_weak_title_only(release):
+        return None
+    return PersistenceTarget.BANGUMI
+
+
+def bangumi_episode_type(release: ParsedRelease) -> str:
+    return "special" if release.media_type in _SPECIAL_MEDIA else "episode"
+
+
+def normalized_season(release: ParsedRelease, *, default: int = 1) -> int:
+    if release.media_type in _SPECIAL_MEDIA:
+        return 0
+    return release.season if release.season is not None else default
+
+
+def preference_identity(
+    release: ParsedRelease, *, default_season: int = 1
+) -> tuple[MediaType, int, int | float] | None:
+    """Return a safe identity for release-preference deduplication."""
+    if (
+        release.release_kind is not ReleaseKind.SINGLE
+        or release.media_type not in _NUMBERED_MEDIA
+        or release.episode is None
+    ):
+        return None
+    return (
+        release.media_type,
+        normalized_season(release, default=default_season),
+        release.episode,
+    )
+
+
+def preference_revision(release: ParsedRelease) -> int:
+    """Rank revisions of the same content; an unmarked release is revision 1."""
+    return release.version if release.version is not None else 1
+
+
+def is_offset_signal(release: ParsedRelease) -> bool:
+    """Whether a release is a safe weekly-episode signal for offset detection."""
+    return bool(
+        release.media_type is MediaType.EPISODE
+        and release.release_kind is ReleaseKind.SINGLE
+        and type(release.episode) is int
+        and release.episode > 0
+    )

backend/src/module/parser/title_parser.py

@@ -9,13 +9,25 @@ from module.models.config import LLM
 from module.parser.analyser import (
     LLMParser,
     mikan_parser,
-    raw_parser,
     tmdb_parser,
     torrent_parser,
 )
 from module.parser.analyser.providers.base import AuthExpiredError
 from module.parser.analyser.providers.credentials import auth_generation
-from module.parser.analyser.raw_parser import _detect_non_episodic_type
+from module.parser.analyser.tokenizer import (
+    MediaType,
+    ParsedRelease,
+    ReleaseKind,
+    parse_release_title_with_trace,
+)
+from module.parser.analyser.tokenizer.candidate import Claims
+from module.parser.release_policy import (
+    PersistenceTarget,
+    bangumi_episode_type,
+    is_weak_title_only,
+    normalized_season,
+    persistence_target,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -206,6 +218,100 @@ def _record_llm_failure(conf: LLM) -> None:
         _llm_breaker_until = time.monotonic() + conf.failure_backoff
 
 
+def _llm_media_type(episode: Episode) -> MediaType:
+    if episode.episode_type == "movie" or episode.is_movie:
+        return MediaType.MOVIE
+    if episode.episode_type == "special":
+        return MediaType.SPECIAL
+    return MediaType.EPISODE
+
+
+def _has_structured_release_claims(claims: Claims) -> bool:
+    """Whether a title-less parse still identified a concrete resource kind."""
+    return any(
+        value is not None
+        for value in (
+            claims.season,
+            claims.episode,
+            claims.episode_end,
+            claims.media_type,
+            claims.release_kind,
+        )
+    )
+
+
+def _merge_llm_release(
+    raw: str,
+    episode: Episode,
+    deterministic: ParsedRelease | None,
+) -> ParsedRelease:
+    """Convert an LLM result without mutating the cached legacy object.
+
+    LLM titles keep their primary-mode semantics.  When deterministic parsing
+    found a structural bundle, its media/cardinality/season/episode/version
+    fields stay atomic so an LLM cannot create hybrids such as RANGE 7-12 or
+    turn a movie/PV into a weekly episode.
+    """
+
+    media_type = _llm_media_type(episode)
+    release_kind = ReleaseKind.SINGLE
+    use_deterministic_structure = False
+    if deterministic is not None:
+        if deterministic.media_type is not MediaType.UNKNOWN:
+            media_type = deterministic.media_type
+        release_kind = deterministic.release_kind
+        use_deterministic_structure = bool(
+            deterministic.media_type is not MediaType.UNKNOWN
+            or deterministic.release_kind is not ReleaseKind.SINGLE
+            or deterministic.season is not None
+            or deterministic.episode is not None
+            or deterministic.episode_end is not None
+        )
+
+    season: int | None = episode.season
+    season_raw: str | None = episode.season_raw
+    episode_number: int | float | None = episode.episode
+    episode_end: int | float | None = None
+    episode_title: str | None = None
+    version: int | None = None
+    if deterministic is not None and use_deterministic_structure:
+        season = deterministic.season
+        season_raw = deterministic.season_raw
+        episode_number = deterministic.episode
+        episode_end = deterministic.episode_end
+        episode_title = deterministic.episode_title
+        version = deterministic.version
+
+    return ParsedRelease(
+        raw=raw,
+        title_en=episode.title_en
+        or (deterministic.title_en if deterministic else None),
+        title_zh=episode.title_zh
+        or (deterministic.title_zh if deterministic else None),
+        title_jp=episode.title_jp
+        or (deterministic.title_jp if deterministic else None),
+        group=episode.group or (deterministic.group if deterministic else None),
+        season=season,
+        season_raw=season_raw,
+        episode=episode_number,
+        episode_end=episode_end,
+        episode_title=episode_title,
+        media_type=media_type,
+        release_kind=release_kind,
+        resolution=episode.resolution
+        or (deterministic.resolution if deterministic else None),
+        source=episode.source or (deterministic.source if deterministic else None),
+        subtitle=episode.sub or (deterministic.subtitle if deterministic else None),
+        codecs=deterministic.codecs if deterministic else (),
+        audio=deterministic.audio if deterministic else (),
+        container=deterministic.container if deterministic else None,
+        version=version,
+        year=deterministic.year if deterministic else None,
+        tags=deterministic.tags if deterministic else (),
+        evidence=deterministic.evidence if deterministic else (),
+    )
+
+
 class TitleParser:
     def __init__(self):
         pass
@@ -254,15 +360,15 @@ class TitleParser:
             logger.warning("Please change bangumi info manually.")
 
     @staticmethod
-    def _resolve_titles(episode: Episode) -> tuple[str | None, str | None]:
-        """Extract official_title and title_raw from an Episode."""
+    def _resolve_titles(release: ParsedRelease) -> tuple[str | None, str | None]:
+        """Extract official_title and title_raw from a generic release."""
         language = settings.rss_parser.language
         titles = {
-            "zh": episode.title_zh,
-            "en": episode.title_en,
-            "jp": episode.title_jp,
+            "zh": release.title_zh,
+            "en": release.title_en,
+            "jp": release.title_jp,
         }
-        title_raw = episode.title_en or episode.title_zh or episode.title_jp
+        title_raw = release.title_en or release.title_zh or release.title_jp
         if titles[language]:
             official_title = titles[language]
         elif titles["zh"]:
@@ -276,35 +382,38 @@ class TitleParser:
         return official_title, title_raw
 
     @staticmethod
-    def _episode_to_bangumi(
-        episode: Episode, official_title: str, title_raw: str
+    def _release_to_bangumi(
+        release: ParsedRelease, official_title: str, title_raw: str
     ) -> Bangumi:
+        episode = release.episode
         return Bangumi(
             official_title=official_title,
             title_raw=title_raw,
-            season=episode.season,
-            season_raw=episode.season_raw,
-            group_name=episode.group,
-            dpi=episode.resolution,
-            source=episode.source,
-            subtitle=episode.sub,
-            eps_collect=False if episode.episode > 1 else True,
+            year=str(release.year) if release.year is not None else None,
+            season=normalized_season(release),
+            season_raw=release.season_raw,
+            group_name=release.group or "",
+            dpi=release.resolution,
+            source=release.source,
+            subtitle=release.subtitle,
+            eps_collect=episode is None or episode <= 1,
             episode_offset=0,
             filter=",".join(settings.rss_parser.filter),
-            episode_type=episode.episode_type,
+            episode_type=bangumi_episode_type(release),
         )
 
     @staticmethod
-    def _episode_to_movie(
-        episode: Episode, official_title: str, title_raw: str
+    def _release_to_movie(
+        release: ParsedRelease, official_title: str, title_raw: str
     ) -> Movie:
         return Movie(
             official_title=official_title,
             title_raw=title_raw,
-            group_name=episode.group,
-            dpi=episode.resolution,
-            source=episode.source,
-            subtitle=episode.sub,
+            year=release.year,
+            group_name=release.group or "",
+            dpi=release.resolution,
+            source=release.source,
+            subtitle=release.subtitle,
             filter=",".join(settings.rss_parser.filter),
         )
 
@@ -312,29 +421,41 @@ class TitleParser:
     async def raw_parser(raw: str) -> Bangumi | Movie | None:
         try:
             llm_conf = _llm_config()
-            episode = None
-            from_llm = False
-            # primary 模式：LLM 优先解析每个标题；失败时用正则兜底，保证不丢标题
+            parse_outcome = parse_release_title_with_trace(raw)
+            deterministic = parse_outcome.result
+            if deterministic is None and _has_structured_release_claims(
+                parse_outcome.trace.claims
+            ):
+                logger.debug("Structured resource has no usable title: %s", raw)
+                return None
+            release: ParsedRelease | None = None
+
+            # primary 模式保留 LLM 标题语义，但确定性结构提示仍参与合并，
+            # 避免关键词误判并确保 range/PV 等准入策略无法被绕过。
             if llm_conf.enable and llm_conf.mode == "primary":
                 episode = await _llm_parse(raw)
-                from_llm = episode is not None
-            if episode is None:
-                episode = raw_parser(raw)
-            # fallback 模式（默认）：正则优先，仅当正则失败时才请求 LLM 兜底
-            if episode is None and llm_conf.enable and llm_conf.mode == "fallback":
-                episode = await _llm_parse(raw)
-                from_llm = episode is not None
-            if episode is None:
-                return None
-            # 正则路径已在剥离字幕组名后的标题上判定过 episode_type，这里只
-            # 给缺少该能力的 LLM 结果补分类；对完整 raw 判定会把组名/标签里
-            # 撞词（如 "[Movie-Fan]"）的普通周更集误判成剧场版/特别篇。
-            if from_llm:
-                detected_episode_type = _detect_non_episodic_type(raw)
-                if detected_episode_type is not None:
-                    episode.episode_type = detected_episode_type
+                if episode is not None:
+                    release = _merge_llm_release(raw, episode, deterministic)
 
-            official_title, title_raw = TitleParser._resolve_titles(episode)
+            if release is None:
+                release = deterministic
+
+            # fallback 只处理“解析失败/只有弱标题”的输入。明确识别出的
+            # PV、range、batch、collection 直接遵守业务拒绝策略。
+            should_fallback = deterministic is None or is_weak_title_only(deterministic)
+            if llm_conf.enable and llm_conf.mode == "fallback" and should_fallback:
+                episode = await _llm_parse(raw)
+                if episode is not None:
+                    release = _merge_llm_release(raw, episode, deterministic)
+
+            if release is None:
+                return None
+            target = persistence_target(release)
+            if target is None:
+                logger.debug("Parsed but did not admit resource: %s", raw)
+                return None
+
+            official_title, title_raw = TitleParser._resolve_titles(release)
             if not title_raw:
                 logger.warning("Cannot extract title_raw from '%s', skipping", raw)
                 return None
@@ -343,12 +464,9 @@ class TitleParser:
                 return None
             logger.debug("RAW:%s >> %s", raw, title_raw)
 
-            # The upstream LLM path exposes its inferred non-episodic type on a
-            # Bangumi.  Keep that contract while routing deterministically parsed
-            # movie markers into the dedicated local Movie table.
-            if episode.episode_type == "movie" and not from_llm:
-                return TitleParser._episode_to_movie(episode, official_title, title_raw)
-            return TitleParser._episode_to_bangumi(episode, official_title, title_raw)
+            if target is PersistenceTarget.MOVIE:
+                return TitleParser._release_to_movie(release, official_title, title_raw)
+            return TitleParser._release_to_bangumi(release, official_title, title_raw)
         except (ValueError, AttributeError, TypeError) as e:
             logger.warning(f"Cannot parse '{raw}': {type(e).__name__}: {e}")
             return None

backend/src/module/rss/engine.py

@@ -10,14 +10,19 @@ from module.conf import settings
 from module.database import Database
 from module.database.bangumi import _groups_are_similar, match_bangumi_in_list
 from module.downloader import AddResult, DownloadClient
-from module.models import Bangumi, Episode, Movie, ResponseModel, RSSItem, Torrent
+from module.models import Bangumi, Movie, ResponseModel, RSSItem, Torrent
 from module.network import RequestContent
 from module.notification.events import (
     DownloadFailureEvent,
     RssFailureEvent,
     SystemEvent,
 )
-from module.parser.analyser.raw_parser import raw_parser
+from module.parser.analyser.tokenizer import (
+    MediaType,
+    ParsedRelease,
+    parse_release_title,
+)
+from module.parser.release_policy import preference_identity, preference_revision
 
 logger = logging.getLogger(__name__)
 
@@ -25,6 +30,9 @@ logger = logging.getLogger(__name__)
 # site at once gets the whole batch rate-limited with HTTP 429 (#1026).
 RSS_PER_HOST_DELAY = 2.0
 
+_PreferenceKey = tuple[int, MediaType, int, int | float]
+_PreferenceRank = tuple[int, int]
+
 
 def _resolution_matches(candidate: str | None, preferred: str | None) -> bool:
     """比较分辨率是否一致（"1080p" 与 "1080" 视为相同）。"""
@@ -35,15 +43,15 @@ def _resolution_matches(candidate: str | None, preferred: str | None) -> bool:
     return normalized_candidate == normalized_preferred
 
 
-def _preference_score(episode: Episode, bangumi: Bangumi) -> int:
+def _preference_score(release: ParsedRelease, bangumi: Bangumi) -> int:
     """候选版本相对番剧发布组/分辨率偏好的匹配得分：每命中一项 +1。"""
     score = 0
     if bangumi.preferred_group and _groups_are_similar(
-        episode.group, bangumi.preferred_group
+        release.group, bangumi.preferred_group
     ):
         score += 1
     if bangumi.preferred_resolution and _resolution_matches(
-        episode.resolution, bangumi.preferred_resolution
+        release.resolution, bangumi.preferred_resolution
     ):
         score += 1
     return score
@@ -193,51 +201,67 @@ class RSSEngine:
         未设置偏好的番剧完全不受影响（沿用旧行为，多字幕组各自下载全部集数）。
 
         规则：
-        - 该集已有下载版本时，新到的版本只有严格优于已下载版本（得分更高）
-          才会被保留，平局或更差一律跳过——不删除已下载的旧版本，只是不再
-          重复下载。
-        - 同一批次内同一集出现多个候选时，只保留得分最高的一个。
-        - 无法从种子名解析出集数的候选不参与去重判断，始终保留（保守回退，
-          避免因解析失败漏下载）。
+        - 内容身份包含媒体类型、规范化季度和集数，防止正片、跨季及特典互相
+          冲突。
+        - 该内容已有下载版本时，新到版本只有偏好得分更高，或偏好得分相同但
+          修订号更高时才保留；偏好匹配始终优先于修订号。
+        - 同一批次内同一内容出现多个候选时，只保留排序最高的一个。
+        - 无法得到安全单集身份的候选不参与去重，始终保留（保守回退，避免因
+          解析失败或把 Movie/合集/PV 当成单集而漏下载）。
 
         返回需要跳过下载的种子对象 id（``id(torrent)``），供调用方在本轮
         ``refresh_rss`` 内部过滤，不做跨请求持久化。
         """
         skip_ids: set[int] = set()
 
-        # 已下载版本：按 (bangumi_id, episode) 记录当前最高得分
-        existing_best: dict[tuple[int, int], int] = {}
+        # 已下载版本：按完整内容身份记录当前最高 (偏好得分, 修订号)。
+        existing_best: dict[_PreferenceKey, _PreferenceRank] = {}
         for bangumi_id, torrents in existing_downloaded.items():
             bangumi = preference_bangumi.get(bangumi_id)
             if not bangumi:
                 continue
             for torrent in torrents:
-                episode = raw_parser(torrent.name)
-                if episode is None:
+                release = parse_release_title(torrent.name)
+                if release is None:
                     continue
-                key = (bangumi_id, episode.episode)
-                score = _preference_score(episode, bangumi)
-                existing_best[key] = max(existing_best.get(key, -1), score)
+                identity = preference_identity(release, default_season=bangumi.season)
+                if identity is None:
+                    continue
+                key = (bangumi_id, identity[0], identity[1], identity[2])
+                rank = (
+                    _preference_score(release, bangumi),
+                    preference_revision(release),
+                )
+                previous = existing_best.get(key)
+                if previous is None or rank > previous:
+                    existing_best[key] = rank
 
-        # 本批次候选：按 (bangumi_id, episode) 分组
-        batch_groups: dict[tuple[int, int], list[tuple[Torrent, int]]] = defaultdict(
-            list
+        # 本批次候选：按 (bangumi_id, media_type, season, episode) 分组。
+        batch_groups: dict[_PreferenceKey, list[tuple[Torrent, _PreferenceRank]]] = (
+            defaultdict(list)
         )
         for torrent, bangumi in matched:
             if bangumi.id is None or bangumi.id not in preference_bangumi:
                 continue
-            episode = raw_parser(torrent.name)
-            if episode is None:
+            release = parse_release_title(torrent.name)
+            if release is None:
                 continue
-            key = (bangumi.id, episode.episode)
-            batch_groups[key].append((torrent, _preference_score(episode, bangumi)))
+            identity = preference_identity(release, default_season=bangumi.season)
+            if identity is None:
+                continue
+            key = (bangumi.id, identity[0], identity[1], identity[2])
+            rank = (
+                _preference_score(release, bangumi),
+                preference_revision(release),
+            )
+            batch_groups[key].append((torrent, rank))
 
         for key, candidates in batch_groups.items():
-            best_score = max(score for _, score in candidates)
-            best_torrent = next(t for t, s in candidates if s == best_score)
+            best_rank = max(rank for _, rank in candidates)
+            best_torrent = next(t for t, rank in candidates if rank == best_rank)
             prior_best = existing_best.get(key)
-            keep_best = prior_best is None or best_score > prior_best
-            for torrent, _score in candidates:
+            keep_best = prior_best is None or best_rank > prior_best
+            for torrent, _rank in candidates:
                 if torrent is best_torrent and keep_best:
                     continue
                 skip_ids.add(id(torrent))

backend/src/test/test_offset_scanner.py

@@ -2,6 +2,8 @@
 
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import pytest
+
 from module.core.offset_scanner import OffsetScanner
 from module.database import Database
 from module.models import Torrent
@@ -48,6 +50,98 @@ class TestGetLatestParsedEpisode:
 
         assert await scanner._get_latest_parsed_episode(bangumi.id) is None
 
+    @pytest.mark.parametrize(
+        "resource_name",
+        [
+            "[TestGroup] Signal Anime - 12.5 [1080p].mkv",
+            "[TestGroup] Signal Anime Movie [1080p].mkv",
+            "[TestGroup] Signal Anime OVA99 [1080p].mkv",
+            "[TestGroup] Signal Anime OAD98 [1080p].mkv",
+            "[TestGroup] Signal Anime SP97 [1080p].mkv",
+            "[TestGroup] Signal Anime [01-99] [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 PV99 [1080p].mkv",
+            "[TestGroup] Signal Anime [1080p].mkv",
+        ],
+        ids=[
+            "fractional",
+            "movie",
+            "ova",
+            "oad",
+            "special",
+            "range",
+            "season-range",
+            "ova-range",
+            "batch",
+            "collection",
+            "pv",
+            "title-only",
+        ],
+    )
+    async def test_non_weekly_resources_are_not_offset_signals(self, resource_name):
+        scanner = OffsetScanner()
+        async with Database() as db:
+            bangumi = make_bangumi(title_raw="Signal Anime")
+            await db.bangumi.add(bangumi)
+            await db.torrent.add(
+                Torrent(
+                    name=resource_name,
+                    url="https://example.com/non-weekly",
+                    bangumi_id=bangumi.id,
+                )
+            )
+
+        assert await scanner._get_latest_parsed_episode(bangumi.id) is None
+
+    async def test_excluded_resources_do_not_override_latest_weekly_episode(self):
+        scanner = OffsetScanner()
+        resource_names = [
+            "[TestGroup] Signal Anime - 05 [1080p].mkv",
+            "[TestGroup] Signal Anime - 99.5 [1080p].mkv",
+            "[TestGroup] Signal Anime Movie [1080p].mkv",
+            "[TestGroup] Signal Anime OVA99 [1080p].mkv",
+            "[TestGroup] Signal Anime OAD98 [1080p].mkv",
+            "[TestGroup] Signal Anime SP97 [1080p].mkv",
+            "[TestGroup] Signal Anime [01-100] [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 PV99 [1080p].mkv",
+            "[TestGroup] Signal Anime [1080p].mkv",
+        ]
+        async with Database() as db:
+            bangumi = make_bangumi(title_raw="Signal Anime")
+            await db.bangumi.add(bangumi)
+            for index, resource_name in enumerate(resource_names):
+                await db.torrent.add(
+                    Torrent(
+                        name=resource_name,
+                        url=f"https://example.com/mixed/{index}",
+                        bangumi_id=bangumi.id,
+                    )
+                )
+
+        assert await scanner._get_latest_parsed_episode(bangumi.id) == 5
+
+    async def test_versioned_weekly_episode_remains_an_offset_signal(self):
+        scanner = OffsetScanner()
+        async with Database() as db:
+            bangumi = make_bangumi(title_raw="Signal Anime")
+            await db.bangumi.add(bangumi)
+            await db.torrent.add(
+                Torrent(
+                    name="[TestGroup] Signal Anime - 12v2 [1080p].mkv",
+                    url="https://example.com/versioned",
+                    bangumi_id=bangumi.id,
+                )
+            )
+
+        assert await scanner._get_latest_parsed_episode(bangumi.id) == 12
+
 
 class TestCheckBangumiUsesRealSignal:
     """Tests for OffsetScanner._check_bangumi consuming the real torrent signal."""
@@ -65,6 +159,26 @@ class TestCheckBangumiUsesRealSignal:
 
         assert event is None
 
+    @pytest.mark.parametrize("episode_type", ("special", "movie"))
+    async def test_skips_non_episode_bangumi_before_tmdb_lookup(
+        self, episode_type: str
+    ):
+        scanner = OffsetScanner()
+        bangumi = make_bangumi(
+            id=1,
+            episode_type=episode_type,
+            season_offset=0,
+            episode_offset=0,
+        )
+
+        with patch(
+            "module.core.offset_scanner.tmdb_parser", new_callable=AsyncMock
+        ) as mock_tmdb:
+            event = await scanner._check_bangumi(bangumi)
+
+        assert event is None
+        mock_tmdb.assert_not_awaited()
+
     async def test_flags_using_real_latest_episode_from_torrent_table(self):
         """A mismatch is only suggested from the actual parsed episode, not a constant."""
         scanner = OffsetScanner()

backend/src/test/test_release_policy.py

@@ -0,0 +1,171 @@
+from dataclasses import replace
+
+import pytest
+
+from module.parser.analyser.tokenizer.result import (
+    MediaType,
+    ParsedRelease,
+    ReleaseKind,
+)
+from module.parser.release_policy import (
+    PersistenceTarget,
+    bangumi_episode_type,
+    has_release_evidence,
+    is_offset_signal,
+    is_weak_title_only,
+    normalized_season,
+    persistence_target,
+    preference_identity,
+    preference_revision,
+)
+
+
+def _release(
+    *,
+    raw: str = "[Group] Anime - 01 [1080p]",
+    group: str | None = "Group",
+    resolution: str | None = None,
+    episode: int | float | None = 1,
+    season: int | None = None,
+    media_type: MediaType = MediaType.EPISODE,
+    release_kind: ReleaseKind = ReleaseKind.SINGLE,
+) -> ParsedRelease:
+    base = ParsedRelease(raw=raw, title_en="Anime")
+    return replace(
+        base,
+        group=group,
+        resolution=resolution,
+        episode=episode,
+        season=season,
+        media_type=media_type,
+        release_kind=release_kind,
+    )
+
+
+def test_persistence_policy_separates_parsing_from_admission() -> None:
+    assert persistence_target(_release()) is PersistenceTarget.BANGUMI
+    assert (
+        persistence_target(_release(media_type=MediaType.MOVIE, episode=None))
+        is PersistenceTarget.MOVIE
+    )
+    assert persistence_target(_release(media_type=MediaType.PV)) is None
+    assert (
+        persistence_target(
+            _release(
+                media_type=MediaType.MOVIE,
+                episode=None,
+                release_kind=ReleaseKind.COLLECTION,
+            )
+        )
+        is None
+    )
+
+
+@pytest.mark.parametrize("release_kind", list(ReleaseKind)[1:])
+def test_non_single_releases_are_not_persisted_by_default(
+    release_kind: ReleaseKind,
+) -> None:
+    assert persistence_target(_release(release_kind=release_kind)) is None
+
+
+@pytest.mark.parametrize(
+    "media_type", (MediaType.PV, MediaType.OPENING, MediaType.ENDING)
+)
+def test_non_content_video_kinds_are_not_persisted(media_type: MediaType) -> None:
+    assert persistence_target(_release(media_type=media_type)) is None
+
+
+@pytest.mark.parametrize(
+    "media_type", (MediaType.OVA, MediaType.OAD, MediaType.SPECIAL)
+)
+def test_special_media_projects_to_bangumi(media_type: MediaType) -> None:
+    release = _release(media_type=media_type)
+
+    assert persistence_target(release) is PersistenceTarget.BANGUMI
+    assert bangumi_episode_type(release) == "special"
+    assert normalized_season(release) == 0
+
+
+def test_title_only_requires_release_evidence() -> None:
+    bare = _release(
+        raw="Random title",
+        group=None,
+        resolution=None,
+        episode=None,
+        media_type=MediaType.UNKNOWN,
+    )
+    assert is_weak_title_only(bare)
+    assert not is_weak_title_only(_release(episode=None, media_type=MediaType.UNKNOWN))
+    assert persistence_target(bare) is None
+    assert (
+        persistence_target(_release(episode=None, media_type=MediaType.UNKNOWN))
+        is PersistenceTarget.BANGUMI
+    )
+
+
+@pytest.mark.parametrize(
+    "release",
+    (
+        _release(group="Group", episode=None, media_type=MediaType.UNKNOWN),
+        _release(
+            group=None,
+            resolution="1080p",
+            episode=None,
+            media_type=MediaType.UNKNOWN,
+        ),
+    ),
+)
+def test_strong_metadata_counts_as_release_evidence(release: ParsedRelease) -> None:
+    assert has_release_evidence(release)
+
+
+def test_preference_identity_keeps_season_and_media_dimensions() -> None:
+    episode = _release(season=2)
+    ova = _release(media_type=MediaType.OVA, season=None)
+
+    assert preference_identity(episode) == (MediaType.EPISODE, 2, 1)
+    assert preference_identity(ova) == (MediaType.OVA, 0, 1)
+    assert preference_identity(episode) != preference_identity(ova)
+
+
+def test_preference_identity_can_inherit_matched_bangumi_season() -> None:
+    implicit = _release(season=None)
+    special = _release(media_type=MediaType.OVA, season=None)
+
+    assert preference_identity(implicit, default_season=2) == (
+        MediaType.EPISODE,
+        2,
+        1,
+    )
+    assert preference_identity(special, default_season=2) == (
+        MediaType.OVA,
+        0,
+        1,
+    )
+
+
+def test_half_episode_has_identity_but_never_drives_offset() -> None:
+    half = _release(episode=12.5, season=2)
+
+    assert preference_identity(half) == (MediaType.EPISODE, 2, 12.5)
+    assert is_offset_signal(half) is False
+
+
+def test_non_single_or_unnumbered_release_has_no_preference_identity() -> None:
+    assert preference_identity(_release(release_kind=ReleaseKind.RANGE)) is None
+    assert preference_identity(_release(episode=None)) is None
+    assert preference_identity(_release(media_type=MediaType.PV)) is None
+
+
+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
+
+
+def test_offset_signal_is_strictly_positive_integer_weekly_episode() -> None:
+    assert is_offset_signal(_release()) is True
+    assert is_offset_signal(_release(episode=12.5)) is False
+    assert is_offset_signal(_release(media_type=MediaType.OVA)) is False
+    assert is_offset_signal(_release(release_kind=ReleaseKind.RANGE)) is False
+    assert is_offset_signal(_release(episode=0)) is False
+    assert is_offset_signal(_release(episode=True)) is False

backend/src/test/test_rss_engine_new.py

@@ -964,6 +964,234 @@ class TestPreferenceDedup:
 
         assert skips == set()
 
+    async def test_same_episode_number_in_different_seasons_is_not_deduped(
+        self, rss_engine
+    ):
+        """Season is part of the content identity, so S1E1 and S2E1 coexist."""
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        season_1 = make_torrent(name="[GroupB] Mushoku Tensei S01E01 [1080p].mkv")
+        season_2 = make_torrent(name="[GroupB] Mushoku Tensei S02E01 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(season_1, bangumi), (season_2, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert skips == set()
+
+    async def test_implicit_season_inherits_matched_bangumi_season(self, rss_engine):
+        """Implicit and explicit S2 releases describe the same matched content."""
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei",
+            season=2,
+            filter="",
+            id=1,
+            preferred_group="GroupA",
+        )
+        implicit = make_torrent(name="[GroupA] Mushoku Tensei - 01 [1080p].mkv")
+        explicit = make_torrent(name="[GroupB] Mushoku Tensei S02E01 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(implicit, bangumi), (explicit, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert id(explicit) in skips
+        assert id(implicit) not in skips
+
+    async def test_existing_explicit_season_blocks_implicit_same_episode(
+        self, rss_engine
+    ):
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei",
+            season=2,
+            filter="",
+            id=1,
+            preferred_group="GroupA",
+        )
+        downloaded = make_torrent(
+            name="[GroupA] Mushoku Tensei S02E01 [1080p].mkv",
+            downloaded=True,
+        )
+        implicit = make_torrent(name="[GroupB] Mushoku Tensei - 01 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(implicit, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={1: [downloaded]},
+        )
+
+        assert id(implicit) in skips
+
+    async def test_fractional_episode_uses_preference_dedup(self, rss_engine):
+        """A half episode is valid content identity even though Offset ignores it."""
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        preferred = make_torrent(name="[GroupA] Mushoku Tensei - 12.5 [1080p].mkv")
+        other = make_torrent(name="[GroupB] Mushoku Tensei - 12.5 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(preferred, bangumi), (other, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert id(other) in skips
+        assert id(preferred) not in skips
+
+    @pytest.mark.parametrize("marker", ["OVA", "OAD", "SP"])
+    async def test_numbered_special_uses_typed_preference_dedup(
+        self, rss_engine, marker
+    ):
+        """Numbered specials dedup only against the same typed special."""
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        preferred = make_torrent(name=f"[GroupA] Mushoku Tensei {marker}01 [1080p].mkv")
+        other = make_torrent(name=f"[GroupB] Mushoku Tensei {marker}01 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(preferred, bangumi), (other, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert id(other) in skips
+        assert id(preferred) not in skips
+
+    async def test_same_number_across_media_types_is_not_deduped(self, rss_engine):
+        """Episode, OVA, OAD and SP identities must not collide at number 1."""
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        candidates = [
+            make_torrent(name="[GroupB] Mushoku Tensei - 01 [1080p].mkv"),
+            make_torrent(name="[GroupB] Mushoku Tensei OVA01 [1080p].mkv"),
+            make_torrent(name="[GroupB] Mushoku Tensei OAD01 [1080p].mkv"),
+            make_torrent(name="[GroupB] Mushoku Tensei SP01 [1080p].mkv"),
+        ]
+
+        skips = RSSEngine._select_preference_skips(
+            [(torrent, bangumi) for torrent in candidates],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert skips == set()
+
+    @pytest.mark.parametrize(
+        "resource",
+        [
+            "Mushoku Tensei Movie [1080p].mkv",
+            "Mushoku Tensei [01-12] [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 PV2 [1080p].mkv",
+            "Mushoku Tensei [1080p].mkv",
+        ],
+        ids=[
+            "movie",
+            "range",
+            "season-range",
+            "ova-range",
+            "batch",
+            "collection",
+            "pv",
+            "title-only",
+        ],
+    )
+    async def test_non_single_episode_resources_are_conservatively_retained(
+        self, rss_engine, resource
+    ):
+        """Resources without a safe single-episode identity never enter dedup."""
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        preferred = make_torrent(name=f"[GroupA] {resource}")
+        other = make_torrent(name=f"[GroupB] {resource}")
+
+        skips = RSSEngine._select_preference_skips(
+            [(preferred, bangumi), (other, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert skips == set()
+
+    async def test_higher_revision_wins_when_preference_score_ties(self, rss_engine):
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        version_1 = make_torrent(name="[GroupA] Mushoku Tensei - 12v1 [1080p].mkv")
+        version_2 = make_torrent(name="[GroupA] Mushoku Tensei - 12v2 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(version_1, bangumi), (version_2, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert id(version_1) in skips
+        assert id(version_2) not in skips
+
+    async def test_downloaded_higher_revision_blocks_older_revision(self, rss_engine):
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        downloaded_v2 = make_torrent(
+            name="[GroupA] Mushoku Tensei - 12v2 [1080p].mkv", downloaded=True
+        )
+        candidate_v1 = make_torrent(name="[GroupA] Mushoku Tensei - 12v1 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(candidate_v1, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={1: [downloaded_v2]},
+        )
+
+        assert id(candidate_v1) in skips
+
+    async def test_newer_revision_replaces_downloaded_older_revision(self, rss_engine):
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        downloaded_v1 = make_torrent(
+            name="[GroupA] Mushoku Tensei - 12v1 [1080p].mkv", downloaded=True
+        )
+        candidate_v2 = make_torrent(name="[GroupA] Mushoku Tensei - 12v2 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(candidate_v2, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={1: [downloaded_v1]},
+        )
+
+        assert id(candidate_v2) not in skips
+
+    async def test_preference_score_outranks_higher_revision(self, rss_engine):
+        """Revision only breaks ties; it must never override user preference."""
+        bangumi = make_bangumi(
+            title_raw="Mushoku Tensei", filter="", id=1, preferred_group="GroupA"
+        )
+        preferred_v1 = make_torrent(name="[GroupA] Mushoku Tensei - 12v1 [1080p].mkv")
+        other_v9 = make_torrent(name="[GroupB] Mushoku Tensei - 12v9 [1080p].mkv")
+
+        skips = RSSEngine._select_preference_skips(
+            [(preferred_v1, bangumi), (other_v9, bangumi)],
+            preference_bangumi={1: bangumi},
+            existing_downloaded={},
+        )
+
+        assert id(other_v9) in skips
+        assert id(preferred_v1) not in skips
+
 
 class TestRefreshRssPreferenceDedup:
     """Integration coverage through refresh_rss end to end."""

backend/src/test/test_title_parser.py

@@ -1,13 +1,15 @@
 from types import SimpleNamespace
+from typing import Literal
 from unittest.mock import AsyncMock, patch
 
 import pytest
 
 from module.conf import settings
-from module.models import Bangumi
+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.title_parser import TitleParser, _llm_config
 
 RAW_TITLE = (
@@ -31,6 +33,10 @@ def _make_llm_episode(episode_type: str = "episode") -> Episode:
     )
 
 
+def _empty_parse_outcome(raw: str = "") -> ParseOutcome:
+    return ParseOutcome(result=None, trace=ParseTrace(raw=raw, normalized=raw))
+
+
 class TestTitleParser:
     async def test_parse_without_llm(self):
         result = await TitleParser.raw_parser(RAW_TITLE)
@@ -83,7 +89,8 @@ class TestRawParserLLMFallbackMode:
             settings, "llm", LLM(enable=True, api_key="k", mode="fallback")
         ):
             with patch(
-                "module.parser.title_parser.raw_parser", return_value=None
+                "module.parser.title_parser.parse_release_title_with_trace",
+                return_value=_empty_parse_outcome(),
             ) as mock_raw:
                 with patch(
                     "module.parser.title_parser._llm_parse",
@@ -101,7 +108,10 @@ class TestRawParserLLMFallbackMode:
     async def test_regex_failure_llm_disabled_returns_none(self):
         """正则失败且 LLM 未启用时直接返回 None。"""
         with patch.object(settings, "llm", LLM(enable=False, mode="fallback")):
-            with patch("module.parser.title_parser.raw_parser", return_value=None):
+            with patch(
+                "module.parser.title_parser.parse_release_title_with_trace",
+                return_value=_empty_parse_outcome(),
+            ):
                 with patch(
                     "module.parser.title_parser._llm_parse", new_callable=AsyncMock
                 ) as mock_llm:
@@ -117,8 +127,8 @@ class TestRawParserLLMPrimaryMode:
     def teardown_method(self):
         title_parser_module.reset_cache()
 
-    async def test_llm_called_first_and_regex_skipped(self):
-        """LLM 成功时结果直接采用，正则解析不再执行。"""
+    async def test_llm_keeps_primary_titles_and_reads_deterministic_hints(self):
+        """LLM 标题优先，同时确定性解析仍提供安全的结构提示。"""
         with patch.object(
             settings, "llm", LLM(enable=True, api_key="k", mode="primary")
         ):
@@ -127,11 +137,14 @@ class TestRawParserLLMPrimaryMode:
                 new_callable=AsyncMock,
                 return_value=_make_llm_episode(),
             ) as mock_llm:
-                with patch("module.parser.title_parser.raw_parser") as mock_raw:
+                with patch(
+                    "module.parser.title_parser.parse_release_title_with_trace",
+                    return_value=_empty_parse_outcome(),
+                ) as mock_raw:
                     result = await TitleParser.raw_parser(RAW_TITLE)
 
         mock_llm.assert_awaited_once()
-        mock_raw.assert_not_called()
+        mock_raw.assert_called_once_with(RAW_TITLE)
         assert result is not None
         assert result.official_title == "LLM Title"
 
@@ -147,8 +160,7 @@ class TestRawParserLLMPrimaryMode:
                 result = await TitleParser.raw_parser("[Group] LLM Title Movie [1080P]")
 
         assert result is not None
-        assert isinstance(result, Bangumi)
-        assert result.episode_type == "movie"
+        assert isinstance(result, Movie)
 
     async def test_llm_failure_falls_back_to_regex(self):
         """LLM 失败（返回 None）时用正则兜底，标题不丢失。"""
@@ -175,12 +187,219 @@ class TestRawParserLLMPrimaryMode:
                 new_callable=AsyncMock,
                 return_value=None,
             ):
-                with patch("module.parser.title_parser.raw_parser", return_value=None):
+                with patch(
+                    "module.parser.title_parser.parse_release_title_with_trace",
+                    return_value=_empty_parse_outcome(),
+                ):
                     result = await TitleParser.raw_parser("unparseable garbage title")
 
         assert result is None
 
 
+class TestParsedReleaseAdmission:
+    """Generic parsing and database admission stay separate and typed."""
+
+    @pytest.mark.parametrize("marker", ("OVA01", "OAD01", "SP01"))
+    async def test_numbered_special_projects_to_special_bangumi(self, marker: str):
+        with patch.object(settings, "llm", LLM(enable=False)):
+            result = await TitleParser.raw_parser(
+                f"[Group] Anime Title {marker} [1080p WEB-DL]"
+            )
+
+        assert isinstance(result, Bangumi)
+        assert result.season == 0
+        assert result.episode_type == "special"
+
+    @pytest.mark.parametrize("marker", ("OVA", "OAD", "SP"))
+    async def test_unnumbered_special_projects_without_episode_sentinel(
+        self, marker: str
+    ):
+        with patch.object(settings, "llm", LLM(enable=False)):
+            result = await TitleParser.raw_parser(
+                f"[Group] Anime Title {marker} [1080p WEB-DL]"
+            )
+
+        assert isinstance(result, Bangumi)
+        assert result.season == 0
+        assert result.episode_type == "special"
+
+    async def test_half_episode_projects_without_integer_coercion(self):
+        with patch.object(settings, "llm", LLM(enable=False)):
+            result = await TitleParser.raw_parser(
+                "[Group] Anime Title - 12.5 [1080p WEB-DL]"
+            )
+
+        assert isinstance(result, Bangumi)
+        assert result.eps_collect is False
+
+    async def test_deterministic_movie_projects_to_movie_and_keeps_year(self):
+        with patch.object(settings, "llm", LLM(enable=False)):
+            result = await TitleParser.raw_parser(
+                "[Group] 劇場版 Anime Title 2024 [1080p BluRay]"
+            )
+
+        assert isinstance(result, Movie)
+        assert result.year == 2024
+
+    @pytest.mark.parametrize(
+        "raw",
+        (
+            "[Group] Anime Title [01-12] [1080p]",
+            "[Group] Anime Title Complete Batch [1080p]",
+            "[Group] Anime Title 全集 [1080p]",
+            "[Group] Anime Title PV2 [1080p]",
+            "[Group] Anime Title NCOP [1080p]",
+            "[Group] Anime Title NCED [1080p]",
+        ),
+    )
+    async def test_recognized_non_single_or_auxiliary_video_is_not_persisted(
+        self, raw: str
+    ):
+        with patch.object(settings, "llm", LLM(enable=False)):
+            assert await TitleParser.raw_parser(raw) is None
+
+    async def test_strong_title_only_release_is_admitted_but_bare_title_is_not(self):
+        with patch.object(settings, "llm", LLM(enable=False)):
+            strong = await TitleParser.raw_parser(
+                "[Group] Plain Anime Title [1080p WEB-DL]"
+            )
+            weak = await TitleParser.raw_parser("Plain Anime Title")
+
+        assert isinstance(strong, Bangumi)
+        assert weak is None
+
+    async def test_llm_movie_uses_same_movie_projection_as_deterministic_path(self):
+        with patch.object(
+            settings, "llm", LLM(enable=True, api_key="k", mode="primary")
+        ):
+            with patch(
+                "module.parser.title_parser._llm_parse",
+                new_callable=AsyncMock,
+                return_value=_make_llm_episode(episode_type="movie"),
+            ):
+                result = await TitleParser.raw_parser("Opaque LLM Movie Title")
+
+        assert isinstance(result, Movie)
+
+    async def test_primary_llm_cannot_override_explicit_episode_from_movie_group(self):
+        raw = "[Movie-Fan] Ordinary Anime - 01 [1080p WEB-DL]"
+        with patch.object(
+            settings, "llm", LLM(enable=True, api_key="k", mode="primary")
+        ):
+            with patch(
+                "module.parser.title_parser._llm_parse",
+                new_callable=AsyncMock,
+                return_value=_make_llm_episode(episode_type="movie"),
+            ):
+                result = await TitleParser.raw_parser(raw)
+
+        assert isinstance(result, Bangumi)
+        assert result.episode_type == "episode"
+
+    async def test_deterministic_episode_bundle_stays_atomic_in_primary_mode(self):
+        raw = "[Group] Anime Title S02E12v2 [1080p WEB-DL]"
+        with patch.object(
+            settings, "llm", LLM(enable=True, api_key="k", mode="primary")
+        ):
+            with patch(
+                "module.parser.title_parser._llm_parse",
+                new_callable=AsyncMock,
+                return_value=_make_llm_episode(),
+            ):
+                result = await TitleParser.raw_parser(raw)
+
+        assert isinstance(result, Bangumi)
+        assert result.season == 2
+        assert result.eps_collect is False
+
+    async def test_deterministic_ova_beats_wrong_llm_movie_type(self):
+        raw = "[Group] Anime Title OVA [1080p WEB-DL]"
+        with patch.object(
+            settings, "llm", LLM(enable=True, api_key="k", mode="primary")
+        ):
+            with patch(
+                "module.parser.title_parser._llm_parse",
+                new_callable=AsyncMock,
+                return_value=_make_llm_episode(episode_type="movie"),
+            ):
+                result = await TitleParser.raw_parser(raw)
+
+        assert isinstance(result, Bangumi)
+        assert result.season == 0
+        assert result.episode_type == "special"
+
+    @pytest.mark.parametrize(
+        "raw",
+        (
+            "[Group] Anime Title [01-12] [1080p]",
+            "[Group] Anime Title PV2 [1080p]",
+            "[Group] Anime Title Complete Batch [1080p]",
+            "[Group] Anime Title 合集 [1080p]",
+        ),
+    )
+    async def test_primary_llm_cannot_bypass_deterministic_admission(self, raw: str):
+        with patch.object(
+            settings, "llm", LLM(enable=True, api_key="k", mode="primary")
+        ):
+            with patch(
+                "module.parser.title_parser._llm_parse",
+                new_callable=AsyncMock,
+                return_value=_make_llm_episode(),
+            ):
+                result = await TitleParser.raw_parser(raw)
+
+        assert result is None
+
+    async def test_fallback_does_not_call_llm_for_known_rejected_kind(self):
+        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]")
+
+        assert result is None
+        mock_llm.assert_not_awaited()
+
+    @pytest.mark.parametrize("mode", ("primary", "fallback"))
+    @pytest.mark.parametrize(
+        "raw",
+        (
+            "[Group] PV [1080p]",
+            "[Group] [01-12] [1080p]",
+            "[Group] OVA [1080p]",
+        ),
+    )
+    async def test_titleless_structured_resource_never_reaches_llm(
+        self, mode: Literal["primary", "fallback"], raw: str
+    ):
+        with patch.object(settings, "llm", LLM(enable=True, api_key="k", mode=mode)):
+            with patch(
+                "module.parser.title_parser._llm_parse",
+                new_callable=AsyncMock,
+                return_value=_make_llm_episode(),
+            ) as mock_llm:
+                result = await TitleParser.raw_parser(raw)
+
+        assert result is None
+        mock_llm.assert_not_awaited()
+
+    async def test_real_weak_title_uses_llm_fallback(self):
+        with patch.object(
+            settings, "llm", LLM(enable=True, api_key="k", mode="fallback")
+        ):
+            with patch(
+                "module.parser.title_parser._llm_parse",
+                new_callable=AsyncMock,
+                return_value=_make_llm_episode(),
+            ) as mock_llm:
+                result = await TitleParser.raw_parser("Bare Weak Anime Title")
+
+        assert isinstance(result, Bangumi)
+        mock_llm.assert_awaited_once()
+
+
 class TestLLMParseErrorHandling:
     """_llm_parse 对 LLM 异常/坏输出的容错：一律返回 None，不向上抛。"""