312 lines
15 KiB
Python
312 lines
15 KiB
Python
#!/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}
|
||
|
||
|
||
def words(text: str) -> list[str]:
|
||
return [w.casefold().strip("-'’") for w in _WORD.findall(text or "") if w.strip("-'’")]
|
||
|
||
|
||
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])
|
||
row = {
|
||
"page": page,
|
||
"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 ""),
|
||
}
|
||
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)
|
||
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),
|
||
"words": {"baseline": words_a, "new": words_b},
|
||
"text_similarity_mean": mean([p["text_similarity"] for p in pages]),
|
||
"pages_with_text_differences": sum("text" in p["flags"] for p in pages),
|
||
"box_recall_mean": mean([p["box_recall"] for p in pages]),
|
||
"box_precision_mean": mean([p["box_precision"] for p in pages]),
|
||
"type_agreement_mean": mean([p["type_agreement"] for p in pages]),
|
||
"order_agreement_mean": mean([p["order_agreement"] for p in pages]),
|
||
"pages_flagged": sum(bool(p["flags"]) for p in pages),
|
||
"figures": {"baseline": sum(p["figures_a"] for p in pages), "new": sum(p["figures_b"] for p in pages)},
|
||
"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)},
|
||
}
|
||
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']}`", "",
|
||
"| measure | value |", "|---|---|",
|
||
f"| pages | {s['pages']} |",
|
||
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"]), 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")
|
||
args = parser.parse_args(argv)
|
||
report = compare(load_magazine(args.baseline), load_magazine(args.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())
|