309 lines
13 KiB
Python
309 lines
13 KiB
Python
"""Forensic: dump & compare PDF text objects — original vs edit-entry vs one-char.
|
|
|
|
Does NOT compare images. Compares:
|
|
font base name, font size, matrix, bbox, unicode text, object count,
|
|
and raw content-stream snippets around Tj/TJ.
|
|
|
|
Also answers Step 4 from static path + runtime FontInfo:
|
|
why measureFace is nullptr for this paragraph's font.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "gateway"))
|
|
import pdfengine # type: ignore
|
|
|
|
|
|
def dump_page_text_objects(doc, page_index: int = 0) -> list[dict]:
|
|
"""Re-extract model and also pull flat glyph list as proxy for text objects."""
|
|
page = doc.get_page(page_index)
|
|
model = page.extract_document_model()
|
|
fonts = doc.get_fonts(page_index, page_index)
|
|
font_by_id = {f.internal_font_id: f for f in fonts}
|
|
|
|
objs = []
|
|
for pi, para in enumerate(model.paragraphs):
|
|
for li, line in enumerate(para.lines):
|
|
for ri, run in enumerate(line.runs):
|
|
fi = font_by_id.get(run.internal_font_id)
|
|
objs.append({
|
|
"para": pi, "line": li, "run": ri,
|
|
"text": run.text,
|
|
"fontName": run.font_name,
|
|
"fontId": run.internal_font_id,
|
|
"fontSize": run.font_size,
|
|
"isEmbedded": run.is_embedded,
|
|
"fontType": run.type,
|
|
"fontFidelity": getattr(run, "font_fidelity", None),
|
|
"x": run.x, "y": run.y, "w": run.w, "h": run.h,
|
|
"objectIndices": list(run.object_indices or []),
|
|
"glyphCount": len(run.glyphs),
|
|
"glyphOrigins": [(round(g.origin_x, 3), round(g.origin_y, 3)) for g in run.glyphs[:40]],
|
|
"glyphUnicodes": [g.unicode for g in run.glyphs[:40]],
|
|
"fontMeta": None if not fi else {
|
|
"ascent": fi.ascent, "descent": fi.descent,
|
|
"capHeight": getattr(fi, "cap_height", None),
|
|
"isEmbedded": fi.is_embedded,
|
|
"isSubset": getattr(fi, "is_subset", None),
|
|
"type": fi.type,
|
|
"fontName": fi.font_name,
|
|
},
|
|
})
|
|
return objs
|
|
|
|
|
|
def extract_content_stream_text_ops(pdf_bytes: bytes) -> dict:
|
|
"""Best-effort parse of content stream for font/text operators."""
|
|
# Find stream ... endstream blocks
|
|
streams = re.findall(rb"stream\r?\n(.*?)\r?\nendstream", pdf_bytes, re.DOTALL)
|
|
ops = []
|
|
for s in streams:
|
|
# Decode latin-1 so binary stays 1:1
|
|
try:
|
|
text = s.decode("latin-1")
|
|
except Exception:
|
|
continue
|
|
# Font select: /F1 16 Tf or /Helv 11 Tf
|
|
for m in re.finditer(r"/([A-Za-z0-9\+\,\-]+)\s+([\d.]+)\s+Tf", text):
|
|
ops.append({"op": "Tf", "fontRes": m.group(1), "size": float(m.group(2))})
|
|
# Text matrix: a b c d e f Tm
|
|
for m in re.finditer(
|
|
r"([\d.\-]+)\s+([\d.\-]+)\s+([\d.\-]+)\s+([\d.\-]+)\s+([\d.\-]+)\s+([\d.\-]+)\s+Tm",
|
|
text,
|
|
):
|
|
ops.append({"op": "Tm", "matrix": [float(x) for x in m.groups()]})
|
|
# Td
|
|
for m in re.finditer(r"([\d.\-]+)\s+([\d.\-]+)\s+Td", text):
|
|
ops.append({"op": "Td", "tx": float(m.group(1)), "ty": float(m.group(2))})
|
|
# Literal strings before Tj
|
|
for m in re.finditer(r"\((?:\\.|[^\\)])*\)\s*Tj", text):
|
|
raw = m.group(0)
|
|
ops.append({"op": "Tj", "raw": raw[:120]})
|
|
for m in re.finditer(r"\[.*?\]\s*TJ", text, re.DOTALL):
|
|
ops.append({"op": "TJ", "raw": m.group(0)[:200]})
|
|
# Font dictionaries
|
|
fonts = []
|
|
for m in re.finditer(
|
|
rb"/Type\s*/Font\s*/Subtype\s*/(\w+)\s*/BaseFont\s*/([^\s/]+)",
|
|
pdf_bytes,
|
|
):
|
|
fonts.append({"subtype": m.group(1).decode(), "baseFont": m.group(2).decode()})
|
|
# Also looser BaseFont
|
|
if not fonts:
|
|
for m in re.finditer(rb"/BaseFont\s*/([^\s/]+)", pdf_bytes):
|
|
fonts.append({"subtype": "?", "baseFont": m.group(1).decode()})
|
|
return {"textOps": ops, "fontDicts": fonts, "streamCount": len(streams)}
|
|
|
|
|
|
def load_layout(path: Path):
|
|
sys.path.insert(0, str(path.parent))
|
|
from forensic_extract import ( # type: ignore
|
|
compute_layout, extract_flat_runs, build_reflow_data,
|
|
)
|
|
return compute_layout, extract_flat_runs, build_reflow_data
|
|
|
|
|
|
def build_payloads(pdf_path: Path):
|
|
compute_layout, extract_flat_runs, build_reflow_data = load_layout(
|
|
Path(__file__).resolve().parent
|
|
)
|
|
doc = pdfengine.PdfDocument.load_from_file(str(pdf_path), "")
|
|
model = doc.get_page(0).extract_document_model()
|
|
para = model.paragraphs[0]
|
|
layout = compute_layout(para)
|
|
dominant_fid = next(
|
|
(r["fid"] for r in layout["seedRuns"] if r["text"].strip() and r["fid"]), ""
|
|
)
|
|
dom = next((r for r in layout["seedRuns"] if r["text"].strip()), layout["seedRuns"][0])
|
|
flat = extract_flat_runs(layout["seedRuns"], dominant_fid, dom["size"], dom["color"])
|
|
para_id = "forensic-obj-cmp"
|
|
|
|
entry_data = build_reflow_data(layout, flat, layout["origLines"], para_id)
|
|
# one char: append x, drop origLines, keep stale advances (mirrors real editor)
|
|
flat_one = [{**flat[0], "text": flat[0]["text"] + "x"}] + flat[1:]
|
|
one_data = build_reflow_data(layout, flat_one, None, para_id)
|
|
|
|
return {
|
|
"entry": {"version": "1.0", "operations": [{
|
|
"id": "e", "type": "reflow_paragraph", "pageIndex": 0, "data": entry_data,
|
|
}]},
|
|
"one": {"version": "1.0", "operations": [{
|
|
"id": "o", "type": "reflow_paragraph", "pageIndex": 0, "data": one_data,
|
|
}]},
|
|
"layout": layout,
|
|
"fonts": [
|
|
{
|
|
"id": f.internal_font_id, "name": f.font_name,
|
|
"embedded": f.is_embedded, "type": f.type,
|
|
"subset": getattr(f, "is_subset", None),
|
|
"ascent": f.ascent, "descent": f.descent,
|
|
}
|
|
for f in doc.get_fonts(0, 0)
|
|
],
|
|
}
|
|
|
|
|
|
def apply_and_save(pdf_path: Path, op: dict, out_path: Path) -> bytes:
|
|
doc = pdfengine.PdfDocument.load_from_file(str(pdf_path), "")
|
|
doc.apply_edits(json.dumps(op))
|
|
data = doc.save_full()
|
|
out_path.write_bytes(data)
|
|
return data
|
|
|
|
|
|
def compare_objs(label_a: str, a: list[dict], label_b: str, b: list[dict]):
|
|
print(f"\n{'='*60}\nCOMPARE {label_a} vs {label_b}\n{'='*60}")
|
|
print(f" object/run count: {len(a)} -> {len(b)}")
|
|
keys = ["text", "fontName", "fontId", "fontSize", "isEmbedded", "fontType",
|
|
"w", "glyphCount"]
|
|
n = max(len(a), len(b))
|
|
for i in range(min(n, 12)):
|
|
oa = a[i] if i < len(a) else None
|
|
ob = b[i] if i < len(b) else None
|
|
if oa is None:
|
|
print(f" [{i}] ADDED: {ob['text']!r} font={ob['fontName']} size={ob['fontSize']}")
|
|
continue
|
|
if ob is None:
|
|
print(f" [{i}] REMOVED: {oa['text']!r}")
|
|
continue
|
|
diffs = []
|
|
for k in keys:
|
|
if oa.get(k) != ob.get(k):
|
|
diffs.append(f"{k}: {oa.get(k)!r} -> {ob.get(k)!r}")
|
|
# origin of first glyph
|
|
goa = oa["glyphOrigins"][0] if oa["glyphOrigins"] else None
|
|
gob = ob["glyphOrigins"][0] if ob["glyphOrigins"] else None
|
|
if goa != gob:
|
|
diffs.append(f"firstOrigin: {goa} -> {gob}")
|
|
status = "CHANGED" if diffs else "same"
|
|
print(f" [{i}] {oa['text']!r:30} {status}")
|
|
for d in diffs:
|
|
print(f" {d}")
|
|
|
|
|
|
def explain_measureface(fonts: list[dict]):
|
|
print(f"\n{'='*60}\nSTEP 4 — why measureFace is nullptr\n{'='*60}")
|
|
for f in fonts:
|
|
print(f" FontInfo: id={f['id']} name={f['name']} embedded={f['embedded']} "
|
|
f"type={f['type']} subset={f['subset']}")
|
|
if not f["embedded"]:
|
|
print(" PATH TAKEN in loadEmissionFont():")
|
|
print(" matchedFontInfo && !isEmbedded")
|
|
print(" -> cacheKey = \"standard_\" + fontName")
|
|
print(" -> useEmbedded = false")
|
|
print(" -> useSystem = false")
|
|
print(" skip embedded/system load branches")
|
|
print(" fall through to FPDFText_LoadStandardFont()")
|
|
print(" measureFace ONLY set from loadedFontDataBuffers_ (TTF bytes)")
|
|
print(" standard fonts never put bytes in that buffer")
|
|
print(" => measureFace remains nullptr BY DESIGN for non-embedded fonts")
|
|
else:
|
|
print(" Embedded path: measureFace set only if getFontData*/LoadFromMemory succeeds")
|
|
print(" or via useSystem branch FONT_METRICS_FIX preserving orig face")
|
|
|
|
|
|
def main():
|
|
pdf_path = Path(sys.argv[1]) if len(sys.argv) > 1 else ROOT / "corpus" / "basic" / "hello_world.pdf"
|
|
out_dir = Path(__file__).resolve().parent
|
|
|
|
print(f"PDF: {pdf_path}")
|
|
payloads = build_payloads(pdf_path)
|
|
explain_measureface(payloads["fonts"])
|
|
|
|
# Original
|
|
orig_doc = pdfengine.PdfDocument.load_from_file(str(pdf_path), "")
|
|
orig_objs = dump_page_text_objects(orig_doc)
|
|
orig_bytes = pdf_path.read_bytes()
|
|
orig_stream = extract_content_stream_text_ops(orig_bytes)
|
|
|
|
print(f"\n{'='*60}\nORIGINAL PDF TEXT OBJECTS\n{'='*60}")
|
|
for o in orig_objs:
|
|
print(f" text={o['text']!r} font={o['fontName']} id={o['fontId']} "
|
|
f"size={o['fontSize']:.2f} emb={o['isEmbedded']} type={o['fontType']} "
|
|
f"w={o['w']:.3f} glyphs={o['glyphCount']} objs={o['objectIndices']}")
|
|
if o["fontMeta"]:
|
|
print(f" meta: ascent={o['fontMeta']['ascent']} descent={o['fontMeta']['descent']} "
|
|
f"subset={o['fontMeta']['isSubset']}")
|
|
print(f" content fontDicts: {orig_stream['fontDicts']}")
|
|
print(f" content Tf/Tj ops ({len(orig_stream['textOps'])}):")
|
|
for op in orig_stream["textOps"][:20]:
|
|
print(f" {op}")
|
|
|
|
# Entry preview (unchanged text)
|
|
print(f"\n--- applying EDIT ENTRY reflow (unchanged text) ---")
|
|
entry_path = out_dir / "_forensic_entry.pdf"
|
|
entry_bytes = apply_and_save(pdf_path, payloads["entry"], entry_path)
|
|
entry_doc = pdfengine.PdfDocument.load_from_memory(entry_bytes, "")
|
|
entry_objs = dump_page_text_objects(entry_doc)
|
|
entry_stream = extract_content_stream_text_ops(entry_bytes)
|
|
|
|
print(f"\n{'='*60}\nEDIT-ENTRY PDF TEXT OBJECTS\n{'='*60}")
|
|
for o in entry_objs[:20]:
|
|
print(f" text={o['text']!r} font={o['fontName']} id={o['fontId']} "
|
|
f"size={o['fontSize']:.2f} emb={o['isEmbedded']} type={o['fontType']} "
|
|
f"w={o['w']:.3f} glyphs={o['glyphCount']} objs={o['objectIndices']}")
|
|
print(f" content fontDicts: {entry_stream['fontDicts']}")
|
|
print(f" content Tf/Tj ops ({len(entry_stream['textOps'])}):")
|
|
for op in entry_stream["textOps"][:30]:
|
|
print(f" {op}")
|
|
|
|
compare_objs("ORIGINAL", orig_objs, "EDIT-ENTRY", entry_objs)
|
|
|
|
# One char
|
|
print(f"\n--- applying ONE-CHAR reflow ---")
|
|
one_path = out_dir / "_forensic_one.pdf"
|
|
one_bytes = apply_and_save(pdf_path, payloads["one"], one_path)
|
|
one_doc = pdfengine.PdfDocument.load_from_memory(one_bytes, "")
|
|
one_objs = dump_page_text_objects(one_doc)
|
|
one_stream = extract_content_stream_text_ops(one_bytes)
|
|
|
|
print(f"\n{'='*60}\nONE-CHAR PDF TEXT OBJECTS\n{'='*60}")
|
|
for o in one_objs[:20]:
|
|
print(f" text={o['text']!r} font={o['fontName']} id={o['fontId']} "
|
|
f"size={o['fontSize']:.2f} emb={o['isEmbedded']} type={o['fontType']} "
|
|
f"w={o['w']:.3f} glyphs={o['glyphCount']} objs={o['objectIndices']}")
|
|
print(f" content fontDicts: {one_stream['fontDicts']}")
|
|
print(f" content Tf/Tj ops ({len(one_stream['textOps'])}):")
|
|
for op in one_stream["textOps"][:30]:
|
|
print(f" {op}")
|
|
|
|
compare_objs("EDIT-ENTRY", entry_objs, "ONE-CHAR", one_objs)
|
|
compare_objs("ORIGINAL", orig_objs, "ONE-CHAR", one_objs)
|
|
|
|
# Summarize first differences
|
|
print(f"\n{'='*60}\nFIRST DIFFERENCES SUMMARY\n{'='*60}")
|
|
print(f" Original run count: {len(orig_objs)}")
|
|
print(f" Edit-entry run count: {len(entry_objs)}")
|
|
print(f" One-char run count: {len(one_objs)}")
|
|
print(f" Original Tj/TJ ops: {sum(1 for o in orig_stream['textOps'] if o['op'] in ('Tj','TJ'))}")
|
|
print(f" Entry Tj/TJ ops: {sum(1 for o in entry_stream['textOps'] if o['op'] in ('Tj','TJ'))}")
|
|
print(f" One-char Tj/TJ ops: {sum(1 for o in one_stream['textOps'] if o['op'] in ('Tj','TJ'))}")
|
|
print(f" Original fontDicts: {orig_stream['fontDicts']}")
|
|
print(f" Entry fontDicts: {entry_stream['fontDicts']}")
|
|
print(f" One-char fontDicts: {one_stream['fontDicts']}")
|
|
|
|
report = {
|
|
"pdf": str(pdf_path),
|
|
"fonts": payloads["fonts"],
|
|
"orig": orig_objs,
|
|
"entry": entry_objs,
|
|
"one": one_objs,
|
|
"streams": {
|
|
"orig": orig_stream,
|
|
"entry": entry_stream,
|
|
"one": one_stream,
|
|
},
|
|
}
|
|
out_json = out_dir / "forensic_obj_compare.json"
|
|
out_json.write_text(json.dumps(report, indent=2), encoding="utf-8")
|
|
print(f"\nWrote {out_json}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|