#!/usr/bin/env python3 """Compare two processed magazine folders (a baseline and a re-import of the same PDF). Read-only. Nothing is guessed by eye: every number comes from the two folders' ``layout.json`` / ``articles.json`` (and, when present, ``quality_report.json``). python scripts/compare_magazines.py BASELINE_DIR NEW_DIR [--out report.json] [--md report.md] What is compared, per page and overall: * text - the narratable words of each page, aligned word by word (difflib), so a re-ordered or re-segmented but identical page still scores 1.0 * boxes - regions matched by overlap (IoU >= 0.5): recall/precision, type agreement * order - of the matched boxes, how many pairs are read in the same order * articles - matched by shared text (word 3-grams); title, author, page span, status * quality - pages needing a decision, unboxed areas, clipped edges, overlaps The JSON report lists every difference so a person (or a review screen) can decide, per difference, which side is better. """ from __future__ import annotations import argparse import difflib import json import re import sys from pathlib import Path from typing import Any TEXT_TYPES = {"title", "heading", "deck", "byline", "body", "quote", "caption", "sidebar", "pullquote"} NARRATED_TYPES = {"title", "heading", "deck", "byline", "body", "quote"} _WORD = re.compile(r"[\wÄÖÜäöüß'’-]+", re.U) _IOU_MATCH = 0.5 _SIMILAR_PAGE = 0.98 def load_magazine(folder: Path) -> dict[str, Any]: folder = Path(folder) layout = json.loads((folder / "layout.json").read_text("utf-8")) articles = json.loads((folder / "articles.json").read_text("utf-8")).get("articles", []) quality: dict[str, Any] = {} if (folder / "quality_report.json").is_file(): quality = json.loads((folder / "quality_report.json").read_text("utf-8")) return {"folder": str(folder), "layout": layout, "articles": articles, "quality": quality, "vision_run": load_vision_run(folder / "vision_run.json")} def load_vision_run(path: Path | None) -> dict[str, Any] | None: """Vision cache manifest of the run that produced a folder (core/vision_cache.py ``last_run.json``, copied next to the folder's layout.json as ``vision_run.json`` or passed with --*-vision-run).""" if path and Path(path).is_file(): try: return json.loads(Path(path).read_text("utf-8")) except (OSError, ValueError): return {"error": "unreadable"} return None def describe_vision_run(run: dict[str, Any] | None) -> str: if not run: return "not recorded (assume live; vision output not frozen)" if "error" in run: return "unreadable manifest" return (f"{run.get('mode', '?')}; models: {', '.join(run.get('models') or []) or 'none'}; " f"hits {run.get('hits', 0)}, recorded {run.get('recorded', 0)}, misses {run.get('misses', 0)}") def comparability_note(a: dict[str, Any] | None, b: dict[str, Any] | None) -> str: modes = {(r or {}).get("mode", "live") for r in (a, b)} if modes == {"replay"} or (modes <= {"record", "replay"} and (b or {}).get("mode") == "replay"): same = sorted((a or {}).get("models") or []) == sorted((b or {}).get("models") or []) return "frozen vision responses" + ("" if same else " (WARNING: different model sets)") return "NOT frozen: at least one side used live vision calls, OCR differences may be model noise" def words(text: str) -> list[str]: return [w.casefold().strip("-'’") for w in _WORD.findall(text or "") if w.strip("-'’")] def page_is_advertisement(layout: dict[str, Any], page: int) -> bool: """Same rule as core.magazine.page_is_advertisement, kept independent of it so this script has no import on the app: flagged by hand, or detected ad regions and nothing narratable. A page either side calls an advertisement is not a fair comparison -- being marked one is itself a one-time manual step a fresh re-import has usually not had yet, so the side that has not skips it fully instead (real OCR text where the other side has almost none), which is a difference in review state, not in pipeline quality.""" meta = next((p for p in layout.get("pages", []) if p.get("page") == page), None) if meta is None: return False if meta.get("advertisement"): return True on_page = [b for b in layout.get("blocks", []) if b.get("page") == page] return (any(b.get("type") == "ad" for b in on_page) and not any(b.get("type") in NARRATED_TYPES for b in on_page)) def page_words(layout: dict[str, Any], page: int) -> list[str]: blocks = sorted((b for b in layout.get("blocks", []) if b["page"] == page and b["type"] in NARRATED_TYPES), key=lambda b: b.get("reading_order", 0)) out: list[str] = [] for block in blocks: out.extend(words(block.get("text", ""))) return out def align(a: list[str], b: list[str]) -> dict[str, Any]: """Word-level alignment. Order-insensitive where it should be: each side is compared as read.""" matcher = difflib.SequenceMatcher(None, a, b, autojunk=False) only_a: list[str] = [] only_b: list[str] = [] for tag, i1, i2, j1, j2 in matcher.get_opcodes(): if tag in ("delete", "replace"): only_a.extend(a[i1:i2]) if tag in ("insert", "replace"): only_b.extend(b[j1:j2]) total = max(1, len(a) + len(b)) same = total - len(only_a) - len(only_b) return {"similarity": 1.0 if not (a or b) else round(same / total, 4), "words_a": len(a), "words_b": len(b), "only_in_a": only_a, "only_in_b": only_b} def iou(a: list[float], b: list[float]) -> float: ax, ay, aw, ah = a bx, by, bw, bh = b w = max(0.0, min(ax + aw, bx + bw) - max(ax, bx)) h = max(0.0, min(ay + ah, by + bh) - max(ay, by)) inter = w * h union = aw * ah + bw * bh - inter return inter / union if union > 0 else 0.0 def match_boxes(blocks_a: list[dict[str, Any]], blocks_b: list[dict[str, Any]]) -> list[tuple[dict[str, Any], dict[str, Any], float]]: pairs = sorted(((iou(x["bbox"], y["bbox"]), i, j) for i, x in enumerate(blocks_a) for j, y in enumerate(blocks_b)), reverse=True) used_a: set[int] = set() used_b: set[int] = set() matched = [] for score, i, j in pairs: if score < _IOU_MATCH: break if i in used_a or j in used_b: continue used_a.add(i) used_b.add(j) matched.append((blocks_a[i], blocks_b[j], score)) return matched def order_agreement(matched: list[tuple[dict[str, Any], dict[str, Any], float]]) -> float | None: """Share of matched box pairs that are read in the same relative order in both.""" if len(matched) < 2: return None concordant = total = 0 for i in range(len(matched)): for j in range(i + 1, len(matched)): da = matched[i][0].get("reading_order", 0) - matched[j][0].get("reading_order", 0) db = matched[i][1].get("reading_order", 0) - matched[j][1].get("reading_order", 0) if da == 0 or db == 0: continue total += 1 concordant += (da > 0) == (db > 0) return round(concordant / total, 4) if total else None def compare_pages(a: dict[str, Any], b: dict[str, Any]) -> list[dict[str, Any]]: pages_a = {p["page"] for p in a["layout"].get("pages", [])} pages_b = {p["page"] for p in b["layout"].get("pages", [])} rows = [] for page in sorted(pages_a | pages_b): blocks_a = [x for x in a["layout"].get("blocks", []) if x["page"] == page and x["type"] in TEXT_TYPES] blocks_b = [x for x in b["layout"].get("blocks", []) if x["page"] == page and x["type"] in TEXT_TYPES] text = align(page_words(a["layout"], page), page_words(b["layout"], page)) matched = match_boxes(blocks_a, blocks_b) images_a = [i for i in a["layout"].get("images", []) if i["page"] == page] images_b = [i for i in b["layout"].get("images", []) if i["page"] == page] figures = match_boxes([{"bbox": i["bbox"]} for i in images_a], [{"bbox": i["bbox"]} for i in images_b]) advertisement = page_is_advertisement(a["layout"], page) or page_is_advertisement(b["layout"], page) row = { "page": page, "advertisement": advertisement, "text_similarity": text["similarity"], "words_a": text["words_a"], "words_b": text["words_b"], "only_in_a": " ".join(text["only_in_a"][:40]), "only_in_b": " ".join(text["only_in_b"][:40]), "boxes_a": len(blocks_a), "boxes_b": len(blocks_b), "boxes_matched": len(matched), "box_recall": round(len(matched) / len(blocks_a), 4) if blocks_a else None, "box_precision": round(len(matched) / len(blocks_b), 4) if blocks_b else None, "type_agreement": round(sum(x["type"] == y["type"] for x, y, _ in matched) / len(matched), 4) if matched else None, "type_changes": sorted({f"{x['type']}->{y['type']}" for x, y, _ in matched if x["type"] != y["type"]}), "order_agreement": order_agreement(matched), "hand_ordered_in_a": bool(next((p for p in a["layout"].get("pages", []) if p["page"] == page), {}).get("order_locked")), "figures_a": len(images_a), "figures_b": len(images_b), "figures_matched": len(figures), "page_only_in": "a" if page not in pages_b else ("b" if page not in pages_a else ""), } # A page either side calls a full-page advertisement is skipped by the quality run on # that side, so its own boxes/text are not comparable pipeline output -- flagging it # "advertisement" instead of running the usual checks keeps it out of the difference # counts and the worst-pages list without hiding that it exists (see markdown()). if advertisement: row["flags"] = ["advertisement"] rows.append(row) continue flags = [] if row["text_similarity"] < _SIMILAR_PAGE: flags.append("text") if row["box_recall"] is not None and row["box_recall"] < 0.9: flags.append("boxes_missing_in_new") if row["box_precision"] is not None and row["box_precision"] < 0.9: flags.append("extra_boxes_in_new") if row["type_agreement"] is not None and row["type_agreement"] < 0.95: flags.append("types") if row["order_agreement"] is not None and row["order_agreement"] < 0.95: flags.append("order") if row["figures_a"] != row["figures_b"]: flags.append("figures") row["flags"] = flags rows.append(row) return rows def _article_text(magazine: dict[str, Any], article: dict[str, Any]) -> list[str]: ids = set(article.get("blocks") or []) blocks = sorted((b for b in magazine["layout"].get("blocks", []) if b["id"] in ids and b["type"] in NARRATED_TYPES), key=lambda b: (b["page"], b.get("reading_order", 0))) out: list[str] = [] for block in blocks: out.extend(words(block.get("text", ""))) return out def _shingles(ws: list[str], n: int = 3) -> set[tuple[str, ...]]: return {tuple(ws[i:i + n]) for i in range(len(ws) - n + 1)} if len(ws) >= n else {tuple(ws)} if ws else set() def compare_articles(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]: texts_a = {x["id"]: _article_text(a, x) for x in a["articles"]} texts_b = {x["id"]: _article_text(b, x) for x in b["articles"]} sh_a = {k: _shingles(v) for k, v in texts_a.items()} sh_b = {k: _shingles(v) for k, v in texts_b.items()} scores = [] for ia, sa in sh_a.items(): for ib, sb in sh_b.items(): if not sa or not sb: continue inter = len(sa & sb) if inter: scores.append((inter / len(sa | sb), ia, ib)) scores.sort(reverse=True) used_a: set[str] = set() used_b: set[str] = set() by_a = {x["id"]: x for x in a["articles"]} by_b = {x["id"]: x for x in b["articles"]} matched = [] for score, ia, ib in scores: if score < 0.3 or ia in used_a or ib in used_b: continue used_a.add(ia) used_b.add(ib) xa, xb = by_a[ia], by_b[ib] diff = align(texts_a[ia], texts_b[ib]) matched.append({ "a": ia, "b": ib, "overlap": round(score, 4), "text_similarity": diff["similarity"], "title_a": xa.get("title", ""), "title_b": xb.get("title", ""), "author_a": xa.get("author", ""), "author_b": xb.get("author", ""), "pages_a": xa.get("pages", []), "pages_b": xb.get("pages", []), "status_a": xa.get("status"), "status_b": xb.get("status"), "same_pages": xa.get("pages", []) == xb.get("pages", []), }) def brief(x: dict[str, Any], texts: dict[str, list[str]]) -> dict[str, Any]: return {"id": x["id"], "title": x.get("title", ""), "pages": x.get("pages", []), "words": len(texts[x["id"]])} return { "matched": matched, "only_in_a": [brief(x, texts_a) for x in a["articles"] if x["id"] not in used_a and texts_a[x["id"]]], "only_in_b": [brief(x, texts_b) for x in b["articles"] if x["id"] not in used_b and texts_b[x["id"]]], } def quality_numbers(m: dict[str, Any]) -> dict[str, Any]: rows = m["quality"].get("pages", []) return { "pages_needing_a_decision": sum(1 for r in rows if r.get("needs_review")), "unboxed_areas": sum(len(r.get("uncovered", [])) for r in rows), "clipped_edges": sum(len(r.get("clipped", [])) for r in rows), "overlapping_pairs": sum(len(r.get("overlaps", [])) for r in rows), "words_to_check": sum(len(r.get("spelling", [])) for r in rows), } def mean(values: list[float]) -> float | None: values = [v for v in values if v is not None] return round(sum(values) / len(values), 4) if values else None def compare(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]: pages = compare_pages(a, b) articles = compare_articles(a, b) # Pages either side calls a full-page advertisement are counted (see pages/words below, # so the totals still add up) but left out of every difference statistic -- being marked # one is a one-time manual step, not pipeline output, so comparing it is comparing review # state, not quality. Reported separately instead of silently dropped. compared = [p for p in pages if not p["advertisement"]] words_a = sum(p["words_a"] for p in pages) words_b = sum(p["words_b"] for p in pages) summary = { "baseline": a["folder"], "new": b["folder"], "pages": len(pages), "pages_skipped_as_advertisement": sum(p["advertisement"] for p in pages), "words": {"baseline": words_a, "new": words_b}, "text_similarity_mean": mean([p["text_similarity"] for p in compared]), "pages_with_text_differences": sum("text" in p["flags"] for p in compared), "box_recall_mean": mean([p["box_recall"] for p in compared]), "box_precision_mean": mean([p["box_precision"] for p in compared]), "type_agreement_mean": mean([p["type_agreement"] for p in compared]), "order_agreement_mean": mean([p["order_agreement"] for p in compared]), "pages_flagged": sum(bool(p["flags"]) for p in compared), "figures": {"baseline": sum(p["figures_a"] for p in compared), "new": sum(p["figures_b"] for p in compared)}, "articles": {"baseline": len(a["articles"]), "new": len(b["articles"]), "matched": len(articles["matched"]), "only_in_baseline": len(articles["only_in_a"]), "only_in_new": len(articles["only_in_b"]), "same_page_span": sum(x["same_pages"] for x in articles["matched"]), "same_title": sum(x["title_a"].strip().casefold() == x["title_b"].strip().casefold() for x in articles["matched"])}, "quality": {"baseline": quality_numbers(a), "new": quality_numbers(b)}, "vision": {"baseline": a.get("vision_run"), "new": b.get("vision_run"), "comparability": comparability_note(a.get("vision_run"), b.get("vision_run"))}, } return {"summary": summary, "pages": pages, "articles": articles} def markdown(report: dict[str, Any], worst: int = 15) -> str: s = report["summary"] lines = ["# Magazine comparison", "", f"* baseline: `{s['baseline']}`", f"* new: `{s['new']}`", f"* vision baseline: {describe_vision_run(s['vision']['baseline'])}", f"* vision new: {describe_vision_run(s['vision']['new'])}", f"* comparability: {s['vision']['comparability']}", "", "| measure | value |", "|---|---|", f"| pages | {s['pages']} |", f"| pages skipped as advertisement (either side) | {s['pages_skipped_as_advertisement']} |", f"| narratable words (baseline / new) | {s['words']['baseline']} / {s['words']['new']} |", f"| text similarity (mean per page) | {s['text_similarity_mean']} |", f"| pages with text differences | {s['pages_with_text_differences']} |", f"| box recall / precision | {s['box_recall_mean']} / {s['box_precision_mean']} |", f"| type agreement | {s['type_agreement_mean']} |", f"| reading-order agreement | {s['order_agreement_mean']} |", f"| figures (baseline / new) | {s['figures']['baseline']} / {s['figures']['new']} |", f"| articles (baseline / new / matched) | {s['articles']['baseline']} / {s['articles']['new']} / {s['articles']['matched']} |", f"| matched articles with the same page span | {s['articles']['same_page_span']} |", f"| pages flagged | {s['pages_flagged']} |", "", "## Quality report", "", "| | baseline | new |", "|---|---|---|"] for key in s["quality"]["baseline"]: lines.append(f"| {key.replace('_', ' ')} | {s['quality']['baseline'][key]} | {s['quality']['new'][key]} |") flagged = sorted((p for p in report["pages"] if p["flags"] and not p["advertisement"]), key=lambda p: (p["text_similarity"], p["page"]))[:worst] lines += ["", f"## The {len(flagged)} pages that differ most", "", "| page | flags | text | boxes b/n | order |", "|---|---|---|---|---|"] for p in flagged: lines.append(f"| {p['page']} | {', '.join(p['flags'])} | {p['text_similarity']} | {p['boxes_a']}/{p['boxes_b']} | {p['order_agreement']} |") for title, key in (("Only in the baseline", "only_in_a"), ("Only in the new import", "only_in_b")): rows = report["articles"][key] lines += ["", f"## Articles {title.lower()} ({len(rows)})", ""] lines += [f"* {r['title'][:60]!r} pages {r['pages']} ({r['words']} words)" for r in rows[:25]] return "\n".join(lines) + "\n" def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("baseline", type=Path) parser.add_argument("new", type=Path) parser.add_argument("--out", type=Path, help="write the full JSON report here") parser.add_argument("--md", type=Path, help="write a short Markdown summary here") parser.add_argument("--baseline-vision-run", type=Path, help="vision cache last_run.json of the baseline import") parser.add_argument("--new-vision-run", type=Path, help="vision cache last_run.json of the new import") args = parser.parse_args(argv) base, new = load_magazine(args.baseline), load_magazine(args.new) if args.baseline_vision_run: base["vision_run"] = load_vision_run(args.baseline_vision_run) if args.new_vision_run: new["vision_run"] = load_vision_run(args.new_vision_run) report = compare(base, new) if args.out: args.out.write_text(json.dumps(report, ensure_ascii=False, indent=1), "utf-8") text = markdown(report) if args.md: args.md.write_text(text, "utf-8") sys.stdout.write(text) return 0 if __name__ == "__main__": raise SystemExit(main())