From bc198ada9fe028c7604b0563929c1fdbed40e95d Mon Sep 17 00:00:00 2001 From: Seungpyo1007 Date: Fri, 25 Sep 2026 06:46:15 +0900 Subject: [PATCH 1/2] Add Wikipedia Commons smartphone image backfill --- app/verify/wikipedia_image_backfill.py | 370 ++++++++++++++++++ tests/verify/test_wikipedia_image_backfill.py | 145 +++++++ 2 files changed, 515 insertions(+) create mode 100644 app/verify/wikipedia_image_backfill.py create mode 100644 tests/verify/test_wikipedia_image_backfill.py diff --git a/app/verify/wikipedia_image_backfill.py b/app/verify/wikipedia_image_backfill.py new file mode 100644 index 0000000..1b1bd88 --- /dev/null +++ b/app/verify/wikipedia_image_backfill.py @@ -0,0 +1,370 @@ +"""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. +""" + +from __future__ import annotations + +import argparse +import json +import re +from collections import Counter +from pathlib import Path +from typing import Any +from urllib.parse import unquote, urlparse + +import httpx +from bs4 import BeautifulSoup + +from app.verify.wikipedia_smartphone_backfill import ( + USER_AGENT, + PoliteWiki, + append_cache, + variant_conflict, +) + +VERSION = 7 +BAD_IMAGE = re.compile( + r"(? str | None: + for url in record.get("source_urls") or []: + if not isinstance(url, str): + continue + parsed = urlparse(url) + if ( + parsed.scheme == "https" + and parsed.hostname + and (parsed.hostname == "wikipedia.org" or parsed.hostname.endswith(".wikipedia.org")) + and parsed.path.startswith("/wiki/") + ): + return url + return None + + +def eligible(root: Path) -> list[tuple[Path, dict[str, Any], str]]: + rows = [] + for path in sorted((root / "data" / "smartphone").rglob("*.json")): + try: + record = json.loads(path.read_text(encoding="utf-8-sig")) + except (ValueError, OSError): + continue + if isinstance(record, dict) and "image_url" in record and record["image_url"] is None: + url = article_url(record) + if url: + rows.append((path, record, url)) + return rows + + +def plain(value: object) -> str: + return BeautifulSoup(str(value or ""), "html.parser").get_text(" ", strip=True) + + +def metadata_value(metadata: dict[str, Any], key: str) -> str: + item = metadata.get(key) + return plain(item.get("value")) if isinstance(item, dict) else "" + + +def filename_matches_model(name: str, filename: str) -> bool: + """Require a distinctive model token in the file name; generic brand photos fail.""" + file_words = re.findall( + r"[a-z]+\d+[a-z]*|\d+[a-z]+|[a-z]+|\d+", + re.sub(r"([a-z])([A-Z])", r"\1 \2", filename.rsplit(".", 1)[0]).lower(), + ) + file_tokens = set(file_words) + file_tokens.update( + left + right + for left, right in zip(file_words, file_words[1:], strict=False) + if left.isalpha() and right.isdigit() + ) + name_tokens = re.findall(r"[a-z]+\d+[a-z]*|\d+[a-z]+|[a-z]+|\d+", name.lower()) + codes = [ + token + for token in name_tokens + if any(char.isdigit() for char in token) and not token.endswith("gb") + ] + if codes: + return any(code in file_tokens for code in codes) + generic = { + "apple", + "samsung", + "google", + "honor", + "huawei", + "motorola", + "nokia", + "xiaomi", + "oppo", + "vivo", + "realme", + "blackberry", + "casio", + "amazon", + "nothing", + "jolla", + "itel", + "oneplus", + "microsoft", + "sony", + "palm", + "phone", + "smartphone", + "mobile", + "edition", + "generation", + "plus", + "ultra", + "pro", + "gb", + "htc", + "galaxy", + "xperia", + "lumia", + } + distinctive = [token for token in name_tokens if token not in generic and len(token) >= 2] + return any(token in file_tokens for token in distinctive) + + +def license_name(metadata: dict[str, Any]) -> str | None: + short = metadata_value(metadata, "LicenseShortName").upper().replace(" ", "-") + terms = " ".join( + metadata_value(metadata, key) for key in ("UsageTerms", "Restrictions", "License") + ) + if BAD_LICENSE.search(short + " " + terms) or not FREE_LICENSE.fullmatch( + short.replace("PUBLIC-DOMAIN", "PUBLIC DOMAIN") + ): + return None + if short.startswith("CC0"): + return "CC0-1.0" + if short == "PUBLIC-DOMAIN": + return "Public Domain" + return short.replace("CC-BY-SA-", "CC-BY-SA-").replace("CC-BY-", "CC-BY-") + + +def load_decisions(path: Path) -> dict[str, dict[str, Any]]: + decisions = {} + if path.exists(): + for line in path.read_text(encoding="utf-8").splitlines(): + try: + item = json.loads(line) + except ValueError: + continue + if item.get("version") in {6, VERSION} and isinstance(item.get("path"), str): + decisions[item["path"]] = item + return decisions + + +class CommonsFetcher(PoliteWiki): + def __init__(self, sleep_s: float = 1.0) -> None: + super().__init__(sleep_s=max(1.0, sleep_s), timeout=20.0) + + def query(self, host: str, params: dict[str, str]) -> dict[str, Any]: + self._pause() + if self._client is None: + self._client = httpx.Client( + timeout=self.timeout, headers={"User-Agent": USER_AGENT}, follow_redirects=True + ) + response = self._client.get( + f"https://{host}/w/api.php", params={"action": "query", "format": "json", **params} + ) + response.raise_for_status() + return response.json() + + +def inspect(url: str, fetcher: CommonsFetcher, name: str = "") -> dict[str, str]: + parsed = urlparse(url) + title = unquote(parsed.path.removeprefix("/wiki/")).replace("_", " ") + try: + result = fetcher.query( + parsed.hostname or "en.wikipedia.org", + {"titles": title, "prop": "pageimages", "piprop": "original", "redirects": "1"}, + ) + pages = result.get("query", {}).get("pages", {}) + page = next(iter(pages.values())) + original = page.get("original") or {} + source = original.get("source") if isinstance(original, dict) else None + filename = page.get("pageimage") + if not filename and isinstance(source, str): + filename = unquote(urlparse(source).path.rsplit("/", 1)[-1]) + if not isinstance(filename, str) or not filename: + return {"reason": "no_image"} + if ( + BAD_IMAGE.search(filename) + or GROUP_IMAGE.search(filename) + or NON_PHONE_MODEL.search(name) + or not filename.lower().endswith((".jpg", ".jpeg", ".webp")) + or ( + name + and ( + not filename_matches_model(name, filename) + or variant_conflict(name, filename.replace("_", " ")) + ) + ) + ): + return {"reason": "logo_like", "file": filename} + commons = fetcher.query( + "commons.wikimedia.org", + { + "titles": f"File:{filename}", + "prop": "imageinfo", + "iiprop": "url|mime|extmetadata", + "iiextmetadatafilter": ( + "LicenseShortName|License|UsageTerms|Restrictions|Artist|Credit|" + "ImageDescription|ObjectName" + ), + }, + ) + file_page = next(iter(commons.get("query", {}).get("pages", {}).values())) + info = (file_page.get("imageinfo") or [None])[0] + if not isinstance(info, dict): + return {"reason": "bad_license", "file": filename} + image_url = info.get("url") + metadata = info.get("extmetadata") or {} + license_id = license_name(metadata) + if ( + not license_id + or not isinstance(image_url, str) + or urlparse(image_url).hostname not in {"upload.wikimedia.org", "commons.wikimedia.org"} + ): + return { + "reason": "bad_license", + "file": filename, + "raw_license": metadata_value(metadata, "LicenseShortName"), + } + if info.get("mime") not in {"image/jpeg", "image/webp"}: + return {"reason": "logo_like", "file": filename} + description = " ".join( + metadata_value(metadata, key) for key in ("ObjectName", "ImageDescription") + ) + if BAD_IMAGE.search(description): + return {"reason": "logo_like", "file": filename} + attribution = metadata_value(metadata, "Artist") or metadata_value(metadata, "Credit") + if not attribution: + return {"reason": "bad_license", "file": filename, "raw_license": license_id} + return { + "reason": "accepted", + "file": filename, + "image_url": image_url.split("?", 1)[0], + "image_license": license_id, + "image_attribution": attribution, + } + except (httpx.HTTPError, ValueError, KeyError, StopIteration, TypeError) as exc: + return {"reason": "error", "error": str(exc)[:200]} + + +def write_image(path: Path, result: dict[str, str]) -> None: + text = path.read_text(encoding="utf-8") + record = json.loads(text) + if record.get("image_url") is not None: + return + replacement = ( + '"image_url": ' + json.dumps(result["image_url"], ensure_ascii=False) + ",\n" + ' "image_license": ' + json.dumps(result["image_license"], ensure_ascii=False) + ",\n" + ' "image_attribution": ' + json.dumps(result["image_attribution"], ensure_ascii=False) + ) + updated, count = re.subn(r'"image_url"\s*:\s*null', lambda _match: replacement, text, count=1) + if count != 1: + raise ValueError(f"missing null image_url in {path}") + path.write_text(updated, encoding="utf-8") + + +def run( + root: Path, + *, + offset: int = 0, + limit: int | None = None, + apply: bool = False, + sleep_s: float = 1.0, + cache_path: Path | None = None, +) -> 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)[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 ( + decision is not None + and decision.get("version") == 6 + and decision.get("article") == article + ): + decision = dict(decision, version=VERSION) + if decision.get("reason") == "accepted" and ( + BAD_IMAGE.search(str(decision.get("file") or "")) + or GROUP_IMAGE.search(str(decision.get("file") or "")) + or NON_PHONE_MODEL.search(str(record.get("name") or "")) + or not filename_matches_model( + str(record.get("name") or ""), str(decision.get("file") or "") + ) + or variant_conflict( + str(record.get("name") or ""), str(decision.get("file") or "").replace("_", " ") + ) + ): + decision["reason"] = "logo_like" + for key in ("image_url", "image_license", "image_attribution"): + decision.pop(key, None) + append_cache(decision, cache_path) + if ( + decision is None + or decision.get("article") != article + or decision.get("reason") == "error" + ): + decision = { + "version": VERSION, + "path": rel, + "name": record.get("name"), + "article": article, + **inspect(article, fetcher, str(record.get("name") or "")), + } + if decision["reason"] != "error": + append_cache(decision, cache_path) + if apply and decision["reason"] == "accepted": + write_image(path, decision) + results.append(decision) + message = ( + f"[{index}/{len(rows)}] {decision['reason']}: " + f"{record.get('name')} ({decision.get('file', '')})" + ) + print(message.encode("ascii", "backslashreplace").decode("ascii"), flush=True) + return results + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data-root", type=Path, required=True) + parser.add_argument("--offset", type=int, default=0) + parser.add_argument("--limit", type=int) + parser.add_argument("--sleep", type=float, default=1.0) + parser.add_argument("--apply", action="store_true") + args = parser.parse_args() + results = run( + args.data_root, offset=args.offset, limit=args.limit, apply=args.apply, sleep_s=args.sleep + ) + print( + json.dumps( + { + "checked": len(results), + "reasons": Counter(row["reason"] for row in results), + "licenses": Counter( + row.get("image_license") for row in results if row["reason"] == "accepted" + ), + }, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/tests/verify/test_wikipedia_image_backfill.py b/tests/verify/test_wikipedia_image_backfill.py new file mode 100644 index 0000000..aa5381b --- /dev/null +++ b/tests/verify/test_wikipedia_image_backfill.py @@ -0,0 +1,145 @@ +"""Offline checks for conservative Commons image selection.""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +from app.verify.wikipedia_image_backfill import ( + BAD_IMAGE, + GROUP_IMAGE, + NON_PHONE_MODEL, + article_url, + eligible, + filename_matches_model, + inspect, + license_name, + write_image, +) + + +def meta(short: str, *, artist: str = "A photographer", terms: str = "") -> dict: + return { + "LicenseShortName": {"value": short}, + "Artist": {"value": artist}, + "UsageTerms": {"value": terms}, + } + + +class FakeFetcher: + def __init__(self, filename: str, metadata: dict) -> None: + self.filename = filename + self.metadata = metadata + self.calls = 0 + + def query(self, host: str, params: dict) -> dict: + self.calls += 1 + if host.endswith("wikipedia.org"): + return { + "query": { + "pages": { + "1": { + "original": { + "source": "https://upload.wikimedia.org/wikipedia/commons/a/aa/" + + self.filename + } + } + } + } + } + return { + "query": { + "pages": { + "1": { + "imageinfo": [ + { + "url": "https://upload.wikimedia.org/wikipedia/commons/a/aa/" + + self.filename, + "mime": "image/jpeg", + "extmetadata": self.metadata, + } + ] + } + } + } + } + + +def test_accepts_only_free_photo_with_attribution() -> None: + fetcher = FakeFetcher("Example_phone.jpg", meta("CC-BY-SA-4.0")) + result = inspect("https://en.wikipedia.org/wiki/Example_phone", fetcher) + assert result["reason"] == "accepted" + assert result["image_license"] == "CC-BY-SA-4.0" + assert result["image_attribution"] == "A photographer" + assert fetcher.calls == 2 + + +def test_rejects_render_and_nonfree() -> None: + render = FakeFetcher("Example_phone.png", meta("CC-BY-SA-4.0")) + assert inspect("https://en.wikipedia.org/wiki/Example_phone", render)["reason"] == "logo_like" + assert render.calls == 1 + nonfree = FakeFetcher("Example_phone.jpg", meta("Fair use")) + assert ( + inspect("https://en.wikipedia.org/wiki/Example_phone", nonfree)["reason"] == "bad_license" + ) + assert license_name(meta("CC-BY-SA-4.0", terms="Non-free media")) 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") + assert not filename_matches_model("3X G750", "damaged_battery.jpg") + assert not filename_matches_model("Galaxy A05", "Samsung_Galaxy_A05s_2024.jpg") + assert not filename_matches_model("OnePlus 12R", "OnePlus_Ace_3_Black.jpg") + assert filename_matches_model("HONOR Magic6 Pro", "Honor_Magic_6_Pro.jpg") + fetcher = FakeFetcher("Honor_headquarter.jpg", meta("CC-BY-SA-4.0")) + assert ( + inspect("https://en.wikipedia.org/wiki/Honor_Magic5_Pro", fetcher, "HONOR Magic5 Pro")[ + "reason" + ] + == "logo_like" + ) + variant = FakeFetcher("OnePlus_7_Pro.jpg", meta("CC-BY-SA-4.0")) + assert ( + inspect("https://en.wikipedia.org/wiki/OnePlus_7", variant, "OnePlus 7")["reason"] + == "logo_like" + ) + assert BAD_IMAGE.search("OnePlus_2_(in_packaging).jpg") + assert BAD_IMAGE.search("Samsung_S26_시리즈.jpg") + assert GROUP_IMAGE.search("Xiaomi_14T_Pro_and_Xiaomi_14T.jpg") + assert NON_PHONE_MODEL.search("Surface 2") + + +def test_writes_only_image_fields() -> None: + path = Path(tempfile.gettempdir()) / "wikipedia_image_backfill_test_phone.json" + before = '{\n "name": "Example",\n "image_url": null,\n "source_urls": ["https://en.wikipedia.org/wiki/Example"]\n}\n' + path.write_text(before, encoding="utf-8") + write_image( + path, + { + "image_url": "https://upload.wikimedia.org/a.jpg", + "image_license": "CC-BY-4.0", + "image_attribution": "Alice", + }, + ) + after = path.read_text(encoding="utf-8") + assert json.loads(after)["source_urls"] == json.loads(before)["source_urls"] + assert ' "name": "Example",\n' in after + assert article_url(json.loads(after)) == "https://en.wikipedia.org/wiki/Example" + path.unlink() + + +def test_eligible_requires_explicit_null_image_url() -> None: + with tempfile.TemporaryDirectory(prefix="wiki_image_eligible_") as folder: + root = Path(folder) + phone_dir = root / "data" / "smartphone" + phone_dir.mkdir(parents=True) + source = ["https://en.wikipedia.org/wiki/Example"] + (phone_dir / "missing.json").write_text( + json.dumps({"source_urls": source}), encoding="utf-8" + ) + (phone_dir / "null.json").write_text( + json.dumps({"image_url": None, "source_urls": source}), encoding="utf-8" + ) + assert [path.name for path, _, _ in eligible(root)] == ["null.json"] From aea6d3d5ab20c22d6e2d1ddef980b63df7f180be Mon Sep 17 00:00:00 2001 From: Seungpyo1007 Date: Fri, 25 Sep 2026 06:52:35 +0900 Subject: [PATCH 2/2] fix(verify): satisfy mypy no-any-return on Commons API response --- app/verify/wikipedia_image_backfill.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/app/verify/wikipedia_image_backfill.py b/app/verify/wikipedia_image_backfill.py index 1b1bd88..9d64487 100644 --- a/app/verify/wikipedia_image_backfill.py +++ b/app/verify/wikipedia_image_backfill.py @@ -178,7 +178,8 @@ def query(self, host: str, params: dict[str, str]) -> dict[str, Any]: f"https://{host}/w/api.php", params={"action": "query", "format": "json", **params} ) response.raise_for_status() - return response.json() + data: dict[str, Any] = response.json() + return data def inspect(url: str, fetcher: CommonsFetcher, name: str = "") -> dict[str, str]: