97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier for the albums->photos extract artifact (ledger lane).
|
|
|
|
Re-fetches both JSONPlaceholder endpoints, re-derives the per-album table, and
|
|
compares the committed CSV + JSON against the fresh extraction (parsed, so line
|
|
endings and quoting do not affect the verdict). Also asserts the row counts,
|
|
the total photo count and the uniform photo_count claim above. Exits 0 only when
|
|
the committed artifacts match the fresh extraction.
|
|
"""
|
|
|
|
import csv
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
from collections import Counter
|
|
|
|
BASE = "/root/troupe-standup/wizard-work/ledger"
|
|
CSV_PATH = BASE + "/2026-10-31-albums-photo-counts.csv"
|
|
JSON_PATH = BASE + "/2026-10-31-albums-photo-counts.json"
|
|
|
|
ALBUMS_URL = "https://jsonplaceholder.typicode.com/albums"
|
|
PHOTOS_URL = "https://jsonplaceholder.typicode.com/photos"
|
|
|
|
|
|
def fetch(url):
|
|
req = urllib.request.Request(url, headers={"User-Agent": "ledger/extract"})
|
|
with urllib.request.urlopen(req, timeout=30) as r:
|
|
return json.load(r)
|
|
|
|
|
|
def read_csv_rows(path):
|
|
with open(path, encoding="utf-8") as f:
|
|
return list(csv.DictReader(f))
|
|
|
|
|
|
def main():
|
|
albums = fetch(ALBUMS_URL)
|
|
photos = fetch(PHOTOS_URL)
|
|
|
|
album_ids = {a["id"] for a in albums}
|
|
unknown = sorted({p["albumId"] for p in photos} - album_ids)
|
|
if unknown:
|
|
print("FAIL: photo albumIds not present in albums: %s" % unknown[:5])
|
|
return 1
|
|
|
|
counts = Counter(p["albumId"] for p in photos)
|
|
total = sum(counts.values())
|
|
|
|
# Fresh extraction result (same field types the extractor writes).
|
|
raw_rows = []
|
|
for a in sorted(albums, key=lambda x: x["id"]):
|
|
raw_rows.append({
|
|
"id": a["id"],
|
|
"userId": a["userId"],
|
|
"title": a["title"],
|
|
"photo_count": counts.get(a["id"], 0),
|
|
})
|
|
|
|
if len(albums) != 100:
|
|
print("FAIL: expected 100 albums, got %d" % len(albums))
|
|
return 1
|
|
if len(photos) != 5000:
|
|
print("FAIL: expected 5000 photos, got %d" % len(photos))
|
|
return 1
|
|
if len(counts) != 100:
|
|
print("FAIL: expected 100 distinct album ids with photos, got %d" % len(counts))
|
|
return 1
|
|
if len(set(counts.values())) != 1:
|
|
print("FAIL: photo_count not uniform, values=%s" % sorted(set(counts.values())))
|
|
return 1
|
|
|
|
# CSV: committed file parsed with DictReader -> all strings.
|
|
csv_rows = []
|
|
for r in raw_rows:
|
|
csv_rows.append({
|
|
"id": str(r["id"]),
|
|
"userId": str(r["userId"]),
|
|
"title": r["title"],
|
|
"photo_count": str(r["photo_count"]),
|
|
})
|
|
if read_csv_rows(CSV_PATH) != csv_rows:
|
|
print("FAIL: CSV does not match fresh extraction")
|
|
return 1
|
|
|
|
# JSON: committed file parsed -> ints for id/userId/photo_count.
|
|
if json.load(open(JSON_PATH, encoding="utf-8")) != raw_rows:
|
|
print("FAIL: JSON does not match fresh extraction")
|
|
return 1
|
|
|
|
print("OK: %d rows match the fresh extraction (uniform photo_count=50, %d photos)"
|
|
% (len(raw_rows), total))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|