v1.1.0: headerless mode for audits printed without row/column headings
Sniffs the format on page 1: printed row numbers -> gridded mode (unchanged); UUID-keyed rows -> new headerless mode, which anchors rows on per-record UUIDs and takes the column map from the page-1 header labels. Handles fused cells (ID+Name+Org+networks), mid-value line wraps, two-timestamp Time Frame ranges, wrapped header labels, and per-export schema drift. Validated against three monthly JC PD audits: 120/120 cross-engine spot checks, 0 anomalies, field-identical to the standalone converter it replaces. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012bQ28wRBiQYcSR9FzV9U7c
This commit is contained in:
daveandClaude Fable 5 committed
1 parent
74b06be742
commit
c9c90920cc
6 files changed
+394
-65
No files matched your search
@@ -0,0 +1,215 @@
|
||||
"""The headerless variant: audit exports printed WITHOUT row/column headings.
|
||||
|
||||
Some Flock audit PDFs come out of "Microsoft Print to PDF" with Excel's
|
||||
headings switched off. Everything the gridded mode anchors on is missing,
|
||||
and this format brings its own traps on top:
|
||||
|
||||
- No gridlines and no printed row numbers, so rows are anchored on the
|
||||
one structural fact left: every row starts with a UUID in the ID
|
||||
column. Column edges come from the page-1 header labels' x-positions.
|
||||
- Adjacent cells FUSE into one text run when columns are tight: ID +
|
||||
Name + Org Name (+ sometimes the networks count) arrive as a single
|
||||
string and must be split apart again. Search Time occasionally fuses
|
||||
with Search Type; the fixed 27-character timestamp gives the split
|
||||
point.
|
||||
- Long header labels wrap onto stacked runs ('Total Networks ' over
|
||||
'Searched'), so header runs sharing an x are merged before matching
|
||||
names. Miss this and a whole column silently lands in the wrong
|
||||
window -- it happened; the strays counter now guards it.
|
||||
- Cells wrap mid-value ('05/05/2026, 04:12:32 ' + 'AM UTC'), and the
|
||||
wrap point moves between exports. Fragments are rejoined by position.
|
||||
- Time Frame is TWO stacked timestamps -- the full start/end range --
|
||||
joined here as 'start - end'.
|
||||
- The schema itself drifts between exports (License Plate present one
|
||||
month, gone the next), so columns are taken from each file's own
|
||||
header, and missing ones are reported, not guessed.
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
import collections
|
||||
import statistics
|
||||
from bisect import bisect_left
|
||||
|
||||
from .page import all_runs, run_text, solid_regions
|
||||
|
||||
CANON = ["ID", "Name", "Org Name", "Total Networks Searched", "Time Frame",
|
||||
"License Plate", "Reason", "Case #", "Filters", "Search Time",
|
||||
"Search Type", "Text Prompt", "Moderation"]
|
||||
UUID = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}')
|
||||
TS = re.compile(r'^\d{2}/\d{2}/\d{4}, \d{2}:\d{2}:\d{2} [AP]M UTC$')
|
||||
TS_PREFIX = re.compile(r'^(\d{2}/\d{2}/\d{4}, \d{2}:\d{2}:\d{2} [AP]M UTC)(.+)$')
|
||||
|
||||
|
||||
def looks_headerless(page0):
|
||||
"""Does page 1 carry UUID-keyed rows? That's this variant's signature."""
|
||||
return any(UUID.match(run_text(r)) for r in all_runs(page0))
|
||||
|
||||
|
||||
def parse_header(page0_runs, header_cut):
|
||||
"""Column names and x-positions from the page-1 header band.
|
||||
|
||||
A long label wraps onto stacked runs at the same x, so runs sharing an
|
||||
x are merged (in y-order) before searching for the canonical names.
|
||||
Char positions are kept so names inside a fused label still yield
|
||||
their own x.
|
||||
"""
|
||||
hdr = [r for r in page0_runs if r["chars"][0][2][1] < header_cut]
|
||||
by_x = collections.defaultdict(list)
|
||||
for r in hdr:
|
||||
x0 = r["chars"][0][2][0]
|
||||
key = next((k for k in by_x if abs(k - x0) < 1.2), x0)
|
||||
by_x[key].append(r)
|
||||
|
||||
col_x = {}
|
||||
for x0, rs in by_x.items():
|
||||
rs.sort(key=lambda r: r["chars"][0][2][1])
|
||||
text = " ".join(run_text(r).strip() for r in rs)
|
||||
cmap, pos = {}, 0
|
||||
for r in rs:
|
||||
raw = run_text(r)
|
||||
lead = len(raw) - len(raw.lstrip())
|
||||
for i in range(len(raw.strip())):
|
||||
cmap[pos + i] = r["chars"][lead + i]
|
||||
pos += len(raw.strip()) + 1
|
||||
for name in CANON:
|
||||
i = text.find(name)
|
||||
if i >= 0 and name not in col_x and i in cmap:
|
||||
col_x[name] = cmap[i][2][0]
|
||||
|
||||
ordered = sorted((x, n) for n, x in col_x.items())
|
||||
names = [n for _, n in ordered]
|
||||
edges = [x for x, _ in ordered] + [1e9]
|
||||
missing = [n for n in CANON if n not in col_x]
|
||||
return names, edges, missing
|
||||
|
||||
|
||||
def convert_headerless(doc, out, n_pages):
|
||||
"""Rebuild every row and write it to `out` (a csv.writer). Returns a
|
||||
stats dict for the checks at the end."""
|
||||
p1 = all_runs(doc[0])
|
||||
anchors1 = sorted(r["chars"][0][2][1] for r in p1 if UUID.match(run_text(r)))
|
||||
header_cut = anchors1[0] - 6.0
|
||||
names, edges, missing = parse_header(p1, header_cut)
|
||||
|
||||
def col_at(x):
|
||||
i = bisect_left(edges, x + 2.5) - 1
|
||||
return names[i] if 0 <= i < len(names) else None
|
||||
|
||||
stats = collections.Counter()
|
||||
stats["columns_missing"] = 0 # ensure keys exist even when zero
|
||||
rows, gaps, bars = [], [], []
|
||||
for pno in range(n_pages):
|
||||
runs = all_runs(doc[pno])
|
||||
anchors = sorted((r["chars"][0][2][1], r) for r in runs
|
||||
if UUID.match(run_text(r)))
|
||||
if not anchors:
|
||||
stats["pages_without_rows"] += 1
|
||||
continue
|
||||
ys = [a[0] for a in anchors]
|
||||
gaps += [b - a for a, b in zip(ys, ys[1:])]
|
||||
page_rows = [collections.defaultdict(list) for _ in anchors]
|
||||
|
||||
# place every text run in its row (nearest anchor at or below it)
|
||||
# and its column (by where the run STARTS)
|
||||
for r in runs:
|
||||
x, y = r["chars"][0][2]
|
||||
if pno == 0 and y < header_cut:
|
||||
continue
|
||||
i = bisect_left(ys, y - 1.0)
|
||||
if i == len(ys):
|
||||
i -= 1
|
||||
if y > ys[-1] + 2.0:
|
||||
stats["runs_below_last_anchor"] += 1
|
||||
c = col_at(x)
|
||||
if c is None:
|
||||
stats["runs_no_column"] += 1
|
||||
continue
|
||||
page_rows[i][c].append((y, x, run_text(r)))
|
||||
|
||||
# blackout bars, same shape test as the gridded mode
|
||||
pitch = statistics.median(gaps) if gaps else 12.0
|
||||
for box in solid_regions(doc[pno]):
|
||||
if box.width <= 1.2 or box.height <= pitch * 0.5:
|
||||
continue
|
||||
yc = (box.y0 + box.y1) / 2
|
||||
i = min(bisect_left(ys, yc - 1.0), len(ys) - 1)
|
||||
bars.append((pno, i, round(box.width, 1)))
|
||||
page_rows[i]["_bar"] = max(page_rows[i].get("_bar") or 0,
|
||||
round(box.width, 1))
|
||||
rows.extend(page_rows)
|
||||
|
||||
# --- assemble cells from fragments ---
|
||||
def join(parts):
|
||||
parts = [p[2].strip() for p in sorted(parts)]
|
||||
parts = [p for p in parts if p]
|
||||
if len(parts) == 2 and all(TS.match(p) for p in parts):
|
||||
return " - ".join(parts) # Time Frame start/end range
|
||||
return re.sub(r'\s+', ' ', " ".join(parts)).strip()
|
||||
|
||||
out_rows, middles = [], []
|
||||
for row in rows:
|
||||
bar = row.pop("_bar", None)
|
||||
# Name/Org come only from splitting the fused ID run; anything
|
||||
# assigned to them directly means a column landed in the wrong
|
||||
# window and would be silently clobbered below.
|
||||
if row.get("Name") or row.get("Org Name"):
|
||||
stats["strays_in_name_org_windows"] += 1
|
||||
cell = {c: join(v) for c, v in row.items()}
|
||||
|
||||
fused = cell.get("ID", "")
|
||||
m = UUID.match(fused)
|
||||
if not m:
|
||||
stats["rows_bad_uuid"] += 1
|
||||
continue
|
||||
rest = fused[36:]
|
||||
digits = re.search(r'(\d+)$', rest)
|
||||
if digits and not cell.get("Total Networks Searched"):
|
||||
cell["Total Networks Searched"] = digits.group(1)
|
||||
rest = rest[:digits.start()]
|
||||
cell["ID"] = m.group(0)
|
||||
cell["_rest"] = rest
|
||||
middles.append(rest)
|
||||
|
||||
st = cell.get("Search Time", "")
|
||||
pm = TS_PREFIX.match(st)
|
||||
if pm:
|
||||
cell["Search Time"] = pm.group(1)
|
||||
cell["Search Type"] = (pm.group(2) + " "
|
||||
+ cell.get("Search Type", "")).strip()
|
||||
stats["searchtime_splits"] += 1
|
||||
cell["_bar"] = bar
|
||||
out_rows.append(cell)
|
||||
|
||||
# Name / Org split: the org is the longest common suffix of every
|
||||
# fused middle -- constant per file, while names vary
|
||||
org = middles[0] if middles else ""
|
||||
for s in middles[1:]:
|
||||
while org and not s.endswith(org):
|
||||
org = org[1:]
|
||||
if len(org) < 4:
|
||||
sys.exit(f"Could not derive a constant Org Name (got {org!r}); "
|
||||
"refusing to split names by guesswork.")
|
||||
for cell in out_rows:
|
||||
cell["Name"] = cell.pop("_rest")[:-len(org)].strip()
|
||||
cell["Org Name"] = org
|
||||
|
||||
out.writerow(CANON + ["reason_redacted", "redaction_bar_pt"])
|
||||
for cell in out_rows:
|
||||
bar = cell.pop("_bar")
|
||||
out.writerow([cell.get(c, "") for c in CANON]
|
||||
+ ["TRUE" if bar else "FALSE", bar or ""])
|
||||
|
||||
tf_ok = tf_single = tf_bad = 0
|
||||
for cell in out_rows:
|
||||
tf = cell.get("Time Frame", "")
|
||||
if " - " in tf:
|
||||
a, b = tf.split(" - ", 1)
|
||||
tf_ok += bool(TS.match(a) and TS.match(b))
|
||||
tf_bad += not (TS.match(a) and TS.match(b))
|
||||
elif tf:
|
||||
tf_single += 1
|
||||
stats["columns_missing"] = len(missing)
|
||||
return {"n_rows": len(out_rows), "org": org, "names": names,
|
||||
"missing": missing, "tf_ok": tf_ok, "tf_single": tf_single,
|
||||
"tf_bad": tf_bad, "bars": bars, "stats": stats}
|
||||
Reference in new issue
Block a user