Add the Commons reference-board research tool
Full build / Import + headless tests (push) Successful in 40s
Full build / Full desktop build + optional release (push) Skipped

This commit is contained in:
zwitschi committed 2026-11-03 14:51:31 +01:00
1 parent f09a5e7937
commit 5b306b1c52
4 files changed
+429 -4

No files matched your search

+293
View File
@@ -0,0 +1,293 @@
#!/usr/bin/env python3
"""Wikimedia Commons reference-board helper for venue documentation.
Lists and fetches licence-cleared photos of a real location and writes the
`SIDECAR.md` rows and the folder `README.md` gallery rows used by
`docs/screens/refs/<venue>/`. Research downloads stay under `tmp/` unless
`--stage refs` copies them into the board folder (still uncommitted; the git
commit is a deliberate human step).
python3 tools/refboard.py search "El Capitan Entertainment Centre" \
[--months 12] [--radius 500] [--at coord] [--json]
python3 tools/refboard.py fetch --from search.json --dir tmp/research/elcapitan \
[--max N]
python3 tools/refboard.py rows --from meta.json --licence-note ... [--append-sidecar PATH]
Licence gate: CC0 / public domain / CC BY are listed; CC BY-SA rows are marked
`(review)` and must record the share-alike decision; anything else is skipped.
Rate limits: the Commons API needs a descriptive User-Agent, metadata is fetched
in one batched `titles=` query, and file downloads are spaced (default 3 s) to
avoid HTTP 429.
"""
import argparse
import json
import re
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
USER_AGENT = "gig-line-dev-refboard/1.0 (development reference collection; contact: repo)"
API = "https://commons.wikimedia.org/w/api.php"
DOWNLOAD_SPACING_S = 3.0
#Licences that can be committed with the stated attribution; anything else is
#research-only.
COMMITTABLE = {
"Public domain": "Public domain",
"CC0": "CC0",
"CC BY 2.0": "CC BY 2.0",
"CC BY 3.0": "CC BY 3.0",
"CC BY 4.0": "CC BY 4.0",
}
# CC BY-SA requires a share-alike review before a file enters the repository, so
# it stays research-only until `--allow-ccbysa` records the decision.
REVIEW_ONLY = {
"CC BY-SA 2.0": "CC BY-SA 2.0 (review)",
"CC BY-SA 3.0": "CC BY-SA 3.0 (review)",
"CC BY-SA 4.0": "CC BY-SA 4.0 (review)",
}
def _strip_html(text: str) -> str:
return re.sub("<[^>]+>", "", text or "").strip()
def normalise_licence(raw: str, allow_ccbysa: bool = False):
"""Return (committed label | None, research-only label | None).
`clean` is set when the file may enter the repository; `review` is set for
share-alike licences that stay under `tmp/` until a review records otherwise.
"""
if raw in COMMITTABLE:
return raw, None
if raw in REVIEW_ONLY:
return (REVIEW_ONLY[raw], None) if allow_ccbysa else (None, REVIEW_ONLY[raw])
return None, raw or None
class Commons:
def __init__(self, user_agent: str = USER_AGENT):
self.user_agent = user_agent
self.last_request = 0.0
def _get(self, params: dict, binary: bool = False, spacing: float = 0.0):
if spacing:
wait = spacing - (time.monotonic() - self.last_request)
if wait > 0:
time.sleep(wait)
self.last_request = time.monotonic()
req = urllib.request.Request(
API if not binary else params["url"],
data=None if binary else urllib.parse.urlencode(params).encode(),
headers={"User-Agent": self.user_agent}, method="GET" if binary else "POST")
with urllib.request.urlopen(req, timeout=90) as resp:
return resp.read() if binary else json.loads(resp.read().decode("utf-8"))
def meta_for_titles(self, titles: list[str], allow_ccbysa: bool = False) -> dict:
rows = {}
for chunk in [titles[i:i + 20] for i in range(0, len(titles), 20)]:
data = self._get({
"action": "query", "format": "json", "titles": "|".join(chunk),
"prop": "imageinfo", "iiprop": "url|size|extmetadata",
"iiurlwidth": "1600", "formatversion": "2",
})
for page in data.get("query", {}).get("pages", []):
if "missing" in page or not page.get("imageinfo"):
continue
info = page["imageinfo"][0]
meta = info.get("extmetadata", {})
def val(key):
return _strip_html(meta.get(key, {}).get("value", ""))
def first_date():
for key in ("DateTimeOriginal", "DateTime", "DateTimeDigitized"):
text = _strip_html(meta.get(key, {}).get("value", ""))
match = re.search(r"\d{4}-\d{2}-\d{2}", text)
if match:
return match.group(0)
return ""
raw_licence = val("LicenseShortName")
clean, review = normalise_licence(raw_licence, allow_ccbysa)
rows[page["title"]] = {
"title": page["title"],
"licence": raw_licence,
"licence_url": val("LicenseUrl"),
"artist": val("Artist") or "unknown",
"credit": val("Credit")[:120],
"date": first_date(),
"description": val("ImageDescription")[:300],
"page": info.get("descriptionurl", ""),
"width": info.get("thumbwidth") or info.get("width", 0),
"height": info.get("thumbheight") or info.get("height", 0),
"url": info.get("thumburl") or info.get("url", ""),
"clean": clean,
"review": review,
}
return rows
def search(self, query: str, limit: int = 12) -> list[str]:
data = self._get({"action": "query", "format": "json", "generator": "search",
"gsrsearch": query, "gsrnamespace": "6", "gsrlimit": str(limit),
"formatversion": "2"})
return [p["title"] for p in data.get("query", {}).get("pages", [])
if not p.get("missing")]
def geosearch(self, coord: str, radius_m: int = 500, limit: int = 60) -> list[str]:
data = self._get({"action": "query", "format": "json", "generator": "geosearch",
"ggscoord": coord, "ggsradius": str(radius_m),
"ggslimit": str(limit), "ggsnamespace": "6",
"formatversion": "2"})
return [p["title"] for p in data.get("query", {}).get("pages", [])
if not p.get("missing")]
def category(self, category: str, limit: int = 50) -> list[str]:
data = self._get({"action": "query", "format": "json", "generator": "categorymembers",
"gcmtitle": category, "gcmtype": "file", "gcmlimit": str(limit),
"formatversion": "2"})
return [p["title"] for p in data.get("query", {}).get("pages", [])
if not p.get("missing")]
def collect(source: str, argument: str, limit: int) -> list[str]:
api = Commons()
if source == "search":
return api.search(argument, limit)
if source == "geo":
coord, radius = (argument.split(",") + [str(500)])[:2] if "," in argument \
else (argument, "500")
return api.geosearch(argument if "," in argument else coord, int(radius), limit)
if source == "category":
return api.category(argument if argument.startswith("Category:")
else f"Category:{argument}", limit)
raise SystemExit(f"unknown source {source}")
def _age_months(date: str, today: str) -> int | None:
if not re.match(r"^\d{4}-\d{2}", date):
return None
years = int(today[:4]) - int(date[:4])
months = int(today[5:7]) - int(date[5:7])
return years * 12 + months
def cmd_search(args) -> int:
titles = collect(args.source, args.query, args.limit)
api = Commons()
rows = api.meta_for_titles(titles, allow_ccbysa=args.allow_ccbysa)
today = time.strftime("%Y-%m-%d")
result = []
for title in titles:
row = rows.get(title)
if not row:
continue
row["age_months"] = _age_months(row["date"], today)
row["recent"] = (args.months <= 0 or
(row["age_months"] is not None and row["age_months"] <= args.months))
row["usable"] = bool(row["clean"] and row["url"]
and int(row["width"] or 0) >= args.min_width)
row["research_only"] = not row["usable"] and bool(row["review"] and row["url"]
and int(row["width"] or 0) >= args.min_width)
result.append(row)
result.sort(key=lambda r: (not r["recent"], r["date"]), reverse=False)
result.sort(key=lambda r: r["date"], reverse=True)
if args.json:
print(json.dumps(result, indent=2))
return 0
for row in result:
age = "?" if row["age_months"] is None else row["age_months"]
flags = "recent" if row["recent"] else " "
flags += " commit" if row["usable"] else " review" if row["research_only"] else " skip"
print(f"{row['date'] or '?':10s} {age:>4}mo {flags:12s} "
f"{row['licence'][:20]:20s} {row['title'][:56]}")
return 0
def cmd_fetch(args) -> int:
rows = json.loads(Path(args.manifest).read_text(encoding="utf-8"))
usable = [r for r in rows if r.get("usable")
or (args.include_review and r.get("research_only"))]
if args.only_recent:
usable = [r for r in usable if r.get("recent")]
out = Path(args.dir)
out.mkdir(parents=True, exist_ok=True)
api = Commons()
manifest = []
for row in usable[:args.max]:
name = re.sub(r"^File:", "", row["title"])
name = re.sub(r"[^A-Za-z0-9._-]+", "_", name)
path = out / name
if path.exists() and not args.force:
print(f"skip existing {path.name}")
else:
data = api._get({"url": row["url"]}, binary=True,
spacing=DOWNLOAD_SPACING_S)
path.write_bytes(data)
print(f"fetched {path.name} {len(data) // 1024} KB")
manifest.append({**row, "file": path.name})
(out / "meta.json").write_text(json.dumps(manifest, indent=2) + "\n")
print(f"meta -> {out / 'meta.json'} ({len(manifest)} entries)")
return 0
def _markdown_filename(title: str, date: str = "") -> str:
stem = re.sub(r"^File:", "", title)
stem = re.sub(r"\.[A-Za-z0-9]+$", "", stem)
slug = re.sub(r"[^A-Za-z0-9]+", "-", stem).strip("-").lower()
return f"{slug}-{date or 'unknown'}.jpg"
def cmd_rows(args) -> int:
rows = json.loads(Path(args.manifest).read_text(encoding="utf-8"))
sidecar = []
readme = []
for row in rows:
name = _markdown_filename(row["title"], row["date"] or "unknown")
licence = row["clean"] or row["licence"]
sidecar.append(
f"| `{name}` | {_strip_html(row['description'])[:80] or row['title']} | "
f"view | <{row['page']}> | {row['artist'][:40]} | {licence} | "
f"{row['date'] or 'n.d.'} | purpose | no |")
readme.append(
f"| ![{row['title']}]({name}) | **{row['title'][:40]}** |")
print("== SIDECAR rows (paste into the Photographs table)\n")
print("\n".join(sidecar))
print("\n== README rows (paste into the board gallery)\n")
print("\n".join(readme))
if args.append_sidecar:
target = Path(args.append_sidecar)
if not target.is_file():
return 1
text = target.read_text(encoding="utf-8")
marker = "| `"
additions = []
existing = {m.group(1) for m in re.finditer(r"^\| `([^`]+)`", text, re.M)}
for row, line in zip(rows, sidecar):
name = line.split("|")[1].strip().strip("`")
if name not in existing:
additions.append(line)
if additions:
last = [l for l in text.splitlines() if l.startswith(marker)][-1]
text = text.replace(last, last + "\n" + "\n".join(additions))
target.write_text(text)
print(f"appended {len(additions)} row(s) to {target}")
return 0
def build_parser():
parser = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = parser.add_subparsers(dest="cmd", required=True)
search = sub.add_parser("search", help="list Commons candidates with licence/date")
search.add_argument("query", help="search text, coord 'lat,lon' with --source geo, "
"or category with --source category")
search.add_argument("--source", choices=("search", "geo", "category"), default="search")
search.add_argument("--limit", type=int, default=20)
search.add_argument("--months", type=int, default=0,
help="flag files captured within this many months (0 = all)")
search.add_argument("--min-width", type=int, default=1200)
search.add_argument("--allow-ccbysa", action="store_true",
help="treat CC BY-SA as committable after a recorded review "
"(default: research-only)")
search.add_argument("--json", action="store_true")
search.set_defaults(handler=cmd_search)
fetch = sub.add_parser("fetch", help="download the usable rows of a manifest")
fetch.add_argument("--from", dest="manifest", required=True)
fetch.add_argument("--dir", required=True)
fetch.add_argument("--max", type=int, default=12)
fetch.add_argument("--only-recent", action="store_true")
fetch.add_argument("--force", action="store_true")
fetch.add_argument("--include-review", action="store_true",
help="also download review-only (CC BY-SA) rows for study under tmp/")
fetch.set_defaults(handler=cmd_fetch)
rows = sub.add_parser("rows", help="emit SIDECAR/README table rows from a manifest")
rows.add_argument("--from", dest="manifest", required=True)
rows.add_argument("--append-sidecar", default="")
rows.set_defaults(handler=cmd_rows)
return parser
def main(argv=None) -> int:
args = build_parser().parse_args(argv)
return args.handler(args)
if __name__ == "__main__":
sys.exit(main())
+103
View File
@@ -0,0 +1,103 @@
"""Tests for the Commons reference-board helper (stdlib only, no network)."""
import json
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest import mock
import refboard as rb
class LicenceGateTests(unittest.TestCase):
def test_committable_licences_pass(self):
for licence in ("Public domain", "CC0", "CC BY 2.0", "CC BY 3.0", "CC BY 4.0"):
clean, review = rb.normalise_licence(licence)
self.assertEqual(clean, licence, licence)
self.assertIsNone(review, licence)
def test_ccbysa_is_research_only_by_default(self):
clean, review = rb.normalise_licence("CC BY-SA 4.0")
self.assertIsNone(clean)
self.assertEqual(review, "CC BY-SA 4.0 (review)")
def test_ccbysa_becomes_committable_only_with_the_flag(self):
clean, review = rb.normalise_licence("CC BY-SA 3.0", allow_ccbysa=True)
self.assertEqual(clean, "CC BY-SA 3.0 (review)")
self.assertIsNone(review)
def test_unknown_licence_is_never_committable(self):
clean, review = rb.normalise_licence("Fair use")
self.assertIsNone(clean)
self.assertEqual(review, "Fair use")
class SearchFilterTests(unittest.TestCase):
@staticmethod
def _row(title, licence="CC0", date="2026-09-30", width=1600):
clean, review = rb.normalise_licence(licence)
return {"title": title, "licence": licence, "clean": clean, "review": review,
"date": date, "url": "https://example/" + title, "width": width,
"artist": "a", "page": "p", "height": 1000, "description": "d",
"credit": "", "licence_url": ""}
def test_recent_and_usable_selection(self):
# `d` is small-width (skipped); `c` is old but CC0, so it is usable and
# sorted after the recent one; only `b` is review-only.
rows = [self._row("a"), self._row("b", licence="CC BY-SA 4.0"),
self._row("c", date="2015-01-01"), self._row("d", width=800)]
with mock.patch.object(rb.Commons, "meta_for_titles", return_value={
r["title"]: r for r in rows}), \
mock.patch.object(rb, "collect", return_value=[r["title"] for r in rows]):
args = mock.Mock(source="geo", query="34.1,-118.3", limit=10, months=12,
min_width=1200, json=True, allow_ccbysa=False)
with mock.patch("builtins.print") as printed:
rb.cmd_search(args)
payload = json.loads(printed.call_args[0][0])
usable = [r["title"] for r in payload if r["usable"]]
review = [r["title"] for r in payload if r.get("research_only")]
self.assertEqual(usable, ["a", "c"])
self.assertEqual(review, ["b"])
def test_dates_pull_the_first_iso_component(self):
text = rb.Commons().meta_for_titles # method exists
self.assertEqual(rb._age_months("2026-09-30", "2026-11-03"), 2)
self.assertIsNone(rb._age_months("", "2026-11-03"))
class FetchTests(unittest.TestCase):
def test_fetch_skips_review_rows_unless_flagged(self):
rows = [{"title": "File:Good.jpg", "usable": True, "url": "u1", "clean": "CC0",
"licence": "CC0", "date": "2026-01-01"},
{"title": "File:ShareAlike.jpg", "usable": False, "research_only": True,
"url": "u2", "clean": None, "review": "CC BY-SA 3.0 (review)",
"licence": "CC BY-SA 3.0", "date": "2026-01-01"}]
with TemporaryDirectory() as tmp:
manifest = Path(tmp) / "meta.json"
manifest.write_text(json.dumps(rows))
args = mock.Mock(manifest=str(manifest), dir=str(Path(tmp) / "out"),
max=10, only_recent=False, force=True, include_review=False)
with mock.patch.object(rb.Commons, "_get", return_value=b"png"):
rb.cmd_fetch(args)
names = sorted(p.name for p in (Path(tmp) / "out").iterdir())
self.assertEqual(names, ["Good.jpg", "meta.json"])
def test_rows_write_sidecar_and_readme_lines(self):
rows = [{"title": "File:Some Photo.jpg", "usable": True, "date": "2026-05-01",
"clean": "CC0", "licence": "CC0", "artist": "An Author",
"page": "https://commons/x", "description": "A photo",
"url": "u", "width": 1600, "height": 900, "review": None,
"credit": "", "licence_url": ""}]
with TemporaryDirectory() as tmp:
manifest = Path(tmp) / "meta.json"
manifest.write_text(json.dumps(rows))
args = mock.Mock(manifest=str(manifest), append_sidecar="")
with mock.patch("builtins.print") as printed:
self.assertEqual(rb.cmd_rows(args), 0)
output = "\n".join(str(c[0][0]) for c in printed.call_args_list)
self.assertIn("| `some-photo-2026-05-01.jpg` |", output)
self.assertIn("![File:Some Photo.jpg](some-photo-2026-05-01.jpg)", output)
if __name__ == "__main__":
unittest.main()