src/ledger/store.py

"""Double-entry ledger backed by SQLite.

Design choices that matter:

* **Money is stored as integer cents.** SQLite has no exact decimal type and
  ``SUM`` over text is meaningless; integers make balance checks and aggregates
  exact. Decimals live only at the boundary (in/out).
* **Every transaction must balance** — total debits == total credits — or ``post``
  refuses it. This is the invariant a double-entry system exists to guarantee.
* **High-level helpers build the entries for you.** ``record_income`` /
  ``record_expense`` translate a freelancer's mental model ("I invoiced 10,000 plus
  IVA") into the correct multi-line journal entry, including IVA trasladado /
  acreditable and client retentions.
* **The engine reads from here, never writes.** ``month_totals`` returns exactly the
  aggregates the tax engine needs.
"""

from __future__ import annotations

import calendar
import sqlite3
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Dict, List, Optional, Sequence, Tuple

from ..engine.money import D, money
from .accounts import ACCOUNT_TYPES, NORMAL_SIDES, SEED_ACCOUNTS, Account


# --- cents <-> Decimal at the boundary ------------------------------------
def to_cents(value) -> int:
    return int((money(value) * 100).to_integral_value())


def from_cents(cents: int) -> Decimal:
    return money(D(cents) / 100)


@dataclass
class Line:
    """One leg of a journal entry. Exactly one of debit/credit is non-zero."""

    account_code: str
    debit: Decimal = Decimal("0.00")
    credit: Decimal = Decimal("0.00")


@dataclass
class MonthTotals:
    """Aggregates for one month — the exact inputs the tax engine expects."""

    year: int
    month: int
    income: Decimal = Decimal("0.00")
    deductible_expenses: Decimal = Decimal("0.00")
    nondeductible_expenses: Decimal = Decimal("0.00")
    iva_trasladado: Decimal = Decimal("0.00")
    iva_acreditable: Decimal = Decimal("0.00")
    iva_retenido: Decimal = Decimal("0.00")
    isr_retenido: Decimal = Decimal("0.00")


@dataclass
class IncomeStatement:
    period: str
    revenue: Decimal
    expenses: Decimal
    net_profit: Decimal


@dataclass
class BalanceSheet:
    as_of: Optional[str]
    assets: Decimal
    liabilities: Decimal
    equity: Decimal  # balancing figure: assets - liabilities
    detail: Dict[str, Decimal] = field(default_factory=dict)


