Self-hosted, verifiable pipeline for extracting structured data (station,

date/time, price per litre, litres, discount, total) from German fuel receipt photos: grayscale + mild bilateral filter + contrast stretch, Tesseract OCR, then LLM structuring via a small local model (Qwen 2.5, served by Ollama). Each step is independently runnable and inspectable through numbered scripts against a shared tooling container.
2026-09-28 19:17:47 +02:00
commit aae57d7a2b
15 changed files with 773 additions and 0 deletions
@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""Test-only tool: compare a directory of extracted.json results (produced
+by running 1_preprocess_image.sh/2_ocr_image.sh/3_extract_data.sh over a
+set of receipts) against the ground-truth spreadsheet, matching rows by
+date and reporting per-field accuracy."""
+import argparse
+import glob
+import json
+import os
+
+from odf.opendocument import load
+from odf.table import Table, TableRow, TableCell
+from odf.text import P
+
+
+def load_ground_truth(ods_path: str) -> dict:
+    doc = load(ods_path)
+    table = doc.spreadsheet.getElementsByType(Table)[0]
+    rows = table.getElementsByType(TableRow)
+    gt = {}
+    for r in rows[1:]:
+        cells = r.getElementsByType(TableCell)
+        vals = []
+        for c in cells:
+            ps = c.getElementsByType(P)
+            vals.append(" ".join(str(p) for p in ps))
+        if not vals or not vals[0]:
+            continue
+        date = vals[0]
+        try:
+            gt[date] = {
+                "time": vals[1],
+                "price_per_litre": float(vals[2].replace(",", ".")) if vals[2] else None,
+                "litres": float(vals[3].replace(",", ".")) if vals[3] else None,
+                "discount": float(vals[4].replace("€", "").replace(",", ".").strip()) if vals[4] else None,
+                "total_price": float(vals[5].replace("€", "").replace(",", ".").strip()) if vals[5] else None,
+            }
+        except (ValueError, IndexError):
+            continue
+    return gt
+
+
+def close(a, b, tol=0.05):
+    if a is None or b is None:
+        return a is None and b is None
+    return abs(abs(a) - abs(b)) <= tol
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-gt", dest="ods_path", required=True, help="path to ground-truth .ods")
+    parser.add_argument("-results", dest="results_dir", required=True,
+                         help="directory containing one subdirectory per receipt, each with an extracted.json")
+    args = parser.parse_args()
+
+    gt = load_ground_truth(args.ods_path)
+
+    fields = ["price_per_litre", "litres", "discount", "total_price"]
+    field_ok = {k: 0 for k in fields}
+    field_total = {k: 0 for k in fields}
+    matched = 0
+    total = 0
+
+    for json_path in sorted(glob.glob(os.path.join(args.results_dir, "*", "extracted.json"))):
+        receipt = os.path.basename(os.path.dirname(json_path))
+        total += 1
+        with open(json_path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        truth = gt.get(data.get("date"))
+        if not truth:
+            print(f"{receipt}: NO DATE MATCH (extracted date={data.get('date')!r})")
+            continue
+
+        matched += 1
+        bad = []
+        for k in fields:
+            field_total[k] += 1
+            ok = close(data.get(k), truth[k])
+            if ok:
+                field_ok[k] += 1
+            else:
+                bad.append(k)
+        status = "OK" if not bad else f"WRONG FIELDS {bad}"
+        print(f"{receipt}: {status}")
+
+    print("\n=== SUMMARY ===")
+    print(f"Receipts matched to ground truth: {matched}/{total}")
+    for k in fields:
+        print(f"  {k}: {field_ok[k]}/{field_total[k]} correct")
+
+
+if __name__ == "__main__":
+    main()