diff --git a/app/verify/wikipedia_image_backfill.py b/app/verify/wikipedia_image_backfill.py index 1b257ae..ae9aa68 100644 --- a/app/verify/wikipedia_image_backfill.py +++ b/app/verify/wikipedia_image_backfill.py @@ -1,7 +1,9 @@ """Backfill freely licensed Commons photos from already cited Wikipedia articles. Run a dry sample first, then use --apply --offset/--limit for sequential batches. -The append-only decision cache is shared between dry runs and apply runs. +Select data/ with --category (default: smartphone). +The append-only decision cache is shared between categories, dry runs and apply +runs; repository-relative paths keep each category's decisions distinct. """ from __future__ import annotations @@ -25,6 +27,7 @@ ) VERSION = 7 +CATEGORIES = ("smartphone", "laptop", "monitor", "tablet", "watch", "pda") BAD_IMAGE = re.compile( r"(? str | None: - for url in record.get("source_urls") or []: + sources = record.get("source_urls") or [] + if not isinstance(sources, list): + return None + for url in sources: if not isinstance(url, str): continue parsed = urlparse(url) @@ -53,17 +59,25 @@ def article_url(record: dict[str, Any]) -> str | None: def eligible( - root: Path, *, include_missing_key: bool = False + root: Path, *, category: str = "smartphone", include_missing_key: bool = False ) -> list[tuple[Path, dict[str, Any], str]]: + if category not in CATEGORIES: + raise ValueError(f"unsupported category: {category}") rows = [] - for path in sorted((root / "data" / "smartphone").rglob("*.json")): + for path in sorted((root / "data" / category).rglob("*.json")): try: record = json.loads(path.read_text(encoding="utf-8-sig")) - except (ValueError, OSError): + except (ValueError, OSError) as exc: + print(f"Skipping {path}: unreadable record ({exc})", flush=True) + continue + if not isinstance(record, dict): + print(f"Skipping {path}: record must be a JSON object", flush=True) continue - if isinstance(record, dict) and ( - ("image_url" in record and record["image_url"] is None) - or (include_missing_key and "image_url" not in record) + if record.get("source_urls") is not None and not isinstance(record["source_urls"], list): + print(f"Skipping {path}: source_urls must be a list", flush=True) + continue + if ("image_url" in record and record["image_url"] is None) or ( + include_missing_key and "image_url" not in record ): url = article_url(record) if url: @@ -269,10 +283,15 @@ def inspect(url: str, fetcher: CommonsFetcher, name: str = "") -> dict[str, str] def write_image(path: Path, result: dict[str, str]) -> None: - text = path.read_bytes().decode("utf-8") + original = path.read_bytes() + text = original.decode("utf-8-sig") record = json.loads(text) + if not isinstance(record, dict): + raise ValueError(f"record must be a JSON object in {path}") if record.get("image_url") is not None: return + if "image_license" in record or "image_attribution" in record: + raise ValueError(f"existing image metadata in {path}") newline = "\r\n" if "\r\n" in text else "\n" replacement = ( '"image_url": ' + json.dumps(result["image_url"], ensure_ascii=False) + ",\n" @@ -287,20 +306,20 @@ def write_image(path: Path, result: dict[str, str]) -> None: if count != 1: raise ValueError(f"missing null image_url in {path}") else: - if "image_license" in record or "image_attribution" in record: - raise ValueError(f"existing image metadata in {path}") match = re.match(r'\{(?P\r?\n)(?P[ \t]+)(?=")', text) if match is None: raise ValueError(f"cannot insert image fields in {path}") indent = match.group("indent") fields = replacement.replace(newline + " ", newline + indent) updated = text[: match.end()] + fields + "," + newline + indent + text[match.end() :] - path.write_bytes(updated.encode("utf-8")) + encoding = "utf-8-sig" if original.startswith(b"\xef\xbb\xbf") else "utf-8" + path.write_bytes(updated.encode(encoding)) def run( root: Path, *, + category: str = "smartphone", offset: int = 0, limit: int | None = None, apply: bool = False, @@ -310,7 +329,7 @@ def run( ) -> list[dict[str, Any]]: cache_path = cache_path or root / "data" / "_verify" / "state" / "wikipedia_image_cache.jsonl" cache = load_decisions(cache_path) - rows = eligible(root, include_missing_key=include_missing_key)[ + rows = eligible(root, category=category, include_missing_key=include_missing_key)[ offset : None if limit is None else offset + limit ] fetcher = CommonsFetcher(sleep_s) @@ -318,6 +337,15 @@ def run( for index, (path, record, article) in enumerate(rows, 1): rel = path.relative_to(root).as_posix() decision = cache.get(rel) + if not isinstance(record.get("name"), str) or not record["name"].strip(): + decision = { + "path": rel, + "reason": "invalid_record", + "error": "name must be a nonempty string", + } + results.append(decision) + print(f"Skipping {rel}: {decision['error']}", flush=True) + continue if ( decision is not None and decision.get("version") == 6 @@ -354,11 +382,15 @@ def run( if decision["reason"] != "error": append_cache(decision, cache_path) if apply and decision["reason"] == "accepted": - write_image(path, decision) + try: + write_image(path, decision) + except (ValueError, OSError) as exc: + decision = dict(decision, reason="invalid_record", error=str(exc)) results.append(decision) message = ( f"[{index}/{len(rows)}] {decision['reason']}: " f"{record.get('name')} ({decision.get('file', '')})" + + (f": {decision['error']}" if decision.get("error") else "") ) print(message.encode("ascii", "backslashreplace").decode("ascii"), flush=True) return results @@ -367,6 +399,7 @@ def run( def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--data-root", type=Path, required=True) + parser.add_argument("--category", choices=CATEGORIES, default="smartphone") parser.add_argument("--offset", type=int, default=0) parser.add_argument("--limit", type=int) parser.add_argument("--sleep", type=float, default=1.0) @@ -375,6 +408,7 @@ def main() -> None: args = parser.parse_args() results = run( args.data_root, + category=args.category, offset=args.offset, limit=args.limit, apply=args.apply, diff --git a/tests/verify/test_wikipedia_image_backfill.py b/tests/verify/test_wikipedia_image_backfill.py index 915db31..52bdd4e 100644 --- a/tests/verify/test_wikipedia_image_backfill.py +++ b/tests/verify/test_wikipedia_image_backfill.py @@ -3,11 +3,16 @@ from __future__ import annotations import json +import sys import tempfile from pathlib import Path +import pytest + +from app.verify import wikipedia_image_backfill as backfill from app.verify.wikipedia_image_backfill import ( BAD_IMAGE, + CATEGORIES, GROUP_IMAGE, NON_PHONE_MODEL, article_url, @@ -86,6 +91,126 @@ def test_rejects_render_and_nonfree() -> None: assert license_name(meta("CC-BY-SA-4.0", terms="Non-free media")) is None +@pytest.mark.parametrize("category", CATEGORIES) +@pytest.mark.parametrize("missing_key", [False, True]) +def test_category_scan_and_apply( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, category: str, missing_key: bool +) -> None: + record = { + "name": "Example X123", + "image_url": None, + "source_urls": ["https://en.wikipedia.org/wiki/Example_X123"], + "specification": {"untouched": True}, + } + if missing_key: + record.pop("image_url") + for folder in CATEGORIES: + directory = tmp_path / "data" / folder + directory.mkdir(parents=True) + (directory / "example.json").write_text(json.dumps(record, indent=2), encoding="utf-8") + assert [ + path.parent.name + for path, _, _ in eligible(tmp_path, category=category, include_missing_key=missing_key) + ] == [category] + monkeypatch.setattr( + backfill, "CommonsFetcher", lambda _: FakeFetcher("Example_X123.jpg", meta("CC-BY-4.0")) + ) + results = backfill.run(tmp_path, category=category, apply=True, include_missing_key=missing_key) + assert [row["reason"] for row in results] == ["accepted"] + for folder in CATEGORIES: + updated = json.loads((tmp_path / "data" / folder / "example.json").read_text()) + assert updated["specification"] == record["specification"] + assert (updated.get("image_url") is not None) == (folder == category) + cache_path = tmp_path / "data" / "_verify" / "state" / "wikipedia_image_cache.jsonl" + assert json.loads(cache_path.read_text())["path"] == f"data/{category}/example.json" + + +def test_malformed_records_skip_with_reason(tmp_path: Path, capsys: pytest.CaptureFixture) -> None: + directory = tmp_path / "data" / "laptop" + directory.mkdir(parents=True) + for name, record in { + "array": [], + "sources": {"image_url": None, "source_urls": 42}, + }.items(): + (directory / f"{name}.json").write_text(json.dumps(record), encoding="utf-8") + (directory / "broken.json").write_text("{", encoding="utf-8") + assert eligible(tmp_path, category="laptop") == [] + output = capsys.readouterr().out + assert "record must be a JSON object" in output + assert "source_urls must be a list" in output + assert "unreadable record" in output + assert article_url({"source_urls": 42}) is None + with pytest.raises(ValueError, match="unsupported category"): + eligible(tmp_path, category="../outside") + + +def test_apply_skips_incompatible_record_and_continues( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + directory = tmp_path / "data" / "watch" + directory.mkdir(parents=True) + record = { + "name": "Example X123", + "image_url": None, + "source_urls": ["https://en.wikipedia.org/wiki/Example_X123"], + } + incompatible = directory / "a.json" + before = json.dumps(dict(record, image_license=None)) + incompatible.write_text(before, encoding="utf-8") + (directory / "b.json").write_text(json.dumps(record), encoding="utf-8") + (directory / "c.json").write_text(json.dumps(dict(record, name=None)), encoding="utf-8") + monkeypatch.setattr( + backfill, "CommonsFetcher", lambda _: FakeFetcher("Example_X123.jpg", meta("CC-BY-4.0")) + ) + results = backfill.run(tmp_path, category="watch", apply=True) + assert [row["reason"] for row in results] == ["invalid_record", "accepted", "invalid_record"] + assert "existing image metadata" in results[0]["error"] + assert "name must be a nonempty string" in results[2]["error"] + assert incompatible.read_text() == before + + +@pytest.mark.parametrize("category", [None, "pda"]) +def test_cli_selects_category(monkeypatch: pytest.MonkeyPatch, category: str | None) -> None: + calls = [] + monkeypatch.setattr(backfill, "run", lambda *args, **kwargs: calls.append(kwargs) or []) + argv = ["backfill", "--data-root", ".", "--apply"] + if category: + argv += ["--category", category] + monkeypatch.setattr(sys, "argv", argv) + backfill.main() + assert calls[0]["category"] == (category or "smartphone") + assert calls[0]["apply"] is True + + +def test_shared_cache_replays_without_crossing_categories( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fetcher = FakeFetcher("Example_X123.jpg", meta("CC-BY-4.0")) + monkeypatch.setattr(backfill, "CommonsFetcher", lambda _: fetcher) + for category in ("laptop", "pda"): + path = tmp_path / "data" / category / "example.json" + path.parent.mkdir(parents=True) + path.write_text( + json.dumps( + { + "name": "Example X123", + "image_url": None, + "source_urls": ["https://en.wikipedia.org/wiki/Example_X123"], + } + ), + encoding="utf-8", + ) + backfill.run(tmp_path, category=category) + assert fetcher.calls == 4 + cache = tmp_path / "data" / "_verify" / "state" / "wikipedia_image_cache.jsonl" + before = cache.read_bytes() + backfill.run(tmp_path, category="laptop", apply=True) + assert fetcher.calls == 4 + assert cache.read_bytes() == before + assert json.loads((tmp_path / "data" / "laptop" / "example.json").read_text())["image_url"] + assert json.loads((tmp_path / "data" / "pda" / "example.json").read_text())["image_url"] is None + + def test_filename_must_name_device() -> None: assert filename_matches_model("HONOR Magic6 Pro", "Honor_Magic_6_Pro.jpg") assert not filename_matches_model("HONOR Magic5 Pro", "Honor_headquarter.jpg")