From da37e91c860206087d7cd743365c84cad724fd7c Mon Sep 17 00:00:00 2001 From: Florian Egger Date: Fri, 21 Aug 2026 16:44:35 +0200 Subject: [PATCH] Add CLI orchestration with confidence-gated XML output - Refactors process_single into inner pipeline with guaranteed client close. - Runs classifier -> OCR branch -> extraction -> review -> XML/sidecar. - Uses review-corrected invoice when available. - Confidence >= 0.75 produces XML; < 0.90 adds a human-review comment. - Adds end-to-end CLI tests covering confidence bands, corrections, and batch mode. --- src/odoo_ocr/cli.py | 17 +++++- tests/test_cli.py | 133 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 tests/test_cli.py diff --git a/src/odoo_ocr/cli.py b/src/odoo_ocr/cli.py index 1bd62e0..ff855ab 100644 --- a/src/odoo_ocr/cli.py +++ b/src/odoo_ocr/cli.py @@ -32,6 +32,16 @@ async def process_single(path: Path, output_dir: Path, settings: Settings) -> di client = OllamaClient(settings) context = ProcessingContext(source_path=path, output_dir=output_dir) + try: + return await _process_single_inner(path, output_dir, settings, client, context) + finally: + await client.aclose() + + +async def _process_single_inner( + path: Path, output_dir: Path, settings: Settings, client: OllamaClient, context: ProcessingContext +) -> dict[str, Any]: + """Run the pipeline stages for one invoice file.""" doc = load_document(path, settings) pages = doc["pages"] @@ -54,6 +64,11 @@ async def process_single(path: Path, output_dir: Path, settings: Settings) -> di review = await review_invoice(extracted, pages, client, settings) context.review_result = review.model_dump(mode="json") + # Prefer the review model's corrected invoice when it provided one. + final_invoice = review.corrected_invoice if review.corrected_invoice is not None else extracted + if review.corrected_invoice is not None: + logger.info("Using corrected invoice from the review stage.") + sidecar = { "source_path": str(path), "classification": classification.model_dump(), @@ -68,7 +83,7 @@ async def process_single(path: Path, output_dir: Path, settings: Settings) -> di json.dump(sidecar, f, indent=2, ensure_ascii=False) if review.confidence >= settings.review.min_confidence_threshold: - xml = build_odoo_xml(extracted, review, settings) + xml = build_odoo_xml(final_invoice, review, settings) with open(f"{out_stem}.xml", "w", encoding="utf-8") as f: f.write(xml) logger.info("Wrote XML: %s", f"{out_stem}.xml") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..b59d1b4 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,133 @@ +"""End-to-end CLI pipeline tests using the fake VLM client.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest +from lxml import etree +from PIL import Image + +from odoo_ocr import cli +from odoo_ocr.config import Settings +from tests.conftest import FakeVLMClient + +EXTRACTION: dict[str, Any] = { + "vendor_name": "Acme", + "vendor_address": None, + "vendor_vat": None, + "invoice_number": "INV-9", + "invoice_date": "2024-01-01", + "due_date": None, + "currency": "EUR", + "payment_terms": None, + "line_items": [ + { + "description": "Widget", + "quantity": 1, + "unit_price": 100.0, + "total_price": 100.0, + "tax_rate": 0, + } + ], + "subtotal": 100.0, + "tax_total": 0.0, + "total": 100.0, + "iban": None, + "raw_ocr_text": "", +} + + +def _responses(confidence: float) -> dict[str, Any]: + return { + "document classifier": { + "category": "scanned_print", + "confidence": 0.9, + "reasoning": "printed invoice", + }, + "OCR engine": {"text": "Invoice INV-9 from Acme, total 100.00 EUR"}, + "invoice data extraction": EXTRACTION, + "invoice review assistant": { + "valid": True, + "confidence": confidence, + "issues": [], + "corrected_invoice": None, + }, + } + + +def _write_invoice(tmp_path: Path) -> Path: + src = tmp_path / "invoice.jpg" + Image.new("RGB", (100, 100), "white").save(src) + return src + + +def _install_fake(monkeypatch: pytest.MonkeyPatch, responses: dict[str, Any]) -> FakeVLMClient: + fake = FakeVLMClient(responses=responses) + monkeypatch.setattr(cli, "OllamaClient", lambda _settings: fake) + return fake + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("confidence", "expect_xml", "expect_review_comment"), + [(0.95, True, False), (0.80, True, True), (0.70, False, False)], +) +async def test_confidence_bands( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + confidence: float, + expect_xml: bool, + expect_review_comment: bool, +) -> None: + src = _write_invoice(tmp_path) + out_dir = tmp_path / "out" + _install_fake(monkeypatch, _responses(confidence)) + + sidecar = await cli.process_single(src, out_dir, Settings()) + + assert sidecar["review"]["confidence"] == confidence + assert (out_dir / "invoice_sidecar.json").exists() + data = json.loads((out_dir / "invoice_sidecar.json").read_text(encoding="utf-8")) + assert data["extracted_invoice"]["vendor_name"] == "Acme" + + xml_file = out_dir / "invoice.xml" + assert xml_file.exists() is expect_xml + if expect_xml: + xml = xml_file.read_text(encoding="utf-8") + assert ("HUMAN REVIEW ADVISED" in xml) is expect_review_comment + root = etree.fromstring(xml.encode("utf-8")) + models = {record.get("model") for record in root.iter("record")} + assert {"res.partner", "account.move", "account.move.line", "account.tax"} <= models + + +@pytest.mark.asyncio +async def test_uses_corrected_invoice_from_review( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + src = _write_invoice(tmp_path) + responses = _responses(0.95) + responses["invoice review assistant"]["corrected_invoice"] = { + **EXTRACTION, + "invoice_number": "INV-999", + } + _install_fake(monkeypatch, responses) + + await cli.process_single(src, tmp_path / "out", Settings()) + + xml = (tmp_path / "out" / "invoice.xml").read_text(encoding="utf-8") + assert "INV-999" in xml + + +@pytest.mark.asyncio +async def test_process_batch(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + _install_fake(monkeypatch, _responses(0.95)) + (tmp_path / "a.jpg").touch() + Image.new("RGB", (10, 10), "white").save(tmp_path / "b.jpg") + + results = await cli.process_batch(tmp_path, tmp_path / "out", Settings()) + + assert len(results) == 1 + assert (tmp_path / "out" / "b.xml").exists()