feat(verify): support Wikipedia image backfill across device categories

Author: Seungpyo1007

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/<category> 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"(?<![A-Za-z0-9])(?:logo|logotype|wordmark|icon|emblem|flag|symbol|diagram|chart|screenshot|placeholder|render|advertisement|battery|headquarters?|building|campus|series|lineup|packaging|시리즈)(?![A-Za-z0-9])",
     re.I,
@@ -38,7 +41,10 @@
 
 
 def article_url(record: dict[str, Any]) -> 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<newline>\r?\n)(?P<indent>[ \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,14 +329,23 @@ 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)
     results = []
     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,

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")