#!/usr/bin/env python3 """Re-run the check behind dataset.json for "What EXIF Data Do Photos Still Carry?". Five legs, all reading their inputs from dataset.json next to this file: pages every URL in inputs.pages is fetched and each recorded quote must still appear on it, after whitespace, quote and entity normalisation. ledger the study's own published DATASET.csv (inputs.ledger.source) is downloaded and every number the post prints from it is recomputed and compared with inputs.ledger.cells. inputs the four test photos of the Google Photos run are rebuilt from the NASA public-domain photo with macOS sips and exiftool, the same steps as device-inputs/make_inputs.sh, and their SHA-256 compared with the recorded ones. Skipped where sips or exiftool is missing. device the recorded Google Photos downloads are checked against each other and against the inputs (bytes, pixel size, which tags survived), and the phone-library counts for internal consistency. Both needed an account or the owner's own library, so they are reported as recorded, not re-run. --downloads DIR re-hashes and re-reads saved downloads. corpus a fresh random draw from Wikimedia Commons read through the API's parsed metadata, plus a subsample of the same files downloaded and read with the JPEG/EXIF reader in this file, compared with the recorded draw within sampling tolerance. Modes (no flag) every leg --recheck pages only --build draw the corpus sample and write its results, the ledger and the pages into dataset.json, then dataset.csv --ledger recompute the ledger, pages and device results without a new corpus draw, and rewrite dataset.json and dataset.csv --csv rewrite dataset.csv from dataset.json Exit code 0 when everything holds, 1 otherwise. """ from __future__ import annotations import csv import hashlib import html import io import json import os import random import re import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from concurrent.futures import ThreadPoolExecutor HERE = os.path.dirname(os.path.abspath(__file__)) UA = "Mozilla/5.0 (compatible; VialloEvidenceCheck/1.0; +https://www.viallo.app)" API = "https://commons.wikimedia.org/w/api.php" POST_DATA = "https://storage.viallo.app/blog/what-exif-data-photos-still-carry/data" BATCHES = 60 # 60 x 50 files, one continuous random stream BATCH_SIZE = 50 CALIBRATION_FILES = 250 HEAD_BYTES = 131072 WORKERS = 3 INSTRUMENT_FILES = 20 MAX_INSTRUMENT_BYTES = 8 * 1024 * 1024 # The study's own list, copied from its published analyze.py, so the group split # here is the one the post measured with. PHONE_MAKES = { "Apple", "samsung", "SAMSUNG", "Xiaomi", "Google", "HUAWEI", "motorola", "Motorola", "Nokia", "LG Electronics", "LGE", "OnePlus", "OPPO", "realme", "ZTE", "vivo", "Sony Ericsson", "HTC", "BlackBerry", "Wiko", "alcatel", "ASUS", "Lenovo", } TYPE_SIZE = {1: 1, 2: 1, 3: 2, 4: 4, 5: 8, 6: 1, 7: 1, 8: 2, 9: 4, 10: 8, 11: 4, 12: 8} TAG_MAKE = 0x010F TAG_MODEL = 0x0110 TAG_EXIF_IFD = 0x8769 TAG_GPS_IFD = 0x8825 TAG_SERIAL = 0xA431 SERIAL_TAGS = ("SerialNumber", "BodySerialNumber", "CameraSerialNumber") GPS_LAT = ("GPSLatitude",) GPS_LON = ("GPSLongitude",) # ---------------------------------------------------------------- text helpers _QUOTES = { "‘": "'", "’": "'", "‚": "'", "‛": "'", "“": '"', "”": '"', "„": '"', "‟": '"', "–": "-", "—": "-", "−": "-", " ": " ", " ": " ", " ": " ", } def normalise(text: str) -> str: """Whitespace collapsed, curly quotes and dashes straightened, entities decoded.""" text = html.unescape(text) for bad, good in _QUOTES.items(): text = text.replace(bad, good) return re.sub(r"\s+", " ", text).strip() def html_to_text(raw: str) -> str: raw = re.sub(r"", " ", raw, flags=re.S | re.I) raw = re.sub(r"", " ", raw, flags=re.S | re.I) raw = re.sub(r"", " ", raw, flags=re.S) return normalise(re.sub(r"<[^>]+>", " ", raw)) def http_get(url: str, byte_range: int | None = None, tries: int = 5, timeout: int = 60): """Status and body. Retries a 429 or a 5xx the way Commons asks a client to.""" delay = 2.0 last = (0, b"") for _ in range(tries): req = urllib.request.Request(url, headers={"User-Agent": UA}) if byte_range: req.add_header("Range", "bytes=0-%d" % (byte_range - 1)) try: with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.status, resp.read() except urllib.error.HTTPError as exc: last = (exc.code, b"") if exc.code == 429: wait = float(exc.headers.get("Retry-After") or delay) elif exc.code >= 500: wait = delay else: return last except Exception as exc: # noqa: BLE001 - a dead URL is a result, not a crash last = (0, str(exc).encode()) wait = delay time.sleep(wait + random.uniform(0, 0.5)) delay = min(delay * 2, 30) return last # ---------------------------------------------------------------- JPEG / EXIF def jpeg_segments(buf: bytes): """APPn segments of a JPEG, in file order. None when the bytes are not a JPEG.""" if len(buf) < 4 or buf[0:2] != b"\xff\xd8": return None pos, segs, n = 2, [], len(buf) while pos + 4 <= n: if buf[pos] != 0xFF: nxt = buf.find(b"\xff", pos) if nxt < 0: break pos = nxt continue marker = buf[pos + 1] if marker == 0xFF: pos += 1 continue if marker in (0xD8, 0x01) or 0xD0 <= marker <= 0xD7: pos += 2 continue if marker in (0xDA, 0xD9): break length = int.from_bytes(buf[pos + 2:pos + 4], "big") if length < 2: break segs.append((marker, buf[pos + 4:pos + 2 + length])) pos += 2 + length return segs def _read_ifd(tiff: bytes, offset: int, endian: str) -> dict: if offset <= 0 or offset + 2 > len(tiff): return {} count = int.from_bytes(tiff[offset:offset + 2], endian) if count > 1000: return {} out, pos = {}, offset + 2 for _ in range(count): if pos + 12 > len(tiff): break tag = int.from_bytes(tiff[pos:pos + 2], endian) typ = int.from_bytes(tiff[pos + 2:pos + 4], endian) num = int.from_bytes(tiff[pos + 4:pos + 8], endian) out[tag] = (typ, num, tiff[pos + 8:pos + 12]) pos += 12 return out def _entry_bytes(tiff: bytes, entry, endian: str) -> bytes: typ, num, raw = entry size = TYPE_SIZE.get(typ, 1) * num if size <= 4: return raw[:size] offset = int.from_bytes(raw, endian) return tiff[offset:offset + size] def _ascii(tiff: bytes, entry, endian: str) -> str: if not entry: return "" return _entry_bytes(tiff, entry, endian).split(b"\x00")[0].decode("latin-1", "replace").strip() def _rationals(tiff: bytes, entry, endian: str): if entry is None: return [] data = _entry_bytes(tiff, entry, endian) out = [] for i in range(0, len(data) - 7, 8): out.append((int.from_bytes(data[i:i + 4], endian), int.from_bytes(data[i + 4:i + 8], endian))) return out def read_exif(buf: bytes) -> dict: """What this check measures in one file. Presence only, never content.""" out = { "is_jpeg": False, "exif": False, "xmp": False, "make": "", "model": "", "gps_block": False, "gps_nonzero": False, "serial": False, } segs = jpeg_segments(buf) if segs is None: return out out["is_jpeg"] = True for marker, seg in segs: if marker != 0xE1: continue if seg.startswith(b"http://ns.adobe.com/xap/1.0/\x00"): out["xmp"] = True continue if not seg.startswith(b"Exif\x00\x00"): continue tiff = seg[6:] if len(tiff) < 8: continue if tiff[0:2] == b"II": endian = "little" elif tiff[0:2] == b"MM": endian = "big" else: continue ifd0 = _read_ifd(tiff, int.from_bytes(tiff[4:8], endian), endian) if not ifd0: continue out["exif"] = True out["make"] = _ascii(tiff, ifd0.get(TAG_MAKE), endian) out["model"] = _ascii(tiff, ifd0.get(TAG_MODEL), endian) serial = False exif_off = ifd0.get(TAG_EXIF_IFD) if exif_off: exif_ifd = _read_ifd(tiff, int.from_bytes(_entry_bytes(tiff, exif_off, endian)[:4], endian), endian) if _ascii(tiff, exif_ifd.get(TAG_SERIAL), endian).strip("\x00 .-"): serial = True if not serial and _ascii(tiff, ifd0.get(TAG_SERIAL), endian).strip("\x00 .-"): serial = True out["serial"] = serial gps_off = ifd0.get(TAG_GPS_IFD) if gps_off: gps_ifd = _read_ifd(tiff, int.from_bytes(_entry_bytes(tiff, gps_off, endian)[:4], endian), endian) if gps_ifd: lat = _rationals(tiff, gps_ifd.get(0x0002), endian) lon = _rationals(tiff, gps_ifd.get(0x0004), endian) if lat or lon: out["gps_block"] = True if any(num != 0 for num, _den in lat + lon): out["gps_nonzero"] = True return out def is_phone(make: str) -> bool: return normalise(make).lower() in {m.lower() for m in PHONE_MAKES} # ---------------------------------------------------------------- corpus def parse_meta(metadata: list) -> dict: """The platform's own parse of a file's EXIF, as the API returns it.""" values = {} for entry in metadata or []: name = entry.get("name") if name: values[name] = entry.get("value") def number(name): try: return float(values.get(name)) except (TypeError, ValueError): return None lat = next((number(k) for k in GPS_LAT if number(k) is not None), None) lon = next((number(k) for k in GPS_LON if number(k) is not None), None) gps_block = lat is not None or lon is not None gps_nonzero = bool(gps_block and ((lat or 0) != 0 or (lon or 0) != 0)) serial = "" for tag in SERIAL_TAGS: value = values.get(tag) if value is not None and str(value).strip("\x00 .-"): serial = str(value).strip() break return { "make": str(values.get("Make") or "").strip(), "model": str(values.get("Model") or "").strip(), "exif": bool(values), "gps_block": gps_block, "gps_nonzero": gps_nonzero, "serial": bool(serial), } def draw_sample(batches: int) -> list[dict]: """The platform's own random generator, one continuous stream, JPEGs only.""" pages, cont = [], None for _ in range(batches): params = { "action": "query", "format": "json", "formatversion": "2", "generator": "random", "grnnamespace": "6", "grnlimit": str(BATCH_SIZE), "prop": "imageinfo", "iiprop": "url|size|mime|metadata", # A unique requestid per call makes the URL unique: without it the CDN # answers the same cached draw every time, and a "fresh draw" becomes a # replay of the last one. maxage/smaxage alone did not stop it. "requestid": "%d-%d" % (int(time.time()), random.randint(0, 999999)), } if cont: params["grncontinue"] = cont status, body = http_get(API + "?" + urllib.parse.urlencode(params), timeout=120) if status != 200: raise SystemExit("random draw failed: HTTP %s" % status) payload = json.loads(body) pages.extend(payload.get("query", {}).get("pages", [])) cont = payload.get("continue", {}).get("grncontinue") if not cont: break sample = [] for page in pages: info = (page.get("imageinfo") or [{}])[0] if info.get("mime") != "image/jpeg": continue row = {"title": page.get("title", ""), "url": info.get("url", "").split("?")[0], "size": info.get("size", 0)} row.update(parse_meta(info.get("metadata"))) sample.append(row) return sample def measure_file(item: dict) -> dict: """What the file itself carries, read from its first HEAD_BYTES bytes.""" status, body = http_get(item["url"], byte_range=HEAD_BYTES) if status not in (200, 206) or not body: return {"http": status} parsed = read_exif(body) parsed["http"] = status return parsed def calibration(sample: list[dict], count: int) -> tuple[list[dict], dict]: """The platform's field against the file itself, on the same files.""" picked = sample[:count] results = [] with ThreadPoolExecutor(max_workers=WORKERS) as pool: fresh = list(pool.map(measure_file, picked)) for item, read in zip(picked, fresh): if read.get("http") not in (200, 206): continue results.append({ "title": item["title"], "url": item["url"], "platform_serial": item["serial"], "file_serial": read["serial"], "platform_gps": item["gps_nonzero"], "file_gps": read["gps_nonzero"], "file_make": read["make"], "platform_make": item["make"], }) agree = sum(1 for r in results if r["platform_serial"] == r["file_serial"] and r["platform_gps"] == r["file_gps"]) stats = { "compared": len(results), "agree_on_both_fields": agree, "agree_share": round(agree / len(results), 4) if results else 0.0, "file_serial_share": round(sum(1 for r in results if r["file_serial"]) / len(results), 4) if results else 0.0, "platform_serial_share": round(sum(1 for r in results if r["platform_serial"]) / len(results), 4) if results else 0.0, "file_location_share": round(sum(1 for r in results if r["file_gps"]) / len(results), 4) if results else 0.0, "platform_location_share": round(sum(1 for r in results if r["platform_gps"]) / len(results), 4) if results else 0.0, } return results, stats def share(rows: list[dict], key: str) -> float: return round(sum(1 for r in rows if r.get(key)) / len(rows), 4) if rows else 0.0 def aggregates(rows: list[dict]) -> dict: jpeg = [r for r in rows if r.get("is_jpeg", True)] phone = [r for r in jpeg if is_phone(r.get("make", ""))] other = [r for r in jpeg if r.get("make", "").strip() and not is_phone(r.get("make", ""))] blocks = [r for r in jpeg if r.get("gps_block")] result = { "files": len(jpeg), "make_share": round(sum(1 for r in jpeg if r.get("make")) / len(jpeg), 4) if jpeg else 0.0, "location_share": share(jpeg, "gps_nonzero"), "gps_block_share": share(jpeg, "gps_block"), "zero_blocks_of_all": round(sum(1 for r in blocks if not r["gps_nonzero"]) / len(jpeg), 4) if jpeg else 0.0, "serial_share": share(jpeg, "serial"), "phone_files": len(phone), "phone_location_share": share(phone, "gps_nonzero"), "phone_serial_share": share(phone, "serial"), "other_files": len(other), "other_location_share": share(other, "gps_nonzero"), "other_serial_share": share(other, "serial"), } makes = {} for row in jpeg: key = row.get("make", "") or "(no make)" entry = makes.setdefault(key, {"files": 0, "location": 0, "serial": 0}) entry["files"] += 1 entry["location"] += 1 if row["gps_nonzero"] else 0 entry["serial"] += 1 if row["serial"] else 0 result["makes"] = { key: {"files": v["files"], "location_share": round(v["location"] / v["files"], 4), "serial_share": round(v["serial"] / v["files"], 4)} for key, v in sorted(makes.items(), key=lambda kv: -kv[1]["files"]) } return result def instrument_check(sample: list[dict]) -> dict: """This file's reader against exiftool 13.55, the study's own second reader.""" exiftool = shutil.which("exiftool") if not exiftool: return {"tool": "exiftool", "compared": 0, "agreed": 0, "note": "exiftool not installed, instrument check not run"} picked = [s for s in sample if 0 < s.get("size", 0) <= MAX_INSTRUMENT_BYTES][:INSTRUMENT_FILES] workdir = tempfile.mkdtemp(prefix="exif-instrument-") agreed = compared = 0 disagreements = [] try: for item in picked: status, body = http_get(item["url"], tries=4, timeout=180) if status != 200 or not body: continue path = os.path.join(workdir, "%d.jpg" % compared) with open(path, "wb") as handle: handle.write(body) mine = read_exif(body) out = subprocess.run( [exiftool, "-n", "-j", "-Make", "-SerialNumber", "-BodySerialNumber", "-GPSLatitude", "-GPSLongitude", path], capture_output=True, text=True, timeout=120) try: theirs = json.loads(out.stdout)[0] except Exception: continue compared += 1 their_make = str(theirs.get("Make", "")).strip() their_serial = bool(str(theirs.get("SerialNumber") or theirs.get("BodySerialNumber") or "").strip("\x00 .-")) their_gps = any(key in theirs and str(theirs[key]) not in ("0", "0.0", "") for key in ("GPSLatitude", "GPSLongitude")) same = (normalise(mine["make"]).lower() == normalise(their_make).lower() and mine["serial"] == their_serial and mine["gps_nonzero"] == their_gps) if same: agreed += 1 else: disagreements.append({"title": item["title"], "make": [mine["make"], their_make], "serial": [mine["serial"], their_serial], "location": [mine["gps_nonzero"], their_gps]}) finally: shutil.rmtree(workdir, ignore_errors=True) return {"tool": "exiftool 13.55", "compared": compared, "agreed": agreed, "disagreements": disagreements} DATASET = os.path.join(HERE, "dataset.json") CSV_OUT = os.path.join(HERE, "dataset.csv") def load() -> dict: with open(DATASET, encoding="utf-8") as handle: return json.load(handle) def save(dataset: dict) -> None: with open(DATASET, "w", encoding="utf-8") as handle: json.dump(dataset, handle, indent=2, ensure_ascii=False) handle.write("\n") # ---------------------------------------------------------------- pages def check_pages(rows: list[dict]) -> tuple[list[str], list[dict]]: failures, results, cache = [], [], {} for row in rows: url = row["url"] if url not in cache: status, body = http_get(url, tries=3, timeout=90) cache[url] = (status, html_to_text(body.decode("utf-8", "replace")) if body else "") status, text = cache[url] found = [] if status != 200: failures.append("%s %s: HTTP %s, page not read" % (row["id"], url, status)) for quote in row["quotes"]: present = status == 200 and normalise(quote) in text found.append({"quote": quote, "present": present}) if status == 200 and not present: failures.append("%s %s: quote gone: %r" % (row["id"], url, quote[:90])) results.append({"id": row["id"], "party": row["party"], "post_claim": row["post_claim"], "url": url, "http": status, "quotes": found}) return failures, results # ---------------------------------------------------------------- ledger _LAST_LEDGER = {} def ledger_rows(source: str) -> list[dict]: status, body = http_get(source, tries=3, timeout=180) if status != 200: raise SystemExit("published DATASET.csv not read: HTTP %s" % status) _LAST_LEDGER["sha256"] = hashlib.sha256(body).hexdigest() return list(csv.DictReader(io.StringIO(body.decode("utf-8", "replace")))) def ledger_counts(rows: list[dict]) -> dict: scored = [r for r in rows if r["excluded"] == "0"] whole = [r for r in scored if r["whole_file_read"] == "1"] def one(r, key): return r[key] == "1" gps = lambda r: one(r, "gps_block") and not one(r, "gps_all_zero") serial = lambda r: one(r, "body_serial_exif") or one(r, "serial_xmp") wide = lambda r: one(r, "exiftool_wide_serial_any") narrow = lambda r: one(r, "exiftool_serial_any") phone = [r for r in scored if is_phone(r["make"])] other = [r for r in scored if r["make"].strip() and not is_phone(r["make"])] def count(rs, fn): return sum(1 for r in rs if fn(r)) nikon_corp = [r for r in scored if r["make"] == "NIKON CORPORATION"] nikon = [r for r in scored if r["make"] == "NIKON"] coolpix = [r for r in nikon_corp + nikon if r["model"].startswith("COOLPIX")] out = { "scored": len(scored), "excluded": len(rows) - len(scored), "exif": count(scored, lambda r: one(r, "has_exif")), "make": count(scored, lambda r: bool(r["make"].strip())), "gps": count(scored, gps), "gps_zero": count(scored, lambda r: one(r, "gps_all_zero")), "serial_file": count(scored, serial), "serial_with_make": count(scored, lambda r: serial(r) and bool(r["make"].strip())), "platform_serial": count(scored, lambda r: one(r, "platform_serial")), "phone_files": len(phone), "phone_loc": count(phone, gps), "phone_ser": count(phone, serial), "other_files": len(other), "other_loc": count(other, gps), "other_ser": count(other, serial), "whole_all": count(rows, lambda r: one(r, "whole_file_read")), "whole_rows": len(whole), "std_serial": count(whole, lambda r: one(r, "body_serial_exif")), "any_two": count(whole, serial), "narrow": count(whole, narrow), "wide": count(whole, wide), "missed": count(whole, lambda r: wide(r) and not serial(r)), "internal_only": count(whole, lambda r: one(r, "exiftool_wide_serial_internal_only")), "exif_only": count(whole, lambda r: one(r, "exiftool_wide_serial_exif") and not one(r, "exiftool_wide_serial_xmp") and not one(r, "exiftool_wide_serial_makernote")), "xmp_only": count(whole, lambda r: one(r, "exiftool_wide_serial_xmp") and not one(r, "exiftool_wide_serial_exif") and not one(r, "exiftool_wide_serial_makernote")), "makernote_only": count(whole, lambda r: one(r, "exiftool_wide_serial_makernote") and not one(r, "exiftool_wide_serial_exif") and not one(r, "exiftool_wide_serial_xmp")), "reader_disagreements": count( whole, lambda r: one(r, "body_serial_exif") != one(r, "exiftool_serial_exif")), "nikon_corp_series": count(nikon_corp, lambda r: r["model"].startswith("NIKON ")), "nikon_corp_coolpix": count(nikon_corp, lambda r: r["model"].startswith("COOLPIX")), "nikon_coolpix": count(nikon, lambda r: r["model"].startswith("COOLPIX")), "nikon_other_names": count(nikon, lambda r: not r["model"].startswith("COOLPIX")), "coolpix_files": len(coolpix), "coolpix_serial": count(coolpix, serial), } makes, models = {}, {} for r in scored: makes.setdefault(r["make"], []).append(r) models.setdefault(r["model"], []).append(r) for name, group in makes.items(): out["make:" + name] = len(group) out["make_loc:" + name] = count(group, gps) out["make_ser:" + name] = count(group, serial) for name, group in models.items(): out["model:" + name] = len(group) out["model_loc:" + name] = count(group, gps) out["model_ser:" + name] = count(group, serial) by_serial = sorted((count(v, serial) for k, v in makes.items() if k.strip()), reverse=True) out["top3_ser"] = sum(by_serial[:3]) overall = out["serial_file"] / len(scored) if scored else 0.0 rates = [(abs(count(v, serial) / len(v) - overall), k) for k, v in makes.items() if k.strip() and len(v) >= 20] out["max_gap"] = round(max(rates)[0], 4) if rates else 0.0 out["low_makes"] = sum(1 for _gap, k in rates if count(makes[k], serial) / len(makes[k]) <= 0.01) return out def check_ledger(spec: dict) -> tuple[list[str], list[dict]]: rows = ledger_rows(spec["source"]) if spec.get("sha256") and _LAST_LEDGER.get("sha256") != spec["sha256"]: print("note: the published DATASET.csv has changed since this check was recorded " "(%s, recorded %s); the cells below are recomputed from the current file" % (_LAST_LEDGER.get("sha256", "")[:12], spec["sha256"][:12])) counts = ledger_counts(rows) tolerance = spec["tolerance"] failures, results = [], [] for cell in spec["cells"]: if cell["selector"] not in counts: failures.append("%s: selector %s found nothing in the published data" % (cell["cell"], cell["selector"])) continue if cell["kind"] == "share": got = round(counts[cell["selector"]] / cell["n"], 4) else: got = counts[cell["selector"]] matched = abs(got - cell["value"]) <= tolerance results.append({"cell": cell["cell"], "what": cell["what"], "post": cell["value"], "recomputed": got, "match": matched, "kind": cell["kind"], "printed": cell.get("printed", True), "metric": cell.get("metric", ""), "why": cell.get("why", "") if not matched else ""}) if matched != (cell["expected"] == "match"): failures.append("%s: post %s, recomputed %s, expected to %s" % (cell["cell"], cell["value"], got, cell["expected"])) return failures, results # ---------------------------------------------------------------- inputs NASA_SRC = ("https://images-assets.nasa.gov/image/KSC-20220119-PH-CSH01_0011/" "KSC-20220119-PH-CSH01_0011~orig.jpg") INPUT_TAGS = [ "-Make=Apple", "-Model=iPhone 15 Pro", "-Software=18.0", "-LensModel=iPhone 15 Pro back triple camera 6.765mm f/1.78", "-DateTimeOriginal=2026:10:25 09:30:00", "-CreateDate=2026:10:25 09:30:00", "-OffsetTimeOriginal=-04:00", "-GPSLatitude=28.586", "-GPSLatitudeRef=N", "-GPSLongitude=80.651", "-GPSLongitudeRef=W", "-GPSAltitude=3", "-GPSAltitudeRef=0", "-ExposureTime=1/120", "-FNumber=1.78", "-ISO=80", "-Orientation#=1", ] def sha256_file(path: str) -> str: with open(path, "rb") as handle: return hashlib.sha256(handle.read()).hexdigest() def rebuild_inputs(workdir: str) -> dict: """The steps of make_inputs.sh, plus the XMP note that makes the -s copies.""" run = lambda *cmd: subprocess.run(cmd, check=True, capture_output=True, cwd=workdir, timeout=300) status, body = http_get(NASA_SRC, tries=3, timeout=300) if status != 200: raise RuntimeError("NASA source photo not read: HTTP %s" % status) with open(os.path.join(workdir, "src.jpg"), "wb") as handle: handle.write(body) run("sips", "-c", "5120", "3840", "src.jpg", "--out", "crop.jpg") run("sips", "-z", "4032", "3024", "-s", "formatOptions", "90", "crop.jpg", "--out", "photo-12mp.jpg") run("sips", "-z", "8064", "6048", "-s", "format", "heic", "crop.jpg", "--out", "photo-48mp.heic") for name in ("photo-12mp.jpg", "photo-48mp.heic"): run("exiftool", "-q", "-overwrite_original", "-all=", name) run("exiftool", "-q", "-overwrite_original", *INPUT_TAGS, name) for name, copy in (("photo-12mp.jpg", "photo-12mp-s.jpg"), ("photo-48mp.heic", "photo-48mp-s.heic")): shutil.copyfile(os.path.join(workdir, name), os.path.join(workdir, copy)) run("exiftool", "-q", "-overwrite_original", "-XMP-dc:Description=storage saver copy", copy) return {name: sha256_file(os.path.join(workdir, name)) for name in ("photo-12mp.jpg", "photo-48mp.heic", "photo-12mp-s.jpg", "photo-48mp-s.heic")} def check_inputs(device: dict) -> tuple[list[str], list[dict]]: if not (shutil.which("sips") and shutil.which("exiftool")): print("inputs: skipped, needs macOS sips and exiftool") return [], [] workdir = tempfile.mkdtemp(prefix="exif-inputs-") failures, results = [], [] try: built = rebuild_inputs(workdir) except Exception as exc: # noqa: BLE001 - a failed rebuild is a result shutil.rmtree(workdir, ignore_errors=True) return ["inputs: rebuild failed: %s" % exc], [] shutil.rmtree(workdir, ignore_errors=True) for name, item in device["google"]["inputs"].items(): same = built.get(name) == item["sha256"] results.append({"file": name, "recorded": item["sha256"], "rebuilt": built.get(name), "match": same}) if not same: failures.append("inputs: %s rebuilt as %s, recorded %s" % (name, (built.get(name) or "")[:12], item["sha256"][:12])) return failures, results # ---------------------------------------------------------------- device KEPT_TAGS = ("model", "lensModel", "dateTimeOriginal", "offsetTimeOriginal") def check_device(device: dict, downloads: str | None) -> tuple[list[str], list[dict]]: """The recorded downloads against the recorded inputs, and against each other.""" failures, rows = [], [] google = device["google"] inputs = google["inputs"] def source(name: str) -> dict: stem = os.path.splitext(name)[0] return next(v for k, v in inputs.items() if os.path.splitext(k)[0] == stem) for run, entry in google["runs"].items(): for item in entry["files"]: sent = source(item["file"]) row = { "run": run, "file_sent": sent["file"], "file_returned": item["file"], "type_sent": sent["type"], "type_returned": item["type"], "bytes_sent": sent["bytes"], "bytes_returned": item["bytes"], "pixels_sent": "%dx%d" % (sent["width"], sent["height"]), "pixels_returned": "%dx%d" % (item["width"], item["height"]), "location_kept": item["gps"], "model_lens_date_timezone_kept": all(item.get(t) == sent.get(t) for t in KEPT_TAGS), "byte_identical": item["sha256"] == sent["sha256"], "pixels_identical": item["imageDataSha256"] == sent["imageDataSha256"], } rows.append(row) if run.startswith("original quality") and not row["byte_identical"]: failures.append("device %s/%s: recorded as the uploaded file but the hash differs" % (run, item["file"])) if run.startswith("storage saver") and (row["byte_identical"] or item["bytes"] >= sent["bytes"]): failures.append("device %s/%s: recorded as re-encoded but not smaller" % (run, item["file"])) if run.startswith("shared link") and item["gps"]: failures.append("device %s/%s: recorded with a location" % (run, item["file"])) if not row["model_lens_date_timezone_kept"]: failures.append("device %s/%s: a camera tag differs from the input" % (run, item["file"])) if downloads: failures.extend(reread_download(downloads, run, item)) library = device["phoneLibrary"] photos = library["photos"] if sum(library["fileTypes"].values()) != photos: failures.append("phone library: file types do not add up to %d photos" % photos) for field, value in library["withField"].items(): if value > photos: failures.append("phone library: %s (%d) exceeds the photo count" % (field, value)) return failures, rows RUN_DIRS = { "original quality, owner download (Shift+D)": "google-original", "storage saver, owner download (Shift+D)": "google-storage-saver", "shared link, signed-out download": "google-link-signed-out", } def reread_download(root: str, run: str, item: dict) -> list[str]: path = os.path.join(root, RUN_DIRS[run], item["file"]) if not os.path.exists(path): return ["device %s: saved download missing" % path] failures = [] if sha256_file(path) != item["sha256"]: failures.append("device %s: sha256 differs from the record" % path) if shutil.which("exiftool"): out = subprocess.run(["exiftool", "-n", "-j", "-Model", "-LensModel", "-DateTimeOriginal", "-OffsetTimeOriginal", "-GPSLatitude", "-ImageWidth", "-ImageHeight", path], capture_output=True, text=True, timeout=120) tags = json.loads(out.stdout)[0] if ("GPSLatitude" in tags) != item["gps"]: failures.append("device %s: location present is %s, recorded %s" % (path, "GPSLatitude" in tags, item["gps"])) if (tags.get("ImageWidth"), tags.get("ImageHeight")) != (item["width"], item["height"]): failures.append("device %s: pixel size differs from the record" % path) if str(tags.get("OffsetTimeOriginal", "")) != item.get("offsetTimeOriginal", ""): failures.append("device %s: time zone offset differs from the record" % path) return failures # ---------------------------------------------------------------- corpus # Same-day draws spread wider than simple random sampling predicts, because uploads come in # batches from one camera; a fresh draw passes inside the recorded range widened by MARGIN. MARGIN = 0.03 CLAIMS = [ ("phone files carry almost no serial", lambda s: s["phone_serial_share"] <= 0.01), ("other devices carry a serial at least 15 points more often than phones", lambda s: s["other_serial_share"] - s["phone_serial_share"] >= 0.15), ("phone files carry a location at least 30 points more often than other devices", lambda s: s["phone_location_share"] - s["other_location_share"] >= 0.30), ] def check_corpus(dataset: dict) -> list[str]: failures = [] span = dataset["results"]["repeatDraws"]["fullDrawRange"] sample = draw_sample(BATCHES) was_drawn = {f["title"] for f in dataset["inputs"]["corpus"]["files"]} replay = sum(1 for s in sample[:50] if s["title"] in was_drawn) if replay > 40: failures.append("corpus: %d of the first 50 titles are the recorded sample, so this is a " "cached replay and not a fresh draw" % replay) stats = aggregates(sample) print("corpus: fresh draw of %d files" % stats["files"]) for key, (low, high) in span.items(): now = stats[key] mark = "ok" if low - MARGIN <= now <= high + MARGIN else "MOVED" print(" %-22s recorded %.4f to %.4f, now %.4f %s" % (key, low, high, now, mark)) if mark != "ok": failures.append("%s is %.4f, outside %.4f to %.4f by more than %.2f" % (key, now, low, high, MARGIN)) for claim, holds in CLAIMS: ok = holds(stats) print(" claim: %s: %s" % (claim, "holds" if ok else "FAILS")) if not ok: failures.append("corpus claim fails: %s" % claim) _rows, calib = calibration(sample, CALIBRATION_FILES) was = dataset["results"]["calibration"] print("calibration: %d files, agreement %.4f (recorded %.4f)" % (calib["compared"], calib["agree_share"], was["agree_share"])) if calib["compared"] < 20: failures.append("calibration: only %d files read" % calib["compared"]) elif calib["agree_share"] < was["agree_share"] - 0.1: failures.append("calibration: agreement fell to %.4f from %.4f" % (calib["agree_share"], was["agree_share"])) return failures # ---------------------------------------------------------------- dataset.csv CSV_FIELDS = ["check", "option", "read_from", "files", "location", "serial", "model_lens_date_timezone", "pixels", "bytes", "value", "post_value", "note"] def pct(value) -> str: return "%.1f%%" % (value * 100) def write_csv(dataset: dict) -> None: results = dataset["results"] corpus, calib = results["corpus"], results["calibration"] drew = "Commons fresh draw %s" % dataset["checkedAt"] rows = [] for cell in results["ledger"]: if not cell.get("printed", True): continue value = pct(cell["recomputed"]) if cell["kind"] == "share" else cell["recomputed"] row = {"check": "ledger", "option": "%s %s" % (cell["cell"], cell["what"]), "read_from": "the study's published DATASET.csv", "value": value, "post_value": pct(cell["post"]) if cell["kind"] == "share" else cell["post"], "note": "match" if cell["match"] else cell["why"]} if cell.get("metric") in ("location", "serial", "files"): row[cell["metric"]] = value rows.append(row) rows.append({"check": "corpus", "option": drew + ", first files of the draw", "read_from": "the file, first 131072 bytes", "files": calib["compared"], "location": pct(calib["file_location_share"]), "serial": pct(calib["file_serial_share"])}) for option, files, loc, ser in [ ("all JPEG files", corpus["files"], corpus["location_share"], corpus["serial_share"]), ("a phone maker", corpus["phone_files"], corpus["phone_location_share"], corpus["phone_serial_share"]), ("every other kind of device", corpus["other_files"], corpus["other_location_share"], corpus["other_serial_share"]), ]: rows.append({"check": "corpus", "option": "%s, %s" % (drew, option), "read_from": "Commons metadata field (imageinfo metadata)", "files": files, "location": pct(loc), "serial": pct(ser)}) for make, value in corpus["makes"].items(): if value["files"] >= 20 and make != "(no make)": rows.append({"check": "corpus", "option": "%s, make %s" % (drew, make), "read_from": "Commons metadata field (imageinfo metadata)", "files": value["files"], "location": pct(value["location_share"]), "serial": pct(value["serial_share"])}) for draw in results.get("repeatDraws", {}).get("draws", []): for group, prefix in (("all JPEG files", ""), ("a phone maker", "phone_"), ("every other kind of device", "other_")): rows.append({"check": "corpus-repeat", "option": "%s, %s: %s" % (drew, draw["run"], group), "read_from": "Commons metadata field (imageinfo metadata)", "files": draw["files"] if not prefix else "", "location": pct(draw[prefix + "location_share"]), "serial": pct(draw[prefix + "serial_share"])}) library = dataset["inputs"]["device"]["phoneLibrary"] fields = library["withField"] loc = next(v for k, v in fields.items() if k.startswith("location")) ser = next(v for k, v in fields.items() if "serial" in k) rows.append({"check": "device", "option": "one iPhone library, %s" % library["checkedAt"], "read_from": library["tool"], "files": library["photos"], "location": "%d (%s)" % (loc, pct(loc / library["photos"])), "serial": ser, "model_lens_date_timezone": "model %d, lens %d, date %d, time zone %d" % ( fields["camera model"], fields["lens model"], fields["capture date (DateTimeOriginal)"], fields["time zone offset (OffsetTimeOriginal)"]), "note": "counts only, approved by the owner; recorded, not re-runnable"}) for row in results["device"]: rows.append({"check": "device", "option": "Google Photos, %s: %s sent, %s returned" % ( row["run"], row["file_sent"], row["type_returned"]), "read_from": "the downloaded file, exiftool", "files": 1, "location": "kept" if row["location_kept"] else "removed", "serial": "none written in the test file", "model_lens_date_timezone": "kept" if row["model_lens_date_timezone_kept"] else "changed", "pixels": "%s to %s" % (row["pixels_sent"], row["pixels_returned"]), "bytes": "%d to %d" % (row["bytes_sent"], row["bytes_returned"]), "note": "byte-identical" if row["byte_identical"] else ( "same pixels, metadata changed" if row["pixels_identical"] else "re-encoded")}) with open(CSV_OUT, "w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS, extrasaction="ignore") writer.writeheader() writer.writerows(rows) # ---------------------------------------------------------------- modes def tools() -> dict: versions = {"python": sys.version.split()[0], "urllib": "stdlib"} for name, args in (("exiftool", ["-ver"]),): path = shutil.which(name) versions[name] = subprocess.run([path, *args], capture_output=True, text=True, timeout=30).stdout.strip() if path else "not installed" versions["sips"] = "macOS built-in" if shutil.which("sips") else "not installed" return versions def build() -> int: dataset = load() sample = draw_sample(BATCHES) print("drew %d JPEG files in %d API calls" % (len(sample), BATCHES)) _rows, calib = calibration(sample, CALIBRATION_FILES) print(" platform and file agree on %d of %d" % (calib["agree_on_both_fields"], calib["compared"])) dataset["instrument"] = instrument_check(sample) dataset["inputs"]["corpus"]["files"] = [{"title": s["title"], "url": s["url"]} for s in sample] dataset["results"]["corpus"] = aggregates(sample) dataset["results"]["calibration"] = calib _f, dataset["results"]["pages"] = check_pages(dataset["inputs"]["pages"]) _f, dataset["results"]["ledger"] = check_ledger(dataset["inputs"]["ledger"]) _f, dataset["results"]["device"] = check_device(dataset["inputs"]["device"], None) dataset["tools"] = tools() save(dataset) write_csv(dataset) print("wrote dataset.json and dataset.csv") return 0 def refresh_ledger() -> int: dataset = load() failures, dataset["results"]["ledger"] = check_ledger(dataset["inputs"]["ledger"]) page_failures, dataset["results"]["pages"] = check_pages(dataset["inputs"]["pages"]) device_failures, dataset["results"]["device"] = check_device(dataset["inputs"]["device"], None) failures += page_failures + device_failures save(dataset) write_csv(dataset) for line in failures: print("FAIL %s" % line) return 0 if not failures else 1 def verify(recheck: bool, downloads: str | None) -> int: dataset = load() failures = [] page_failures, pages = check_pages(dataset["inputs"]["pages"]) print("pages: %d rows, %s" % (len(pages), "ok" if not page_failures else "%d failure(s)" % len(page_failures))) failures.extend(page_failures) if not recheck: ledger_failures, ledger = check_ledger(dataset["inputs"]["ledger"]) matched = sum(1 for c in ledger if c["match"]) print("ledger: %d of %d cells reproduce from the published DATASET.csv" % (matched, len(ledger))) for cell in ledger: if not cell["match"]: print(" %s post %s, recomputed %s (expected: %s)" % (cell["cell"], cell["post"], cell["recomputed"], cell["why"])) failures.extend(ledger_failures) input_failures, built = check_inputs(dataset["inputs"]["device"]) if built: print("inputs: %d of %d rebuilt files match the recorded SHA-256" % (sum(1 for b in built if b["match"]), len(built))) failures.extend(input_failures) device_failures, _rows = check_device(dataset["inputs"]["device"], downloads) print("device: recorded runs %s%s" % ("consistent" if not device_failures else "INCONSISTENT", ", saved downloads re-read" if downloads else "")) failures.extend(device_failures) failures.extend(check_corpus(dataset)) for line in failures: print("FAIL %s" % line) print("RESULT %s" % ("PASS" if not failures else "FAIL")) return 0 if not failures else 1 def main() -> int: started = time.time() downloads = None if "--downloads" in sys.argv: downloads = sys.argv[sys.argv.index("--downloads") + 1] if "--build" in sys.argv: code = build() elif "--ledger" in sys.argv: code = refresh_ledger() elif "--csv" in sys.argv: write_csv(load()) code = 0 else: code = verify(recheck="--recheck" in sys.argv, downloads=downloads) print("done in %.1fs" % (time.time() - started)) return code if __name__ == "__main__": sys.exit(main())