dodil/erp-brick-ar

DODIL ERP brick: accounts receivable

src/brick_ar/infrastructure/odoo.py

"""The reader port for the Odoo slice (migration.odoo.reads): read-only, search_read only, never a write.

`OdooReader` is the shape the migration use case needs; tests hand it fake rows (no live Odoo). `XmlRpcOdoo` is the one real
adapter, configured by ODOO_URL, ODOO_DB, ODOO_USER and ODOO_KEY (an API key), and used only from the use case's default.
"""
import os
import xmlrpc.client
from decimal import Decimal
from typing import Protocol

MOVE_FIELDS = ["id", "name", "move_type", "state", "payment_state", "partner_id", "commercial_partner_id", "invoice_date",
               "invoice_date_due", "date", "invoice_payment_term_id", "currency_id", "amount_untaxed", "amount_tax",
               "amount_total", "amount_residual", "amount_untaxed_signed", "amount_total_signed", "amount_residual_signed",
               "ref", "payment_reference", "invoice_origin", "reversed_entry_id", "journal_id", "company_id",
               "partner_shipping_id", "create_date", "write_date"]
LINE_FIELDS = ["id", "move_id", "sequence", "display_type", "product_id", "name", "quantity", "product_uom_id", "price_unit",
               "discount", "price_subtotal", "price_total", "tax_ids", "tax_line_id", "tax_base_amount", "balance",
               "amount_currency", "date_maturity", "sale_line_ids", "account_id"]


class OdooReader(Protocol):
    db: str

    def company_currency(self) -> str: ...
    def moves(self, company_id, include_drafts: bool) -> list: ...
    def lines(self, move_ids: list) -> dict: ...
    def payment_terms(self, term_ids: list) -> tuple: ...       # (terms, lines)
    def partners(self, partner_ids: list) -> list: ...
    def receivable_balance(self, cutover) -> dict: ...


class XmlRpcOdoo:
    def __init__(self, url=None, db=None, user=None, key=None):
        self.url, self.db = (url or os.environ["ODOO_URL"]).rstrip("/"), db or os.environ["ODOO_DB"]
        self._user, self._key = user or os.environ["ODOO_USER"], key or os.environ["ODOO_KEY"]
        self._uid = None

    def _call(self, model, method, args, **kw):
        if self._uid is None:
            self._uid = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/common").authenticate(self.db, self._user, self._key, {})
        proxy = xmlrpc.client.ServerProxy(f"{self.url}/xmlrpc/2/object", allow_none=True)
        return proxy.execute_kw(self.db, self._uid, self._key, model, method, args, kw)

    def _read(self, model, domain, fields):
        return self._call(model, "search_read", [domain], fields=fields)

    def company_currency(self) -> str:
        return self._call("res.company", "search_read", [[]], fields=["currency_id"], limit=1)[0]["currency_id"][1]

    def moves(self, company_id, include_drafts: bool) -> list:
        states = ["posted", "cancel"] + (["draft"] if include_drafts else [])
        dom = [("move_type", "in", ["out_invoice", "out_refund"]), ("state", "in", states)]
        if company_id:
            dom.append(("company_id", "=", company_id))
        return self._read("account.move", dom, MOVE_FIELDS)

    def lines(self, move_ids: list) -> dict:
        out: dict = {}
        for i in range(0, len(move_ids), 500):
            for ln in self._read("account.move.line", [("move_id", "in", move_ids[i:i + 500])], LINE_FIELDS):
                out.setdefault(ln["move_id"][0], []).append(ln)
        return out

    def payment_terms(self, term_ids: list) -> tuple:
        terms = self._read("account.payment.term", [("id", "in", term_ids)], ["id", "name", "active", "line_ids"]) if term_ids else []
        lines = self._read("account.payment.term.line", [("payment_id", "in", term_ids)], []) if term_ids else []
        return terms, lines

    def partners(self, partner_ids: list) -> list:
        return self._read("res.partner", [("id", "in", partner_ids)],
                          ["id", "name", "email", "street", "street2", "zip", "city", "property_payment_term_id"]) if partner_ids else []

    def receivable_balance(self, cutover) -> dict:
        rows = self._read("account.move.line", [("account_id.account_type", "=", "asset_receivable"), ("parent_state", "=", "posted"),
                                                ("date", "<=", str(cutover))], ["balance", "currency_id"])
        out: dict = {}
        for r in rows:
            cur = r["currency_id"][1] if r.get("currency_id") else self.company_currency()
            out[cur] = out.get(cur, Decimal(0)) + Decimal(str(r["balance"]))
        return {k: f"{v:.2f}" for k, v in out.items()}