breakpointsoftware/document-parser

1from __future__ import annotations2 3import argparse4import json5import os6from pathlib import Path7from typing import Any8 9from document_processing import extract_text, is_image_document, load_documents, to_data_uri10from firebase_processed_files import build_firebase_tracker11from receipt_ai import extract_receipt_json, extract_receipt_json_from_image12from receipt_results import build_empty_result13 14 15def parse_args() -> argparse.Namespace:16    parser = argparse.ArgumentParser(description="Extract purchase receipt data from documents using GPT.")17    parser.add_argument("--input", default="documents", help="Folder containing documents to process.")18    parser.add_argument("--output", help="Optional output JSON file. Prints to stdout when omitted.")19    parser.add_argument("--model", default=os.getenv("OPENAI_MODEL", "gpt-4o"), help="OpenAI model name.")20    return parser.parse_args()21 22 23def main() -> int:24    from dotenv import load_dotenv25    from openai import OpenAI26 27    load_dotenv()28    args = parse_args()29    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))30 31    input_folder = Path(args.input)32    files = load_documents(input_folder)33    results: list[dict[str, Any]] = []34    skipped_files: list[str] = []35    firebase_warning: str | None = None36    firebase_tracker = build_firebase_tracker()37 38    for file_path in files:39        relative_name = file_path.relative_to(input_folder).as_posix()40        file_hash: str | None = None41 42        if firebase_tracker is not None:43            try:44                file_hash = firebase_tracker.compute_file_hash(file_path)45                if firebase_tracker.is_processed(file_hash):46                    skipped_files.append(relative_name)47                    continue48            except Exception as exc:49                firebase_warning = f"Firebase tracking unavailable. Processing remaining files without deduplication. Details: {exc}"50                firebase_tracker = None51 52        if is_image_document(file_path):53            image_data_uri = to_data_uri(file_path)54            result = extract_receipt_json_from_image(client, args.model, file_path, image_data_uri)55        else:56            text = extract_text(file_path)57            if not text.strip():58                results.append(build_empty_result(relative_name))59                continue60 61            result = extract_receipt_json(client, args.model, file_path, text)62 63        result["source_file"] = relative_name64        results.append(result)65 66        if firebase_tracker is not None and file_hash is not None:67            try:68                firebase_tracker.mark_processed(file_hash, relative_name)69            except Exception:70                pass71 72    payload: dict[str, Any] = {"input_folder": str(input_folder), "results": results}73    if skipped_files:74        payload["skipped_already_processed"] = skipped_files75    if firebase_warning:76        payload["firebase_warning"] = firebase_warning77 78    rendered = json.dumps(payload, indent=2, ensure_ascii=False)79 80    if args.output:81        output_path = Path(args.output)82        output_path.parent.mkdir(parents=True, exist_ok=True)83        output_path.write_text(rendered, encoding="utf-8")84    else:85        print(rendered)86 87    return 088 89 90if __name__ == "__main__":91    raise SystemExit(main())