dodil/erp-brick-payments

DODIL ERP brick: payments

src/brick_payments/domain/odoo_map.py

"""Odoo -> payments, as pure functions (PAY-44, module.yaml migration.odoo.maps): which allocation a reconciliation
partial becomes, and the keys everything is known by. No I/O: the use case reads, this decides.

Keys follow shared.yaml's migrated_form - "odoo:<db>:<model>:<id>" - so a test copy and the live database can never
collide, and a re-run names the same row. ar names its invoices and credit notes "odoo:<db>:<account.move id>"."""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional

from brick_payments.domain.money import money
from brick_payments.domain.settlement import credit_ref


def key(db: str, model: str, odoo_id) -> str:
    return f"odoo:{db}:{model}:{odoo_id}"


def ar_id(db: str, move_id) -> str:
    """The id ar's migration gave an invoice, credit note (and, by the same rule, write-off) for an account.move."""
    return f"odoo:{db}:{move_id}"


@dataclass(frozen=True)
class Planned:
    """One allocation a partial becomes."""
    partial_id: int
    source_kind: str
    source_id: str
    invoice_id: str                 # for a refund: the credit it was paid out of ("receipt:<id>" / "credit_note:<id>")
    amount: Decimal
    currency: str
    source_amount: Optional[Decimal]
    allocated_on: Optional[date]
    cross_currency: bool = False
    rate: Optional[Decimal] = None  # receipt-currency units per invoice-currency unit, for a cross-currency receipt


def classify(partial: dict, lines: dict, db: str, receipt_ids: set, refund_ids: set, company_currency: str) -> tuple:
    """(Planned, None) or (None, why-not). `receipt_ids`/`refund_ids`: the keys of the receipts and refunds being loaded."""
    d, c = lines.get(partial["debit_line"]), lines.get(partial["credit_line"])
    if d is None or c is None:
        return None, "a reconciliation names a receivable line that was not read"
    when = date.fromisoformat(str(partial["max_date"])[:10]) if partial.get("max_date") else None
    amount = money(partial["debit_amount_currency"])
    cur = d.get("currency") or company_currency

    def source_of(line: dict) -> Optional[tuple]:
        if line.get("payment_id"):
            sid = key(db, "account.payment", line["payment_id"])
            return ("receipt", sid) if sid in receipt_ids else None
        if line.get("statement_line_id"):
            sid = key(db, "account.bank.statement.line", line["statement_line_id"])
            return ("receipt", sid) if sid in receipt_ids else None
        if line.get("move_type") == "out_refund":
            return ("credit_note", ar_id(db, line["move_id"]))
        return None

    if d.get("move_type") == "out_invoice":
        invoice = ar_id(db, d["move_id"])
        src = source_of(c)
        if src is None:
            if c.get("payment_id") or c.get("statement_line_id"):
                return None, "the receipt behind it is not one this migration loads (a draft, outbound or unposted payment)"
            src = ("writeoff", ar_id(db, c["move_id"]))     # an entry move: ar's migrated write-off
        cc = c.get("currency") or company_currency
        cross = src[0] == "receipt" and cc != cur
        src_amount = money(partial["credit_amount_currency"]) if cross else None
        rate = (src_amount / amount).quantize(Decimal("0.0001")) if cross and amount else None
        return Planned(partial["id"], src[0], src[1], invoice, amount, cur, src_amount, when, cross, rate), None
    if d.get("payment_id"):
        refund = key(db, "account.payment", d["payment_id"])
        if refund in refund_ids:
            src = source_of(c)
            if src is None or src[0] not in ("receipt", "credit_note"):
                return None, "a refund is paid out of a receipt or a credit note, and this one is neither"
            return Planned(partial["id"], "refund", refund, credit_ref(src[0], src[1]), amount, cur, None, when), None
    return None, "neither an invoice settled nor a refund paid out of a credit"