class Ledger:
    """A per-user double-entry ledger. Use ``:memory:`` for tests."""

    def __init__(self, path: str = ":memory:", seed: bool = True):
        # check_same_thread=False: a UI like Gradio runs each event in a worker
        # thread, so the connection must be usable across threads. Python's sqlite3
        # runs in serialized mode, so a single shared connection is thread-safe.
        self.conn = sqlite3.connect(path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA foreign_keys = ON")
        self._create_schema()
        if seed:
            self.seed_default_accounts()

    # --- lifecycle --------------------------------------------------------
    def close(self) -> None:
        self.conn.close()

    def __enter__(self) -> "Ledger":
        return self

    def __exit__(self, *exc) -> None:
        self.close()

    def _create_schema(self) -> None:
        self.conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS accounts (
                id INTEGER PRIMARY KEY,
                code TEXT UNIQUE NOT NULL,
                name TEXT NOT NULL,
                type TEXT NOT NULL,
                normal_side TEXT NOT NULL,
                role TEXT
            );
            CREATE TABLE IF NOT EXISTS transactions (
                id INTEGER PRIMARY KEY,
                date TEXT NOT NULL,
                description TEXT NOT NULL,
                kind TEXT,
                deductible INTEGER NOT NULL DEFAULT 0,
                iva_treatment TEXT,
                iva_rate TEXT,
                cfdi_uuid TEXT
            );
            CREATE TABLE IF NOT EXISTS entries (
                id INTEGER PRIMARY KEY,
                transaction_id INTEGER NOT NULL REFERENCES transactions(id) ON DELETE CASCADE,
                account_id INTEGER NOT NULL REFERENCES accounts(id),
                debit_cents INTEGER NOT NULL DEFAULT 0,
                credit_cents INTEGER NOT NULL DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS idx_tx_date ON transactions(date);
            CREATE INDEX IF NOT EXISTS idx_entry_tx ON entries(transaction_id);
            """
        )
        self.conn.commit()

    # --- accounts ---------------------------------------------------------
    def seed_default_accounts(self) -> None:
        for a in SEED_ACCOUNTS:
            self.ensure_account(a.code, a.name, a.type, a.normal_side, a.role)

    def ensure_account(
        self, code, name, type, normal_side, role=None
    ) -> int:
        if type not in ACCOUNT_TYPES:
            raise ValueError(f"unknown account type {type!r}")
        if normal_side not in NORMAL_SIDES:
            raise ValueError(f"unknown normal_side {normal_side!r}")
        row = self.conn.execute(
            "SELECT id FROM accounts WHERE code = ?", (code,)
        ).fetchone()
        if row:
            return row["id"]
        cur = self.conn.execute(
            "INSERT INTO accounts(code, name, type, normal_side, role) VALUES (?,?,?,?,?)",
            (code, name, type, normal_side, role),
        )
        self.conn.commit()
        return cur.lastrowid

    def _account_id(self, code: str) -> int:
        row = self.conn.execute(
            "SELECT id FROM accounts WHERE code = ?", (code,)
        ).fetchone()
        if not row:
            raise KeyError(f"no account with code {code!r}")
        return row["id"]

    def code_for_role(self, role: str) -> str:
        row = self.conn.execute(
            "SELECT code FROM accounts WHERE role = ? ORDER BY code LIMIT 1", (role,)
        ).fetchone()
        if not row:
            raise KeyError(f"no account with role {role!r}")
        return row["code"]

    # --- low-level posting ------------------------------------------------
    def post(
        self,
        date: str,
        description: str,
        lines: Sequence[Line],
        kind: Optional[str] = None,
        deductible: bool = False,
        iva_treatment: Optional[str] = None,
        iva_rate=None,
        cfdi_uuid: Optional[str] = None,
    ) -> int:
        """Post a balanced journal entry. Raises if debits != credits."""
        debit_total = sum(to_cents(l.debit) for l in lines)
        credit_total = sum(to_cents(l.credit) for l in lines)
        if debit_total != credit_total:
            raise ValueError(
                f"unbalanced entry: debits {from_cents(debit_total)} "
                f"!= credits {from_cents(credit_total)}"
            )
        if debit_total == 0:
            raise ValueError("empty entry (zero debits and credits)")

        cur = self.conn.execute(
            "INSERT INTO transactions(date, description, kind, deductible, "
            "iva_treatment, iva_rate, cfdi_uuid) VALUES (?,?,?,?,?,?,?)",
            (
                date,
                description,
                kind,
                1 if deductible else 0,
                iva_treatment,
                None if iva_rate is None else str(D(iva_rate)),
                cfdi_uuid,
            ),
        )
        tx_id = cur.lastrowid
        for l in lines:
            self.conn.execute(
                "INSERT INTO entries(transaction_id, account_id, debit_cents, credit_cents) "
                "VALUES (?,?,?,?)",
                (tx_id, self._account_id(l.account_code), to_cents(l.debit), to_cents(l.credit)),
            )
        self.conn.commit()
        return tx_id

    # --- high-level helpers ----------------------------------------------
    def record_income(
        self,
        date: str,
        description: str,
        amount,  # taxable base (pre-IVA subtotal)
        iva_rate=Decimal("0.16"),
        income_code: Optional[str] = None,
        settle_role: str = "bank",
        isr_retenido=0,
        iva_retenido=0,
        cfdi_uuid: Optional[str] = None,
    ) -> int:
        """Book a sale. Builds: Dr cash/retentions, Cr income + IVA trasladado."""
        base = money(amount)
        rate = D(iva_rate)
        iva = money(base * rate)
        isr_ret = money(isr_retenido)
        iva_ret = money(iva_retenido)
        net_cash = money(base + iva - isr_ret - iva_ret)

        income_code = income_code or self.code_for_role("income")
        lines: List[Line] = [Line(self.code_for_role(settle_role), debit=net_cash)]
        if isr_ret > 0:
            lines.append(Line(self.code_for_role("isr_retenido"), debit=isr_ret))
        if iva_ret > 0:
            lines.append(Line(self.code_for_role("iva_retenido"), debit=iva_ret))
        lines.append(Line(income_code, credit=base))
        if iva > 0:
            lines.append(Line(self.code_for_role("iva_trasladado"), credit=iva))

        treatment = "zero" if rate == 0 else "standard"
        return self.post(
            date, description, lines, kind="income",
            iva_treatment=treatment, iva_rate=rate, cfdi_uuid=cfdi_uuid,
        )

    def record_expense(
        self,
        date: str,
        description: str,
        amount,  # taxable base (pre-IVA subtotal)
        iva_rate=Decimal("0.16"),
        expense_code: Optional[str] = None,
        deductible: bool = True,
        settle_role: str = "bank",
        cfdi_uuid: Optional[str] = None,
    ) -> int:
        """Book a purchase. Deductible → IVA is acreditable; otherwise expensed whole."""
        base = money(amount)
        rate = D(iva_rate)
        iva = money(base * rate)
        expense_code = expense_code or self.code_for_role("expense")

        lines: List[Line] = []
        if deductible:
            lines.append(Line(expense_code, debit=base))
            if iva > 0:
                lines.append(Line(self.code_for_role("iva_acreditable"), debit=iva))
            lines.append(Line(self.code_for_role(settle_role), credit=money(base + iva)))
        else:
            # Non-deductible: no acreditable IVA; the whole outlay hits the expense.
            total = money(base + iva)
            lines.append(Line(expense_code, debit=total))
            lines.append(Line(self.code_for_role(settle_role), credit=total))

        treatment = "zero" if rate == 0 else "standard"
        return self.post(
            date, description, lines, kind="expense", deductible=deductible,
            iva_treatment=treatment, iva_rate=rate, cfdi_uuid=cfdi_uuid,
        )

    # --- aggregates -------------------------------------------------------
    @staticmethod
    def _month_range(year: int, month: int) -> Tuple[str, str]:
        last = calendar.monthrange(year, month)[1]
        return f"{year:04d}-{month:02d}-01", f"{year:04d}-{month:02d}-{last:02d}"

    def _sum(self, sql: str, params: tuple) -> Decimal:
        row = self.conn.execute(sql, params).fetchone()
        return from_cents(row[0] or 0)

    def month_totals(self, year: int, month: int) -> MonthTotals:
        start, end = self._month_range(year, month)
        win = (start, end)

        income = self._sum(
            "SELECT SUM(e.credit_cents) FROM entries e "
            "JOIN accounts a ON a.id = e.account_id "
            "JOIN transactions t ON t.id = e.transaction_id "
            "WHERE a.type='income' AND t.date BETWEEN ? AND ?",
            win,
        )
        deductible = self._sum(
            "SELECT SUM(e.debit_cents) FROM entries e "
            "JOIN accounts a ON a.id = e.account_id "
            "JOIN transactions t ON t.id = e.transaction_id "
            "WHERE a.type='expense' AND t.deductible=1 AND t.date BETWEEN ? AND ?",
            win,
        )
        nondeductible = self._sum(
            "SELECT SUM(e.debit_cents) FROM entries e "
            "JOIN accounts a ON a.id = e.account_id "
            "JOIN transactions t ON t.id = e.transaction_id "
            "WHERE a.type='expense' AND t.deductible=0 AND t.date BETWEEN ? AND ?",
            win,
        )

        def role_sum(role: str, side: str) -> Decimal:
            return self._sum(
                f"SELECT SUM(e.{side}_cents) FROM entries e "
                "JOIN accounts a ON a.id = e.account_id "
                "JOIN transactions t ON t.id = e.transaction_id "
                "WHERE a.role=? AND t.date BETWEEN ? AND ?",
                (role, start, end),
            )

        return MonthTotals(
            year=year,
            month=month,
            income=income,
            deductible_expenses=deductible,
            nondeductible_expenses=nondeductible,
            iva_trasladado=role_sum("iva_trasladado", "credit"),
            iva_acreditable=role_sum("iva_acreditable", "debit"),
            iva_retenido=role_sum("iva_retenido", "debit"),
            isr_retenido=role_sum("isr_retenido", "debit"),
        )

    def list_transactions(self, year: int, month: int) -> List[dict]:
        """Rows for the UI ledger table: date, description, kind, deductible, amount."""
        start, end = self._month_range(year, month)
        rows = self.conn.execute(
            "SELECT t.id, t.date, t.description, t.kind, t.deductible, "
            "SUM(e.debit_cents) d, SUM(e.credit_cents) c "
            "FROM transactions t JOIN entries e ON e.transaction_id = t.id "
            "WHERE t.date BETWEEN ? AND ? GROUP BY t.id ORDER BY t.date",
            (start, end),
        ).fetchall()
        out = []
        for r in rows:
            total = max(r["d"] or 0, r["c"] or 0)  # balanced entry → debits == credits
            out.append({
                "date": r["date"],
                "description": r["description"],
                "kind": r["kind"] or "",
                "deductible": "sí" if r["deductible"] else ("—" if r["kind"] != "expense" else "no"),
                "amount": str(from_cents(total)),
            })
        return out

    # --- statements -------------------------------------------------------
    def income_statement(self, year: int, month: Optional[int] = None) -> IncomeStatement:
        if month is None:
            start, end = f"{year:04d}-01-01", f"{year:04d}-12-31"
            period = str(year)
        else:
            start, end = self._month_range(year, month)
            period = f"{year:04d}-{month:02d}"
        revenue = self._sum(
            "SELECT SUM(e.credit_cents) FROM entries e JOIN accounts a ON a.id=e.account_id "
            "JOIN transactions t ON t.id=e.transaction_id "
            "WHERE a.type='income' AND t.date BETWEEN ? AND ?",
            (start, end),
        )
        expenses = self._sum(
            "SELECT SUM(e.debit_cents) FROM entries e JOIN accounts a ON a.id=e.account_id "
            "JOIN transactions t ON t.id=e.transaction_id "
            "WHERE a.type='expense' AND t.date BETWEEN ? AND ?",
            (start, end),
        )
        return IncomeStatement(period, revenue, expenses, money(revenue - expenses))

    def account_balance(self, code: str, as_of: Optional[str] = None) -> Decimal:
        aid = self._account_id(code)
        row = self.conn.execute("SELECT type, normal_side FROM accounts WHERE id=?", (aid,)).fetchone()
        clause, params = "WHERE e.account_id=?", [aid]
        if as_of:
            clause += " AND t.date <= ?"
            params.append(as_of)
        r = self.conn.execute(
            "SELECT SUM(e.debit_cents) d, SUM(e.credit_cents) c FROM entries e "
            "JOIN transactions t ON t.id=e.transaction_id " + clause,
            tuple(params),
        ).fetchone()
        debit, credit = from_cents(r["d"] or 0), from_cents(r["c"] or 0)
        return money(debit - credit) if row["normal_side"] == "debit" else money(credit - debit)

    def balance_sheet(self, as_of: Optional[str] = None) -> BalanceSheet:
        def type_total(acct_type: str) -> Decimal:
            rows = self.conn.execute(
                "SELECT code FROM accounts WHERE type=?", (acct_type,)
            ).fetchall()
            return money(sum((self.account_balance(r["code"], as_of) for r in rows), Decimal("0")))

        assets = type_total("asset")
        liabilities = type_total("liability")
        equity = money(assets - liabilities)  # balancing figure (incl. retained earnings)
        return BalanceSheet(
            as_of=as_of,
            assets=assets,
            liabilities=liabilities,
            equity=equity,
            detail={"assets": assets, "liabilities": liabilities, "equity": equity},
        )