275 lines
12 KiB
Python
275 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Scratch harness: reproduce the multi-edit font-collision scramble, and (after the
|
|
fix) verify it's gone. Reads engine output through pybind (NOT HTTP/stdin) to avoid the
|
|
Windows cp1252 mojibake trap. Writes a UTF-8 report to _reflow_repro.out.txt.
|
|
|
|
Run: gateway/.venv/Scripts/python.exe tests/edits/_reflow_repro.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "gateway"))
|
|
import pdfengine # noqa: E402
|
|
|
|
PDF = ROOT / "Mo-Faishal-Qureshi.pdf"
|
|
OUT = Path(__file__).resolve().parent / "_reflow_repro.out.txt"
|
|
report = io.StringIO()
|
|
|
|
|
|
def log(*a):
|
|
report.write(" ".join(str(x) for x in a) + "\n")
|
|
|
|
|
|
def para_text(p):
|
|
return " ".join(r.text for l in p.lines for r in l.runs)
|
|
|
|
|
|
def para_fonts(p):
|
|
return sorted({r.internal_font_id for l in p.lines for r in l.runs if r.text.strip()})
|
|
|
|
|
|
def build_reflow_op(p, page_index, new_runs=None):
|
|
"""Build a reflow_paragraph op for paragraph p. If new_runs is None, re-emit the
|
|
paragraph's own runs unchanged (so a correct engine is a no-op on text)."""
|
|
runs = []
|
|
obj_idx = []
|
|
for l in p.lines:
|
|
for r in l.runs:
|
|
obj_idx.extend(list(r.object_indices))
|
|
if r.text == "":
|
|
continue
|
|
runs.append({
|
|
"text": r.text,
|
|
"internalFontId": r.internal_font_id,
|
|
"fontSize": r.font_size,
|
|
"color": "#000000",
|
|
})
|
|
if new_runs is not None:
|
|
runs = new_runs
|
|
baselines = [l.baseline_y for l in p.lines]
|
|
first_baseline = baselines[0] if baselines else (p.y + p.h)
|
|
if len(baselines) >= 2:
|
|
leading = abs(baselines[0] - baselines[1])
|
|
else:
|
|
leading = p.lines[0].h if p.lines else 14.0
|
|
if leading <= 0:
|
|
leading = 14.0
|
|
return {
|
|
"id": "rf", "type": "reflow_paragraph", "pageIndex": page_index,
|
|
"data": {
|
|
"objectIndices": sorted(set(obj_idx)),
|
|
"runs": runs,
|
|
"columnLeft": p.x,
|
|
"columnRight": p.x + p.w,
|
|
"firstBaselineY": first_baseline,
|
|
"leading": leading,
|
|
"oldLineCount": len(p.lines),
|
|
"align": "left",
|
|
},
|
|
}
|
|
|
|
|
|
def dump_structure():
|
|
doc = pdfengine.PdfDocument.load_from_memory(PDF.read_bytes(), "")
|
|
npages = 0
|
|
while True:
|
|
try:
|
|
doc.get_page(npages)
|
|
npages += 1
|
|
except Exception:
|
|
break
|
|
log(f"PDF: {PDF.name} pages={npages}")
|
|
for pi in range(npages):
|
|
m = doc.get_page(pi).extract_document_model()
|
|
log(f"\n=== PAGE {pi} ({m.width:.0f}x{m.height:.0f}) paragraphs={len(m.paragraphs)} ===")
|
|
for i, p in enumerate(m.paragraphs):
|
|
t = para_text(p)
|
|
log(f" [{pi}.{i}] lines={len(p.lines)} fonts={para_fonts(p)}")
|
|
log(f" text={t[:90]!r}")
|
|
|
|
|
|
def find_para(model, needle):
|
|
for i, p in enumerate(model.paragraphs):
|
|
if needle in para_text(p):
|
|
return i, p
|
|
return -1, None
|
|
|
|
|
|
def repro(page_index=0, a_needle="Java Backend Developer", b_needle="Core Java",
|
|
survive=("Core", "Java")):
|
|
"""Faithful app flow: edit paragraph A (modified), re-extract, then re-emit
|
|
paragraph B UNCHANGED. B's text must survive intact in the committed PDF."""
|
|
# ---- CONTROL: B-only on a fresh doc (proves B alone is fine) ----
|
|
doc = pdfengine.PdfDocument.load_from_memory(PDF.read_bytes(), "")
|
|
m0 = doc.get_page(page_index).extract_document_model()
|
|
bi, pb = find_para(m0, b_needle)
|
|
log(f"\n--- CONTROL: re-emit B [{page_index}.{bi}] unchanged (no prior edit) ---")
|
|
log(f" B text(before) = {para_text(pb)[:80]!r}")
|
|
doc.apply_edits(json.dumps({"version": "1.0", "operations": [build_reflow_op(pb, page_index)]}))
|
|
m_ctrl = pdfengine.PdfDocument.load_from_memory(doc.save_full(), "").get_page(page_index).extract_document_model()
|
|
_, pb_ctrl = find_para(m_ctrl, survive[0])
|
|
ctrl_text = para_text(pb_ctrl) if pb_ctrl else "<B PARA NOT FOUND>"
|
|
log(f" B text(after) = {ctrl_text[:80]!r}")
|
|
ctrl_ok = all(s in ctrl_text for s in survive)
|
|
log(f" CONTROL survive={survive} -> {'OK' if ctrl_ok else 'LOST'}")
|
|
|
|
# ---- TEST: edit A first, re-extract, then re-emit B unchanged ----
|
|
doc = pdfengine.PdfDocument.load_from_memory(PDF.read_bytes(), "")
|
|
m0 = doc.get_page(page_index).extract_document_model()
|
|
ai, pa = find_para(m0, a_needle)
|
|
log(f"\n--- TEST: edit A [{page_index}.{ai}] THEN re-emit B unchanged (same doc) ---")
|
|
# Modify A: prepend a word to its first run.
|
|
a_runs = []
|
|
first = True
|
|
for l in pa.lines:
|
|
for r in l.runs:
|
|
if r.text == "":
|
|
continue
|
|
txt = ("EDITED " + r.text) if first else r.text
|
|
first = False
|
|
a_runs.append({"text": txt, "internalFontId": r.internal_font_id,
|
|
"fontSize": r.font_size, "color": "#000000"})
|
|
doc.apply_edits(json.dumps({"version": "1.0",
|
|
"operations": [build_reflow_op(pa, page_index, new_runs=a_runs)]}))
|
|
# Re-extract (frontend re-fetches the model after each edit) and re-locate B.
|
|
m1 = doc.get_page(page_index).extract_document_model()
|
|
bi2, pb2 = find_para(m1, b_needle)
|
|
if pb2 is None:
|
|
log(f" !! B not found after edit A (searched {b_needle!r})")
|
|
else:
|
|
log(f" B text(before 2nd edit) = {para_text(pb2)[:80]!r}")
|
|
doc.apply_edits(json.dumps({"version": "1.0", "operations": [build_reflow_op(pb2, page_index)]}))
|
|
# In-memory (pre-save) B text: localizes corruption to emission vs save-merge.
|
|
m_inmem = doc.get_page(page_index).extract_document_model()
|
|
_, pb_inmem = find_para(m_inmem, survive[0])
|
|
log(f" B text(in-memory, pre-save) = {(para_text(pb_inmem) if pb_inmem else '<NOT FOUND>')[:80]!r}")
|
|
m2 = pdfengine.PdfDocument.load_from_memory(doc.save_full(), "").get_page(page_index).extract_document_model()
|
|
_, pb_test = find_para(m2, survive[0])
|
|
test_text = para_text(pb_test) if pb_test else "<B PARA NOT FOUND>"
|
|
log(f" B text(after) = {test_text[:80]!r}")
|
|
test_ok = all(s in test_text for s in survive)
|
|
log(f" TEST survive={survive} -> {'OK' if test_ok else 'SCRAMBLED/LOST'}")
|
|
if not test_ok:
|
|
# Show the corrupted bullets paragraph (locate by leftover 'Languages'/'ang' or dump near B).
|
|
for i, p in enumerate(m2.paragraphs):
|
|
t = para_text(p)
|
|
if "ang" in t or "OOP" in t or "Collec" in t or "Stream" in t or (b_needle[:3] in t):
|
|
log(f" >> scrambled B candidate [{page_index}.{i}] = {t[:90]!r}")
|
|
log(f"\n==> CONTROL={'OK' if ctrl_ok else 'FAIL'} TEST={'OK' if test_ok else 'FAIL'} "
|
|
f"(bug reproduced if CONTROL=OK and TEST=FAIL)")
|
|
|
|
|
|
def stress():
|
|
"""Sequentially reflow EVERY paragraph on both pages (re-extracting between each, like the
|
|
real app), then assert distinctive substrings from each survive — catches any font scramble
|
|
across non-subset AND subset (BCDKEE+Calibri...) fonts under heavy multi-edit."""
|
|
# Distinctive substrings to verify survive on each page after all edits.
|
|
checks = {
|
|
0: ["Professional", "Backend", "Core", "Java", "Frameworks", "Microservices",
|
|
"TapQwik", "Present", "LG", "Commerce"],
|
|
1: ["Architected", "Wego", "Booking", "Education", "Bachelor", "Certification",
|
|
"Kafka", "Problem", "Ownership"],
|
|
}
|
|
# Edit (reflow-in-place) the paragraph CONTAINING each anchor, sequentially, re-extracting
|
|
# between edits — exactly how a user edits N paragraphs in one session.
|
|
anchors = {
|
|
0: ["Backend", "Core Java", "Mar 2023", "LG"],
|
|
1: ["Architected", "Wego", "Java Development", "Problem"],
|
|
}
|
|
failures = []
|
|
for pi in (0, 1):
|
|
doc = pdfengine.PdfDocument.load_from_memory(PDF.read_bytes(), "")
|
|
log(f"\n=== STRESS page {pi}: sequentially reflowing {len(anchors[pi])} targeted paragraphs ===")
|
|
for anchor in anchors[pi]:
|
|
m = doc.get_page(pi).extract_document_model()
|
|
_, p = find_para(m, anchor)
|
|
if p is None:
|
|
log(f" anchor {anchor!r}: paragraph not found (possibly merged by a prior edit)")
|
|
continue
|
|
try:
|
|
doc.apply_edits(json.dumps({"version": "1.0", "operations": [build_reflow_op(p, pi)]}))
|
|
except Exception as exc:
|
|
log(f" anchor {anchor!r}: apply failed: {exc}")
|
|
final = para_textall(doc.get_page(pi).extract_document_model())
|
|
for needle in checks[pi]:
|
|
if needle not in final:
|
|
failures.append((pi, needle))
|
|
present = [n for n in checks[pi] if n in final]
|
|
log(f" page {pi}: {len(present)}/{len(checks[pi])} substrings survived: "
|
|
f"missing={[n for n in checks[pi] if n not in final]}")
|
|
log(f"\n==> STRESS {'PASS' if not failures else 'FAIL ' + str(failures)}")
|
|
|
|
|
|
def para_textall(model):
|
|
return " ".join(r.text for p in model.paragraphs for l in p.lines for r in l.runs)
|
|
|
|
|
|
def render_after_edits():
|
|
"""Apply A (Summary) then B (bullets) on one doc, save, reload, render page 0 to PNG —
|
|
GROUND TRUTH: is the committed PDF visually corrupt, or only text extraction?"""
|
|
from PIL import Image # noqa: PLC0415
|
|
doc = pdfengine.PdfDocument.load_from_memory(PDF.read_bytes(), "")
|
|
m0 = doc.get_page(0).extract_document_model()
|
|
_, pa = find_para(m0, "Java Backend Developer")
|
|
a_runs = []
|
|
first = True
|
|
for l in pa.lines:
|
|
for r in l.runs:
|
|
if r.text == "":
|
|
continue
|
|
a_runs.append({"text": ("EDITED " + r.text) if first else r.text,
|
|
"internalFontId": r.internal_font_id, "fontSize": r.font_size, "color": "#000000"})
|
|
first = False
|
|
doc.apply_edits(json.dumps({"version": "1.0", "operations": [build_reflow_op(pa, 0, new_runs=a_runs)]}))
|
|
m1 = doc.get_page(0).extract_document_model()
|
|
_, pb = find_para(m1, "Core Java")
|
|
doc.apply_edits(json.dumps({"version": "1.0", "operations": [build_reflow_op(pb, 0)]}))
|
|
out = doc.save_full()
|
|
img = pdfengine.PdfDocument.load_from_memory(out, "").get_page(0).render(150)
|
|
data = img.data
|
|
png = Path(__file__).resolve().parent / "_reflow_render.png"
|
|
n = img.width * img.height
|
|
if len(data) >= 8 and data[:8] == b"\x89PNG\r\n\x1a\n":
|
|
png.write_bytes(data) # already PNG-encoded
|
|
else:
|
|
mode = "RGBA" if len(data) == n * 4 else ("RGB" if len(data) == n * 3 else None)
|
|
from PIL import Image # noqa: PLC0415
|
|
Image.frombytes(mode, (img.width, img.height), data).convert("RGB").save(png)
|
|
log(f"rendered committed page0 -> {png} ({img.width}x{img.height}, {len(data)} bytes)")
|
|
|
|
|
|
def probe_fonts():
|
|
doc = pdfengine.PdfDocument.load_from_memory(PDF.read_bytes(), "")
|
|
for pi in (0, 1):
|
|
fonts = doc.get_page(pi).get_fonts()
|
|
log(f"\n=== PAGE {pi} fonts ({len(fonts)}) ===")
|
|
for f in fonts:
|
|
log(f" font_name={f.font_name!r:34} internal_id={f.internal_font_id!r:28} "
|
|
f"type={f.type!r:12} flags={f.flags:<4} subset={f.is_subset} tag={f.subset_tag!r} "
|
|
f"embedded={f.is_embedded}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mode = sys.argv[1] if len(sys.argv) > 1 else "dump"
|
|
if mode == "dump":
|
|
dump_structure()
|
|
elif mode == "repro":
|
|
repro()
|
|
elif mode == "repro_diff":
|
|
# A = Calibri body (Summary); B = a Calibri-Bold-only heading ("Projects").
|
|
# Different base fonts -> if B survives, the scramble is same-/BaseFont aliasing.
|
|
repro(a_needle="Java Backend Developer", b_needle="Projects", survive=("Projects",))
|
|
elif mode == "probe":
|
|
probe_fonts()
|
|
elif mode == "render":
|
|
render_after_edits()
|
|
elif mode == "stress":
|
|
stress()
|
|
OUT.write_text(report.getvalue(), encoding="utf-8")
|
|
print(f"wrote {OUT}")
|