dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/checks.py
"""Checks that look before someone relies on a number (v4): the VAT-number check of a customer (TAX-13: HMRC's lookup, an AI
comparison of the name it returns, a person's confirmation when HMRC cannot answer) and the AI review of a draft VAT return
(findings for the tax admin; nothing changes, locking never waits on a finding).
"""
import hashlib
import json
from decimal import Decimal
from typing import Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, text
from brick_tax.application import returns as R
from brick_tax.application import use_cases as uc
from brick_tax.application.inputs import Key, Reason
from brick_tax.application.use_cases import (
RETURN_PREPARE,
SETTINGS_WRITE,
SUGGEST,
TAX,
VAT_NUMBER_CHECK,
)
from brick_tax.domain.errors import unmet
from brick_tax.infrastructure import ai_steps, seams
from brick_tax.infrastructure.hmrc import HmrcError
def _current(repo, bp_id: int) -> dict:
row = repo.parties.effective(bp_id, uc._today(repo))
if row is None:
raise Refused(404, f"no tax treatment recorded for customer {bp_id}")
return row
@TAX.use_case(kind="command", route="POST /parties/{bp_id}/check-vat-number", permission=VAT_NUMBER_CHECK, idempotent=True)
def check_party_vat_number(caller: dict, bp_id: BigId) -> dict:
"""Check a customer's VAT number with HMRC (the system runs it when a treatment that needs a number is saved, and again when the
last check is older than vat_number_check_days before a reverse-charge document issues): valid | invalid | could_not_check
(a non-GB number, or HMRC unavailable), with HMRC's name and address and the consultation number kept on the party's row. In
suggest mode the AI compares HMRC's name with the partner's and flags a mismatch for the biller; it never changes the treatment."""
def unit(repo):
row = _current(repo, int(bp_id))
number = (row["vat_number"] or "").replace(" ", "").upper()
t = repo.treatments.get(row["treatment_code"])
if not number or not (t and t["needs_vat_number"]):
raise unmet("vat_number_to_check", "only a treatment that needs a VAT number, with the number saved, is checked", "Customer tax > the VAT number")
settings = uc._settings_now(repo)
result, name, address, consultation = "could_not_check", "", "", ""
if number.startswith("GB"):
try:
got = seams.hmrc.check_vat_number(settings.hmrc_environment, number[2:])
except HmrcError:
got, result = None, "could_not_check"
else:
result = "valid" if got else "invalid"
if got:
name, address, consultation = got.get("name", ""), got.get("address", ""), got.get("consultation_number", "")
flag, ai = None, None
if result == "valid" and settings.check_vat_number_mode != "off":
partner = seams.partners.customer(int(bp_id), caller) or {}
answer = ai_steps.ask("check_vat_number", settings.check_vat_number_mode, f"tax_party:{bp_id}", 1, {
"vat_number": number, "hmrc_name": name, "hmrc_address": address, "partner_name": partner.get("legal_name", ""),
"partner_address": partner.get("address", "")}, fields=("name_mismatch",))
ai = answer.as_dict()
if answer.kind == "done" and (answer.suggestion or {}).get("name_mismatch"):
flag = answer.rationale or "HMRC's name differs from the partner's"
repo.parties.update(int(bp_id), row["valid_from"], checked_on=uc._today(repo), check_result=result, check_name=name,
check_address=address, consultation_number=consultation, check_flag=flag, confirmed_by=None, confirm_reason=None)
return {**uc._party_view(repo.parties.effective(int(bp_id), uc._today(repo))), "ai": ai}
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /parties/{bp_id}/confirm-vat-number", permission=SETTINGS_WRITE)
def confirm_party_vat_number(caller: dict, bp_id: BigId, reason: Reason) -> dict:
"""A person confirms a number HMRC could not check (a non-GB number, or HMRC unavailable), with the reason; recorded with who
and when. A number HMRC found valid or invalid is not confirmed this way."""
if not (reason or "").strip():
raise Refused(422, "confirming a VAT number needs a reason")
def unit(repo):
row = _current(repo, int(bp_id))
if row.get("check_result") != "could_not_check":
raise unmet("could_not_check", "only a number whose check could not be made is confirmed by a person",
"check the number first" if not row.get("check_result") else "the check already answered")
repo.parties.update(int(bp_id), row["valid_from"], confirmed_by=uc._who(caller), confirm_reason=reason)
return uc._party_view(repo.parties.effective(int(bp_id), uc._today(repo)))
return uc._atomic(unit)
# ═══ the AI review of a draft return ════════════════════════════════════════════════════════════
def _finding_view(r: dict) -> dict:
return {"finding_no": r["finding_no"], "evidence": r["evidence"], "box": r["box"], "model": r["model"], "decision": r["decision"] or "open",
"decided_by": r["decided_by"] or None, "note": r["note"] or None}
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/ai-review", permission=SUGGEST)
def ai_review_vat_return(caller: dict, period_key: Key) -> dict:
"""The system asks the AI to review a DRAFT return against the last four periods: a box that moved far, a document with an
unusual code. Findings are listed for the tax admin with their evidence; NOTHING changes and locking never waits on a finding.
Only figures go to the sovereign model (boxes, entry sizes and codes), no customer names. Off when ai.review_vat_return.mode is
off; with no provider connected the tax admin reviews without findings."""
def unit(repo):
vrn = R._vrn(repo)
ret, version = R._load_return(repo, vrn, period_key)
ret.editable()
mode = uc._settings_now(repo).review_vat_return_mode
if mode == "off":
raise Refused(409, "the AI review of VAT returns is switched off (ai.review_vat_return.mode): the tax admin reviews without findings")
previous = [r for r in repo.returns.all(vrn) if r.period_key < period_key][-4:]
entries = sorted(repo.entries.where(vrn=vrn, period_key=period_key), key=lambda e: abs(Decimal(str(e["vat_amount"] or 0))), reverse=True)[:10]
context = {f"box{i + 1}": str(v) for i, v in enumerate(ret.boxes.values())}
for n, p in enumerate(reversed(previous), 1):
context |= {f"prev{n}_box{i + 1}": str(v) for i, v in enumerate(p.boxes.values())}
context["largest_entries"] = json.dumps([{"tax_code": e["tax_code"], "direction": e["direction"], "kind": e["entry_kind"],
"vat": str(e["vat_amount"]), "net": str(e["net_amount"])} for e in entries])
digest = hashlib.sha256(json.dumps(context, sort_keys=True).encode()).hexdigest()[:16]
existing = repo.findings.where(vrn=vrn, period_key=period_key)
if any(f["inputs_digest"] == digest for f in existing):
return {"ai": {"kind": "replay"}, "findings": [_finding_view(f) for f in existing]}
answer = ai_steps.ask("ai_review_vat_return", mode, f"tax_vat_return:{vrn}:{period_key}", version, context, fields=("findings",))
found = (answer.suggestion or {}).get("findings") if answer.kind == "done" else None
no = max([f["finding_no"] for f in existing], default=0)
for f in found or []:
no += 1
repo.findings.put({"vrn": vrn, "period_key": period_key, "finding_no": no, "evidence": str(f.get("evidence", ""))[:1000],
"box": int(f["box"]) if str(f.get("box", "")).isdigit() else None, "model": answer.model, "inputs_digest": digest,
"decision": ""})
return {"ai": answer.as_dict(), "findings": [_finding_view(f) for f in repo.findings.where(vrn=vrn, period_key=period_key)]}
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /vat-returns/{period_key}/findings/{finding_no}/decide", permission=RETURN_PREPARE)
def decide_vat_finding(caller: dict, period_key: Key, finding_no: int, decision: Literal["dismiss", "acted"], note: Optional[text(500)] = None) -> dict:
"""The tax admin dismisses a finding or records that they acted on it (kept with who and when). Locking never waits on one."""
def unit(repo):
vrn = R._vrn(repo)
f = repo.findings.get(vrn=vrn, period_key=period_key, finding_no=finding_no)
if f is None:
raise Refused(404, f"no finding {finding_no} on {period_key}")
repo.findings.put({**{k: v for k, v in f.items() if k not in ("created_at", "updated_at")}, "decision": decision,
"decided_by": uc._who(caller), "decided_at": R._utcnow(), "note": note or ""})
return _finding_view(repo.findings.get(vrn=vrn, period_key=period_key, finding_no=finding_no))
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /vat-returns/{period_key}/findings", permission=RETURN_PREPARE)
def list_vat_findings(caller: dict, period_key: Key) -> dict:
"""The AI review's findings on a return with the tax admin's decisions."""
with uc.repository() as repo:
return {"period_key": period_key, "findings": [_finding_view(f) for f in repo.findings.where(vrn=R._vrn(repo), period_key=period_key)]}