From a20b99a1fd7c2b75649d36ba6ee86de721f74733 Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 16 Oct 2026 13:59:09 -0700 Subject: [PATCH 01/46] fix(metadata): trust validated catalogs over stale summary counts --- .../services/metadata_series_adoption.py | 16 ++++--- tests/api/test_source_series_add_api.py | 46 +++++++++++++++++-- .../metadata_identity/test_series_adoption.py | 16 +++++++ .../metadata_identity/test_series_refresh.py | 33 ++++++++++--- .../test_metadata_series_adoption_fetch.py | 42 +++++++++++++---- 5 files changed, 130 insertions(+), 23 deletions(-) diff --git a/src/pullbox/services/metadata_series_adoption.py b/src/pullbox/services/metadata_series_adoption.py index f3a01d4c..0395c494 100644 --- a/src/pullbox/services/metadata_series_adoption.py +++ b/src/pullbox/services/metadata_series_adoption.py @@ -163,8 +163,6 @@ async def fetch_source_series_bundle( raise SeriesAdoptionError( "The series profile belongs to a different source identity." ) - if profile.issue_count is not None and profile.issue_count > max_issues: - raise SeriesAdoptionError("The issue catalog exceeds the interactive add limit.") issues: list[ProviderIssueRead] = [] seen: set[str] = set() page = 1 @@ -181,9 +179,8 @@ async def fetch_source_series_bundle( current = result.data if total is None: total = current.total - if current.total != total or ( - profile.issue_count is not None and profile.issue_count != total - ): + # Summary counts can lag; completeness comes from the paginated catalog. + if current.total != total: raise SeriesCatalogCountError( "The provider catalog count changed. Retry the add." ) @@ -208,6 +205,14 @@ async def fetch_source_series_bundle( if current.next_page is None: if len(issues) != total: raise SeriesAdoptionError("The issue catalog is incomplete. Retry the add.") + if profile.issue_count is not None and profile.issue_count != total: + logger.info( + "metadata_series_summary_count_disagreement", + source=source.value, + external_id=identifier, + summary_count=profile.issue_count, + catalog_count=total, + ) return SourceSeriesBundle( profile, tuple(issues), source_revision, total, started_at ) @@ -285,7 +290,6 @@ def _validate_bundle(bundle: SourceSeriesBundle) -> list[tuple[float, str]]: or len(catalog) > 10000 or type(bundle.catalog_total) is not int or bundle.catalog_total != len(catalog) - or (profile.issue_count is not None and profile.issue_count != len(catalog)) ): raise SeriesAdoptionError("The series profile or issue catalog is incomplete.") if bundle.excluded_issues: diff --git a/tests/api/test_source_series_add_api.py b/tests/api/test_source_series_add_api.py index 64ed1966..0e947088 100644 --- a/tests/api/test_source_series_add_api.py +++ b/tests/api/test_source_series_add_api.py @@ -43,13 +43,12 @@ async def source_add_setup(authenticated_client, sec_db, monkeypatch, tmp_path): root_id = root.id calls = [] responses = [] + profile = {**series_row(8), "name": "Fixture 8", "cv_id": None, "gcd_id": None} def handle(request): calls.append(request) if request.url.path == "/api/series/8/": - return httpx.Response( - 200, json={**series_row(8), "name": "Fixture 8", "cv_id": None, "gcd_id": None} - ) + return httpx.Response(200, json=profile) assert request.url.path == "/api/series/8/issue_list/" if responses: return responses.pop(0) @@ -106,6 +105,7 @@ async def covers_dir(_session): }, "calls": calls, "responses": responses, + "profile": profile, "events": events, "root": root_path, } @@ -136,6 +136,46 @@ async def test_add_native_series_uses_full_server_catalog_and_is_repeat_safe( assert await session.scalar(select(func.count()).select_from(Issue)) == 3 +@pytest.mark.parametrize("invalid_after_preview", [False, True]) +async def test_stale_summary_preview_and_final_add_revalidate_the_catalog( + authenticated_client, source_add_setup, sec_db, invalid_after_preview +): + fixture = source_add_setup + fixture["profile"]["issue_count"] = 1 + complete = envelope([issue_row(100, "1"), issue_row(101, "2")]) + fixture["responses"].append(httpx.Response(200, json=complete)) + preview = await authenticated_client.post( + "/api/v1/metadata/series/preview", + json={key: value for key, value in fixture["body"].items() if key != "source_revision"}, + headers=csrf(authenticated_client), + ) + assert preview.status_code == 200, preview.text + assert preview.json()["issues"]["data"]["total"] == 2 + assert preview.json()["series"]["data"]["issue_count"] == 1 + assert not list(fixture["root"].iterdir()) + latest = {**complete, "count": 3} if invalid_after_preview else complete + fixture["responses"].append(httpx.Response(200, json=latest)) + added = await authenticated_client.post( + "/api/v1/series", json=fixture["body"], headers=csrf(authenticated_client) + ) + assert added.status_code == (409 if invalid_after_preview else 201), added.text + assert [call.url.path for call in fixture["calls"]] == [ + "/api/series/8/", + "/api/series/8/issue_list/", + "/api/series/8/", + "/api/series/8/issue_list/", + ] + async with sec_db() as session: + assert await session.scalar(select(func.count()).select_from(Issue)) == ( + 0 if invalid_after_preview else 2 + ) + if invalid_after_preview: + assert not fixture["events"] and not list(fixture["root"].iterdir()) + else: + assert added.json()["issue_count"] == 2 + assert len(fixture["events"]) == 1 + + @pytest.mark.parametrize( "changes", [ diff --git a/tests/integration/metadata_identity/test_series_adoption.py b/tests/integration/metadata_identity/test_series_adoption.py index 5e6f3579..55f0c7d0 100644 --- a/tests/integration/metadata_identity/test_series_adoption.py +++ b/tests/integration/metadata_identity/test_series_adoption.py @@ -115,6 +115,22 @@ async def test_new_add_returns_canonical_snapshots_matching_persisted_metadata(i assert item.values.issue_number_text == issue.issue_number_text +async def test_stale_summary_adopts_complete_catalog_and_derives_saved_count(identity_probe_db): + _, factory, _ = identity_probe_db + data = bundle(numbers=("1", "2")) + data.series.issue_count = 1 + async with factory.begin() as session: + try: + result = await adopt_source_series_bundle(session, data) + except SeriesAdoptionError as exc: + pytest.fail(f"A stale summary cannot veto the complete catalog: {exc}") + assert result.series.issue_count == result.snapshot.values.issue_count == 2 + origin = next(item for item in result.snapshot.origins if item.field == "issue_count") + assert origin.derivation == "catalog" + assert await session.scalar(select(func.count()).select_from(Issue)) == 2 + assert data.series.issue_count == 1 + + async def test_representative_metron_cover_survives_add_and_session_reload(identity_probe_db): _, factory, _ = identity_probe_db adapter = CatalogAdapter(1) diff --git a/tests/integration/metadata_identity/test_series_refresh.py b/tests/integration/metadata_identity/test_series_refresh.py index 7372c0c4..e141b420 100644 --- a/tests/integration/metadata_identity/test_series_refresh.py +++ b/tests/integration/metadata_identity/test_series_refresh.py @@ -841,7 +841,7 @@ def from_source(source): return factory, series_id, original, adapters, instance -async def test_refresh_falls_back_from_stale_local_count_to_exact_api_catalog( +async def test_refresh_accepts_complete_catalog_with_stale_summary_without_fallback( comicvine_count_disagreement, ): from pullbox.services.metadata_series_refresh import refresh_series_catalog_from_sources @@ -853,9 +853,7 @@ async def test_refresh_falls_back_from_stale_local_count_to_exact_api_catalog( assert result.series.issue_count == 5 assert result.created_issue_ids == () assert result.series.path == "/reference/Civil War - Unmasked" - assert [(o.source, o.status) for o in result.outcomes] == [ - (Source.COMICVINE_LOCAL, SourceStatus.INCOMPATIBLE_RESPONSE) - ] + assert result.outcomes == () await session.commit() async with factory() as session: issues = list(await session.scalars(select(Issue).order_by(Issue.id))) @@ -863,14 +861,37 @@ async def test_refresh_falls_back_from_stale_local_count_to_exact_api_catalog( assert set(await session.scalars(select(IssueExternalIdentity.identity_namespace))) == { Namespace.COMICVINE } - assert ("issues", "42", 1) in adapters[Source.COMICVINE_API].calls + assert not any(call[0] == "issues" for call in adapters[Source.COMICVINE_API].calls) @pytest.mark.parametrize("fallback", ["disabled", "rate_limited"]) -async def test_count_mismatch_without_consistent_fallback_is_actionable_and_atomic( +async def test_changing_page_total_without_consistent_fallback_is_actionable_and_atomic( comicvine_count_disagreement, fallback ): factory, series_id, original, adapters, instance = comicvine_count_disagreement + local = adapters[Source.COMICVINE_LOCAL] + + async def changing_pages(external_id, *, page=1, validator=None): + start, stop = (0, 100) if page == 1 else (100, 102) + return MetadataFetch( + status=SourceStatus.OK, + data=MetadataPage( + results=[ + local.data.issues[0].model_copy( + update={ + "external_id": str(1000 + index), + "issue_number_text": str(index + 1), + "issue_number_key": str(index + 1), + } + ) + for index in range(start, stop) + ], + total=101 if page == 1 else 102, + next_page=2 if page == 1 else None, + ), + ) + + local.issues = changing_pages if fallback == "disabled": instance.runtime[Source.COMICVINE_API].policy.enabled = False else: diff --git a/tests/unit/test_metadata_series_adoption_fetch.py b/tests/unit/test_metadata_series_adoption_fetch.py index 9606121e..de9ffec4 100644 --- a/tests/unit/test_metadata_series_adoption_fetch.py +++ b/tests/unit/test_metadata_series_adoption_fetch.py @@ -150,20 +150,46 @@ async def test_cross_page_consistency_is_required(mutation): await fetch(adapter) -async def test_series_and_issue_count_disagreement_is_not_complete(): - adapter = CatalogAdapter(0) - adapter.pages[1] = MetadataFetch( - status=SourceStatus.OK, data=MetadataPage(results=[issue_row()], total=1) +@pytest.mark.parametrize("summary_count", [0, 1, 3, 10001, None]) +@pytest.mark.parametrize("catalog_count", [2, 101]) +async def test_stale_summary_does_not_reject_a_complete_catalog(summary_count, catalog_count): + adapter = CatalogAdapter(catalog_count) + profile = row(adapter.source).model_copy(update={"issue_count": summary_count}) + try: + result = await fetch(adapter, profile=profile) + except SeriesAdoptionError as exc: + pytest.fail(f"A complete catalog must not be rejected for its stale summary: {exc}") + assert result.catalog_total == len(result.issues) == catalog_count + assert result.series.issue_count == summary_count, "Keep the original provider evidence" + assert adapter.calls == [ + ("issues", "42", page) for page in range(1, (catalog_count + 99) // 100 + 1) + ] + + +@pytest.mark.parametrize("problem", ["changed_total", "incomplete", "repeat_id", "wrong_parent"]) +async def test_stale_summary_never_excuses_an_invalid_catalog(problem): + adapter = CatalogAdapter(101) + profile = row(adapter.source).model_copy(update={"issue_count": 1}) + item = issue_row(external_id="1100", issue_number_text="101", issue_number_key="101") + total = 102 if problem == "changed_total" else 101 + if problem == "repeat_id": + item.external_id = "1000" + elif problem == "wrong_parent": + item.series_external_id = "99" + adapter.pages[2] = MetadataFetch( + status=SourceStatus.OK, + data=MetadataPage(results=[] if problem == "incomplete" else [item], total=total), ) - with pytest.raises(SeriesAdoptionError, match="count"): - await fetch(adapter) + with pytest.raises(SeriesAdoptionError): + await fetch(adapter, profile=profile) + assert adapter.calls[-1] == ("issues", "42", 2) -async def test_resource_cap_stops_before_issue_traversal(): +async def test_resource_cap_stops_after_first_page_without_traversing_catalog(): adapter = CatalogAdapter(201) with pytest.raises(SeriesAdoptionError, match="limit"): await fetch(adapter, max_issues=200) - assert adapter.calls == [("series", "42")] + assert adapter.calls == [("series", "42"), ("issues", "42", 1)] async def test_total_deadline_bounds_traversal_and_closes_adapter(): From 4235b087c5e4e605df2c6ff25cc7231ebd87f576 Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 16 Oct 2026 14:13:58 -0700 Subject: [PATCH 02/46] fix(metadata): safely extend catalogs from verified linked sources --- .../services/metadata_issue_catalog.py | 34 +- .../services/metadata_series_refresh.py | 77 ++-- .../test_linked_catalog_extension.py | 344 ++++++++++++++++++ 3 files changed, 428 insertions(+), 27 deletions(-) create mode 100644 tests/integration/metadata_identity/test_linked_catalog_extension.py diff --git a/src/pullbox/services/metadata_issue_catalog.py b/src/pullbox/services/metadata_issue_catalog.py index 2a5047e8..a5234b55 100644 --- a/src/pullbox/services/metadata_issue_catalog.py +++ b/src/pullbox/services/metadata_issue_catalog.py @@ -41,6 +41,7 @@ async def apply_issue_batch( now: datetime, *, complete: bool = False, + reconcile_crosswalks: bool = False, replace_managed: bool = False, release_facts: tuple[IssueReleaseFacts, ...] = (), ) -> tuple[int, ...]: @@ -50,6 +51,8 @@ async def apply_issue_batch( checkpoint updates. This writer never starts provider or filesystem work. """ source = batch.source + if reconcile_crosswalks and not complete: + raise IssueCatalogConflictError("Linked issue reconciliation requires a complete catalog.") parent = ExternalIdentityRef( source.identity_namespace, MetadataEntityKind.SERIES, batch.series_external_id ) @@ -67,6 +70,7 @@ async def apply_issue_batch( except ValueError as exc: raise IssueCatalogConflictError(str(exc)) from exc by_identity = {identity: item for item in state.issues for identity in item.identities} + parent_namespaces = {identity.namespace for identity in state.series.identities} offered = { ExternalIdentityRef(source.identity_namespace, MetadataEntityKind.ISSUE, item.external_id) for item in batch.issues @@ -87,6 +91,27 @@ async def apply_issue_batch( source.identity_namespace, MetadataEntityKind.ISSUE, metadata.external_id ) existing = by_identity.get(identity) + if reconcile_crosswalks: + # Native membership plus a verified local crosswalk proves overlap; + # equal issue numbers alone never authorize attachment. + matches = { + by_identity[claim].local_id: by_identity[claim] + for claim in _identities(metadata, MetadataEntityKind.ISSUE) + if claim in by_identity and claim.namespace in parent_namespaces + } + if len(matches) > 1: + raise IssueCatalogConflictError( + "Provider crosswalks identify different library issues. Review their matches." + ) + if matches: + existing = next(iter(matches.values())) + if any( + claim.namespace is identity.namespace and claim != identity + for claim, _, _ in existing.claims + ): + raise IssueCatalogConflictError( + "This issue already has a different provider identity. Review its match." + ) if existing is not None and existing.values.issue_number_text != text: raise IssueCatalogConflictError( "A provider issue was renumbered. Review the issue match; existing files were kept." @@ -98,7 +123,7 @@ async def apply_issue_batch( ) snapshot = assemble_metadata( MetadataEntityKind.ISSUE, - existing.identities if existing else (identity,), + tuple(dict.fromkeys((*existing.identities, identity))) if existing else (identity,), [metadata.model_copy(update={"issue_number_text": text})], state.policies, now=now, @@ -114,6 +139,8 @@ async def apply_issue_batch( ) prepared.append((existing, metadata, number, text, snapshot)) new_metadata = tuple(metadata for old, metadata, *_rest in prepared if old is None) + if len(state.issues) + len(new_metadata) > 10000: + raise IssueCatalogConflictError("The combined catalog exceeds the 10,000 issue limit.") if new_metadata: await _require_unowned_issues(session, replace(batch, issues=new_metadata)) series = await session.get(Series, state.series.local_id) @@ -149,7 +176,12 @@ async def apply_issue_batch( ) for issue, old, metadata, _ in pending if old is None + or ExternalIdentityRef( + source.identity_namespace, MetadataEntityKind.ISSUE, metadata.external_id + ) + not in old.identities ], + require_current_ownership=True, ) for issue, old, metadata, _ in pending: if old is None: diff --git a/src/pullbox/services/metadata_series_refresh.py b/src/pullbox/services/metadata_series_refresh.py index 4fccb893..a2397fd7 100644 --- a/src/pullbox/services/metadata_series_refresh.py +++ b/src/pullbox/services/metadata_series_refresh.py @@ -228,17 +228,17 @@ async def refresh_series_catalog_from_sources( for item in fetched.outcomes if item.status not in {SourceStatus.OK, SourceStatus.NOT_QUERIED} } - bundle = await _catalog(registry, before, profiles, failed) + bundles = await _catalogs(registry, before, profiles, failed) + profiles.update((item.series.source, item.series) for item in bundles) profile = profiles.get(MetadataSource.METRON_API) if ( profile is not None and not profile.image_url - and bundle.series.source is not profile.source + and not any(item.series.source is profile.source for item in bundles) ): page = await registry.issues(profile.source, profile.external_id) if page.status is SourceStatus.OK and page.data is not None: profiles[profile.source] = with_representative_cover(profile, page.data.results) - profiles[bundle.series.source] = bundle.series snapshot = assemble_metadata( MetadataEntityKind.SERIES, before.series.identities, @@ -287,13 +287,15 @@ async def refresh_series_catalog_from_sources( result = await _apply( session, current, - bundle, + bundles, snapshot, now, replace_managed=replace_managed, release_facts=rechecked_issues, ) - if bundle.catalog_started_at is not None: + for bundle in bundles: + if bundle.catalog_started_at is None: + continue checkpoint = next( (item for item in current.checkpoints if item.source is bundle.series.source), None, @@ -319,8 +321,8 @@ async def refresh_series_catalog_from_sources( logger.info( "metadata_series_source_refreshed", series_id=series_id, - source=bundle.series.source.value, - catalog_count=bundle.catalog_total, + sources=[item.series.source.value for item in bundles], + catalog_count=result.series.issue_count, ) return replace(result, outcomes=tuple(failed.values())) except SeriesRefreshError: @@ -341,17 +343,19 @@ async def refresh_series_catalog_from_sources( ) from exc -async def _catalog( +async def _catalogs( registry: MetadataSourceRegistry, state: SeriesRefreshState, profiles: dict[MetadataSource, ProviderSeriesRead], failed: dict[MetadataSource, SourceOutcome], -) -> SourceSeriesBundle: +) -> tuple[SourceSeriesBundle, ...]: known = {item.namespace: item.external_id for item in state.series.identities} inconsistent_counts = False catalog_namespaces = { identity.namespace for issue in state.issues for identity in issue.identities } + bundles: list[SourceSeriesBundle] = [] + loaded_namespaces = set() sources = sorted( registry.runtime, key=lambda source: ( @@ -362,11 +366,20 @@ async def _catalog( ), ) for source in sources: + profile = profiles.get(source) + linked_parent = profile is not None and any( + identity in state.series.identities for identity in profile.cross_identities + ) if ( source in failed or source.identity_namespace not in known - # A series link authorizes descriptive enrichment, not issue rematching. - or (catalog_namespaces and source.identity_namespace not in catalog_namespaces) + or source.identity_namespace in loaded_namespaces + # A manual series link alone cannot authorize automatic issue matching. + or ( + catalog_namespaces + and source.identity_namespace not in catalog_namespaces + and not linked_parent + ) or registry._unavailable(source, capability=SourceCapability.ISSUE_LIST) ): continue @@ -380,10 +393,11 @@ async def _catalog( ) bundle = apply_catalog_exclusions(bundle, state.catalog_exclusions) _validate_bundle(bundle) - return bundle + bundles.append(bundle) + loaded_namespaces.add(source.identity_namespace) except SeriesCatalogCountError: - # Do not accept an incomplete catalog or mistake a stale summary for - # an identity dispute. Another eligible source must pass every check. + # Changing page totals cannot establish completeness. Another eligible + # source must still pass every catalog and identity check. inconsistent_counts = True failed[source] = SourceOutcome(source=source, status=SourceStatus.INCOMPATIBLE_RESPONSE) logger.info( @@ -403,6 +417,8 @@ async def _catalog( failed[source] = SourceOutcome( source=source, status=exc.status, retry_after_seconds=exc.retry_after_seconds ) + if bundles: + return tuple(bundles) if inconsistent_counts: raise SeriesRefreshError( "A metadata source reported inconsistent issue counts, and no configured source " @@ -441,7 +457,7 @@ def _with_derived( async def _apply( session: AsyncSession, state: SeriesRefreshState, - bundle: SourceSeriesBundle, + bundles: tuple[SourceSeriesBundle, ...], snapshot: MetadataSnapshot, now: datetime, *, @@ -451,17 +467,26 @@ async def _apply( series = await session.get(Series, state.series.local_id) assert series is not None await apply_series_metadata_values(session, series, snapshot.values) - created = await apply_issue_batch( - session, - state, - SourceIssueBatch( - bundle.series.source, bundle.series.external_id, bundle.issues, bundle.source_revision - ), - now, - complete=True, - replace_managed=replace_managed, - release_facts=release_facts, - ) + created: tuple[int, ...] = () + current = state + for index, bundle in enumerate(bundles): + if index: + current = await read_series_refresh_state(session, series.id) + created += await apply_issue_batch( + session, + current, + SourceIssueBatch( + bundle.series.source, + bundle.series.external_id, + bundle.issues, + bundle.source_revision, + ), + now, + complete=True, + reconcile_crosswalks=True, + replace_managed=replace_managed, + release_facts=release_facts, + ) series.issue_count = len(state.issues) + len(created) snapshot = _with_derived(snapshot, "issue_count", series.issue_count, now) origins = {item.field: item for item in snapshot.origins} diff --git a/tests/integration/metadata_identity/test_linked_catalog_extension.py b/tests/integration/metadata_identity/test_linked_catalog_extension.py new file mode 100644 index 00000000..b91981bd --- /dev/null +++ b/tests/integration/metadata_identity/test_linked_catalog_extension.py @@ -0,0 +1,344 @@ +"""Linked catalogs extend proven membership without replacing library identity.""" + +from dataclasses import replace +from datetime import UTC, date, datetime, timedelta + +import pytest +from sqlalchemy import func, select, update + +from pullbox.core.metadata_identity import ExternalIdentityRef +from pullbox.core.metadata_identity import IdentityNamespace as Namespace +from pullbox.core.metadata_identity import MetadataEntityKind as Kind +from pullbox.core.metadata_identity import MetadataSource as Source +from pullbox.core.metadata_identity_state import IdentityVerificationAction as Action +from pullbox.models import Issue, Series +from pullbox.models.issue import IssueStatus +from pullbox.models.library import LibraryFile, LibraryRoot +from pullbox.models.metadata_identity import ( + IssueExternalIdentity, + IssueIdentityEvent, + SeriesExternalIdentity, +) +from pullbox.models.metadata_source import MetadataSourceConfig +from pullbox.models.reader import IssueReaderState +from pullbox.models.user import User +from pullbox.schemas.metadata_sources import SourceCapability +from pullbox.services.metadata_discovery import MetadataSourceRegistry +from pullbox.services.metadata_identity_attachment import attach_verified_identities +from pullbox.services.metadata_identity_review import record_identity_observation +from pullbox.services.metadata_series_adoption import _request, adopt_source_series_bundle +from pullbox.services.metadata_series_refresh import ( + SeriesRefreshError, + refresh_series_catalog_from_sources, +) +from tests.integration.metadata_identity.test_identity_review import decide +from tests.integration.metadata_identity.test_series_adoption import ( # noqa: F401 + bundle, + configured_sources, +) +from tests.integration.metadata_identity.test_series_refresh import RefreshAdapter +from tests.unit.test_metadata_discovery import registration, runtime + + +@pytest.fixture +async def linked_catalog(identity_probe_db, tmp_path): + _, factory, _ = identity_probe_db + metron = bundle(numbers=tuple(str(i) for i in range(1, 8))) + metron.series.cross_identities = [ExternalIdentityRef(Namespace.COMICVINE, Kind.SERIES, "500")] + metron.series.description = "Secondary description" + for index, issue in enumerate(metron.issues): + issue.external_id = str(200 + index) + issue.title = f"Secondary issue {index + 1}" + if index < 5: + issue.cross_identities = [ + ExternalIdentityRef(Namespace.COMICVINE, Kind.ISSUE, str(100 + index)) + ] + metron.issues[-1].store_date = date.today() + timedelta(days=30) + cv = replace( + metron, + series=metron.series.model_copy( + update={ + "source": Source.COMICVINE_API, + "identity_namespace": Namespace.COMICVINE, + "external_id": "500", + "issue_count": 5, + "cross_identities": [], + "description": "Preferred description", + } + ), + issues=tuple( + issue.model_copy( + update={ + "source": Source.COMICVINE_API, + "identity_namespace": Namespace.COMICVINE, + "series_external_id": "500", + "external_id": str(100 + index), + "cross_identities": [], + "title": f"Preferred issue {index + 1}", + } + ) + for index, issue in enumerate(metron.issues[:5]) + ), + catalog_total=5, + ) + artifact = tmp_path / "owned.cbz" + artifact.write_bytes(b"owned comic must remain untouched") + async with factory.begin() as session: + result = await adopt_source_series_bundle(session, cv, monitored=True) + series_id = result.series.id + result.series.path = str(tmp_path) + issues = list(await session.scalars(select(Issue).order_by(Issue.id))) + issues[0].status = IssueStatus.OWNED + issues[1].title = "My issue title" + issues[2].status = IssueStatus.SKIPPED + issues[2].manual_skip = True + original = [(i.id, i.comicvine_id, i.status, i.manual_skip) for i in issues] + root = LibraryRoot(name="Reference", path=str(tmp_path)) + user = User(username="catalog-reader", password_hash="not-a-live-account") + session.add_all([root, user]) + await session.flush() + session.add( + LibraryFile( + issue_id=issues[0].id, + library_root_id=root.id, + file_path=str(artifact), + file_name=artifact.name, + file_format="cbz", + file_size=artifact.stat().st_size, + file_modified_at=datetime.now(UTC), + storage_mode="referenced", + ) + ) + session.add( + IssueReaderState( + user_id=user.id, + issue_id=issues[0].id, + last_page_index=17, + page_count=24, + ) + ) + await attach_verified_identities( + session, + [ + _request( + metron, + metron.series, + ExternalIdentityRef(Namespace.METRON, Kind.SERIES, "42"), + series_id, + ) + ], + ) + adapters = {Source.COMICVINE_API: RefreshAdapter(cv), Source.METRON_API: RefreshAdapter(metron)} + for source, adapter in adapters.items(): + adapter.source = source + registry = MetadataSourceRegistry( + [runtime(source, revision=1, priority=index) for index, source in enumerate(adapters, 1)], + factories={ + source: registration(adapter, capabilities=list(SourceCapability)) + for source, adapter in adapters.items() + }, + ) + return factory, series_id, original, artifact, adapters, registry + + +@pytest.mark.parametrize("replace_managed", [False, True]) +async def test_linked_catalog_extends_five_to_seven_preserving_existing_state( + linked_catalog, replace_managed +): + factory, series_id, original, artifact, adapters, registry = linked_catalog + before = artifact.stat() + async with factory() as session: + result = await refresh_series_catalog_from_sources( + session, series_id, registry=registry, replace_managed=replace_managed + ) + assert result.series.issue_count == 7, ( + "The linked catalog must supply the two missing issues" + ) + assert len(result.created_issue_ids) == 2 + assert result.series.description == "Preferred description" + assert result.series.path == str(artifact.parent) + await session.commit() + async with factory() as session: + issues = list(await session.scalars(select(Issue).order_by(Issue.id))) + assert [(i.id, i.comicvine_id, i.status, i.manual_skip) for i in issues[:5]] == original + assert issues[0].title == "Preferred issue 1" + assert issues[1].title == "My issue title" + assert [i.issue_number_text for i in issues] == [str(i) for i in range(1, 8)] + assert issues[-1].store_date == adapters[Source.METRON_API].data.issues[-1].store_date + assert all(i.status is IssueStatus.WANTED for i in issues[5:]) + assert all(i.comicvine_id is None for i in issues[5:]) + claims = list(await session.scalars(select(IssueExternalIdentity))) + assert len(claims) == 12 + assert all(c.verification_state.value == "verified" for c in claims) + reader = await session.scalar(select(IssueReaderState)) + assert reader.last_page_index == 17 and reader.page_count == 24 + assert (await session.scalar(select(LibraryFile))).file_path == str(artifact) + events = await session.scalar(select(func.count()).select_from(IssueIdentityEvent)) + assert artifact.read_bytes() == b"owned comic must remain untouched" + assert (artifact.stat().st_ino, artifact.stat().st_mtime_ns) == ( + before.st_ino, + before.st_mtime_ns, + ) + async with factory() as session: + repeated = await refresh_series_catalog_from_sources( + session, series_id, registry=registry, replace_managed=replace_managed + ) + assert repeated.created_issue_ids == () and repeated.series.issue_count == 7 + await session.commit() + async with factory() as session: + assert await session.scalar(select(func.count()).select_from(Issue)) == 7 + assert await session.scalar(select(func.count()).select_from(IssueIdentityEvent)) == events + + +@pytest.mark.parametrize( + "problem", ["number_only", "wrong_parent", "renumbered", "wrong_crosswalk"] +) +async def test_linked_catalog_conflict_is_atomic_and_does_not_reassign(linked_catalog, problem): + factory, series_id, original, artifact, adapters, registry = linked_catalog + data = adapters[Source.METRON_API].data + if problem == "number_only": + data.issues[0].cross_identities = [] + elif problem == "wrong_parent": + data.issues[0].series_external_id = "999" + elif problem == "renumbered": + data.issues[0].issue_number_text = "99" + data.issues[0].issue_number_key = "99" + else: + data.issues[0].cross_identities = [ + ExternalIdentityRef(Namespace.COMICVINE, Kind.ISSUE, "999") + ] + async with factory() as session: + with pytest.raises(SeriesRefreshError): + await refresh_series_catalog_from_sources(session, series_id, registry=registry) + await session.commit() + async with factory() as session: + assert (await session.get(Series, series_id)).issue_count == 5 + issues = list(await session.scalars(select(Issue).order_by(Issue.id))) + assert [(i.id, i.comicvine_id, i.status, i.manual_skip) for i in issues] == original + assert await session.scalar(select(func.count()).select_from(IssueExternalIdentity)) == 5 + assert artifact.read_bytes() == b"owned comic must remain untouched" + + +async def test_linked_catalog_cannot_reactivate_a_rejected_overlap(linked_catalog): + factory, series_id, original, artifact, adapters, registry = linked_catalog + data = adapters[Source.METRON_API].data + issue_id = original[0][0] + async with factory.begin() as session: + user_id = await session.scalar(select(User.id)) + saved = await record_identity_observation( + session, + _request( + data, + data.issues[0], + ExternalIdentityRef(Namespace.METRON, Kind.ISSUE, "200"), + issue_id, + observation=True, + ), + ) + await decide(session, Kind.ISSUE, issue_id, saved.event_id, user_id, Action.REJECT) + async with factory() as session: + with pytest.raises(SeriesRefreshError): + await refresh_series_catalog_from_sources(session, series_id, registry=registry) + await session.commit() + async with factory() as session: + assert (await session.get(Series, series_id)).issue_count == 5 + assert await session.scalar(select(func.count()).select_from(IssueExternalIdentity)) == 5 + event = await session.scalar( + select(IssueIdentityEvent).order_by(IssueIdentityEvent.id.desc()) + ) + assert event.verification_state.value == "rejected" + assert artifact.read_bytes() == b"owned comic must remain untouched" + + +async def test_linked_catalog_preserves_priorities_when_supplement_is_first(linked_catalog): + factory, series_id, original, _, _, registry = linked_catalog + async with factory.begin() as session: + for source, priority in ((Source.METRON_API, 0), (Source.COMICVINE_API, 10)): + priorities = {"core": priority, "issues": priority} + await session.execute( + update(MetadataSourceConfig) + .where(MetadataSourceConfig.source == source.value) + .values(domain_priorities=priorities) + ) + registry.runtime[source] = runtime(source, revision=1, domain_priorities=priorities) + async with factory() as session: + result = await refresh_series_catalog_from_sources( + session, series_id, registry=registry, replace_managed=True + ) + assert result.series.issue_count == 7 and len(result.created_issue_ids) == 2 + assert result.series.description == "Secondary description" + await session.commit() + async with factory() as session: + issues = list(await session.scalars(select(Issue).order_by(Issue.id))) + assert issues[0].title == "Secondary issue 1" + assert issues[1].title == "My issue title" + assert [(i.id, i.comicvine_id, i.status, i.manual_skip) for i in issues[:5]] == original + + +async def test_incomplete_supplement_cannot_partially_extend_catalog(linked_catalog, monkeypatch): + factory, series_id, _, _, adapters, registry = linked_catalog + adapter = adapters[Source.METRON_API] + original_issues = adapter.issues + + async def missing_member(external_id, *, page=1, validator=None): + result = await original_issues(external_id, page=page, validator=validator) + result.data.results.pop() + return result + + monkeypatch.setattr(adapter, "issues", missing_member) + async with factory() as session: + with pytest.raises(SeriesRefreshError): + await refresh_series_catalog_from_sources(session, series_id, registry=registry) + await session.commit() + async with factory() as session: + assert (await session.get(Series, series_id)).issue_count == 5 + assert await session.scalar(select(func.count()).select_from(Issue)) == 5 + assert await session.scalar(select(func.count()).select_from(IssueExternalIdentity)) == 5 + + +async def test_linked_catalog_extension_respects_paused_series(linked_catalog): + factory, series_id, _, _, _, registry = linked_catalog + async with factory.begin() as session: + (await session.get(Series, series_id)).monitored = False + async with factory() as session: + result = await refresh_series_catalog_from_sources(session, series_id, registry=registry) + assert len(result.created_issue_ids) == 2 + await session.commit() + async with factory() as session: + added = list( + await session.scalars(select(Issue).where(Issue.id.in_(result.created_issue_ids))) + ) + assert all(issue.status is IssueStatus.SKIPPED for issue in added) + + +async def test_unverified_series_link_cannot_extend_catalog(linked_catalog): + factory, series_id, _, _, adapters, registry = linked_catalog + async with factory.begin() as session: + claim = await session.scalar( + select(SeriesExternalIdentity).where( + SeriesExternalIdentity.identity_namespace == Namespace.METRON, + ) + ) + claim.verification_state = "stale" + async with factory() as session: + result = await refresh_series_catalog_from_sources(session, series_id, registry=registry) + assert result.created_issue_ids == () and result.series.issue_count == 5 + await session.commit() + assert adapters[Source.METRON_API].calls == [] + + +async def test_later_linked_catalog_cannot_drop_a_previously_verified_member(linked_catalog): + factory, series_id, _, artifact, adapters, registry = linked_catalog + async with factory() as session: + result = await refresh_series_catalog_from_sources(session, series_id, registry=registry) + assert result.series.issue_count == 7 + await session.commit() + data = adapters[Source.METRON_API].data + adapters[Source.METRON_API].data = replace(data, issues=data.issues[:-1], catalog_total=6) + async with factory() as session: + with pytest.raises(SeriesRefreshError, match="removed existing issue identities"): + await refresh_series_catalog_from_sources(session, series_id, registry=registry) + await session.commit() + async with factory() as session: + assert await session.scalar(select(func.count()).select_from(Issue)) == 7 + assert artifact.read_bytes() == b"owned comic must remain untouched" From 6aeafb6b4edf85e4a49b56e148557b290d651cec Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 16 Oct 2026 14:22:39 -0700 Subject: [PATCH 03/46] fix(metadata): normalize advisory counts before assembly --- src/pullbox/services/metadata_assembly.py | 9 +++++++++ tests/api/test_source_series_add_api.py | 7 ++++--- .../metadata_identity/test_series_adoption.py | 9 ++++++--- .../metadata_identity/test_series_refresh.py | 6 +++++- tests/unit/test_metadata_assembly.py | 18 ++++++++++++++++++ 5 files changed, 42 insertions(+), 7 deletions(-) diff --git a/src/pullbox/services/metadata_assembly.py b/src/pullbox/services/metadata_assembly.py index f2d21326..9da63ed5 100644 --- a/src/pullbox/services/metadata_assembly.py +++ b/src/pullbox/services/metadata_assembly.py @@ -3,6 +3,8 @@ from collections.abc import Sequence from datetime import UTC, datetime +from pydantic import ValidationError + from pullbox.core.issue_numbers import parse_issue_number_text from pullbox.core.metadata_identity import ExternalIdentityRef, MetadataEntityKind, MetadataSource from pullbox.schemas.metadata_snapshot import ( @@ -61,6 +63,13 @@ def _candidate_values( for key, value in candidate.model_dump().items() if key in MetadataValues.model_fields } + if isinstance(candidate, ProviderSeriesRead): + try: + MetadataValues(issue_count=candidate.issue_count) + except ValidationError: + # Summary counts are advisory; retain the raw provider row as evidence. + values["issue_count"] = None + diagnostics.add(f"{candidate.source.value}:invalid_advisory_issue_count") if candidate.image_url and ( candidate.source in {MetadataSource.GCD_LOCAL, MetadataSource.GCD_API_V2} or not allowed_artwork_url(candidate.image_url) diff --git a/tests/api/test_source_series_add_api.py b/tests/api/test_source_series_add_api.py index 0e947088..2e6f2744 100644 --- a/tests/api/test_source_series_add_api.py +++ b/tests/api/test_source_series_add_api.py @@ -137,11 +137,12 @@ async def test_add_native_series_uses_full_server_catalog_and_is_repeat_safe( @pytest.mark.parametrize("invalid_after_preview", [False, True]) +@pytest.mark.parametrize("summary_count", [1, 1000001, 1000000000]) async def test_stale_summary_preview_and_final_add_revalidate_the_catalog( - authenticated_client, source_add_setup, sec_db, invalid_after_preview + authenticated_client, source_add_setup, sec_db, invalid_after_preview, summary_count ): fixture = source_add_setup - fixture["profile"]["issue_count"] = 1 + fixture["profile"]["issue_count"] = summary_count complete = envelope([issue_row(100, "1"), issue_row(101, "2")]) fixture["responses"].append(httpx.Response(200, json=complete)) preview = await authenticated_client.post( @@ -151,7 +152,7 @@ async def test_stale_summary_preview_and_final_add_revalidate_the_catalog( ) assert preview.status_code == 200, preview.text assert preview.json()["issues"]["data"]["total"] == 2 - assert preview.json()["series"]["data"]["issue_count"] == 1 + assert preview.json()["series"]["data"]["issue_count"] == summary_count assert not list(fixture["root"].iterdir()) latest = {**complete, "count": 3} if invalid_after_preview else complete fixture["responses"].append(httpx.Response(200, json=latest)) diff --git a/tests/integration/metadata_identity/test_series_adoption.py b/tests/integration/metadata_identity/test_series_adoption.py index 55f0c7d0..a6bf226f 100644 --- a/tests/integration/metadata_identity/test_series_adoption.py +++ b/tests/integration/metadata_identity/test_series_adoption.py @@ -115,10 +115,13 @@ async def test_new_add_returns_canonical_snapshots_matching_persisted_metadata(i assert item.values.issue_number_text == issue.issue_number_text -async def test_stale_summary_adopts_complete_catalog_and_derives_saved_count(identity_probe_db): +@pytest.mark.parametrize("summary_count", [1, -1, 1000001, 1000000000]) +async def test_stale_summary_adopts_complete_catalog_and_derives_saved_count( + identity_probe_db, summary_count +): _, factory, _ = identity_probe_db data = bundle(numbers=("1", "2")) - data.series.issue_count = 1 + data.series.issue_count = summary_count async with factory.begin() as session: try: result = await adopt_source_series_bundle(session, data) @@ -128,7 +131,7 @@ async def test_stale_summary_adopts_complete_catalog_and_derives_saved_count(ide origin = next(item for item in result.snapshot.origins if item.field == "issue_count") assert origin.derivation == "catalog" assert await session.scalar(select(func.count()).select_from(Issue)) == 2 - assert data.series.issue_count == 1 + assert data.series.issue_count == summary_count async def test_representative_metron_cover_survives_add_and_session_reload(identity_probe_db): diff --git a/tests/integration/metadata_identity/test_series_refresh.py b/tests/integration/metadata_identity/test_series_refresh.py index e141b420..b7aba57e 100644 --- a/tests/integration/metadata_identity/test_series_refresh.py +++ b/tests/integration/metadata_identity/test_series_refresh.py @@ -841,12 +841,15 @@ def from_source(source): return factory, series_id, original, adapters, instance +@pytest.mark.parametrize("summary_count", [4, 1000001, 1000000000]) async def test_refresh_accepts_complete_catalog_with_stale_summary_without_fallback( - comicvine_count_disagreement, + comicvine_count_disagreement, summary_count ): from pullbox.services.metadata_series_refresh import refresh_series_catalog_from_sources factory, series_id, original, adapters, instance = comicvine_count_disagreement + for adapter in adapters.values(): + adapter.data.series.issue_count = summary_count async with factory() as session: result = await refresh_series_catalog_from_sources(session, series_id, registry=instance) assert result.series.description == "Updated description" @@ -862,6 +865,7 @@ async def test_refresh_accepts_complete_catalog_with_stale_summary_without_fallb Namespace.COMICVINE } assert not any(call[0] == "issues" for call in adapters[Source.COMICVINE_API].calls) + assert all(adapter.data.series.issue_count == summary_count for adapter in adapters.values()) @pytest.mark.parametrize("fallback", ["disabled", "rate_limited"]) diff --git a/tests/unit/test_metadata_assembly.py b/tests/unit/test_metadata_assembly.py index 5fecf6f4..dedb3651 100644 --- a/tests/unit/test_metadata_assembly.py +++ b/tests/unit/test_metadata_assembly.py @@ -45,6 +45,24 @@ def origin(snapshot, field): return next(item for item in snapshot.origins if item.field == field) +@pytest.mark.parametrize("count", [-1, 1000001, 1000000000]) +def test_out_of_range_series_count_is_advisory_and_raw_evidence_is_preserved(count): + candidate = row(METRON).model_copy(update={"issue_count": count}) + result = assemble([candidate]) + assert result.values.title == candidate.title + assert result.values.issue_count is None + assert result.diagnostics == ("metron_api:invalid_advisory_issue_count",) + assert candidate.issue_count == count + + +@pytest.mark.parametrize("count", [None, 0, 1, 1000000]) +def test_valid_advisory_series_count_is_preserved(count): + candidate = row(METRON).model_copy(update={"issue_count": count}) + result = assemble([candidate]) + assert result.values.issue_count == count + assert result.diagnostics == () + + def test_derived_values_are_refreshable_without_becoming_user_overrides(): before = MetadataSnapshot( entity_kind=Kind.SERIES, From e5cd83c1ad421b4b41590245db725d6ca47e9083 Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 16 Oct 2026 14:57:39 -0700 Subject: [PATCH 04/46] fix(metadata): hydrate linked Metron issue identity evidence --- .../services/metadata_catalog_crosswalks.py | 76 ++++++ .../services/metadata_series_refresh.py | 38 ++- .../test_linked_catalog_extension.py | 219 +++++++++++++++++- 3 files changed, 315 insertions(+), 18 deletions(-) create mode 100644 src/pullbox/services/metadata_catalog_crosswalks.py diff --git a/src/pullbox/services/metadata_catalog_crosswalks.py b/src/pullbox/services/metadata_catalog_crosswalks.py new file mode 100644 index 00000000..0aa69b8b --- /dev/null +++ b/src/pullbox/services/metadata_catalog_crosswalks.py @@ -0,0 +1,76 @@ +"""Bounded identity evidence for Metron's lightweight issue-list rows.""" + +from dataclasses import replace + +from pullbox.core.issue_numbers import parse_issue_number_text +from pullbox.core.metadata_identity import ExternalIdentityRef, MetadataEntityKind, MetadataSource +from pullbox.schemas.metadata_sources import SourceStatus +from pullbox.services.metadata_discovery import MetadataSourceRegistry +from pullbox.services.metadata_issue_catalog import IssueCatalogConflictError +from pullbox.services.metadata_series_adoption import SourceSeriesBundle, _validate_bundle +from pullbox.services.metadata_series_refresh_state import SeriesRefreshState + +_MAX_DETAIL_READS = 25 + + +async def hydrate_metron_catalog_crosswalks( + registry: MetadataSourceRegistry, + state: SeriesRefreshState, + bundles: tuple[SourceSeriesBundle, ...], +) -> tuple[SourceSeriesBundle, ...]: + """Fetch proof, never infer a match from an overlapping issue designation. + + Already verified native issues need no detail read. Only the crosswalks + augment the list row; the normal atomic writer still decides ownership. + """ + known = {identity for issue in state.issues for identity in issue.identities} + numbers = {issue.values.issue_number_text for issue in state.issues} + for bundle in bundles: + if bundle.series.source is not MetadataSource.METRON_API: + numbers.update(text for _, text in _validate_bundle(bundle)) + result = [] + for bundle in bundles: + source = bundle.series.source + if source is not MetadataSource.METRON_API: + result.append(bundle) + continue + pending = [ + row + for row, (_, text) in zip(bundle.issues, _validate_bundle(bundle), strict=True) + if text in numbers + and not row.cross_identities + and ExternalIdentityRef( + source.identity_namespace, MetadataEntityKind.ISSUE, row.external_id + ) + not in known + ] + if len(pending) > _MAX_DETAIL_READS: + raise IssueCatalogConflictError( + "The linked Metron catalog needs more issue identity checks than one refresh " + "can safely perform. Review the issue matches; no library metadata was changed." + ) + checked = {} + for row in pending: + fetched = await registry.issue(source, row.external_id) + detail = fetched.data + if fetched.status is not SourceStatus.OK or detail is None: + raise IssueCatalogConflictError( + "Metron issue identities could not be verified. Check source availability " + "and retry the refresh; no library metadata was changed." + ) + if detail.series_external_id != row.series_external_id or parse_issue_number_text( + detail.issue_number_text + ) != parse_issue_number_text(row.issue_number_text): + raise IssueCatalogConflictError( + "Metron issue details disagree with its catalog. Review the issue match; " + "no library metadata was changed." + ) + checked[row.external_id] = row.model_copy( + update={"cross_identities": detail.cross_identities} + ) + enriched = replace( + bundle, issues=tuple(checked.get(row.external_id, row) for row in bundle.issues) + ) + _validate_bundle(enriched) + result.append(enriched) + return tuple(result) diff --git a/src/pullbox/services/metadata_series_refresh.py b/src/pullbox/services/metadata_series_refresh.py index a2397fd7..fa58efc3 100644 --- a/src/pullbox/services/metadata_series_refresh.py +++ b/src/pullbox/services/metadata_series_refresh.py @@ -30,6 +30,7 @@ from pullbox.services.metadata_assembly import assemble_metadata from pullbox.services.metadata_baselines import MetadataBaselineWrite, save_metadata_baselines from pullbox.services.metadata_catalog_checkpoints import save_full_catalog_checkpoint +from pullbox.services.metadata_catalog_crosswalks import hydrate_metron_catalog_crosswalks from pullbox.services.metadata_catalog_review import apply_catalog_exclusions from pullbox.services.metadata_discovery import MetadataSourceRegistry from pullbox.services.metadata_entity_values import apply_series_metadata_values @@ -229,6 +230,7 @@ async def refresh_series_catalog_from_sources( if item.status not in {SourceStatus.OK, SourceStatus.NOT_QUERIED} } bundles = await _catalogs(registry, before, profiles, failed) + bundles = await hydrate_metron_catalog_crosswalks(registry, before, bundles) profiles.update((item.series.source, item.series) for item in bundles) profile = profiles.get(MetadataSource.METRON_API) if ( @@ -366,30 +368,44 @@ async def _catalogs( ), ) for source in sources: - profile = profiles.get(source) - linked_parent = profile is not None and any( - identity in state.series.identities for identity in profile.cross_identities - ) if ( source in failed or source.identity_namespace not in known or source.identity_namespace in loaded_namespaces - # A manual series link alone cannot authorize automatic issue matching. - or ( - catalog_namespaces - and source.identity_namespace not in catalog_namespaces - and not linked_parent - ) or registry._unavailable(source, capability=SourceCapability.ISSUE_LIST) ): + if source in failed: + logger.info( + "metadata_series_catalog_unavailable", + series_id=state.series.local_id, + source=source.value, + status=failed[source].status.value, + ) continue try: + profile = profiles.get(source) + if catalog_namespaces and source.identity_namespace not in catalog_namespaces: + # Descriptive fields may already be complete. Catalog eligibility + # still needs the linked source's independently verified parent. + if profile is None: + fetched = await registry.series(source, known[source.identity_namespace]) + if fetched.status is not SourceStatus.OK or fetched.data is None: + raise SeriesAdoptionError( + "The linked series identity could not be checked.", + status=fetched.status, + retry_after_seconds=fetched.retry_after_seconds, + ) + profile = profiles[source] = fetched.data + if not any( + identity in state.series.identities for identity in profile.cross_identities + ): + continue bundle = await fetch_source_series_bundle( registry, source, known[source.identity_namespace], source_revision=registry.runtime[source].policy.revision, - profile=profiles.get(source), + profile=profile, ) bundle = apply_catalog_exclusions(bundle, state.catalog_exclusions) _validate_bundle(bundle) diff --git a/tests/integration/metadata_identity/test_linked_catalog_extension.py b/tests/integration/metadata_identity/test_linked_catalog_extension.py index b91981bd..9fbc5c90 100644 --- a/tests/integration/metadata_identity/test_linked_catalog_extension.py +++ b/tests/integration/metadata_identity/test_linked_catalog_extension.py @@ -22,8 +22,8 @@ from pullbox.models.metadata_source import MetadataSourceConfig from pullbox.models.reader import IssueReaderState from pullbox.models.user import User -from pullbox.schemas.metadata_sources import SourceCapability -from pullbox.services.metadata_discovery import MetadataSourceRegistry +from pullbox.schemas.metadata_sources import MetadataFetch, SourceCapability, SourceStatus +from pullbox.services.metadata_discovery import MetadataSourceError, MetadataSourceRegistry from pullbox.services.metadata_identity_attachment import attach_verified_identities from pullbox.services.metadata_identity_review import record_identity_observation from pullbox.services.metadata_series_adoption import _request, adopt_source_series_bundle @@ -41,15 +41,16 @@ @pytest.fixture -async def linked_catalog(identity_probe_db, tmp_path): +async def linked_catalog(identity_probe_db, tmp_path, request): _, factory, _ = identity_probe_db + existing_count = getattr(request, "param", 5) metron = bundle(numbers=tuple(str(i) for i in range(1, 8))) metron.series.cross_identities = [ExternalIdentityRef(Namespace.COMICVINE, Kind.SERIES, "500")] metron.series.description = "Secondary description" for index, issue in enumerate(metron.issues): issue.external_id = str(200 + index) issue.title = f"Secondary issue {index + 1}" - if index < 5: + if index < existing_count: issue.cross_identities = [ ExternalIdentityRef(Namespace.COMICVINE, Kind.ISSUE, str(100 + index)) ] @@ -61,7 +62,7 @@ async def linked_catalog(identity_probe_db, tmp_path): "source": Source.COMICVINE_API, "identity_namespace": Namespace.COMICVINE, "external_id": "500", - "issue_count": 5, + "issue_count": existing_count, "cross_identities": [], "description": "Preferred description", } @@ -77,9 +78,9 @@ async def linked_catalog(identity_probe_db, tmp_path): "title": f"Preferred issue {index + 1}", } ) - for index, issue in enumerate(metron.issues[:5]) + for index, issue in enumerate(metron.issues[:existing_count]) ), - catalog_total=5, + catalog_total=existing_count, ) artifact = tmp_path / "owned.cbz" artifact.write_bytes(b"owned comic must remain untouched") @@ -141,6 +142,210 @@ async def linked_catalog(identity_probe_db, tmp_path): return factory, series_id, original, artifact, adapters, registry +class ThinMetronCatalog(RefreshAdapter): + """Metron's list omits crosswalks that its native issue details provide.""" + + def __init__(self, data): + super().__init__( + replace( + data, + issues=tuple(i.model_copy(update={"cross_identities": []}) for i in data.issues), + ) + ) + self.details = {i.external_id: i.model_copy(deep=True) for i in data.issues} + self.detail_failure = None + + async def issue(self, external_id, *, validator=None): + self.calls.append(("issue", external_id)) + if self.detail_failure: + raise MetadataSourceError(self.detail_failure, retry_after_seconds=60) + return MetadataFetch(status=SourceStatus.OK, data=self.details[external_id]) + + +@pytest.mark.parametrize("linked_catalog", [6], indirect=True) +async def test_thin_linked_catalog_adds_seventh_issue_using_native_detail_proof(linked_catalog): + factory, series_id, original, artifact, adapters, registry = linked_catalog + adapter = ThinMetronCatalog(adapters[Source.METRON_API].data) + registry.factories = { + **registry.factories, + Source.METRON_API: registration(adapter, capabilities=list(SourceCapability)), + } + async with factory() as session: + result = await refresh_series_catalog_from_sources(session, series_id, registry=registry) + assert result.series.issue_count == 7 + assert len(result.created_issue_ids) == 1 + await session.commit() + async with factory() as session: + issues = list(await session.scalars(select(Issue).order_by(Issue.id))) + assert [(i.id, i.comicvine_id, i.status, i.manual_skip) for i in issues[:6]] == original + assert issues[1].title == "My issue title" + assert issues[-1].issue_number_text == "7" and issues[-1].comicvine_id is None + assert issues[-1].store_date == adapter.details["206"].store_date + assert (await session.scalar(select(IssueReaderState))).last_page_index == 17 + assert (await session.scalar(select(LibraryFile))).file_path == str(artifact) + assert [c for c in adapter.calls if c[0] == "issue"] == [ + ("issue", str(200 + i)) for i in range(6) + ] + adapter.calls.clear() + async with factory() as session: + result = await refresh_series_catalog_from_sources(session, series_id, registry=registry) + assert result.created_issue_ids == () and result.series.issue_count == 7 + await session.commit() + assert not [c for c in adapter.calls if c[0] == "issue"] + assert artifact.read_bytes() == b"owned comic must remain untouched" + + +async def test_thin_catalog_checks_overlap_added_by_another_catalog_in_the_same_refresh( + linked_catalog, +): + factory, series_id, original, _, adapters, registry = linked_catalog + metron = adapters[Source.METRON_API].data + metron.issues[5].cross_identities = [ + ExternalIdentityRef(Namespace.COMICVINE, Kind.ISSUE, "105") + ] + cv = adapters[Source.COMICVINE_API] + cv.data = replace( + cv.data, + catalog_total=6, + issues=( + *cv.data.issues, + metron.issues[5].model_copy( + update={ + "source": Source.COMICVINE_API, + "identity_namespace": Namespace.COMICVINE, + "series_external_id": "500", + "external_id": "105", + "cross_identities": [], + } + ), + ), + ) + adapter = ThinMetronCatalog(metron) + registry.factories = { + **registry.factories, + Source.METRON_API: registration(adapter, capabilities=list(SourceCapability)), + } + async with factory() as session: + result = await refresh_series_catalog_from_sources(session, series_id, registry=registry) + assert result.series.issue_count == 7 and len(result.created_issue_ids) == 2 + await session.commit() + async with factory() as session: + issues = list(await session.scalars(select(Issue).order_by(Issue.id))) + assert [(i.id, i.comicvine_id, i.status, i.manual_skip) for i in issues[:5]] == original + assert issues[5].comicvine_id == 105 + assert issues[6].comicvine_id is None + assert await session.scalar(select(func.count()).select_from(IssueExternalIdentity)) == 13 + assert [c for c in adapter.calls if c[0] == "issue"] == [ + ("issue", str(200 + i)) for i in range(6) + ] + + +@pytest.mark.parametrize( + "problem", + ["number_only", "wrong_parent", "wrong_id", "renumbered", "wrong_crosswalk", "rate_limited"], +) +async def test_thin_catalog_requires_consistent_detail_identity_and_keeps_atomicity( + linked_catalog, problem +): + factory, series_id, original, artifact, adapters, registry = linked_catalog + adapter = ThinMetronCatalog(adapters[Source.METRON_API].data) + detail = adapter.details["200"] + if problem == "number_only": + detail.cross_identities = [] + elif problem == "wrong_parent": + detail.series_external_id = "999" + elif problem == "wrong_id": + detail.external_id = "999" + elif problem == "renumbered": + detail.issue_number_text = detail.issue_number_key = "99" + elif problem == "wrong_crosswalk": + detail.cross_identities = [ExternalIdentityRef(Namespace.COMICVINE, Kind.ISSUE, "999")] + else: + adapter.detail_failure = SourceStatus.RATE_LIMITED + registry.factories = { + **registry.factories, + Source.METRON_API: registration(adapter, capabilities=list(SourceCapability)), + } + async with factory() as session: + with pytest.raises(SeriesRefreshError): + await refresh_series_catalog_from_sources(session, series_id, registry=registry) + await session.commit() + async with factory() as session: + assert (await session.get(Series, series_id)).issue_count == 5 + issues = list(await session.scalars(select(Issue).order_by(Issue.id))) + assert [(i.id, i.comicvine_id, i.status, i.manual_skip) for i in issues] == original + assert await session.scalar(select(func.count()).select_from(IssueExternalIdentity)) == 5 + if problem == "rate_limited": + assert [c for c in adapter.calls if c[0] == "issue"] == [("issue", "200")] + assert artifact.read_bytes() == b"owned comic must remain untouched" + + +async def test_thin_catalog_detail_budget_is_checked_before_requests(linked_catalog): + factory, series_id, original, _, adapters, registry = linked_catalog + data = adapters[Source.METRON_API].data + issues = tuple( + data.issues[0].model_copy( + update={ + "external_id": str(200 + i), + "issue_number_text": str(i + 1), + "issue_number_key": str(i + 1), + "cross_identities": [], + } + ) + for i in range(26) + ) + adapter = ThinMetronCatalog(replace(data, issues=issues, catalog_total=26)) + cv = adapters[Source.COMICVINE_API] + cv.data = replace( + cv.data, + issues=tuple( + issue.model_copy( + update={ + "source": Source.COMICVINE_API, + "identity_namespace": Namespace.COMICVINE, + "series_external_id": "500", + "external_id": str(100 + i), + } + ) + for i, issue in enumerate(issues) + ), + catalog_total=26, + ) + registry.factories = { + **registry.factories, + Source.METRON_API: registration(adapter, capabilities=list(SourceCapability)), + } + async with factory() as session: + with pytest.raises(SeriesRefreshError, match="more issue identity checks"): + await refresh_series_catalog_from_sources(session, series_id, registry=registry) + await session.commit() + assert not [c for c in adapter.calls if c[0] == "issue"] + async with factory() as session: + assert await session.scalar(select(func.count()).select_from(Issue)) == len(original) + + +async def test_linked_catalog_is_checked_even_when_preferred_profile_fills_every_field( + linked_catalog, +): + factory, series_id, _, _, adapters, registry = linked_catalog + profile = adapters[Source.COMICVINE_API].data.series + profile.sort_title = profile.title + profile.publisher = "DC" + profile.year_start = 2025 + profile.year_end = 2026 + profile.series_type = "standard" + profile.status = "ended" + profile.image_url = "https://comicvine.gamespot.com/cover.jpg" + profile.volume = "1" + profile.language = "en" + async with factory() as session: + result = await refresh_series_catalog_from_sources(session, series_id, registry=registry) + assert result.series.issue_count == 7, "Catalog extension cannot depend on missing fields" + assert len(result.created_issue_ids) == 2 + assert ("series", "42") in adapters[Source.METRON_API].calls + assert ("issues", "42", 1) in adapters[Source.METRON_API].calls + + @pytest.mark.parametrize("replace_managed", [False, True]) async def test_linked_catalog_extends_five_to_seven_preserving_existing_state( linked_catalog, replace_managed From 53571607b36fc44ca26253ca6ae687e81fbec91f Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 16 Oct 2026 15:31:27 -0700 Subject: [PATCH 05/46] fix(import): review manual file metadata before paired writing --- src/pullbox/api/v1/issues.py | 43 +++++ src/pullbox/core/archive_metadata_fields.py | 10 ++ src/pullbox/schemas/issue.py | 6 + .../services/archive_metadata_preservation.py | 2 + .../archive_metadata_reconciliation.py | 2 + .../services/issue_file_metadata_review.py | 6 +- src/pullbox/services/issue_import_service.py | 26 +++ .../services/manual_import_metadata_review.py | 138 +++++++++++++++ .../services/manual_paired_metadata.py | 22 ++- src/pullbox/tasks/issue_import_task.py | 1 + src/pullbox/ui/static/js/pullbox.js | 87 +++++++++- .../ui/templates/pages/issue_detail.html | 1 + .../partials/issue_import_progress_modal.html | 37 +++- tests/api/test_issue_import_file.py | 61 +++++++ .../e2e/test_manual_import_metadata_review.py | 119 +++++++++++++ .../test_manual_import_metadata_review.py | 160 ++++++++++++++++++ tests/unit/test_comicinfo_aliases.py | 63 +++++++ tests/unit/test_issue_import_task.py | 9 + 18 files changed, 773 insertions(+), 20 deletions(-) create mode 100644 src/pullbox/services/manual_import_metadata_review.py create mode 100644 tests/e2e/test_manual_import_metadata_review.py create mode 100644 tests/integration/metadata_identity/test_manual_import_metadata_review.py create mode 100644 tests/unit/test_comicinfo_aliases.py diff --git a/src/pullbox/api/v1/issues.py b/src/pullbox/api/v1/issues.py index d9871504..71c5595e 100644 --- a/src/pullbox/api/v1/issues.py +++ b/src/pullbox/api/v1/issues.py @@ -31,6 +31,7 @@ IssueFileDeleteResponse, IssueResponse, IssueUpdate, + ManualFileImportPreviewRequest, ManualFileImportProgressResponse, ManualFileImportRequest, ManualFileImportResponse, @@ -69,6 +70,11 @@ prepare_manual_issue_import, ) from pullbox.services.issue_service import IssueService +from pullbox.services.manual_import_metadata_review import ( + ManualImportMetadataPreview, + preview_manual_import_metadata, +) +from pullbox.services.manual_paired_metadata import read_manual_metadata_plan from pullbox.services.release_validator import ( ValidationResult, ) @@ -1144,6 +1150,42 @@ async def download_issue( } +@router.post( + "/{issue_id}/import-file/metadata-preview", + response_model=ManualImportMetadataPreview | None, +) +async def preview_import_file_metadata( + issue_id: int, + body: ManualFileImportPreviewRequest, + _user: AuthenticatedUser, + session: DbSession, +) -> ManualImportMetadataPreview | None: + """Review the selected source without registering it or modifying either file.""" + from pullbox.config import get_settings + from pullbox.services.issue_file_metadata import file_metadata_error + + try: + prepared = await prepare_manual_issue_import( + session, + issue_id=issue_id, + file_path=body.file_path, + move_to_library=body.move_to_library, + ) + if not ( + get_settings().metadata_paired_import_writer_enabled + and prepared.ingest_policy.update_embedded_comicinfo_from_match + ): + return None + plan = await read_manual_metadata_plan(session, issue_id) + source = prepared.source_path + await session.rollback() + return (await preview_manual_import_metadata(plan, source, body.choices)).preview + except ManualIssueImportError as exc: + raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc + except (ValueError, OSError) as exc: + raise HTTPException(status_code=409, detail=file_metadata_error(exc)) from exc + + @router.post( "/{issue_id}/import-file", status_code=201, @@ -1171,6 +1213,7 @@ async def import_file_for_issue( session, prepared, allow_resource_safety_exception=body.allow_resource_safety_exception, + metadata_approval=body.metadata_approval, ) library_file = result.library_file except ManualIssueImportError as exc: diff --git a/src/pullbox/core/archive_metadata_fields.py b/src/pullbox/core/archive_metadata_fields.py index 617925d1..d83fdb66 100644 --- a/src/pullbox/core/archive_metadata_fields.py +++ b/src/pullbox/core/archive_metadata_fields.py @@ -1,5 +1,7 @@ """The credit-role overlap between ComicInfo and richer canonical metadata.""" +from xml.etree import ElementTree as ET + from pullbox.schemas.metadata_credits import MetadataCredit, parse_credits COMICINFO_ROLES = { @@ -25,6 +27,14 @@ COMICINFO_ONLY_FORMATS = frozenset({"special", "compendium", "deluxe", "volume"}) +def normalize_comicinfo_aliases(root: ET.Element) -> None: + """Normalize only known aliases in a parsed copy; retain duplicates for rejection.""" + aliases = {"Issue": "Number", "writer": "Writer"} + for node in root: + if isinstance(node.tag, str) and node.tag in aliases: + node.tag = aliases[node.tag] + + def archive_format_label(value: str | None) -> str | None: if value in COMICINFO_ONLY_FORMATS: return value.title() diff --git a/src/pullbox/schemas/issue.py b/src/pullbox/schemas/issue.py index 1cf8a864..1d78e201 100644 --- a/src/pullbox/schemas/issue.py +++ b/src/pullbox/schemas/issue.py @@ -6,6 +6,7 @@ from pydantic import BaseModel, ConfigDict, Field from pullbox.models.issue import IssueStatus +from pullbox.schemas.issue_file_metadata import FileMetadataApproval, FileMetadataReview class IssueUpdate(BaseModel): @@ -54,6 +55,11 @@ class ManualFileImportRequest(BaseModel): False, description="Allow one explicit resource safety exception for this manual import attempt.", ) + metadata_approval: FileMetadataApproval | None = None + + +class ManualFileImportPreviewRequest(ManualFileImportRequest, FileMetadataReview): + """Inspect the selected file and choose only values read from that file or library.""" class ManualFileImportResponse(BaseModel): diff --git a/src/pullbox/services/archive_metadata_preservation.py b/src/pullbox/services/archive_metadata_preservation.py index 53c2ee28..e1163396 100644 --- a/src/pullbox/services/archive_metadata_preservation.py +++ b/src/pullbox/services/archive_metadata_preservation.py @@ -3,6 +3,7 @@ from xml.etree import ElementTree as ET from pullbox.core.archive_metadata import ArchiveMetadataFiles +from pullbox.core.archive_metadata_fields import normalize_comicinfo_aliases from pullbox.core.metadata_xml import MetadataXmlError, parse_metadata_xml from pullbox.core.metroninfo_schema import validate_metroninfo_xml @@ -84,6 +85,7 @@ def preserved_roots(files: ArchiveMetadataFiles) -> tuple[ET.Element, ET.Element else ET.Element(name) ) if name == "ComicInfo": + normalize_comicinfo_aliases(root) _validate_comicinfo(root) elif member.payload is not None: if root.get("version") not in {None, "1.0", "1.1"}: diff --git a/src/pullbox/services/archive_metadata_reconciliation.py b/src/pullbox/services/archive_metadata_reconciliation.py index c5b0fa37..9eb04f09 100644 --- a/src/pullbox/services/archive_metadata_reconciliation.py +++ b/src/pullbox/services/archive_metadata_reconciliation.py @@ -13,6 +13,7 @@ COMICINFO_ROLES, canonical_archive_format, comicinfo_credits, + normalize_comicinfo_aliases, ) from pullbox.core.comicinfo_sanitizer import references_stale_retailer from pullbox.core.issue_numbers import normalize_issue_number_text @@ -179,6 +180,7 @@ def _read_document(member: MetadataFile, document: str) -> EmbeddedMetadataValue except MetadataXmlError as exc: diagnostics.append(ArchiveMetadataDiagnostic(document, exc.code, "ComicInfo")) return EmbeddedMetadataValues(diagnostics=tuple(diagnostics)) + normalize_comicinfo_aliases(root) return _ComicInfoReader(root, diagnostics).read() diff --git a/src/pullbox/services/issue_file_metadata_review.py b/src/pullbox/services/issue_file_metadata_review.py index c9081936..146df856 100644 --- a/src/pullbox/services/issue_file_metadata_review.py +++ b/src/pullbox/services/issue_file_metadata_review.py @@ -17,6 +17,7 @@ from pullbox.services.archive_metadata_preservation import ArchiveMetadataRenderError from pullbox.services.archive_metadata_reconciliation import ArchiveMetadataReconciliation from pullbox.services.archive_metadata_rendering import _series_separator_key +from pullbox.services.metadata_series_refresh_state import SeriesRefreshState _FIELDS = { "series": ( @@ -51,7 +52,7 @@ def _equal(field: str, source: FileMetadataSource, old: object, new: object) -> def review_metadata_fields( - binding: ArchiveMetadataBinding, + binding: ArchiveMetadataBinding | SeriesRefreshState, archive: ArchiveMetadataReconciliation, series: MetadataSnapshot, issue: MetadataSnapshot, @@ -61,8 +62,9 @@ def review_metadata_fields( conflicts: list[FileMetadataConflict] = [] snapshots = {"series": series, "issue": issue} seen: set[str] = set() + metadata = binding.metadata if isinstance(binding, ArchiveMetadataBinding) else binding for kind, fields in _FIELDS.items(): - current = binding.metadata.series if kind == "series" else binding.metadata.issues[0] + current = metadata.series if kind == "series" else metadata.issues[0] snapshot = snapshots[kind] values = snapshot.values.model_dump() origins = {item.field: item for item in snapshot.origins} diff --git a/src/pullbox/services/issue_import_service.py b/src/pullbox/services/issue_import_service.py index 9891c2c5..479fbf2b 100644 --- a/src/pullbox/services/issue_import_service.py +++ b/src/pullbox/services/issue_import_service.py @@ -27,6 +27,7 @@ from pullbox.models.series import Series from pullbox.services.archive_metadata_binding import ArchiveMetadataBindingError from pullbox.services.issue_file_service import resolve_configured_utility_trash_dir +from pullbox.services.manual_import_metadata_review import preview_manual_import_metadata from pullbox.services.manual_paired_metadata import ( materialize_manual_metadata, read_manual_metadata_plan, @@ -41,6 +42,7 @@ from sqlalchemy.ext.asyncio import AsyncSession + from pullbox.schemas.issue_file_metadata import FileMetadataApproval from pullbox.services.direct_pack_paired_metadata import StagedPackMember @@ -158,6 +160,7 @@ async def execute_manual_issue_import( prepared: PreparedManualIssueImport, *, use_paired_metadata: bool = True, + metadata_approval: FileMetadataApproval | None = None, staged_pack_member: StagedPackMember | None = None, allow_resource_safety_exception: bool = False, preparation_progress_callback: Callable[[str, int, int, str], Any] | None = None, @@ -177,7 +180,12 @@ async def execute_manual_issue_import( else None ) plan = None + reviewed = None source_signature = None + if metadata_approval is not None and (not paired or staged_pack_member is not None): + raise ManualIssueImportError( + status_code=409, detail="Import settings changed; preview metadata again." + ) if staged_pack_member is not None: prepared = await prepare_manual_issue_import( session, @@ -223,6 +231,23 @@ async def execute_manual_issue_import( source_signature = await asyncio.to_thread( build_file_identity_signature, prepared.source_path ) + if metadata_approval is not None: + try: + reviewed = await preview_manual_import_metadata( + plan, prepared.source_path, metadata_approval.choices + ) + if ( + not reviewed.preview.ready + or reviewed.preview.review_key != metadata_approval.review_key + ): + raise ValueError("Approval changed") + except ValueError: + raise ManualIssueImportError( + status_code=409, + detail=( + "The file, metadata or choices changed. Preview and review metadata again." + ), + ) from None async def converter_with_progress( source: Path, @@ -286,6 +311,7 @@ async def paired_progress(stage: str, current: int, total: int, unit: str) -> No allow_resource_safety_exception=allow_resource_safety_exception, cancellation_check=cancellation_check, progress_callback=paired_progress, + reviewed=reviewed, ) except (ArchiveMetadataStagingError, ValueError) as exc: if isinstance(exc, ArchiveMetadataStagingError) and str(exc) != "metadata_conflict": diff --git a/src/pullbox/services/manual_import_metadata_review.py b/src/pullbox/services/manual_import_metadata_review.py new file mode 100644 index 00000000..9abf7a3a --- /dev/null +++ b/src/pullbox/services/manual_import_metadata_review.py @@ -0,0 +1,138 @@ +"""Read-only descriptive review for a manually selected source archive.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from pydantic import BaseModel, TypeAdapter + +from pullbox.core.archive import ArchiveError +from pullbox.core.archive_metadata import ( + MAX_METADATA_BYTES, + ArchiveMetadataFiles, + MetadataFile, + read_archive_metadata, +) +from pullbox.core.library_file_ownership import build_file_identity_signature +from pullbox.schemas.issue_file_metadata import FileMetadataConflict # noqa: TC001 - Pydantic field +from pullbox.services.archive_metadata_binding import ( + assemble_archive_metadata_state, + select_archive_primary_identity, +) +from pullbox.services.archive_metadata_reconciliation import reconcile_archive_metadata +from pullbox.services.archive_metadata_rendering import ( + ArchiveMetadataRenderError, + render_archive_metadata, +) +from pullbox.services.issue_file_metadata_review import review_metadata_fields + +if TYPE_CHECKING: + from pathlib import Path + + from pullbox.core.metadata_identity import ExternalIdentityRef + from pullbox.schemas.issue_file_metadata import FileMetadataChoices + from pullbox.schemas.metadata_snapshot import MetadataSnapshot + from pullbox.services.manual_paired_metadata import ManualMetadataPlan + + +class ManualImportMetadataPreview(BaseModel): + ready: bool + review_key: str + context_key: str + conflicts: list[FileMetadataConflict] + + +@dataclass(frozen=True) +class ReviewedManualMetadata: + series: MetadataSnapshot + issue: MetadataSnapshot + primary: ExternalIdentityRef | None + preview: ManualImportMetadataPreview + + +async def preview_manual_import_metadata( + plan: ManualMetadataPlan, source: Path, choices: FileMetadataChoices +) -> ReviewedManualMetadata: + """Read bounded XML off-thread; no publication, database writes or provider calls.""" + try: + return await asyncio.to_thread(_preview, plan, source, choices) + except ArchiveError as exc: + raise ArchiveMetadataRenderError("archive_unreadable") from exc + + +def _preview( + plan: ManualMetadataPlan, source: Path, choices: FileMetadataChoices +) -> ReviewedManualMetadata: + fingerprint = build_file_identity_signature(source) + kind = source.suffix.lstrip(".").casefold() + files = ( + ArchiveMetadataFiles(MetadataFile("ComicInfo.xml"), MetadataFile("MetronInfo.xml")) + if kind == "pdf" + else read_archive_metadata( + source, + "cbz" if kind == "zip" else kind, + max_solid_scan_bytes=MAX_METADATA_BYTES if kind in {"cbz", "zip"} else plan.limit, + ) + ) + archive = reconcile_archive_metadata(files) + series, issue = assemble_archive_metadata_state(plan.metadata, archive, now=datetime.now(UTC)) + primary = select_archive_primary_identity(plan.metadata, archive) + series, issue, conflicts = review_metadata_fields( + plan.metadata, archive, series, issue, choices + ) + ready = all(item.selected is not None for item in conflicts) + try: + render_archive_metadata( + series, + issue, + files, + primary_identity=primary, + previous_series=plan.metadata.series.baseline, + previous_issue=plan.metadata.issues[0].baseline, + ) + except ArchiveMetadataRenderError as exc: + if ( + ready + or exc.code != "unreconciled_field" + or not any( + item.key.split(".")[1] == exc.field and item.selected is None for item in conflicts + ) + ): + raise + for conflict in conflicts: + if conflict.key == "issue.page_count": + conflict.note = ( + "Archive page counts can include covers and extras; catalog counts may differ. " + "Choose the value for this library copy. The catalog record stays unchanged." + ) + elif conflict.key == "issue.cover_date": + conflict.note = ( + "The file date may describe publication rather than the catalog cover date. " + "Pullbox will not guess a missing day or change the store date." + ) + # Bind approval to the exact plan, source, XML and choices, not observation timestamps. + evidence = TypeAdapter(type(plan)).dump_json(plan) + evidence += json.dumps( + {"path": str(source), "fingerprint": fingerprint}, sort_keys=True + ).encode() + evidence += (files.comicinfo.payload or b"") + (files.metroninfo.payload or b"") + context_key = hashlib.sha256(evidence).hexdigest() + evidence += json.dumps(choices, sort_keys=True).encode() + if build_file_identity_signature(source) != fingerprint: + raise ArchiveMetadataRenderError("source_changed") + return ReviewedManualMetadata( + series, + issue, + primary, + ManualImportMetadataPreview( + ready=ready, + review_key=hashlib.sha256(evidence).hexdigest(), + context_key=context_key, + conflicts=conflicts, + ), + ) diff --git a/src/pullbox/services/manual_paired_metadata.py b/src/pullbox/services/manual_paired_metadata.py index 852c5806..2b328dd7 100644 --- a/src/pullbox/services/manual_paired_metadata.py +++ b/src/pullbox/services/manual_paired_metadata.py @@ -27,6 +27,7 @@ require_mutable_library_target, ) from pullbox.core.library_policy import ( + LibraryIngestPolicy, load_effective_library_ingest_policy, load_library_ingest_policy, ) @@ -43,7 +44,10 @@ require_unowned_metadata_file, require_verified_archive_metadata, ) -from pullbox.services.metadata_series_refresh_state import read_series_refresh_state +from pullbox.services.metadata_series_refresh_state import ( + SeriesRefreshState, + read_series_refresh_state, +) from pullbox.utilities.executors.archive_metadata_staging import stage_cbz_metadata_interruptible if TYPE_CHECKING: @@ -51,8 +55,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker - from pullbox.core.library_policy import LibraryIngestPolicy - from pullbox.services.metadata_series_refresh_state import SeriesRefreshState + from pullbox.services.manual_import_metadata_review import ReviewedManualMetadata from pullbox.utilities.executors.archive_metadata_staging import StagedArchiveMetadata from pullbox.utilities.executors.archive_subprocess import ControlCheck, ProgressCallback @@ -168,6 +171,7 @@ async def materialize_manual_metadata( allow_resource_safety_exception: bool = False, cancellation_check: ControlCheck | None = None, progress_callback: ProgressCallback | None = None, + reviewed: ReviewedManualMetadata | None = None, ) -> bool: """Stage both XML files privately, revalidate, then publish without overwrite.""" directories = await asyncio.to_thread(_directories, target, Path(plan.root_path)) @@ -179,6 +183,7 @@ async def materialize_manual_metadata( allow_resource_safety_exception=allow_resource_safety_exception, cancellation_check=cancellation_check, progress_callback=progress_callback, + reviewed=reviewed, ) as staged, factory() as reader, ): @@ -208,6 +213,7 @@ async def stage_manual_metadata( allow_resource_safety_exception: bool = False, cancellation_check: ControlCheck | None = None, progress_callback: ProgressCallback | None = None, + reviewed: ReviewedManualMetadata | None = None, ) -> AsyncIterator[StagedArchiveMetadata]: """Reuse the verified reconciler before an existing owner registers a batch.""" limit = plan.limit @@ -216,7 +222,10 @@ async def stage_manual_metadata( if allow_resource_safety_exception: members = await asyncio.to_thread(ArchiveReader(source).list_members) limit = max(limit, sum(member.size for member in members) + 2 * MAX_METADATA_BYTES) - series, issue = assemble_archive_metadata_state(plan.metadata, None, now=datetime.now(UTC)) + if reviewed is None: + series, issue = assemble_archive_metadata_state(plan.metadata, None, now=datetime.now(UTC)) + else: + series, issue = reviewed.series, reviewed.issue async with stage_cbz_metadata_interruptible( source, staging_parent, @@ -224,7 +233,10 @@ async def stage_manual_metadata( issue, max_uncompressed_bytes=limit, block_dangerous=plan.block_dangerous, - metadata_state=plan.metadata, + metadata_state=plan.metadata if reviewed is None else None, + primary_identity=reviewed.primary if reviewed is not None else None, + previous_series=plan.metadata.series.baseline, + previous_issue=plan.metadata.issues[0].baseline, cancellation_check=cancellation_check, progress_callback=progress_callback, ) as staged: diff --git a/src/pullbox/tasks/issue_import_task.py b/src/pullbox/tasks/issue_import_task.py index 882c4ffd..da980100 100644 --- a/src/pullbox/tasks/issue_import_task.py +++ b/src/pullbox/tasks/issue_import_task.py @@ -241,6 +241,7 @@ async def check_import_control() -> None: session, prepared, allow_resource_safety_exception=request.allow_resource_safety_exception, + metadata_approval=request.metadata_approval, preparation_progress_callback=emit_stage_progress, transfer_progress_callback=lambda current, total: emit_stage_progress( stage="transferring", diff --git a/src/pullbox/ui/static/js/pullbox.js b/src/pullbox/ui/static/js/pullbox.js index fe0cab66..c85a46d6 100644 --- a/src/pullbox/ui/static/js/pullbox.js +++ b/src/pullbox/ui/static/js/pullbox.js @@ -19187,6 +19187,9 @@ function issueDetailPage(config) { importing: false, cancellingImport: false, importReplacementConfirmed: false, + importMetadataPreview: null, + importMetadataChoices: {}, + importPreviewController: null, importModalOpen: false, importPollTimer: null, importState: "idle", @@ -19281,6 +19284,8 @@ function issueDetailPage(config) { this.form.file_size = selection.size || 0; this.form.file_ext = selection.ext || ""; this.importReplacementConfirmed = false; + this.importMetadataPreview = null; + this.importMetadataChoices = {}; this.submitImport(); }, @@ -19288,6 +19293,19 @@ function issueDetailPage(config) { return this.importModalOpen; }, + trapImportFocus: function (event) { + var dialog = event.currentTarget; + var elements = Array.from(dialog.querySelectorAll('button:not([disabled]), input:not([disabled]), a[href], [tabindex="0"]')) + .filter(function (element) { return element.getClientRects().length > 0; }); + var first = elements[0], last = elements[elements.length - 1]; + if (!first) { event.preventDefault(); return; } + if (event.shiftKey && (document.activeElement === first || document.activeElement === dialog)) { + event.preventDefault(); last.focus(); + } else if (!event.shiftKey && (document.activeElement === last || document.activeElement === dialog)) { + event.preventDefault(); first.focus(); + } + }, + closeImportModal: function () { if (this.importState === "running") { return; @@ -19303,6 +19321,8 @@ function issueDetailPage(config) { completed: "Complete", failed: "Needs attention", safety_blocked: "Safety review", + metadata_review: "Metadata review", + checking: "Checking metadata", cancelled: "Cancelled", }; return labels[this.importState] || "Importing"; @@ -19488,6 +19508,14 @@ function issueDetailPage(config) { cancelIssueImport: async function () { if (this.cancellingImport || this.importState === "completed") return; + if (this.importState === "checking") { + if (this.importPreviewController) this.importPreviewController.abort(); + return; + } + if (this.importState === "metadata_review") { + this.closeImportModal(); + return; + } if (this.importState === "failed" || this.importState === "safety_blocked" || this.importState === "cancelled") { this.closeImportModal(); return; @@ -19531,7 +19559,17 @@ function issueDetailPage(config) { return (bytes / 1048576).toFixed(1) + " MB"; }, - submitImport: async function () { + canImportReviewedMetadata: function () { + var self = this; + return !self.importing && self.importMetadataPreview && + self.importMetadataPreview.conflicts.every(function (conflict) { + return conflict.options.some(function (option) { + return option.selectable && self.importMetadataChoices[conflict.key] === option.source; + }); + }); + }, + + submitImport: async function (approveMetadata) { if (!this.form.file_path || this.importing) return; if (cfg.hasLibraryFile && !this.importReplacementConfirmed) { @@ -19564,17 +19602,45 @@ function issueDetailPage(config) { }); try { + var approval = null; + if (cfg.pairedImportMetadata && cfg.importMetadataPreviewUrl) { + this.importState = "checking"; + this.importMessage = "Checking file metadata..."; + this.importPreviewController = new AbortController(); + var previewResponse = await fetch(cfg.importMetadataPreviewUrl, { + method: "POST", + headers: { "Content-Type": "application/json", "X-CSRF-Token": this.csrfToken() }, + signal: this.importPreviewController.signal, + body: JSON.stringify({ file_path: this.form.file_path, move_to_library: this.form.move_to_library, choices: this.importMetadataChoices }), + }); + var preview = await previewResponse.json(); + if (!previewResponse.ok) throw new Error(preview.detail || "Unable to review file metadata."); + var changed = preview && this.importMetadataPreview && + preview.context_key !== this.importMetadataPreview.context_key; + if (changed) this.importMetadataChoices = {}; + this.importMetadataPreview = preview; + if (preview && (changed || (preview.conflicts.length && (!approveMetadata || !preview.ready)))) { + this.importState = "metadata_review"; + this.importMessage = changed ? "File or catalog changed. Review the current values." : "Review file metadata"; + this.importing = false; + return; + } + if (preview) approval = { review_key: preview.review_key, choices: this.importMetadataChoices }; + } + this.importState = "running"; + var requestBody = { + file_path: this.form.file_path, + move_to_library: this.form.move_to_library, + allow_resource_safety_exception: allowSafetyException, + }; + if (approval) requestBody.metadata_approval = approval; var response = await fetch(cfg.importStartUrl, { method: "POST", headers: { "Content-Type": "application/json", "X-CSRF-Token": this.csrfToken(), }, - body: JSON.stringify({ - file_path: this.form.file_path, - move_to_library: this.form.move_to_library, - allow_resource_safety_exception: allowSafetyException, - }), + body: JSON.stringify(requestBody), }); if (!response.ok) { @@ -19594,18 +19660,25 @@ function issueDetailPage(config) { } catch (error) { this.stopImportPolling(); this.importing = false; + if (error.name === "AbortError") { + this.importState = "cancelled"; + this.importModalOpen = false; + return; + } this.importState = "failed"; this.importMessage = "Import failed."; this.importErrorMessage = error.message || "An error occurred during import"; this.dispatchToast(this.importErrorMessage, "error"); + } finally { + this.importPreviewController = null; } }, retryImportWithSafetyException: function () { if (!this.form.file_path || this.importing) return; this.importAllowSafetyException = true; - this.submitImport(); + this.submitImport(Boolean(this.importMetadataPreview)); }, deleteIssueFile: async function () { diff --git a/src/pullbox/ui/templates/pages/issue_detail.html b/src/pullbox/ui/templates/pages/issue_detail.html index cde5cb72..3b18325b 100644 --- a/src/pullbox/ui/templates/pages/issue_detail.html +++ b/src/pullbox/ui/templates/pages/issue_detail.html @@ -20,6 +20,7 @@ searchUrl: {{ ('/api/v1/issues/' ~ issue.id ~ '/download')|tojson|forceescape }}, deleteFileUrl: {{ ('/api/v1/issues/' ~ issue.id ~ '/file')|tojson|forceescape }}, importStartUrl: {{ ('/api/v1/issues/' ~ issue.id ~ '/import-file/start')|tojson|forceescape }}, + importMetadataPreviewUrl: {{ ('/api/v1/issues/' ~ issue.id ~ '/import-file/metadata-preview')|tojson|forceescape }}, importCancelUrl: {{ ('/api/v1/issues/' ~ issue.id ~ '/import-file/cancel')|tojson|forceescape }}, importProgressUrl: {{ ('/api/v1/issues/' ~ issue.id ~ '/import-file/progress')|tojson|forceescape }}, readerManifestUrl: {{ ('/api/v1/reader/issues/' ~ issue.id ~ '/manifest')|tojson|forceescape }}, diff --git a/src/pullbox/ui/templates/partials/issue_import_progress_modal.html b/src/pullbox/ui/templates/partials/issue_import_progress_modal.html index 1fe237d4..8e49c3ef 100644 --- a/src/pullbox/ui/templates/partials/issue_import_progress_modal.html +++ b/src/pullbox/ui/templates/partials/issue_import_progress_modal.html @@ -2,7 +2,10 @@ class="modal-shell" x-cloak x-show="showImportModal()" + x-data="{ importFocusReturn: null }" + x-init="$watch('importModalOpen', value => { if (value) { importFocusReturn = document.activeElement; $nextTick(() => $refs.importDialog.focus({ preventScroll: true })); } else if (importFocusReturn && importFocusReturn.isConnected) { importFocusReturn.focus({ preventScroll: true }); } })" data-testid="issue-import-modal" + @keydown.escape.window="if (showImportModal() && importState !== 'running') cancelIssueImport()" > diff --git a/tests/api/test_issue_import_file.py b/tests/api/test_issue_import_file.py index ea9fbbdf..203ddcc6 100644 --- a/tests/api/test_issue_import_file.py +++ b/tests/api/test_issue_import_file.py @@ -123,6 +123,67 @@ async def _override_db() -> AsyncGenerator[AsyncSession, None]: # ── Helpers ──────────────────────────────────────────────────────────── +async def test_manual_import_metadata_preview_and_approval( + client, _db_factory, tmp_path, monkeypatch +): + from pullbox.config import get_settings + from tests.integration.metadata_identity.test_manual_import_metadata_review import gargoyle_case + + monkeypatch.setenv("PULLBOX_METADATA_PAIRED_IMPORT_WRITER_ENABLED", "true") + get_settings.cache_clear() + try: + issue_id, source = await gargoyle_case(_db_factory, tmp_path) + original = source.read_bytes() + endpoint = f"/api/v1/issues/{issue_id}/import-file" + response = await client.post( + endpoint + "/metadata-preview", json={"file_path": str(source)} + ) + assert response.status_code == 200, response.text + assert not response.json()["ready"] + choices = {item["key"]: "library" for item in response.json()["conflicts"]} + response = await client.post( + endpoint + "/metadata-preview", json={"file_path": str(source), "choices": choices} + ) + assert response.status_code == 200 and response.json()["ready"], response.text + assert source.read_bytes() == original + response = await client.post( + endpoint, + json={ + "file_path": str(source), + "metadata_approval": { + "review_key": response.json()["review_key"], + "choices": choices, + }, + }, + ) + assert response.status_code == 201, response.text + assert source.read_bytes() == original + finally: + get_settings.cache_clear() + + +async def test_manual_metadata_preview_reports_unreadable_archive( + client, _db_factory, tmp_path, monkeypatch +): + from pullbox.config import get_settings + from tests.integration.metadata_identity.test_manual_import_metadata_review import gargoyle_case + + monkeypatch.setenv("PULLBOX_METADATA_PAIRED_IMPORT_WRITER_ENABLED", "true") + get_settings.cache_clear() + try: + issue_id, source = await gargoyle_case(_db_factory, tmp_path) + source.write_bytes(b"broken zip") + response = await client.post( + f"/api/v1/issues/{issue_id}/import-file/metadata-preview", + json={"file_path": str(source)}, + ) + assert response.status_code == 409, response.text + assert "cannot be read" in response.json()["detail"] + assert source.read_bytes() == b"broken zip" + finally: + get_settings.cache_clear() + + async def _create_series_and_issue( factory: async_sessionmaker[AsyncSession], *, diff --git a/tests/e2e/test_manual_import_metadata_review.py b/tests/e2e/test_manual_import_metadata_review.py new file mode 100644 index 00000000..d7ca9225 --- /dev/null +++ b/tests/e2e/test_manual_import_metadata_review.py @@ -0,0 +1,119 @@ +"""Manual import offers choices before starting background file work.""" + +import os +from pathlib import Path +from unittest.mock import patch + +import pytest +from playwright.sync_api import expect + +from tests.e2e.accessibility import assert_no_axe_violations +from tests.e2e.conftest import _run_async_blocking, _running_live_server + +pytestmark = pytest.mark.e2e + + +@pytest.fixture(scope="session") +def live_server(): + with ( + patch.dict(os.environ, {"PULLBOX_METADATA_PAIRED_IMPORT_WRITER_ENABLED": "true"}), + _running_live_server() as server, + ): + yield server + + +@pytest.fixture(autouse=True) +def enable_manual_metadata(seeded_server): + from pullbox.database import get_session_factory + from pullbox.models.config import SystemConfig + + async def configure(): + async with get_session_factory().begin() as session: + await session.merge( + SystemConfig( + key="update_embedded_comicinfo_from_match_on_import", + value="true", + value_type="bool", + ) + ) + + _run_async_blocking(configure()) + + +@pytest.mark.parametrize("theme,width", [("light", 1280), ("dark", 390), ("dark", 1280)]) +def test_manual_import_reviews_fields_before_start(authed_page, seeded_server, theme, width): + page = authed_page + page.set_viewport_size({"width": width, "height": 1000}) + starts = [] + endpoint = "**/api/v1/issues/2/import-file" + + def preview(route): + selected = route.request.post_data_json.get("choices", {}).get("issue.page_count") + route.fulfill( + json={ + "ready": bool(selected), + "review_key": ("b" if selected else "a") * 64, + "context_key": "c" * 64, + "conflicts": [ + { + "key": "issue.page_count", + "label": "Issue page count", + "selected": selected, + "note": "Archive page counts can include covers and extras.", + "options": [ + {"source": "library", "value": "48", "selectable": True}, + {"source": "ComicInfo.xml", "value": "54", "selectable": True}, + ], + } + ], + } + ) + + def start(route): + starts.append(route.request.post_data_json) + route.fulfill(status=202, json={"issue_id": 2, "state": "running", "message": "Importing"}) + + page.route(endpoint + "/metadata-preview", preview) + page.route(endpoint + "/start", start) + page.route( + endpoint + "/progress", + lambda route: route.fulfill( + json={ + "issue_id": 2, + "state": "running", + "current_file_progress_pct": 40, + } + ), + ) + page.goto(f"{seeded_server}/issues/2") + page.evaluate("theme => applyTheme(theme)", theme) + expect(page.locator("html")).to_have_attribute("data-theme", theme) + page.evaluate("""() => Alpine.$data(document.querySelector('[data-testid="issue-detail-page"]')) + .applyImportFileSelection({path:'/imports/fixture.cbz', name:'fixture.cbz', size:100})""") + dialog = page.get_by_test_id("issue-import-modal") + expect(dialog.get_by_text("Issue page count", exact=True)).to_be_visible() + assert not starts + button = dialog.get_by_role("button", name="Import reviewed file", exact=True) + expect(button).to_be_disabled() + dialog.get_by_role("radio", name="ComicInfo.xml: 54", exact=True).check() + expect(button).to_be_enabled() + button.focus() + page.keyboard.press("Tab") + assert dialog.evaluate("el => el.contains(document.activeElement)") + assert_no_axe_violations( + page, name=f"manual-import-review-{theme}", include=['[data-testid="issue-import-modal"]'] + ) + page.screenshot(path=str(Path("/private/tmp") / f"bug8-review-{theme}-{width}.png")) + button.click() + expect(dialog.get_by_role("progressbar")).to_have_attribute("aria-valuenow", "40") + assert starts == [ + { + "file_path": "/imports/fixture.cbz", + "move_to_library": True, + "allow_resource_safety_exception": False, + "metadata_approval": { + "review_key": "b" * 64, + "choices": {"issue.page_count": "ComicInfo.xml"}, + }, + } + ] diff --git a/tests/integration/metadata_identity/test_manual_import_metadata_review.py b/tests/integration/metadata_identity/test_manual_import_metadata_review.py new file mode 100644 index 00000000..19a2de01 --- /dev/null +++ b/tests/integration/metadata_identity/test_manual_import_metadata_review.py @@ -0,0 +1,160 @@ +"""Manual import reviews descriptive differences without changing its source.""" + +from datetime import date +from xml.etree import ElementTree as ET +from zipfile import ZipFile + +import pytest +from sqlalchemy import delete, select + +from pullbox.models import Issue, LibraryFile +from pullbox.models.creator import Creator, IssueCreator +from pullbox.models.download import DownloadHistory +from pullbox.schemas.issue_file_metadata import FileMetadataApproval +from pullbox.services.issue_import_service import ( + ManualIssueImportError, + execute_manual_issue_import, + prepare_manual_issue_import, +) +from pullbox.services.manual_import_metadata_review import preview_manual_import_metadata +from pullbox.services.manual_paired_metadata import read_manual_metadata_plan +from tests.integration.metadata_identity.test_download_paired_metadata import completed_download +from tests.integration.metadata_identity.test_manual_paired_metadata import ( + paired_manual_writer, # noqa: F401 +) + +pytestmark = pytest.mark.usefixtures("paired_manual_writer") + + +async def gargoyle_case(factory, tmp_path): + download_id, issue_id, source = await completed_download(factory, tmp_path) + async with factory.begin() as session: + await session.execute(delete(DownloadHistory).where(DownloadHistory.id == download_id)) + issue = await session.get(Issue, issue_id) + issue.page_count = 48 + issue.release_date = date(2025, 2, 1) + issue.store_date = date(2024, 12, 26) + creator = Creator(name="Catalog Writer") + session.add(creator) + await session.flush() + session.add(IssueCreator(issue_id=issue_id, creator_id=creator.id, role="writer")) + with ZipFile(source, "w") as archive: + for page in range(54): + archive.writestr(f"page-{page:03}.jpg", f"page {page}".encode()) + archive.writestr( + "ComicInfo.xml", + "Canonical series50-X" + "Local Writer54" + "202412", + ) + return issue_id, source + + +async def test_manual_review_requires_choices_then_imports_paired_copy(identity_probe_db, tmp_path): + _, factory, _ = identity_probe_db + issue_id, source = await gargoyle_case(factory, tmp_path) + original = source.read_bytes() + async with factory() as session: + plan = await read_manual_metadata_plan(session, issue_id) + review = await preview_manual_import_metadata(plan, source, {}) + assert not review.preview.ready + conflicts = {item.key: item for item in review.preview.conflicts} + assert {"issue.page_count", "issue.cover_date", "issue.credits"} <= conflicts.keys() + assert [(o.value, o.selectable) for o in conflicts["issue.cover_date"].options] == [ + ("2025-02-01", True), + ("2024-12-?", False), + ] + assert "archive" in conflicts["issue.page_count"].note.lower() + choices = {item.key: "library" for item in review.preview.conflicts} + choices["issue.page_count"] = "ComicInfo.xml" + choices["issue.credits"] = "ComicInfo.xml" + approved = await preview_manual_import_metadata(plan, source, choices) + assert approved.preview.ready + assert source.read_bytes() == original + async with factory() as session: + prepared = await prepare_manual_issue_import( + session, issue_id=issue_id, file_path=str(source), move_to_library=True + ) + result = await execute_manual_issue_import( + session, + prepared, + metadata_approval=FileMetadataApproval( + review_key=approved.preview.review_key, choices=choices + ), + ) + await session.commit() + with ZipFile(result.library_file.file_path) as archive: + ci = ET.fromstring(archive.read("ComicInfo.xml")) + mi = ET.fromstring(archive.read("MetronInfo.xml")) + assert ci.findtext("PageCount") == mi.findtext("PageCount") == "54" + assert ci.findtext("Year") == "2025" and ci.findtext("Month") == "2" + assert mi.findtext("CoverDate") == "2025-02-01" + assert mi.findtext("StoreDate") == "2024-12-26" + assert ci.findtext("Writer") == mi.findtext("Credits/Credit/Creator") == "Local Writer" + assert ci.find("Issue") is None and ci.find("writer") is None + assert archive.read("page-053.jpg") == b"page 53" + assert (await session.get(Issue, issue_id)).page_count == 48 + assert source.read_bytes() == original + + +@pytest.mark.parametrize("change", ["file", "metadata", "choices", "unresolved"]) +async def test_manual_review_rejects_stale_or_incomplete_approval( + identity_probe_db, tmp_path, change +): + _, factory, _ = identity_probe_db + issue_id, source = await gargoyle_case(factory, tmp_path) + async with factory() as session: + plan = await read_manual_metadata_plan(session, issue_id) + preview = await preview_manual_import_metadata(plan, source, {}) + choices = {item.key: "library" for item in preview.preview.conflicts} + review = await preview_manual_import_metadata(plan, source, choices) + if change == "file": + with ZipFile(source, "a") as archive: + archive.writestr("new-page.jpg", b"new") + elif change == "metadata": + async with factory.begin() as session: + (await session.get(Issue, issue_id)).page_count = 49 + elif change == "choices": + choices["issue.page_count"] = "ComicInfo.xml" + else: + review, choices = preview, {} + before = source.read_bytes() + async with factory() as session: + prepared = await prepare_manual_issue_import( + session, issue_id=issue_id, file_path=str(source), move_to_library=True + ) + with pytest.raises(ManualIssueImportError, match=r"review|preview"): + await execute_manual_issue_import( + session, + prepared, + metadata_approval=FileMetadataApproval( + review_key=review.preview.review_key, choices=choices + ), + ) + await session.rollback() + assert source.read_bytes() == before + async with factory() as session: + assert await session.scalar(select(LibraryFile.id)) is None + assert not list((tmp_path / "comics").rglob("*.cbz")) + + +@pytest.mark.parametrize( + "xml", + [ + "99", + "50-X99", + "50-Xhttps://comicvine.gamespot.com/issue/4000-999999/", + ], +) +async def test_manual_review_cannot_override_issue_identity(identity_probe_db, tmp_path, xml): + _, factory, _ = identity_probe_db + issue_id, source = await gargoyle_case(factory, tmp_path) + with ZipFile(source, "w") as archive: + archive.writestr("page.jpg", b"page") + archive.writestr("ComicInfo.xml", f"{xml}") + before = source.read_bytes() + async with factory() as session: + plan = await read_manual_metadata_plan(session, issue_id) + with pytest.raises(ValueError): + await preview_manual_import_metadata(plan, source, {}) + assert source.read_bytes() == before diff --git a/tests/unit/test_comicinfo_aliases.py b/tests/unit/test_comicinfo_aliases.py new file mode 100644 index 00000000..37c66f45 --- /dev/null +++ b/tests/unit/test_comicinfo_aliases.py @@ -0,0 +1,63 @@ +"""Known ComicInfo aliases remain evidence, not permission to replace metadata.""" + +from xml.etree import ElementTree as ET + +import pytest + +from pullbox.services.archive_metadata_reconciliation import reconcile_archive_metadata +from pullbox.services.archive_metadata_rendering import ( + ArchiveMetadataRenderError, + render_archive_metadata, +) +from tests.unit.test_archive_metadata_rendering import files, snapshots + + +def test_known_aliases_are_read_and_rendered_as_canonical_fields(): + incoming = files( + "50-XLocal Writer" + ) + original = incoming.comicinfo.payload + evidence = reconcile_archive_metadata(incoming) + assert evidence.issue.issue_number_text == "50-X" + assert evidence.issue.credits[0].name == "Local Writer" + assert not evidence.diagnostics + series, issue = snapshots(credits=({"name": "Local Writer", "role": "writer"},)) + rendered = render_archive_metadata(series, issue, incoming) + ci = ET.fromstring(rendered.comicinfo) + mi = ET.fromstring(rendered.metroninfo) + assert ci.findtext("Number").upper() == mi.findtext("Number").upper() == "50-X" + assert ci.findtext("Writer") == mi.findtext("Credits/Credit/Creator") == "Local Writer" + assert ci.find("Issue") is None and ci.find("writer") is None + assert b"" in rendered.comicinfo + assert incoming.comicinfo.payload == original + + +@pytest.mark.parametrize( + "xml", + [ + "50-X50-X", + "50-X3", + "50-X50-X", + "OneTwo", + '50-X', + "One", + "keep me", + '50-X', + ], +) +def test_aliases_do_not_allow_ambiguous_or_unsupported_content(xml): + with pytest.raises(ArchiveMetadataRenderError): + render_archive_metadata(*snapshots(), files(f"{xml}")) + + +def test_alias_issue_number_still_blocks_the_wrong_issue(): + with pytest.raises(ArchiveMetadataRenderError, match="issue_number_conflict"): + render_archive_metadata(*snapshots(), files("3")) + + +def test_alias_writer_cannot_be_silently_replaced(): + with pytest.raises(ArchiveMetadataRenderError, match="unreconciled_field"): + render_archive_metadata( + *snapshots(credits=({"name": "Provider Writer", "role": "writer"},)), + files("Local Writer"), + ) diff --git a/tests/unit/test_issue_import_task.py b/tests/unit/test_issue_import_task.py index 7b0a1e05..ad498b98 100644 --- a/tests/unit/test_issue_import_task.py +++ b/tests/unit/test_issue_import_task.py @@ -136,6 +136,7 @@ async def fake_execute_manual_issue_import( prepared_arg, *, allow_resource_safety_exception=False, + metadata_approval=None, preparation_progress_callback=None, transfer_progress_callback=None, comicinfo_progress_callback=None, @@ -143,6 +144,7 @@ async def fake_execute_manual_issue_import( ): assert prepared_arg is prepared assert allow_resource_safety_exception is True + assert metadata_approval is None assert preparation_progress_callback is not None assert transfer_progress_callback is not None assert comicinfo_progress_callback is not None @@ -218,12 +220,15 @@ async def fake_execute_manual_issue_import( prepared_arg, *, allow_resource_safety_exception=False, + metadata_approval=None, preparation_progress_callback=None, transfer_progress_callback=None, comicinfo_progress_callback=None, cancellation_check=None, ): assert prepared_arg is prepared + assert metadata_approval.review_key == "b" * 64 + assert metadata_approval.choices == {"issue.page_count": "ComicInfo.xml"} assert comicinfo_progress_callback is not None def emit_from_worker_thread() -> None: @@ -272,6 +277,10 @@ def emit_from_worker_thread() -> None: "file_path": "/tmp/imports/Aliens Epic Collection.cbz", "move_to_library": True, "allow_resource_safety_exception": False, + "metadata_approval": { + "review_key": "b" * 64, + "choices": {"issue.page_count": "ComicInfo.xml"}, + }, }, ) From 2c0b12d8a43d0cdfd72eb47c709644d9baa96269 Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 16 Oct 2026 15:46:53 -0700 Subject: [PATCH 06/46] fix(ui): constrain manual import review modal to viewport --- .../partials/issue_import_progress_modal.html | 58 +++++++------- .../e2e/test_manual_import_metadata_review.py | 80 +++++++++++++++++++ 2 files changed, 110 insertions(+), 28 deletions(-) diff --git a/src/pullbox/ui/templates/partials/issue_import_progress_modal.html b/src/pullbox/ui/templates/partials/issue_import_progress_modal.html index 8e49c3ef..de167769 100644 --- a/src/pullbox/ui/templates/partials/issue_import_progress_modal.html +++ b/src/pullbox/ui/templates/partials/issue_import_progress_modal.html @@ -1,3 +1,4 @@ + diff --git a/tests/e2e/test_manual_import_metadata_review.py b/tests/e2e/test_manual_import_metadata_review.py index d7ca9225..370d65a2 100644 --- a/tests/e2e/test_manual_import_metadata_review.py +++ b/tests/e2e/test_manual_import_metadata_review.py @@ -117,3 +117,83 @@ def start(route): }, } ] + + +@pytest.mark.parametrize( + "theme,width,height", [("light", 1280, 720), ("dark", 390, 844), ("light", 844, 390)] +) +def test_long_manual_review_scrolls_between_visible_header_and_actions( + authed_page, seeded_server, theme, width, height +): + page = authed_page + page.set_viewport_size({"width": width, "height": height}) + conflicts = [ + { + "key": "issue.cover_date", + "label": "Issue cover date", + "options": [ + {"source": "library", "value": "2025-02-01", "selectable": True}, + {"source": "ComicInfo.xml", "value": "2024-12-?", "selectable": False}, + ], + }, + { + "key": "issue.page_count", + "label": "Issue page count", + "options": [ + {"source": "library", "value": "48", "selectable": True}, + {"source": "ComicInfo.xml", "value": "54", "selectable": True}, + ], + }, + { + "key": "issue.credits", + "label": "Issue credits", + "options": [ + { + "source": "library", + "value": "\n".join(f"Catalog creator {index} (artist)" for index in range(13)), + "selectable": True, + }, + {"source": "ComicInfo.xml", "value": "Rafael Grampa (writer)", "selectable": True}, + ], + }, + ] + page.route( + "**/api/v1/issues/2/import-file/metadata-preview", + lambda route: route.fulfill( + json={ + "ready": False, + "review_key": "a" * 64, + "context_key": "b" * 64, + "conflicts": conflicts, + } + ), + ) + page.goto(f"{seeded_server}/issues/2") + page.evaluate("theme => applyTheme(theme)", theme) + shell_scroll = page.locator("#content").evaluate("el => el.scrollTop") + page.evaluate("""() => Alpine.$data(document.querySelector('[data-testid="issue-detail-page"]')) + .applyImportFileSelection({path:'/imports/gargoyle-003.cbz', name:'gargoyle-003.cbz', size:183465435})""") + modal = page.get_by_test_id("issue-import-modal") + expect(modal.get_by_text("Issue credits", exact=True)).to_be_visible() + panel = modal.get_by_role("dialog") + expect(panel).to_be_in_viewport(ratio=1) + header = panel.locator(":scope > .modal-header") + footer = panel.locator(":scope > .modal-footer") + body = panel.locator(":scope > .modal-body") + expect(header).to_be_in_viewport(ratio=1) + expect(footer).to_be_in_viewport(ratio=1) + assert body.evaluate("el => el.scrollHeight > el.clientHeight && el.clientHeight > 40") + header_before, footer_before = header.bounding_box(), footer.bounding_box() + modal.get_by_role("radio", name="library: 2025-02-01", exact=True).check() + modal.get_by_role("radio", name="ComicInfo.xml: 54", exact=True).check() + modal.get_by_role("radio", name="library: Catalog creator 0", exact=False).check() + assert body.evaluate("el => el.scrollTop > 0") + assert header.bounding_box() == header_before + assert footer.bounding_box() == footer_before + assert page.locator("#content").evaluate("el => el.scrollTop") == shell_scroll + button = footer.get_by_role("button", name="Import reviewed file", exact=True) + expect(button).to_be_enabled() + expect(button).to_be_in_viewport(ratio=1) + page.screenshot(path=f"/private/tmp/bug8-scroll-{theme}-{width}x{height}.png") + footer.get_by_role("button", name="Close", exact=True).click() + expect(modal).not_to_be_visible() From 5dbd1668bd5b26ba74c936be34af1641cb2d2486 Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 16 Oct 2026 16:18:58 -0700 Subject: [PATCH 07/46] fix: recover retained direct-download metadata without reacquiring --- src/pullbox/api/v1/intervention_metadata.py | 137 +++++ src/pullbox/api/v1/issue_metadata_links.py | 5 + src/pullbox/api/v1/router.py | 2 + .../services/direct_metadata_recovery.py | 550 ++++++++++++++++++ src/pullbox/services/intervention_service.py | 4 + .../services/issue_file_metadata_recovery.py | 28 +- src/pullbox/services/issue_import_service.py | 9 +- src/pullbox/ui/intervention_filter_helpers.py | 4 + .../ui/static/js/issue-file-metadata.js | 22 +- src/pullbox/ui/static/js/pullbox.js | 4 + .../partials/intervention_content.html | 1 + .../templates/partials/intervention_list.html | 5 + .../partials/issue_file_metadata.html | 27 +- .../utilities/executors/file_metadata.py | 29 +- tests/e2e/test_issue_file_metadata.py | 82 +++ .../test_direct_metadata_recovery.py | 392 +++++++++++++ tests/ui/test_intervention_filter_helpers.py | 22 + 17 files changed, 1303 insertions(+), 20 deletions(-) create mode 100644 src/pullbox/api/v1/intervention_metadata.py create mode 100644 src/pullbox/services/direct_metadata_recovery.py create mode 100644 tests/integration/metadata_identity/test_direct_metadata_recovery.py diff --git a/src/pullbox/api/v1/intervention_metadata.py b/src/pullbox/api/v1/intervention_metadata.py new file mode 100644 index 00000000..e3cda9d8 --- /dev/null +++ b/src/pullbox/api/v1/intervention_metadata.py @@ -0,0 +1,137 @@ +"""Explicit retained-file review; no provider requests or acquisition replanning.""" + +import json + +from fastapi import APIRouter, HTTPException +from sqlalchemy import JSON, Text, cast, select, type_coerce + +from pullbox.api.deps import DbSession, InteractiveOperatorUser +from pullbox.api.v1.issue_metadata_links import LocalId, file_job_status +from pullbox.models.pending_match import PendingMatch, PendingMatchStatus +from pullbox.schemas.issue_file_metadata import FileMetadataApproval, FileMetadataReview +from pullbox.services.direct_metadata_recovery import ( + DirectMetadataRecoveryError, + DirectMetadataRecoveryPreview, + prepare_direct_metadata_recovery, +) +from pullbox.services.issue_file_metadata import file_metadata_error +from pullbox.services.library_mutation_coordination import lock_file_mutation_admission +from pullbox.utilities.import_guards import ensure_no_active_import_file_mutation +from pullbox.utilities.models import JobState, JobType, UtilityJob +from pullbox.utilities.router import _get_manager, _schedule_dispatch + +router = APIRouter(prefix="/intervention", tags=["intervention"]) + + +def _error(exc: ValueError | OSError) -> HTTPException: + return HTTPException( + 409, str(exc) if isinstance(exc, DirectMetadataRecoveryError) else file_metadata_error(exc) + ) + + +@router.post("/{pending_id}/file-metadata/preview", response_model=DirectMetadataRecoveryPreview) +async def preview( + pending_id: LocalId, + session: DbSession, + _user: InteractiveOperatorUser, + body: FileMetadataReview | None = None, +) -> DirectMetadataRecoveryPreview: + try: + return ( + await prepare_direct_metadata_recovery( + session, pending_id, choices=body.choices if body else None + ) + ).preview + except (ValueError, OSError) as exc: + raise _error(exc) from exc + + +@router.post("/{pending_id}/file-metadata/write", status_code=202) +async def write( + pending_id: LocalId, + body: FileMetadataApproval, + session: DbSession, + user: InteractiveOperatorUser, +) -> dict[str, str]: + try: + prepared = await prepare_direct_metadata_recovery(session, pending_id, choices=body.choices) + if not prepared.preview.ready or prepared.preview.review_key != body.review_key: + raise DirectMetadataRecoveryError( + "Review each conflicting field and update the preview before recovering this file." + ) + await lock_file_mutation_admission(session) + await ensure_no_active_import_file_mutation(session) + document = JSON().with_variant(Text(), "sqlite") + existing = await session.scalar( + select(UtilityJob) + .where( + UtilityJob.job_type == JobType.FILE_METADATA, + UtilityJob.state.in_( + ( + JobState.QUEUED, + JobState.RUNNING, + JobState.PAUSING, + JobState.PAUSED, + JobState.CANCELLING, + ) + ), + type_coerce(cast(UtilityJob.config, document), JSON)["issue_id"].as_integer() + == prepared.owner.issue_id, + ) + .limit(1) + ) + if existing is not None: + config = json.loads(existing.config) + if ( + config.get("recovery_pending_id") != pending_id + or config.get("review_key") != body.review_key + ): + raise DirectMetadataRecoveryError( + "A metadata job already owns this issue. " + "Wait for it to finish or cancel it before retrying." + ) + return {"job_id": existing.id, "state": existing.state} + manager = _get_manager() + job = await manager.create_job( + session, + JobType.FILE_METADATA, + f"Recover file metadata: {prepared.preview.file_name}", + { + "issue_id": prepared.owner.issue_id, + "recovery_pending_id": pending_id, + "review_key": body.review_key, + "choices": body.choices, + }, + created_by=user.username, + ) + await session.commit() + _schedule_dispatch(manager) + return {"job_id": job.id, "state": job.state} + except (ValueError, OSError) as exc: + raise _error(exc) from exc + + +@router.get("/{pending_id}/file-metadata/job") +async def job( + pending_id: LocalId, session: DbSession, _user: InteractiveOperatorUser +) -> dict[str, object]: + document = JSON().with_variant(Text(), "sqlite") + row = await session.scalar( + select(UtilityJob) + .where( + UtilityJob.job_type == JobType.FILE_METADATA, + type_coerce(cast(UtilityJob.config, document), JSON)["recovery_pending_id"].as_integer() + == pending_id, + ) + .order_by(UtilityJob.created_at.desc(), UtilityJob.id.desc()) + .limit(1) + ) + pending = await session.get(PendingMatch, pending_id) + result: dict[str, object] = await file_job_status(session, row) if row else {"job": None} + result["resolved"] = bool( + pending + and pending.status == PendingMatchStatus.APPROVED + and (pending.match_details or {}).get("resolution_reason") + in {"metadata_completed", "metadata_superseded"} + ) + return result diff --git a/src/pullbox/api/v1/issue_metadata_links.py b/src/pullbox/api/v1/issue_metadata_links.py index 1cd8f639..5d6489d7 100644 --- a/src/pullbox/api/v1/issue_metadata_links.py +++ b/src/pullbox/api/v1/issue_metadata_links.py @@ -180,6 +180,11 @@ async def file_job( ) if job is None: return {"job": None} + return await file_job_status(session, job) + + +async def file_job_status(session: DbSession, job: UtilityJob) -> dict[str, object]: + """One progress projection for both issue writes and retained-download recovery.""" item = await session.scalar( select(UtilityJobItem).where(UtilityJobItem.job_id == job.id).limit(1) ) diff --git a/src/pullbox/api/v1/router.py b/src/pullbox/api/v1/router.py index fbd13aec..bf22ec9b 100644 --- a/src/pullbox/api/v1/router.py +++ b/src/pullbox/api/v1/router.py @@ -22,6 +22,7 @@ from pullbox.api.v1.import_safety_bulk import router as import_safety_bulk_router from pullbox.api.v1.indexers import router as indexers_router from pullbox.api.v1.intervention import router as intervention_router +from pullbox.api.v1.intervention_metadata import router as intervention_metadata_router from pullbox.api.v1.issue_metadata_links import router as issue_metadata_links_router from pullbox.api.v1.issues import router as issues_router from pullbox.api.v1.library import router as library_router @@ -53,6 +54,7 @@ v1_router.include_router(story_arc_placements_router) v1_router.include_router(issues_router) v1_router.include_router(issue_metadata_links_router) +v1_router.include_router(intervention_metadata_router) v1_router.include_router(library_router) v1_router.include_router(metadata_identities_router) v1_router.include_router(metadata_sources_router) diff --git a/src/pullbox/services/direct_metadata_recovery.py b/src/pullbox/services/direct_metadata_recovery.py new file mode 100644 index 00000000..5b97a1d5 --- /dev/null +++ b/src/pullbox/services/direct_metadata_recovery.py @@ -0,0 +1,550 @@ +"""Review retained direct-download files without reacquiring them.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +from dataclasses import asdict, dataclass, replace +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from pydantic import BaseModel, Field, TypeAdapter +from sqlalchemy import select + +from pullbox.config import get_settings +from pullbox.core.library_file_ownership import build_file_identity_signature +from pullbox.models import Issue, LibraryFile +from pullbox.models.direct_acquisition import ( + DirectAcquisitionAttempt, + DirectAcquisitionState, + DirectArtifactAttempt, + DirectArtifactFailureClass, + DirectArtifactState, +) +from pullbox.models.download import DownloadHistory, DownloadState +from pullbox.models.issue import IssueStatus +from pullbox.models.pending_match import PendingMatch, PendingMatchStatus +from pullbox.schemas.issue_file_metadata import ( + FileMetadataApproval, + FileMetadataChange, + FileMetadataConflict, +) +from pullbox.services.direct_acquisition_state import ( + advance_acquisition_progress, + transition_acquisition, + transition_artifact, +) +from pullbox.services.direct_download_history_adapter import sync_direct_download_history +from pullbox.services.issue_file_metadata import prepare_file_metadata, write_file_metadata +from pullbox.services.issue_import_service import ( + execute_manual_issue_import, + prepare_manual_issue_import, +) +from pullbox.services.library_mutation_coordination import lock_file_mutation_admission +from pullbox.services.manual_import_metadata_review import preview_manual_import_metadata +from pullbox.services.manual_paired_metadata import read_manual_metadata_plan +from pullbox.services.post_processing_operation_progress import ( + project_post_processing_operation_progress, +) +from pullbox.utilities.import_guards import ensure_no_active_import_file_mutation +from pullbox.utilities.models import UtilityJob, UtilityJobItem + +if TYPE_CHECKING: + from uuid import UUID + + from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + + from pullbox.schemas.issue_file_metadata import FileMetadataChoices + from pullbox.services.archive_metadata_binding import ArchiveMetadataTarget + from pullbox.utilities.executors.archive_subprocess import ControlCheck, ProgressCallback + + +class DirectMetadataRecoveryError(ValueError): + """An operator-actionable recovery error without a download URL or secret.""" + + +class DirectMetadataRecoveryPreview(BaseModel): + issue_id: int + file_name: str + review_key: str + unchanged: bool = False + ready: bool + conflicts: list[FileMetadataConflict] = Field(default_factory=list) + changes: list[FileMetadataChange] = Field(default_factory=list) + converts_to_cbz: bool = False + restores_copy: bool = False + superseded: bool = False + + +@dataclass(frozen=True) +class RecoveryOwner: + pending_id: int + issue_id: int + attempt_id: int + artifact_id: int + history_id: int + plan_revision: int + marker: dict[str, Any] + safety_review: object + file_id: int | None + file_path: str | None + source: Path + superseded: bool + + +@dataclass(frozen=True) +class PreparedRecovery: + owner: RecoveryOwner + metadata_key: str + preview: DirectMetadataRecoveryPreview + file_target: ArchiveMetadataTarget | None + + +async def _read_owner(session: AsyncSession, pending_id: int) -> RecoveryOwner: + pending = await session.get(PendingMatch, pending_id, populate_existing=True) + if ( + pending is None + or pending.status != PendingMatchStatus.PENDING + or (pending.match_details or {}).get("failure_code") != "direct_metadata_review_required" + ): + raise DirectMetadataRecoveryError( + "This metadata recovery is no longer pending. Refresh the queue." + ) + attempt_id = pending.match_details.get("direct_attempt_id") + if type(attempt_id) is not int: + raise DirectMetadataRecoveryError("The saved download identity is unavailable.") + attempt = await session.get(DirectAcquisitionAttempt, attempt_id, populate_existing=True) + if ( + attempt is None + or attempt.issue_id != pending.issue_id + or attempt.state is not DirectAcquisitionState.INTERVENTION + or attempt.failure_class is not DirectArtifactFailureClass.POST_PROCESS + or attempt.failure_code != "direct_metadata_review_required" + ): + raise DirectMetadataRecoveryError( + "The download changed. Refresh the queue before reviewing metadata." + ) + selected = list( + await session.scalars( + select(DirectArtifactAttempt) + .where( + DirectArtifactAttempt.acquisition_attempt_id == attempt.id, + DirectArtifactAttempt.is_selected.is_(True), + ) + .limit(2) + ) + ) + marker = (attempt.plan_snapshot or {}).get("paired_metadata_copy") + if ( + len(selected) != 1 + or not isinstance(marker, dict) + or marker.get("artifact_id") != selected[0].id + or marker.get("plan_revision") != attempt.plan_revision + or type(marker.get("file_id")) is not int + or selected[0].state is not DirectArtifactState.INTERVENTION + or selected[0].failure_code != "direct_metadata_review_required" + or not selected[0].quarantine_path + ): + raise DirectMetadataRecoveryError( + "The retained download evidence changed. Review the issue before retrying." + ) + histories = list( + await session.scalars( + select(DownloadHistory) + .where( + DownloadHistory.external_id == f"direct:{attempt.id}", + ) + .limit(2) + ) + ) + if ( + len(histories) != 1 + or histories[0].issue_id != pending.issue_id + or histories[0].imported_at is not None + or histories[0].state is not DownloadState.FAILED + ): + raise DirectMetadataRecoveryError( + "The download history changed. Refresh the queue before retrying." + ) + files = list( + await session.scalars( + select(LibraryFile).where(LibraryFile.issue_id == pending.issue_id).limit(2) + ) + ) + if len(files) > 1: + raise DirectMetadataRecoveryError( + "This issue has multiple registered files. Reconcile them before recovery." + ) + file = files[0] if files else None + if file is None: + issue = await session.get(Issue, pending.issue_id) + if issue is None or issue.status is not IssueStatus.WANTED: + raise DirectMetadataRecoveryError( + "The library copy is missing. Mark this issue Wanted " + "before explicitly restoring it." + ) + return RecoveryOwner( + pending.id, + pending.issue_id, + attempt.id, + selected[0].id, + histories[0].id, + attempt.plan_revision, + dict(marker), + (attempt.plan_snapshot or {}).get("safety_review"), + file.id if file else None, + file.file_path if file else None, + Path(selected[0].quarantine_path), + file is not None and (file.id != marker["file_id"] or attempt.library_file_id != file.id), + ) + + +async def _check_source(owner: RecoveryOwner) -> None: + try: + signature = await asyncio.to_thread(build_file_identity_signature, owner.source) + valid = ( + not owner.source.is_symlink() + and owner.source.parent.name == f"artifact-{owner.artifact_id}" + and owner.source.parent.parent.name == f"attempt-{owner.attempt_id}" + and signature == owner.marker.get("source_signature") + ) + except OSError: + valid = False + if not valid: + raise DirectMetadataRecoveryError( + "The retained download is missing or changed. Restore the original file before " + "retrying; Pullbox will not download it again." + ) + + +async def prepare_direct_metadata_recovery( + session: AsyncSession, + pending_id: int, + *, + choices: FileMetadataChoices | None = None, +) -> PreparedRecovery: + if not get_settings().metadata_paired_direct_writer_enabled: + raise DirectMetadataRecoveryError( + "Enable direct-download paired metadata writing before recovering this file." + ) + owner = await _read_owner(session, pending_id) + await ensure_no_active_import_file_mutation(session) + await session.commit() + await _check_source(owner) + file_target = None + if owner.file_id is not None: + prepared = await prepare_file_metadata(session, owner.issue_id, choices=choices) + file_target = prepared.target + if prepared.target.binding.library_file_id != owner.file_id: + raise DirectMetadataRecoveryError("The library file changed. Preview again.") + signature = await asyncio.to_thread(build_file_identity_signature, prepared.target.path) + owner = replace( + owner, superseded=owner.superseded or signature != owner.marker.get("copy_signature") + ) + preview = DirectMetadataRecoveryPreview( + issue_id=owner.issue_id, + **prepared.preview.model_dump(exclude={"file_id"}), + superseded=owner.superseded, + ) + if owner.superseded and (not preview.ready or not preview.unchanged): + raise DirectMetadataRecoveryError( + "A different library copy now owns this issue. Review its file metadata on " + "the issue page first; this recovery will not overwrite it." + ) + else: + if not get_settings().metadata_paired_import_writer_enabled: + raise DirectMetadataRecoveryError( + "Enable paired import metadata writing before restoring the retained download." + ) + plan = await read_manual_metadata_plan(session, owner.issue_id) + if plan.existing_file is not None or not plan.policy.update_embedded_comicinfo_from_match: + raise DirectMetadataRecoveryError( + "The import settings or library binding changed. " + "Enable embedded metadata updates and preview again." + ) + await session.commit() + reviewed = await preview_manual_import_metadata(plan, owner.source, choices or {}) + preview = DirectMetadataRecoveryPreview( + issue_id=owner.issue_id, + file_name=owner.source.name, + review_key=reviewed.preview.review_key, + ready=reviewed.preview.ready, + conflicts=reviewed.preview.conflicts, + restores_copy=True, + converts_to_cbz=owner.source.suffix.lower() in {".cbr", ".cb7", ".pdf"}, + ) + metadata_key = preview.review_key + preview.review_key = hashlib.sha256( + json.dumps( + {"owner": asdict(owner), "metadata_key": metadata_key}, + default=str, + sort_keys=True, + ).encode() + ).hexdigest() + return PreparedRecovery(owner, metadata_key, preview, file_target) + + +async def _complete( + session: AsyncSession, + owner: RecoveryOwner, + file: LibraryFile, + operation: UUID, + review_key: str, +) -> None: + await lock_file_mutation_admission(session) + await ensure_no_active_import_file_mutation(session) + current = await _read_owner(session, owner.pending_id) + # The approved writer may convert the owned path or restore its missing registration. + expected = replace( + owner, + file_id=file.id, + file_path=file.file_path, + superseded=current.superseded if owner.file_id is None else owner.superseded, + ) + # Supersession can also be detected from changed bytes at the same registered path. + if ( + replace(current, superseded=owner.superseded if owner.file_id else current.superseded) + != expected + ): + raise DirectMetadataRecoveryError( + "Recovery ownership changed before completion. The saved file needs review." + ) + pending = await session.get(PendingMatch, owner.pending_id) + attempt = await session.get(DirectAcquisitionAttempt, owner.attempt_id) + artifact = await session.get(DirectArtifactAttempt, owner.artifact_id) + assert pending is not None and attempt is not None and artifact is not None + now = datetime.now(UTC) + reason = "metadata_superseded" if owner.superseded else "metadata_completed" + if owner.superseded: + transition_acquisition(attempt, DirectAcquisitionState.FAILED, at=now) + transition_artifact(artifact, DirectArtifactState.FAILED, at=now) + attempt.error_message = ( + "Superseded by a verified library copy; " + "the previous transfer was not marked successful." + ) + artifact.error_message = attempt.error_message + else: + transition_acquisition(attempt, DirectAcquisitionState.POST_PROCESSING, at=now) + transition_artifact(artifact, DirectArtifactState.VALIDATING, at=now) + transition_acquisition(attempt, DirectAcquisitionState.COMPLETED, at=now) + transition_artifact(artifact, DirectArtifactState.COMPLETED, at=now) + attempt.library_file_id = file.id + attempt.failure_class = artifact.failure_class = None + attempt.failure_code = artifact.failure_code = None + attempt.error_message = artifact.error_message = None + issue = await session.get(Issue, owner.issue_id) + assert issue is not None + issue.status = IssueStatus.OWNED + advance_acquisition_progress( + attempt, + revision=attempt.progress_revision + 1, + snapshot={ + "schema_version": 1, + "stage": "completed" if not owner.superseded else "superseded", + "percent": 100, + "artifact_attempt_id": artifact.id, + }, + ) + history = await sync_direct_download_history( + session, attempt, artifact, at=now, final_path=file.file_path + ) + await project_post_processing_operation_progress(session, history, None) + pending.status = PendingMatchStatus.APPROVED + pending.resolved_at = now + pending.resolved_by = "metadata_recovery" + pending.match_details = { + **pending.match_details, + "resolution_reason": reason, + "metadata_operation_id": str(operation), + "metadata_review_key": review_key, + "resolved_library_file_id": file.id, + } + + +async def run_direct_metadata_recovery( + factory: async_sessionmaker[AsyncSession], + pending_id: int, + review_key: str, + operation: UUID, + *, + choices: FileMetadataChoices, + limit: int, + check_control: ControlCheck, + progress: ProgressCallback, + job_id: str | None = None, +) -> str: + async with factory() as reader: + pending = await reader.get(PendingMatch, pending_id) + if ( + pending is not None + and pending.status == PendingMatchStatus.APPROVED + and ( + pending.match_details.get("metadata_operation_id") == str(operation) + and pending.match_details.get("metadata_review_key") == review_key + ) + ): + return "already_completed" + prepared = await prepare_direct_metadata_recovery(reader, pending_id, choices=choices) + if not prepared.preview.ready or prepared.preview.review_key != review_key: + raise DirectMetadataRecoveryError( + "The file, metadata, or recovery state changed. Preview again before writing." + ) + owner = prepared.owner + + async def control() -> None: + await check_control() + async with factory() as reader: + current = await _read_owner(reader, pending_id) + if replace(current, superseded=owner.superseded) != owner: + raise DirectMetadataRecoveryError( + "The recovery or library binding changed. No further file changes will be made." + ) + await _check_source(owner) + + await control() + if job_id is not None: + async with factory.begin() as writer: + item = await writer.get(UtilityJobItem, operation.hex) + if item is None or item.job_id != job_id: + raise DirectMetadataRecoveryError("The recovery job no longer owns this operation.") + item.before_state = json.dumps( + { + "direct_metadata_recovery": { + "owner": asdict(owner), + "review_key": review_key, + } + }, + default=str, + ) + if owner.file_id is None: + async with factory() as writer: + plan = await prepare_manual_issue_import( + writer, issue_id=owner.issue_id, file_path=str(owner.source), move_to_library=True + ) + result = await execute_manual_issue_import( + writer, + plan, + metadata_approval=FileMetadataApproval( + review_key=prepared.metadata_key, choices=choices + ), + cancellation_check=control, + preparation_progress_callback=progress, + comicinfo_progress_callback=progress, + ) + await _complete(writer, owner, result.library_file, operation, review_key) + await writer.commit() + return "restored" + outcome = ( + "superseded" + if owner.superseded + else await write_file_metadata( + factory, + owner.issue_id, + prepared.metadata_key, + operation, + limit=limit, + choices=choices, + check_control=control, + progress=progress, + job_id=job_id, + ) + ) + if job_id is not None and await recover_direct_metadata_operation(factory, operation, job_id): + return outcome + async with factory.begin() as writer: + if outcome in {"superseded", "unchanged"}: + from pullbox.services.archive_metadata_binding import revalidate_archive_metadata_target + + assert prepared.file_target is not None + await lock_file_mutation_admission(writer) + await revalidate_archive_metadata_target(writer, prepared.file_target) + await asyncio.to_thread(prepared.file_target.check_unchanged) + file = await writer.get(LibraryFile, owner.file_id) + if file is None or file.issue_id != owner.issue_id: + raise DirectMetadataRecoveryError("The library binding changed before completion.") + await _complete(writer, owner, file, operation, review_key) + return outcome + + +async def recover_direct_metadata_operation( + factory: async_sessionmaker[AsyncSession], + operation: UUID, + job_id: str, +) -> bool: + """A lost acknowledgement needs the exact writer journal, never OWNED alone.""" + from pullbox.models.archive_metadata_publication import PublicationState + from pullbox.services.archive_metadata_binding import revalidate_archive_metadata_target + from pullbox.services.archive_metadata_publication import ( + inspect_archive_publication, + load_archive_publication, + ) + from pullbox.services.issue_file_metadata import recover_file_metadata + from pullbox.services.native_file_metadata import recover_native_file_metadata + + async with factory() as session: + job = await session.get(UtilityJob, job_id) + item = await session.get(UtilityJobItem, operation.hex) + if job is None or item is None or item.job_id != job_id: + return False + config = json.loads(job.config) + saved = json.loads(item.before_state or "{}").get("direct_metadata_recovery") + if not saved: + return False + owner = TypeAdapter(RecoveryOwner).validate_python(saved["owner"]) + if ( + config.get("recovery_pending_id") != owner.pending_id + or config.get("issue_id") != owner.issue_id + or config.get("review_key") != saved["review_key"] + ): + raise DirectMetadataRecoveryError( + "The recovery job evidence changed. Review the queue entry again." + ) + pending = await session.get(PendingMatch, owner.pending_id) + if ( + pending is not None + and pending.status == PendingMatchStatus.APPROVED + and pending.match_details.get("metadata_operation_id") == str(operation) + ): + return True + # Restoration commits its new registration and recovery result together. + if owner.file_id is None: + return False + receipt = await load_archive_publication(session, operation) + await _check_source(owner) + if receipt is not None: + if ( + receipt.plan.metadata_job_id != job_id + or receipt.plan.target.binding.library_file_id != owner.file_id + ): + raise DirectMetadataRecoveryError( + "The completed metadata journal belongs to a different file." + ) + if await recover_file_metadata(factory, operation) != PublicationState.FINALIZED: + return False + async with factory() as session: + receipt = await load_archive_publication(session, operation) + assert receipt is not None + inspected = await inspect_archive_publication(receipt) + if inspected.digest != receipt.plan.output_digest: + raise DirectMetadataRecoveryError( + "The written file changed after publication. Review it before completing recovery." + ) + elif await recover_native_file_metadata(factory, owner.issue_id, operation) is None: + return False + async with factory() as session: + prepared = await prepare_file_metadata(session, owner.issue_id) + if not prepared.preview.ready or not prepared.preview.unchanged: + raise DirectMetadataRecoveryError( + "The recovered file metadata changed. Preview the queue entry again." + ) + async with factory.begin() as session: + await lock_file_mutation_admission(session) + await revalidate_archive_metadata_target(session, prepared.target) + await asyncio.to_thread(prepared.target.check_unchanged) + file = await session.get(LibraryFile, owner.file_id) + if file is None or file.issue_id != owner.issue_id: + raise DirectMetadataRecoveryError("The library binding changed after publication.") + await _complete(session, owner, file, operation, saved["review_key"]) + return True diff --git a/src/pullbox/services/intervention_service.py b/src/pullbox/services/intervention_service.py index 054a6ccd..c0f9dead 100644 --- a/src/pullbox/services/intervention_service.py +++ b/src/pullbox/services/intervention_service.py @@ -527,6 +527,10 @@ async def retry_direct_recovery( prior = await session.get(DirectAcquisitionAttempt, attempt_id) if prior is None or prior.issue_id != pending_match.issue_id: raise ValueError("Pending match has an invalid direct attempt reference") + if prior.failure_code == "direct_metadata_review_required": + raise ValueError( + "Review the retained file metadata instead of refreshing download routes." + ) fresh = DirectAcquisitionAttempt( request_key=f"direct-recovery:{uuid4().hex}", diff --git a/src/pullbox/services/issue_file_metadata_recovery.py b/src/pullbox/services/issue_file_metadata_recovery.py index 089473e3..5f400d98 100644 --- a/src/pullbox/services/issue_file_metadata_recovery.py +++ b/src/pullbox/services/issue_file_metadata_recovery.py @@ -89,6 +89,14 @@ async def _recover_native_metadata_jobs(factory: async_sessionmaker[AsyncSession if type(issue_id) is not int or issue_id <= 0: continue operation = UUID(item_id) + if payload.get("recovery_pending_id") is not None: + from pullbox.services.direct_metadata_recovery import ( + recover_direct_metadata_operation, + ) + + if await recover_direct_metadata_operation(factory, operation, job_id): + await _complete_metadata_item(factory, operation, job_id) + continue if await recover_native_file_metadata(factory, issue_id, operation) is None: continue await _complete_metadata_item(factory, operation, job_id) @@ -137,6 +145,24 @@ async def recover_file_metadata_jobs(factory: async_sessionmaker[AsyncSession]) continue try: state = await recover_file_metadata(factory, receipt.operation_id) + if state is not PublicationState.FINALIZED: + continue + async with factory() as session: + job = await session.get(UtilityJob, receipt.plan.metadata_job_id) + is_recovery = ( + job is not None + and json.loads(job.config).get("recovery_pending_id") is not None + ) + if is_recovery: + # A settled archive alone does not resolve its download recovery entry. + from pullbox.services.direct_metadata_recovery import ( + recover_direct_metadata_operation, + ) + + if not await recover_direct_metadata_operation( + factory, receipt.operation_id, receipt.plan.metadata_job_id + ): + continue except (ValueError, OSError) as exc: logger.warning( "file_metadata_recovery_needs_review", @@ -150,8 +176,6 @@ async def recover_file_metadata_jobs(factory: async_sessionmaker[AsyncSession]) job.error_message = file_metadata_error(exc) await project_utility_operation_progress(session, job) continue - if state is not PublicationState.FINALIZED: - continue await _complete_metadata_item( factory, receipt.operation_id, receipt.plan.metadata_job_id ) diff --git a/src/pullbox/services/issue_import_service.py b/src/pullbox/services/issue_import_service.py index 479fbf2b..dd346c49 100644 --- a/src/pullbox/services/issue_import_service.py +++ b/src/pullbox/services/issue_import_service.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import inspect from dataclasses import dataclass, replace from pathlib import Path from typing import TYPE_CHECKING, Any @@ -262,7 +263,9 @@ async def converter_with_progress( async def conversion_progress(stage: str, current: int, total: int, unit: str) -> None: if preparation_progress_callback is not None: - preparation_progress_callback(stage, current, total, unit) + reported = preparation_progress_callback(stage, current, total, unit) + if inspect.isawaitable(reported): + await reported return await convert_file_interruptible( source, @@ -298,7 +301,9 @@ async def materialize_cbz_with_progress( async def paired_progress(stage: str, current: int, total: int, unit: str) -> None: if progress_callback is not None: - progress_callback(stage, current, total, unit) + reported = progress_callback(stage, current, total, unit) + if inspect.isawaitable(reported): + await reported try: return await materialize_manual_metadata( diff --git a/src/pullbox/ui/intervention_filter_helpers.py b/src/pullbox/ui/intervention_filter_helpers.py index d10d9f66..6ac95be3 100644 --- a/src/pullbox/ui/intervention_filter_helpers.py +++ b/src/pullbox/ui/intervention_filter_helpers.py @@ -328,6 +328,8 @@ def build_intervention_item_meta(pending_match: Any) -> dict[str, object]: "rejection_reason": str(details.get("rejection_reason") or "").strip(), "lane": lane, "recovery_summary": _direct_recovery_summary(details) if lane == "recovery" else "", + "metadata_recovery": lane == "recovery" + and details.get("failure_code") == "direct_metadata_review_required", } @@ -349,6 +351,8 @@ def _direct_source_label(details: dict[str, object]) -> str: def _direct_recovery_summary(details: dict[str, object]) -> str: """Describe the acquisition failure without implying the semantic match was wrong.""" + if details.get("failure_code") == "direct_metadata_review_required": + return "Downloaded file needs metadata review; no new download is needed" host = _DIRECT_ARTIFACT_HOST_LABELS.get( str(details.get("artifact_host_kind") or "").strip(), "Artifact host", diff --git a/src/pullbox/ui/static/js/issue-file-metadata.js b/src/pullbox/ui/static/js/issue-file-metadata.js index 04583a7b..0a36fe71 100644 --- a/src/pullbox/ui/static/js/issue-file-metadata.js +++ b/src/pullbox/ui/static/js/issue-file-metadata.js @@ -1,10 +1,11 @@ /* Preview and approve one managed comic; native conversion is explicit. */ function issueFileMetadata(issueId) { - var endpoint = "/api/v1/issues/" + issueId + "/file-metadata"; + var endpoint = issueId ? "/api/v1/issues/" + issueId + "/file-metadata" : ""; return { + recovery: !issueId, resolved: false, open: false, busy: false, error: "", preview: null, job: null, choices: {}, choicesDirty: false, timer: null, disposed: false, trigger: null, generation: 0, jobGeneration: 0, - init: function () { this.loadJob(); }, + init: function () { if (endpoint) this.loadJob(); }, destroy: function () { this.disposed = true; this.generation += 1; clearTimeout(this.timer); }, request: async function (url, body) { var response = await fetch(url, body === undefined ? { cache: "no-store" } : { @@ -23,7 +24,9 @@ function issueFileMetadata(issueId) { var data = await this.request(endpoint + "/job"); if (this.disposed || generation !== this.jobGeneration) return; this.job = data.job; + if (this.recovery) this.resolved = !!data.resolved; if (wasActive && this.job && this.job.state === "COMPLETED") { + if (this.recovery) return; window.dispatchEvent(new CustomEvent("issue-file-metadata-complete")); var target = document.getElementById("issue-metadata-content"); if (target) await htmx.ajax("GET", "/htmx/issues/" + issueId + "/metadata", { target: target, swap: "morph:outerHTML" }); @@ -31,6 +34,20 @@ function issueFileMetadata(issueId) { } catch (error) { if (!this.disposed) this.error = error.message; } if (!this.disposed && this.active()) this.timer = setTimeout(() => this.loadJob(), 1000); }, + showRecovery: async function (pendingId) { + if (!this.recovery || this.busy || !Number.isSafeInteger(pendingId) || pendingId <= 0) return; + clearTimeout(this.timer); + this.jobGeneration += 1; + this.generation += 1; + endpoint = "/api/v1/intervention/" + pendingId + "/file-metadata"; + this.preview = null; this.job = null; this.choices = {}; this.error = ""; this.resolved = false; + this.trigger = document.activeElement; + this.open = true; this.busy = true; + this.$nextTick(() => this.$refs.dialog.focus({ preventScroll: true })); + await this.loadJob(); + this.busy = false; + if (this.open && !this.active() && !this.resolved && !(this.job && this.job.state === "PAUSED")) await this.loadPreview(); + }, show: async function () { this.trigger = document.activeElement; this.open = true; @@ -41,6 +58,7 @@ function issueFileMetadata(issueId) { this.open = false; this.generation += 1; if (this.trigger && this.trigger.isConnected) this.trigger.focus({ preventScroll: true }); + if (this.recovery && this.resolved) htmx.trigger(document.body, "intervention:refresh"); }, choose: function (key, source) { this.choices[key] = source; diff --git a/src/pullbox/ui/static/js/pullbox.js b/src/pullbox/ui/static/js/pullbox.js index c85a46d6..51ada6a2 100644 --- a/src/pullbox/ui/static/js/pullbox.js +++ b/src/pullbox/ui/static/js/pullbox.js @@ -20804,6 +20804,10 @@ function interventionQueueRefreshEnabled() { return false; } + if (document.querySelector("[data-metadata-recovery-open='true']")) { + return false; + } + if (document.querySelector("[data-intervention-queue-expanded='true']")) { return false; } diff --git a/src/pullbox/ui/templates/partials/intervention_content.html b/src/pullbox/ui/templates/partials/intervention_content.html index 353dab8e..aad6d848 100644 --- a/src/pullbox/ui/templates/partials/intervention_content.html +++ b/src/pullbox/ui/templates/partials/intervention_content.html @@ -18,5 +18,6 @@ x-data="interventionPage({ csrfToken: '{{ csrf_token }}', bulkActionsEnabled: {{ 'false' if tab == 'recovery' else 'true' }} })" > {% include "partials/intervention_results.html" %} + {% with recovery_mode=true %}{% include "partials/issue_file_metadata.html" %}{% endwith %} {% endif %} diff --git a/src/pullbox/ui/templates/partials/intervention_list.html b/src/pullbox/ui/templates/partials/intervention_list.html index 823eca3b..f0412ef0 100644 --- a/src/pullbox/ui/templates/partials/intervention_list.html +++ b/src/pullbox/ui/templates/partials/intervention_list.html @@ -92,6 +92,10 @@ {% if meta.lane == 'recovery' %} + {% if meta.metadata_recovery %} + + {% else %}
{% endif %} + {% endif %}
+
+ {% if not recovery_mode|default(false) %}

File metadata is being written in the background.

+ {% endif %}