diff --git a/MPCAutofill/cardpicker/local_phash.py b/MPCAutofill/cardpicker/local_phash.py index c10c8945a..9f1deff02 100644 --- a/MPCAutofill/cardpicker/local_phash.py +++ b/MPCAutofill/cardpicker/local_phash.py @@ -383,6 +383,161 @@ def submit_next() -> None: return BackfillResult(dry_run=dry_run, total_candidates=total, hashed=hashed, failed=failed) +DEFAULT_CANONICAL_HASH_BACKFILL_BATCH_SIZE = 500 +# Matches SCRYFALL_CDN.max_concurrency (harvest_fetch_limiter.py) - a wider pool just queues +# threads behind that destination's own concurrency semaphore, never raises real throughput. +DEFAULT_CANONICAL_HASH_BACKFILL_WORKERS = 5 + + +@dataclass(frozen=True) +class CanonicalHashBackfillResult: + dry_run: bool = False + allow_remote: bool = False + total_backlog: int = 0 + total_candidates: int = 0 + hashed: int = 0 + failed: int = 0 + skipped_no_local_url: int = 0 + elapsed_seconds: float = 0.0 + + +def run_canonical_hash_backfill( + dry_run: bool = False, + batch_size: int = DEFAULT_CANONICAL_HASH_BACKFILL_BATCH_SIZE, + workers: int = DEFAULT_CANONICAL_HASH_BACKFILL_WORKERS, + limit: Optional[int] = None, + nice: bool = True, + progress_every: int = 1000, + queue_depth_batches: int = DEFAULT_PIPELINE_QUEUE_DEPTH_BATCHES, + allow_remote: bool = False, +) -> CanonicalHashBackfillResult: + """ + Completes the local phash reference corpus (docs/features/catalog-completion-plan.md): + hashes every `CanonicalCard` row still at the unset sentinel (`image_hash == 0` - see + `get_or_compute_canonical_hash`'s own docstring on why 0 always means "never computed", not + "computed as zero"). Idempotent and resumable by construction, same NULL/sentinel-filter-as- + checkpoint discipline as `run_content_phash_backfill` above: a plain re-invocation after a + kill just filters out everything already hashed and picks up where it left off. + + LOCAL-ONLY BY DEFAULT. `CanonicalPrintingMetadata.art_crop_url` is local-first (Stage B, + 2026-07-19) and, as of issue #339's closure, populated on 113,224/113,224 printings - so the + entire corpus is computable with zero network calls to Scryfall's REST API, only its image + CDN (for the hash fetch itself, which cannot be avoided - the point is skipping the REST + round-trip that measured as 93.6% of a Stage B wall-clock probe, not skipping the fetch that + actually produces the hash). A candidate whose local URL is genuinely missing is counted in + `skipped_no_local_url` and left alone - NOT silently routed to the live Scryfall REST API - + unless the caller explicitly opts in via `allow_remote` (off by default; the escape hatch + exists for closing a residual gap deliberately, not as an implicit fallback). + + Same pipelined shape as `run_content_phash_backfill`: one long-lived `workers`-thread pool + for the whole run, a sliding submission window of `batch_size * queue_depth_batches` futures + kept full via `concurrent.futures.wait(..., return_when=FIRST_COMPLETED)`, checkpoint-flush + per batch as fetches complete. See that function's own docstring for the out-of-order- + completion safety argument (identical here: each row's persist is independent). + + No separate rate-limit parameter here (unlike `run_content_phash_backfill`'s + `rate_limit_per_sec`): `_fetch_and_hash` already goes through + `harvest_fetch_limiter.rate_limited_get(SCRYFALL_CDN, ...)`, which paces and bounds + concurrency at the destination-limiter layer - a second, uncoordinated pacer on top of that + would just add a redundant ceiling, not a better one. + + `elapsed_seconds` covers only THIS invocation's own selection (respecting `limit`) - the + caller (the management command) is responsible for extrapolating a full-backlog wall-clock + estimate from `total_backlog`/`total_candidates`/`elapsed_seconds` when `limit` narrows the + run to a sample. + """ + if nice: + try: + os.nice(15) + except (AttributeError, PermissionError, OSError): + logger.warning("os.nice unavailable in this environment - --nice throttling is CPU-yield-only") + + base_queryset = CanonicalCard.objects.filter(image_hash=0).select_related("printing_metadata") + total_backlog = base_queryset.count() + queryset = base_queryset.order_by("pk") + if limit is not None: + queryset = queryset[:limit] + all_canonicals = list(queryset) + total = len(all_canonicals) + print(f"{total_backlog} canonical printing/s with no image_hash yet ({total} selected this run).") + + hashed = 0 + failed = 0 + skipped_no_local_url = 0 + processed = 0 + to_persist: list[CanonicalCard] = [] + window_size = max(batch_size * queue_depth_batches, workers) + start_time = time.monotonic() + + def _resolve_url(canonical: CanonicalCard) -> Optional[str]: + local_url = _local_art_crop_url(canonical) + if local_url is not None: + return local_url + if allow_remote: + return _fetch_scryfall_art_crop_url(str(canonical.identifier)) + return None + + def _fetch_one(canonical: CanonicalCard) -> tuple[Optional[int], bool]: + """Returns (hash_or_None, had_url) - had_url False means no art-crop URL was available + (locally, or via the live REST fallback if allow_remote), distinct from a fetch/hash + failure on a URL that did resolve.""" + url = _resolve_url(canonical) + if url is None: + return None, False + return _fetch_and_hash(url), True + + with ThreadPoolExecutor(max_workers=workers) as executor: + pending: dict["Future[tuple[Optional[int], bool]]", CanonicalCard] = {} + canonical_iter = iter(all_canonicals) + + def submit_next() -> None: + canonical = next(canonical_iter, None) + if canonical is not None: + pending[executor.submit(_fetch_one, canonical)] = canonical + + for _ in range(window_size): + submit_next() + + while pending: + done, _ = wait(pending.keys(), return_when=FIRST_COMPLETED) + for done_future in done: + canonical = pending.pop(done_future) + image_hash, had_url = done_future.result() + processed += 1 + submit_next() # keep the window full - fetch stays ahead of persist + + if not had_url: + skipped_no_local_url += 1 + elif image_hash is not None: + canonical.image_hash = image_hash + to_persist.append(canonical) + hashed += 1 + else: + failed += 1 + + if len(to_persist) >= batch_size: + if not dry_run: + CanonicalCard.objects.bulk_update(to_persist, ["image_hash"], batch_size=batch_size) + to_persist = [] + + if progress_every and processed % progress_every < len(done): + print(f" ... {processed}/{total} canonical printings processed") + + if to_persist and not dry_run: + CanonicalCard.objects.bulk_update(to_persist, ["image_hash"], batch_size=batch_size) + + return CanonicalHashBackfillResult( + dry_run=dry_run, + allow_remote=allow_remote, + total_backlog=total_backlog, + total_candidates=total, + hashed=hashed, + failed=failed, + skipped_no_local_url=skipped_no_local_url, + elapsed_seconds=time.monotonic() - start_time, + ) + + @dataclass(frozen=True) class PhashMatch: candidate: "CandidatePrinting" @@ -444,11 +599,15 @@ def find_best_match( "DEFAULT_BACKFILL_WORKERS", "DEFAULT_PIPELINE_QUEUE_DEPTH_BATCHES", "DEFAULT_BACKFILL_RATE_LIMIT_PER_SEC", + "DEFAULT_CANONICAL_HASH_BACKFILL_BATCH_SIZE", + "DEFAULT_CANONICAL_HASH_BACKFILL_WORKERS", "PhashMatch", "BackfillResult", + "CanonicalHashBackfillResult", "get_or_compute_canonical_hash", "compute_card_art_hash", "compute_content_phash_for_card", "run_content_phash_backfill", + "run_canonical_hash_backfill", "find_best_match", ] diff --git a/MPCAutofill/cardpicker/management/commands/local_backfill_canonical_hash.py b/MPCAutofill/cardpicker/management/commands/local_backfill_canonical_hash.py new file mode 100644 index 000000000..bc806912e --- /dev/null +++ b/MPCAutofill/cardpicker/management/commands/local_backfill_canonical_hash.py @@ -0,0 +1,129 @@ +from typing import Any + +from django.core.management.base import BaseCommand + +from cardpicker.local_phash import ( + DEFAULT_CANONICAL_HASH_BACKFILL_BATCH_SIZE, + DEFAULT_CANONICAL_HASH_BACKFILL_WORKERS, + DEFAULT_PIPELINE_QUEUE_DEPTH_BATCHES, + run_canonical_hash_backfill, +) + + +class Command(BaseCommand): + help = ( + "Completes the local phash reference corpus (docs/features/catalog-completion-plan.md): " + "computes and persists CanonicalCard.image_hash for every printing still at the unset " + "sentinel (0). Local-only by default - CanonicalPrintingMetadata.art_crop_url covers " + "113,224/113,224 printings (issue #339), so this needs zero calls to Scryfall's REST " + "API, only its image CDN for the hash fetch itself. A printing with no local art-crop " + "URL is reported and skipped, not silently routed to a live REST call - see " + "--allow-remote. Idempotent and resumable by construction (filters on image_hash=0, so " + "a plain re-invocation after a kill just picks up where it left off) - no separate " + "--resume flag needed, same discipline as local_backfill_content_phash." + ) + + def add_arguments(self, parser: Any) -> None: + parser.add_argument( + "--dry-run", + action="store_true", + default=False, + help="Fetch and hash without writing anything to the database. Reports how many " + "printings would be hashed, how many would be skipped for a missing local URL, and " + "an estimated wall-clock for the full backlog.", + ) + parser.add_argument( + "--batch-size", + type=int, + default=DEFAULT_CANONICAL_HASH_BACKFILL_BATCH_SIZE, + help=f"Printings persisted per checkpoint-flush bulk_update. " + f"Default: {DEFAULT_CANONICAL_HASH_BACKFILL_BATCH_SIZE}.", + ) + parser.add_argument( + "--workers", + type=int, + default=DEFAULT_CANONICAL_HASH_BACKFILL_WORKERS, + help="Fetch thread pool size, long-lived for the whole run - sized to " + "SCRYFALL_CDN's own max_concurrency (harvest_fetch_limiter.py), not for raw " + f"parallelism; a wider pool just queues behind that destination's semaphore. " + f"Default: {DEFAULT_CANONICAL_HASH_BACKFILL_WORKERS}.", + ) + parser.add_argument( + "--queue-depth-batches", + type=int, + default=DEFAULT_PIPELINE_QUEUE_DEPTH_BATCHES, + help=f"How many batches' worth of fetches can be in flight (fetched-but-not-yet-" + f"persisted) at once - bounds memory, decoupled from --workers. " + f"Default: {DEFAULT_PIPELINE_QUEUE_DEPTH_BATCHES}.", + ) + parser.add_argument( + "--limit", + type=int, + default=None, + help="Only process this many candidates still at image_hash=0 (for testing/" + "sampling). Default: no limit, process the entire backlog.", + ) + parser.add_argument( + "--allow-remote", + action="store_true", + default=False, + help="For a printing with no local art_crop_url, fall back to a live Scryfall REST " + "call (the pre-Stage-B behaviour) instead of skipping it. Default: off - this " + "backfill is local-only by design; use this only to deliberately close a residual " + "gap, not as a routine flag.", + ) + parser.add_argument( + "--nice", + action="store_true", + default=True, + help="Lower this process's CPU scheduling priority (default: on).", + ) + parser.add_argument("--no-nice", action="store_false", dest="nice") + # --skip-checks is deliberately NOT defined here - see local_backfill_content_phash.py's + # matching comment; Django's BaseCommand already adds it natively. + + def handle(self, *args: Any, **kwargs: Any) -> None: + dry_run = kwargs["dry_run"] + batch_size = kwargs["batch_size"] + workers = kwargs["workers"] + queue_depth_batches = kwargs["queue_depth_batches"] + limit = kwargs["limit"] + allow_remote = kwargs["allow_remote"] + nice = kwargs["nice"] + + mode = "DRY RUN" if dry_run else "WRITE" + self.stdout.write( + f"[{mode}] local_backfill_canonical_hash --batch-size={batch_size} " + f"--workers={workers} --queue-depth-batches={queue_depth_batches} " + f"--limit={limit} --allow-remote={allow_remote} --nice={nice}" + ) + + result = run_canonical_hash_backfill( + dry_run=dry_run, + batch_size=batch_size, + workers=workers, + queue_depth_batches=queue_depth_batches, + limit=limit, + allow_remote=allow_remote, + nice=nice, + ) + + self.stdout.write( + f"Selected {result.total_candidates}/{result.total_backlog} candidate/s still at " f"image_hash=0." + ) + self.stdout.write( + f"Hashed {result.hashed}, skipped_no_local_url={result.skipped_no_local_url}, " + f"failed={result.failed} (failed stays at the sentinel - will retry on next " + f"invocation)." + ) + if result.total_candidates > 0: + rate = result.total_candidates / result.elapsed_seconds if result.elapsed_seconds > 0 else 0.0 + self.stdout.write(f"Elapsed {result.elapsed_seconds:.1f}s ({rate:.2f} printing/s).") + if rate > 0: + estimated_seconds = result.total_backlog / rate + self.stdout.write( + f"Estimated wall-clock for the full backlog ({result.total_backlog}): " + f"{estimated_seconds:.1f}s (~{estimated_seconds / 60:.1f} min)." + ) + if dry_run: + self.stdout.write("Dry run - nothing written.") diff --git a/MPCAutofill/cardpicker/tests/test_local_phash.py b/MPCAutofill/cardpicker/tests/test_local_phash.py index b1b414bc8..7dd7017df 100644 --- a/MPCAutofill/cardpicker/tests/test_local_phash.py +++ b/MPCAutofill/cardpicker/tests/test_local_phash.py @@ -86,3 +86,226 @@ def test_none_when_neither_source_has_a_url(self, db, monkeypatch): monkeypatch.setattr(module, "_fetch_scryfall_art_crop_url", lambda scryfall_id: None) assert module.get_or_compute_canonical_hash(canonical) is None + + +class TestCanonicalHashBackfill: + """Completes the local phash reference corpus (docs/features/catalog-completion-plan.md): + the one-time backfill for existing image_hash=0 CanonicalCard rows, mirroring + run_content_phash_backfill's checkpoint discipline (test_local_identify_printing_tags.py's + TestContentPhashBackfill) for this table instead of Card.""" + + def test_hashes_every_unhashed_canonical_and_persists_the_result(self, db, monkeypatch): + canonical_a = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical_a, art_crop_url="https://example.test/a.jpg") + canonical_b = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical_b, art_crop_url="https://example.test/b.jpg") + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 42) + + result = module.run_canonical_hash_backfill(nice=False) + + assert result.total_backlog == 2 + assert result.total_candidates == 2 + assert result.hashed == 2 + assert result.failed == 0 + assert result.skipped_no_local_url == 0 + canonical_a.refresh_from_db() + canonical_b.refresh_from_db() + assert canonical_a.image_hash == 42 + assert canonical_b.image_hash == 42 + + def test_already_hashed_canonicals_are_not_touched(self, db, monkeypatch): + already_hashed = CanonicalCardFactory(image_hash=99) + CanonicalPrintingMetadataFactory(canonical_card=already_hashed, art_crop_url="https://example.test/a.jpg") + called: list[str] = [] + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: called.append(url) or 42) + + result = module.run_canonical_hash_backfill(nice=False) + + assert result.total_backlog == 0 + assert result.total_candidates == 0 + assert called == [] + already_hashed.refresh_from_db() + assert already_hashed.image_hash == 99 + + def test_missing_local_url_is_skipped_not_fetched_remotely(self, db, monkeypatch): + canonical = CanonicalCardFactory(image_hash=0) # no CanonicalPrintingMetadata row at all + + def _fail(*args: object, **kwargs: object) -> None: + raise AssertionError("should never be called - allow_remote defaults to False") + + monkeypatch.setattr(module, "_fetch_scryfall_art_crop_url", _fail) + monkeypatch.setattr(module, "_fetch_and_hash", _fail) + + result = module.run_canonical_hash_backfill(nice=False) + + assert result.hashed == 0 + assert result.skipped_no_local_url == 1 + canonical.refresh_from_db() + assert canonical.image_hash == 0 + + def test_allow_remote_falls_back_to_rest_when_no_local_url(self, db, monkeypatch): + canonical = CanonicalCardFactory(image_hash=0) + + monkeypatch.setattr( + module, "_fetch_scryfall_art_crop_url", lambda scryfall_id: "https://example.test/fallback.jpg" + ) + monkeypatch.setattr( + module, "_fetch_and_hash", lambda url: 7 if url == "https://example.test/fallback.jpg" else None + ) + + result = module.run_canonical_hash_backfill(nice=False, allow_remote=True) + + assert result.hashed == 1 + assert result.skipped_no_local_url == 0 + canonical.refresh_from_db() + assert canonical.image_hash == 7 + + def test_a_failed_hash_stays_at_sentinel_and_is_counted_as_failed(self, db, monkeypatch): + canonical = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical, art_crop_url="https://example.test/a.jpg") + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: None) + + result = module.run_canonical_hash_backfill(nice=False) + + assert result.hashed == 0 + assert result.failed == 1 + canonical.refresh_from_db() + assert canonical.image_hash == 0 + + def test_dry_run_writes_nothing(self, db, monkeypatch): + canonical = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical, art_crop_url="https://example.test/a.jpg") + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 42) + + result = module.run_canonical_hash_backfill(dry_run=True, nice=False) + + assert result.hashed == 1 + canonical.refresh_from_db() + assert canonical.image_hash == 0 + + def test_a_second_invocation_only_processes_what_the_first_missed(self, db, monkeypatch): + # simulates a kill mid-backfill and a plain re-invocation - the image_hash=0 filter is + # the checkpoint, no separate --resume flag needed. + already_hashed = CanonicalCardFactory(image_hash=42) + CanonicalPrintingMetadataFactory(canonical_card=already_hashed, art_crop_url="https://example.test/a.jpg") + still_unhashed = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=still_unhashed, art_crop_url="https://example.test/b.jpg") + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 7) + + result = module.run_canonical_hash_backfill(nice=False) + + assert result.total_candidates == 1 + already_hashed.refresh_from_db() + still_unhashed.refresh_from_db() + assert already_hashed.image_hash == 42 # untouched + assert still_unhashed.image_hash == 7 # newly hashed + + def test_limit_narrows_the_run_but_reports_the_full_backlog(self, db, monkeypatch): + for _ in range(3): + canonical = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical, art_crop_url="https://example.test/x.jpg") + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 1) + + result = module.run_canonical_hash_backfill(nice=False, limit=1) + + assert result.total_backlog == 3 + assert result.total_candidates == 1 + + +class TestCanonicalHashBackfillCommandCLI: + """Mirrors TestBackfillCommandCLI (test_local_identify_printing_tags.py) - exercises the + real add_arguments()/parser path, not just run_canonical_hash_backfill() directly.""" + + def test_real_cli_invocation_with_no_flags_at_all(self, db, monkeypatch): + from django.core.management import call_command + + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 1) + call_command("local_backfill_canonical_hash", "--limit=0") + + def test_skip_checks_flag_does_not_conflict_with_djangos_own(self, db, monkeypatch): + from django.core.management import call_command + + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 1) + call_command("local_backfill_canonical_hash", "--skip-checks", "--limit=0") + + def test_dry_run_flag_reports_without_writing(self, db, monkeypatch, capsys): + from django.core.management import call_command + + canonical = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical, art_crop_url="https://example.test/a.jpg") + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 42) + + call_command("local_backfill_canonical_hash", "--skip-checks", "--dry-run") + captured = capsys.readouterr() + + assert "Dry run - nothing written." in captured.out + canonical.refresh_from_db() + assert canonical.image_hash == 0 + + def test_allow_remote_flag_wires_through(self, db, monkeypatch, capsys): + from django.core.management import call_command + + canonical = CanonicalCardFactory(image_hash=0) # no local metadata row + + monkeypatch.setattr( + module, "_fetch_scryfall_art_crop_url", lambda scryfall_id: "https://example.test/fallback.jpg" + ) + monkeypatch.setattr(module, "_fetch_and_hash", lambda url: 9) + + call_command("local_backfill_canonical_hash", "--skip-checks", "--allow-remote") + captured = capsys.readouterr() + + assert "--allow-remote=True" in captured.out + canonical.refresh_from_db() + assert canonical.image_hash == 9 + + +class TestCanonicalHashBackfillPipelineOutOfOrder: + """Same proof as TestPipelinedBackfillOutOfOrder (test_local_identify_printing_tags.py) for + this backfill's own pipeline: completion order isn't submission order once more than one + worker thread is in flight, and persistence must be keyed by which future belongs to which + canonical, not by completion position.""" + + def test_persists_correctly_when_completion_order_differs_from_submission_order(self, db, monkeypatch): + import time + + canonical_a = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical_a, art_crop_url="https://example.test/a.jpg") + canonical_b = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical_b, art_crop_url="https://example.test/b.jpg") + canonical_c = CanonicalCardFactory(image_hash=0) + CanonicalPrintingMetadataFactory(canonical_card=canonical_c, art_crop_url="https://example.test/c.jpg") + + # canonical_a is submitted first (lower pk, since the queryset orders by pk) but its + # fetch is made to finish LAST - proves the persisted result is keyed by which canonical + # the future belongs to, not by submission/completion position. + delays = { + "https://example.test/a.jpg": 0.3, + "https://example.test/b.jpg": 0.05, + "https://example.test/c.jpg": 0.15, + } + hashes = { + "https://example.test/a.jpg": 111, + "https://example.test/b.jpg": 222, + "https://example.test/c.jpg": 333, + } + + def slow_variable_hash(url: str) -> int: + time.sleep(delays[url]) + return hashes[url] + + monkeypatch.setattr(module, "_fetch_and_hash", slow_variable_hash) + + # workers=3 keeps all three in flight simultaneously (window_size = max(batch_size * + # queue_depth_batches, workers) = max(1, 3) = 3), so completion order is purely + # determined by the delays above (b, then c, then a) - the reverse of submission order. + result = module.run_canonical_hash_backfill(nice=False, batch_size=1, workers=3, queue_depth_batches=1) + + assert result.hashed == 3 + assert result.failed == 0 + canonical_a.refresh_from_db() + canonical_b.refresh_from_db() + canonical_c.refresh_from_db() + assert canonical_a.image_hash == 111 + assert canonical_b.image_hash == 222 + assert canonical_c.image_hash == 333