Add OCR naming and PNG-to-PDF export (modules 5-7, single-page only)

- ocr/: Tesseract-based vendor/date/total extraction. PSM 6 (single
  uniform text block) instead of the default fixes receipts where
  item-name/price columns were otherwise split into separate blocks and
  dropped. Vendor line picked by median (not max) word height within the
  top fraction of the crop, breaking ties by topmost line - max() was
  fooled by single descender/ascender glyphs (commas, parens) inflating
  one word's bounding box.
- naming/: merge OCR metadata into YYYY-MM-DD_vendor.pdf filenames,
  degrading gracefully to UNSORTED_<timestamp>.pdf; per-run dedup.
- pdf/: img2pdf-based multi-page-capable assembly (images_to_pdf).
- pipeline.run_export(): OCR + name + PDF each detected crop from a
  detect() report, writes pdf/ and export_summary.csv.
- New `paperpod export <video>` CLI command.
- Verified against synthetic Home Depot/Metro/Petro-Canada receipts:
  3/3 correct vendor, date, and total after tuning.

Known limitation (documented in README): no page-flip/multi-page
grouping yet (that's events/, still unbuilt) - every detected document
becomes its own single-page PDF.
This commit is contained in:
2026-07-07 18:04:06 -04:00
parent 8f5be59b7e
commit 2c28a623fb
14 changed files with 723 additions and 12 deletions
+84
View File
@@ -10,6 +10,7 @@ from __future__ import annotations
import csv
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
@@ -18,6 +19,9 @@ import numpy as np
from paperpod.capture.video import iter_frames, probe_video
from paperpod.config import Config, config_to_dict
from paperpod.naming.filename import build_name, dedupe_stem
from paperpod.ocr.extract import run_ocr
from paperpod.pdf.build import images_to_pdf
from paperpod.vision.document import detect_document, warp_document
from paperpod.vision.enhance import enhance_for_ocr
from paperpod.vision.motion import Segment, find_segments, motion_score, prepare_gray
@@ -160,3 +164,83 @@ def run_detection(video_path: str | Path, cfg: Config, out_dir: str | Path) -> d
with open(out_dir / "report.json", "w") as f:
json.dump(report, f, indent=2)
return report
def run_export(report: dict[str, Any], cfg: Config, out_dir: str | Path) -> dict[str, Any]:
"""OCR-name and PDF-export every detected crop from a detect() report.
NOTE: there is no page-grouping yet (module 3 / events/ is not built),
so every detected crop becomes its own single-page PDF, even if it was
physically one side of a multi-page stack. pod_id == window_id for now.
Writes:
pdf/<name>.pdf one PDF per detected document
export_summary.csv timestamp, pod_id, page_count, source_of_name,
ocr_vendor, ocr_date, ocr_total, final_filename
Returns the summary rows (also embedded back for programmatic use).
"""
out_dir = Path(out_dir)
pdf_dir = out_dir / "pdf"
pdf_dir.mkdir(parents=True, exist_ok=True)
video_path = Path(report["video"])
duration_s = report["video_meta"]["duration_s"]
try:
recording_start = datetime.fromtimestamp(video_path.stat().st_mtime) - timedelta(
seconds=duration_s
)
except FileNotFoundError:
recording_start = datetime.now() - timedelta(seconds=duration_s)
used_stems: set[str] = set()
rows: list[dict[str, Any]] = []
for window in report["stable_windows"]:
if not window.get("document_found"):
continue
crop_path = out_dir / window["crop_path"]
image = cv2.imread(str(crop_path))
if image is None:
continue
ocr = run_ocr(
image,
lang=cfg.ocr.lang,
vendor_top_fraction=cfg.ocr.vendor_top_fraction,
min_word_confidence=cfg.ocr.min_word_confidence,
psm=cfg.ocr.psm,
)
event_time = recording_start + timedelta(seconds=window["t_start"])
naming = build_name(ocr.date, ocr.vendor, event_time, cfg.naming.max_vendor_len)
stem = dedupe_stem(naming.stem, used_stems)
pdf_path = pdf_dir / f"{stem}.pdf"
images_to_pdf([crop_path], pdf_path, title=stem)
rows.append(
{
"timestamp": event_time.isoformat(timespec="seconds"),
"pod_id": window["window_id"],
"page_count": 1,
"source_of_name": naming.source,
"ocr_vendor": ocr.vendor or "",
"ocr_date": ocr.date or "",
"ocr_total": ocr.total or "",
"ocr_confidence": round(ocr.mean_confidence, 1),
"final_filename": pdf_path.name,
"pdf_path": str(pdf_path.relative_to(out_dir)),
}
)
with open(out_dir / "export_summary.csv", "w", newline="") as f:
fieldnames = [
"timestamp", "pod_id", "page_count", "source_of_name",
"ocr_vendor", "ocr_date", "ocr_total", "ocr_confidence",
"final_filename", "pdf_path",
]
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
return {"rows": rows, "pdf_dir": str(pdf_dir)}