File size: 5,457 Bytes
7ff7119 | 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 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """Domain check registry — 14 deterministic rules with a unified API.
The ``risk_subgraph`` uses the Send API to fan out (per-doc, per-applicable-check)
pairs; each Send invokes an ``apply_domain_check`` node which looks up and runs
the check from this registry.
Two SEPARATE entry points (skipped from dispatch via the ``SKIP_FROM_DISPATCH`` set):
* ``check_06_evidence_score``: per-doc info, called directly after classification
* ``check_12_duplicate_invoice``: package-level O(n²), called from a separate
node in the ``risk_subgraph``
"""
from __future__ import annotations
from domain_checks.base import DomainCheck, is_empty, make_risk
from domain_checks.check_01_invoice_mandatory import InvoiceMandatoryCheck
from domain_checks.check_02_tax_cdv import TaxCDVCheck, compute_cdv, validate_tax_cdv
from domain_checks.check_03_contract_completeness import ContractCompletenessCheck
from domain_checks.check_04_proportionality import ProportionalityCheck
from domain_checks.check_05_rounded_amounts import RoundedAmountsCheck
from domain_checks.check_06_evidence_score import EvidenceScoreCheck, get_evidence_score
from domain_checks.check_07_materiality import MaterialityCheck
from domain_checks.check_08_gdpr_28 import GDPR28Check
from domain_checks.check_09_dd_red_flags import DDRedFlagsCheck
from domain_checks.check_10_incoterms import INCOTERMS_2020, IncotermsCheck
from domain_checks.check_11_ifrs_har import IFRSHARCheck
from domain_checks.check_12_duplicate_invoice import (
DuplicateInvoiceCheck,
check_duplicate_invoices,
)
from domain_checks.check_13_aml_sanctions import AMLSanctionsCheck
from domain_checks.check_14_contract_dates import ContractDatesCheck
# Unified registry of all 14 checks. The risk_subgraph's domain_dispatch_node
# iterates this list and Send-fans-out the (doc, check) pairs. Skipped
# checks (06: evidence score, 12: duplicate detection) are called via separate
# entry points.
CHECK_REGISTRY: list[DomainCheck] = [
InvoiceMandatoryCheck(), # 01: HU VAT Act §169 (HU jurisdiction)
TaxCDVCheck(), # 02: HU Tax Procedure Act §22 mod-11 (HU jurisdiction)
ContractCompletenessCheck(), # 03: Universal contract completeness
ProportionalityCheck(), # 04: Universal contract proportionality
RoundedAmountsCheck(), # 05: ISA 240
EvidenceScoreCheck(), # 06: ISA 500 (separate entry point)
MaterialityCheck(), # 07: ISA 320
GDPR28Check(), # 08: GDPR Article 28
DDRedFlagsCheck(), # 09: M&A DD best practice
IncotermsCheck(), # 10: Incoterms 2020
IFRSHARCheck(), # 11: IFRS / national GAAP comparison
DuplicateInvoiceCheck(), # 12: ISA 240 package-level (separate entry point)
AMLSanctionsCheck(), # 13: AML / Sanctions screening
ContractDatesCheck(), # 14: Contract date best practice
]
# Skipped check_ids (NOT Send-fanned out; called by separate nodes)
SKIP_FROM_DISPATCH = {"check_06_evidence_score", "check_12_duplicate_invoice"}
def get_check(check_id: str) -> DomainCheck | None:
"""Look up a check by check_id."""
for c in CHECK_REGISTRY:
if c.check_id == check_id:
return c
return None
def get_applied_standards(risks) -> list[str]:
"""Return the list of standards/regulations actually applied to the package.
The UI footer only shows standards that had at least one risk finding,
OR that always run (e.g. ISA 500 evidence score).
"""
# Standards that always run (universal, every jurisdiction)
always = {"ISA 500"}
# Standards referenced in actual risks (i.e. triggered)
from_risks: set[str] = set()
for r in risks or []:
if hasattr(r, "regulation"):
reg = r.regulation
elif isinstance(r, dict):
reg = r.get("regulation") or r.get("jogszabaly") # legacy compat
else:
reg = None
if reg:
from_risks.add(reg)
all_standards = always | from_risks
# Sorted display order for the UI footer
order = [
"HU VAT Act §169", "HU Tax Procedure Act §22",
"Universal contract completeness", "Universal contract proportionality",
"ISA 240", "ISA 240 (duplicate invoice)",
"ISA 500", "ISA 320",
"GDPR Article 28", "M&A DD best practice",
"Incoterms 2020", "IFRS / national GAAP comparison",
"AML / Sanctions screening",
"Contract date best practice",
"EU VAT Directive",
]
result = [s for s in order if s in all_standards]
# Append any standards not in the fixed order
for s in sorted(all_standards):
if s and s not in result:
result.append(s)
return result
__all__ = [
"DomainCheck",
"CHECK_REGISTRY",
"SKIP_FROM_DISPATCH",
"get_check",
"get_applied_standards",
"is_empty",
"make_risk",
# Check classes
"InvoiceMandatoryCheck",
"TaxCDVCheck",
"ContractCompletenessCheck",
"ProportionalityCheck",
"RoundedAmountsCheck",
"EvidenceScoreCheck",
"MaterialityCheck",
"GDPR28Check",
"DDRedFlagsCheck",
"IncotermsCheck",
"IFRSHARCheck",
"DuplicateInvoiceCheck",
"AMLSanctionsCheck",
"ContractDatesCheck",
# Helpers
"compute_cdv",
"validate_tax_cdv",
"get_evidence_score",
"INCOTERMS_2020",
"check_duplicate_invoices",
]
|