Query Bangumi collection torrent coverage

README.md

@@ -70,8 +70,10 @@ The MCP server provides tools to:
 - update Bangumi subjects and add, move, group, or remove torrent hashes
 - resolve torrent hashes to live qBittorrent names and metadata
 - list aggregates using SQLite filters
+- list mirrored Bangumi collection subjects by collection type and local torrent
+  coverage; the default reports wish/doing subjects without torrents
 - search aggregates semantically
-- run configured aggregate audits
+- run configured aggregate audits, including collection coverage checks
 - synchronize the Bangumi collection mirror and search index, then run audits
 - check health and rebuild the search index directly
 
@@ -150,6 +152,7 @@ To enable OpenCode to directly interact with Bangumi,
   - [x] Expose aggregate collection summary resource
 - [ ] Bangumi integration
   - [x] Mirror a configured user's anime collections with TTL-controlled sync
+  - [x] Query and audit mirrored subjects by local torrent coverage
   - [ ] Synchronize missing Bangumi collections to local catalog
 - [ ] Torrent management
   - [x] Group torrent hashes within aggregates

env.example

@@ -16,8 +16,10 @@ BANGUMI_COLLECTION_TTL_SECONDS=21600
 
 # Aggregate defaults and audit selection
 AGGREGATE_CATEGORY=anime
-AUDIT_CHECKS=torrent_mapping
+AUDIT_CHECKS=torrent_mapping,collection_coverage
 AUDIT_CATEGORIES=anime,RSS,prowlarr
+AUDIT_BANGUMI_COLLECTION_TYPES=wish,doing
+AUDIT_BANGUMI_COLLECTION_LOCAL_STATES=unmapped,empty
 
 # Semantic search
 EMBEDDING_MODEL=Qwen/Qwen3-Embedding-0.6B

src/commands/mcp_server.py

@@ -13,6 +13,11 @@
 from lib.models import ResponsePayload
 from lib.models.aggregates import Aggregate
 from lib.models.audit import AuditReport
+from lib.models.bangumi import (
+    BangumiCollectionLocalState,
+    BangumiCollectionSubjectCoverage,
+    BangumiCollectionType,
+)
 from lib.models.health import HealthCheckReport
 from lib.models.qbittorrent import QbittorrentTorrent
 from lib.models.search import AggregateSearchResults, SearchIndexRebuildResult
@@ -159,6 +164,26 @@ def list_aggregates(
             aggregates,
         )
 
