Invoice-Downloader/document_acceptance.py at main · EthanYoQ/Invoice-Downloader · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from __future__ import annotations

from typing import Any


class DocumentAcceptanceService:
    """Run the established provider/entity acceptance gate before archive."""

    def __init__(self, api: Any) -> None:
        self.api = api

    @staticmethod
    def normalized_snapshot(fields: dict[str, Any] | None) -> dict[str, Any] | None:
        if not fields or not isinstance(fields, dict):
            return None
        raw_date = str(fields.get("Date", ""))
        clean_date = (
            raw_date.replace("/", "")
            .replace("-", "")
            .replace("年", "")
            .replace("月", "")
            .replace("日", "")
            .strip()
        )
        raw_amount = str(fields.get("Amount", ""))
        clean_amount = (
            raw_amount.replace(",", "")
            .replace("¥", "")
            .replace("￥", "")
            .replace("元", "")
            .replace(" ", "")
            .strip()
        )
        return {
            "Date": clean_date or "未知",
            "Amount": clean_amount or "未知",
            "Purchaser": str(fields.get("Purchaser", "")),
            "Seller": str(fields.get("Seller", "")),
            "Type": str(fields.get("Type", "")),
            "InvoiceCode": str(fields.get("InvoiceCode", "")).strip(),
            "InvoiceNumber": str(fields.get("InvoiceNumber", "")).strip(),
            "is_invoice": fields.get("is_invoice", True),
        }

    @staticmethod
    def write_canonical_fields(
        info_json: dict[str, Any], snapshot: dict[str, Any] | None
    ) -> None:
        if not snapshot:
            return
        for field in (
            "Date",
            "Amount",
            "Purchaser",
            "Seller",
            "InvoiceCode",
            "InvoiceNumber",
        ):
            info_json[field] = snapshot[field]

    def evaluate(
        self,
        metadata: dict[str, Any],
        info_json: dict[str, Any],
        pdf_path: str,
        *,
        pdf_health: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        snapshot = self.normalized_snapshot(info_json)
        if pdf_health is None:
            pdf_health = self.api._inspect_pdf_health(pdf_path)
        return self.api._evaluate_document_acceptance(
            metadata,
            info_json,
            snapshot,
            pdf_health,
            pdf_path,
        )