+    @mcp.tool
+    @health_gated
+    def list_bangumi_collection_subjects(
+        collection_types: list[BangumiCollectionType] | None = None,
+        local_states: list[BangumiCollectionLocalState] | None = None,
+    ) -> ResponsePayload[list[BangumiCollectionSubjectCoverage]]:
+        """List mirrored Bangumi collection subjects by local torrent coverage.
+
+        By default, return wish and doing subjects that either have no Aggregate
+        mapping or only map to Aggregates without torrents. Pass collection_types
+        and local_states to compose other queries, such as on-hold or dropped
+        subjects that still have tracked torrents. Only current collection rows are
+        returned; entries removed from the remote collection are excluded.
+        """
+        subjects = context.indexed.list_bangumi_collection_subjects(
+            collection_types,
+            local_states,
+        )
+        return success("Listed Bangumi collection subjects", subjects)
+
     @mcp.tool
     def get_torrent_info(
         hashes: list[str],

src/config.py

@@ -8,6 +8,13 @@
 
 from dotenv import dotenv_values
 
+from lib.models.bangumi import (
+    DEFAULT_BANGUMI_COLLECTION_LOCAL_STATES,
+    DEFAULT_BANGUMI_COLLECTION_TYPES,
+    BangumiCollectionLocalState,
+    BangumiCollectionType,
+)
+
 PROJECT_ROOT = Path(__file__).resolve().parents[1]
 
 ENVIRON_SOURCES = [
@@ -167,6 +174,8 @@ class Config:
     aggregate_category: str
     audit_checks: tuple[str, ...]
     audit_categories: tuple[str, ...]
+    audit_bangumi_collection_types: tuple[BangumiCollectionType, ...]
+    audit_bangumi_collection_local_states: tuple[BangumiCollectionLocalState, ...]
     bangumi: BangumiConfig
     qbittorrent: QbittorrentConfig
     search: SearchConfig
@@ -179,7 +188,9 @@ def from_env(cls, environs: dict[str, str]) -> Config:
             aggregate_category=environs.get("AGGREGATE_CATEGORY", "anime"),
             audit_checks=tuple(
                 check.strip()
-                for check in environs.get("AUDIT_CHECKS", "torrent_mapping").split(",")
+                for check in environs.get(
+                    "AUDIT_CHECKS", "torrent_mapping,collection_coverage"
+                ).split(",")
                 if check.strip()
             ),
             audit_categories=tuple(
@@ -189,6 +200,26 @@ def from_env(cls, environs: dict[str, str]) -> Config:
                 ).split(",")
                 if category.strip()
             ),
+            audit_bangumi_collection_types=parse_bangumi_collection_types(
+                environs.get(
+                    "AUDIT_BANGUMI_COLLECTION_TYPES",
+                    ",".join(
+                        collection_type.name.lower()
+                        for collection_type in DEFAULT_BANGUMI_COLLECTION_TYPES
+                    ),
+                )
+            ),
+            audit_bangumi_collection_local_states=(
+                parse_bangumi_collection_local_states(
+                    environs.get(
+                        "AUDIT_BANGUMI_COLLECTION_LOCAL_STATES",
+                        ",".join(
+                            local_state.value
+                            for local_state in (DEFAULT_BANGUMI_COLLECTION_LOCAL_STATES)
+                        ),
+                    )
+                )
+            ),
             bangumi=BangumiConfig.from_env(environs),
             qbittorrent=QbittorrentConfig.from_env(environs),
             search=SearchConfig.from_env(environs),
@@ -198,3 +229,39 @@ def from_env(cls, environs: dict[str, str]) -> Config:
 
 def load_config() -> Config:
     return Config.from_env(load_environs())
+
+
+def parse_bangumi_collection_types(
+    value: str,
+) -> tuple[BangumiCollectionType, ...]:
+    try:
+        collection_types = tuple(
+            BangumiCollectionType[item.strip().upper()]
+            for item in value.split(",")
+            if item.strip()
+        )
+    except KeyError as exc:
+        raise ValueError(
+            "AUDIT_BANGUMI_COLLECTION_TYPES must contain collection type names."
+        ) from exc
+    if not collection_types:
+        raise ValueError("AUDIT_BANGUMI_COLLECTION_TYPES cannot be empty.")
+    return collection_types
+
+
+def parse_bangumi_collection_local_states(
+    value: str,
+) -> tuple[BangumiCollectionLocalState, ...]:
+    try:
+        local_states = tuple(
+            BangumiCollectionLocalState(item.strip().lower())
+            for item in value.split(",")
+            if item.strip()
+        )
+    except ValueError as exc:
+        raise ValueError(
+            "AUDIT_BANGUMI_COLLECTION_LOCAL_STATES must contain local state names."
+        ) from exc
+    if not local_states:
+        raise ValueError("AUDIT_BANGUMI_COLLECTION_LOCAL_STATES cannot be empty.")
+    return local_states

src/lib/audit/checks/__init__.py

@@ -1,3 +1,4 @@
+from lib.audit.checks.collection_coverage import CollectionCoverageAuditor
 from lib.audit.checks.torrent_mapping import TorrentMappingAuditor
 
-__all__ = ["TorrentMappingAuditor"]
+__all__ = ["CollectionCoverageAuditor", "TorrentMappingAuditor"]

src/lib/audit/checks/collection_coverage.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+from lib.audit.context import AuditContext
+from lib.models.audit import AuditFinding, AuditSeverity
+from lib.models.bangumi import (
+    DEFAULT_BANGUMI_COLLECTION_LOCAL_STATES,
+    DEFAULT_BANGUMI_COLLECTION_TYPES,
+    BangumiCollectionLocalState,
+    BangumiCollectionSubjectCoverage,
+    BangumiCollectionType,
+)
+
+if TYPE_CHECKING:
+    from pydantic import JsonValue
+
+
+class CollectionCoverageAuditor:
+    name = "collection_coverage"
+
+    def __init__(
+        self,
+        collection_types: tuple[BangumiCollectionType, ...] = (
+            DEFAULT_BANGUMI_COLLECTION_TYPES
+        ),
+        local_states: tuple[BangumiCollectionLocalState, ...] = (
+            DEFAULT_BANGUMI_COLLECTION_LOCAL_STATES
+        ),
+    ) -> None:
+        self.collection_types = collection_types
+        self.local_states = local_states
+
+    def audit(self, context: AuditContext) -> list[AuditFinding]:
+        return [
+            self._finding(coverage)
+            for coverage in context.get_collection_subject_coverage(
+                self.collection_types,
+                self.local_states,
+            )
+        ]
+
+    def _finding(
+        self,
+        coverage: BangumiCollectionSubjectCoverage,
+    ) -> AuditFinding:
+        snapshot = coverage.subject.snapshot
+        subject_name = snapshot.name if snapshot is not None else ""
+        subject_name_cn = snapshot.name_cn if snapshot is not None else ""
+        aggregate_names = [aggregate.short_name for aggregate in coverage.aggregates]
+        match coverage.local_state:
+            case BangumiCollectionLocalState.UNMAPPED:
+                code = "collection.unmapped"
+                message = "Collected Bangumi subject is not mapped to an aggregate."
+                severity = AuditSeverity.WARNING
+            case BangumiCollectionLocalState.EMPTY:
+                code = "collection.empty"
+                message = "Collected Bangumi subject has no tracked torrents."
+                severity = AuditSeverity.WARNING
+            case BangumiCollectionLocalState.WITH_TORRENTS:
+                code = "collection.with_torrents"
+                message = "Collected Bangumi subject has tracked torrents."
+                severity = AuditSeverity.INFO
+        return AuditFinding(
+            auditor=self.name,
+            code=code,
+            severity=severity,
+            message=message,
+            aggregate_short_name=(
+                aggregate_names[0] if len(aggregate_names) == 1 else None
+            ),
+            metadata={
+                "subject_id": coverage.subject.subject_id,
+                "subject_name": subject_name,
+                "subject_name_cn": subject_name_cn,
+                "collection_type": coverage.collection_type.name.lower(),
+                "local_state": coverage.local_state.value,
+                "aggregates": cast("JsonValue", aggregate_names),
+                "torrent_count": coverage.torrent_count,
+            },
+        )

src/lib/audit/context.py

@@ -2,8 +2,14 @@
 
 from dataclasses import dataclass, field
 
-from lib.audit.protocols import AuditQbittorrentClient
+from lib.audit.exceptions import AuditSkipped
+from lib.audit.protocols import AuditQbittorrentClient, CollectionCoverageProvider
 from lib.models.aggregates import Aggregate
+from lib.models.bangumi import (
+    BangumiCollectionLocalState,
+    BangumiCollectionSubjectCoverage,
+    BangumiCollectionType,
+)
 from lib.models.qbittorrent import QbittorrentTorrent, QbittorrentTorrentFile
 from lib.sql.repositories import AggregateRepository
 
@@ -13,6 +19,7 @@ class AuditContext:
     repository: AggregateRepository
     qbit: AuditQbittorrentClient
     categories: tuple[str, ...]
+    collection_coverage_provider: CollectionCoverageProvider | None = None
     _aggregates: tuple[Aggregate, ...] | None = field(default=None, init=False)
     _qbit_torrents: tuple[QbittorrentTorrent, ...] | None = field(
         default=None,
@@ -22,6 +29,13 @@ class AuditContext:
         default_factory=dict,
         init=False,
     )
+    _collection_coverages: dict[
+        tuple[
+            tuple[BangumiCollectionType, ...],
+            tuple[BangumiCollectionLocalState, ...],
+        ],
+        tuple[BangumiCollectionSubjectCoverage, ...],
+    ] = field(default_factory=dict, init=False)
     _qbit_authenticated: bool = field(default=False, init=False)
 
     def get_aggregates(self) -> tuple[Aggregate, ...]:
@@ -36,6 +50,23 @@ def get_qbittorrent_torrents(self) -> tuple[QbittorrentTorrent, ...]:
             self._qbit_torrents = tuple(self.qbit.get_all_torrents())
         return self._qbit_torrents
 
+    def get_collection_subject_coverage(
+        self,
+        collection_types: tuple[BangumiCollectionType, ...],
+        local_states: tuple[BangumiCollectionLocalState, ...],
+    ) -> tuple[BangumiCollectionSubjectCoverage, ...]:
+        if self.collection_coverage_provider is None:
+            raise AuditSkipped("BANGUMI_USERNAME is not configured.")
+        key = (collection_types, local_states)
+        if key not in self._collection_coverages:
+            self._collection_coverages[key] = tuple(
+                self.collection_coverage_provider.list_subject_coverage(
+                    collection_types,
+                    local_states,
+                )
+            )
+        return self._collection_coverages[key]
+
     def get_torrent_files(
         self,
         torrent_hash: str,

src/lib/audit/factory.py

@@ -2,11 +2,21 @@
 
 from collections.abc import Callable
 
-from lib.audit.checks import TorrentMappingAuditor
+from lib.audit.checks import CollectionCoverageAuditor, TorrentMappingAuditor
 from lib.audit.context import AuditContext
-from lib.audit.protocols import AggregateAuditor, AuditQbittorrentClient
+from lib.audit.protocols import (
+    AggregateAuditor,
+    AuditQbittorrentClient,
+    CollectionCoverageProvider,
+)
 from lib.audit.runner import AuditRunner
 from lib.models.audit import AuditFinding
+from lib.models.bangumi import (
+    DEFAULT_BANGUMI_COLLECTION_LOCAL_STATES,
+    DEFAULT_BANGUMI_COLLECTION_TYPES,
+    BangumiCollectionLocalState,
+    BangumiCollectionType,
+)
 from lib.sql.repositories import AggregateRepository
 
 type AuditorFactory = Callable[[], AggregateAuditor]
@@ -30,17 +40,32 @@ def create_audit_runner(
     *,
     categories: tuple[str, ...],
     auditor_names: tuple[str, ...],
+    collection_coverage_provider: CollectionCoverageProvider | None = None,
+    collection_types: tuple[BangumiCollectionType, ...] = (
+        DEFAULT_BANGUMI_COLLECTION_TYPES
+    ),
+    collection_local_states: tuple[BangumiCollectionLocalState, ...] = (
+        DEFAULT_BANGUMI_COLLECTION_LOCAL_STATES
+    ),
 ) -> AuditRunner:
     context = AuditContext(
         repository=repository,
         qbit=qbit,
         categories=categories,
+        collection_coverage_provider=collection_coverage_provider,
     )
+    auditor_factories = {
+        **AUDITOR_FACTORIES,
+        CollectionCoverageAuditor.name: lambda: CollectionCoverageAuditor(
+            collection_types,
+            collection_local_states,
+        ),
+    }
     return AuditRunner(
         context,
         [
-            AUDITOR_FACTORIES[name]()
-            if name in AUDITOR_FACTORIES
+            auditor_factories[name]()
+            if name in auditor_factories
             else UnknownAuditor(name)
             for name in auditor_names
         ],