"""Finance self-service configuration surfaces. Everything a finance analyst tunes lives here and is editable by the ``finance`` role (NOT reserved to global admin) — variance thresholds, the NetSuite consolidation scope, comparison modes, the AI toggle + prompt, the statement-line catalog (labels / sections / order / owners / per-line threshold overrides / active), the Summary KPI catalog, and the NS account→line mapping. (The NetSuite API *credentials* remain admin-managed encrypted integration config — those are secrets, not finance settings.) """ from __future__ import annotations import uuid from datetime import UTC, datetime from decimal import Decimal from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import NotFoundError from app.domains.finance_flux import constants as c from app.domains.finance_flux.models import ( FinanceFluxAccountMap, FinanceFluxConfig, FinanceFluxKpiDef, FinanceFluxLineDef, ) _CONFIG_FIELDS = ( "amount_threshold", "pct_threshold", "currency", "consolidation_subsidiary", "accounting_book", "comparison_modes", "closed_periods_only", "ai_enabled", "ai_prompt_template", ) async def _config_row(db: AsyncSession) -> FinanceFluxConfig: row = (await db.execute( select(FinanceFluxConfig).order_by(FinanceFluxConfig.created_at) )).scalars().first() if row is None: row = FinanceFluxConfig( id=uuid.uuid4(), amount_threshold=Decimal(c.DEFAULT_AMOUNT_THRESHOLD), pct_threshold=Decimal(c.DEFAULT_PCT_THRESHOLD), currency=c.DEFAULT_CURRENCY, comparison_modes=[c.COMPARE_MOM, c.COMPARE_YOY], ) db.add(row) await db.flush() return row def _dec2(x) -> str | None: """Stable 2dp string regardless of whether the value is freshly set in Python (``25000``) or loaded from the Numeric column (``25000.00``).""" if x is None: return None return str(Decimal(x).quantize(Decimal("0.01"))) def _serialize_config(r: FinanceFluxConfig) -> dict: return { "amount_threshold": _dec2(r.amount_threshold), "pct_threshold": _dec2(r.pct_threshold), "currency": r.currency, "consolidation_subsidiary": r.consolidation_subsidiary, "accounting_book": r.accounting_book, "comparison_modes": r.comparison_modes or [], "closed_periods_only": r.closed_periods_only, "ai_enabled": r.ai_enabled, "ai_prompt_template": r.ai_prompt_template, } async def get_config(db: AsyncSession) -> dict: return _serialize_config(await _config_row(db)) async def update_config(db: AsyncSession, *, editor_id=None, **fields) -> dict: row = await _config_row(db) for k in _CONFIG_FIELDS: if k in fields and fields[k] is not None: setattr(row, k, fields[k]) row.updated_by_person_id = editor_id row.updated_at = datetime.now(UTC) await db.flush() return _serialize_config(row) # ── Statement-line catalog ────────────────────────────────────────── def _serialize_line_def(d: FinanceFluxLineDef) -> dict: return { "id": str(d.id), "statement": d.statement, "line_key": d.line_key, "label": d.label, "section": d.section, "row_type": d.row_type, "sort_order": d.sort_order, "owner_person_id": str(d.owner_person_id) if d.owner_person_id else None, "owner_label": d.owner_label, "threshold_amount": _dec2(d.threshold_amount), "threshold_pct": _dec2(d.threshold_pct), "is_active": d.is_active, } async def list_line_defs(db: AsyncSession, *, statement: str | None = None) -> list[dict]: stmt = select(FinanceFluxLineDef).order_by( FinanceFluxLineDef.statement, FinanceFluxLineDef.sort_order) if statement: stmt = stmt.where(FinanceFluxLineDef.statement == statement) return [_serialize_line_def(d) for d in (await db.execute(stmt)).scalars()] _LINE_DEF_EDITABLE = ( "label", "section", "sort_order", "owner_person_id", "owner_label", "threshold_amount", "threshold_pct", "is_active", ) async def update_line_def(db: AsyncSession, *, line_def_id: str, **fields) -> dict: row = await db.get(FinanceFluxLineDef, uuid.UUID(str(line_def_id))) if row is None: raise NotFoundError("line definition not found") for k in _LINE_DEF_EDITABLE: if k in fields and fields[k] is not None: setattr(row, k, fields[k]) await db.flush() return _serialize_line_def(row) # ── Summary KPI catalog ───────────────────────────────────────────── def _serialize_kpi(k: FinanceFluxKpiDef) -> dict: return { "id": str(k.id), "kpi_key": k.kpi_key, "label": k.label, "statement": k.statement, "source_line_key": k.source_line_key, "sort_order": k.sort_order, "higher_is_better": k.higher_is_better, "is_active": k.is_active, } async def list_kpi_defs(db: AsyncSession) -> list[dict]: return [_serialize_kpi(k) for k in (await db.execute( select(FinanceFluxKpiDef).order_by(FinanceFluxKpiDef.sort_order) )).scalars()] _KPI_EDITABLE = ("label", "source_line_key", "sort_order", "higher_is_better", "is_active") async def update_kpi_def(db: AsyncSession, *, kpi_def_id: str, **fields) -> dict: row = await db.get(FinanceFluxKpiDef, uuid.UUID(str(kpi_def_id))) if row is None: raise NotFoundError("KPI definition not found") for k in _KPI_EDITABLE: if k in fields and fields[k] is not None: setattr(row, k, fields[k]) await db.flush() return _serialize_kpi(row) # ── NS account → line mapping ─────────────────────────────────────── def _serialize_map(m: FinanceFluxAccountMap) -> dict: return { "id": str(m.id), "ns_account_id": m.ns_account_id, "account_number": m.account_number, "account_name": m.account_name, "account_type": m.account_type, "statement": m.statement, "line_key": m.line_key, "sign": m.sign, "is_active": m.is_active, } async def list_account_map( db: AsyncSession, *, statement: str | None = None, unmapped_only: bool = False, ) -> list[dict]: stmt = select(FinanceFluxAccountMap).order_by(FinanceFluxAccountMap.account_number) if statement: stmt = stmt.where(FinanceFluxAccountMap.statement == statement) if unmapped_only: stmt = stmt.where(FinanceFluxAccountMap.line_key.is_(None)) return [_serialize_map(m) for m in (await db.execute(stmt)).scalars()] async def update_account_map_entry( db: AsyncSession, *, entry_id: str, line_key=None, sign=None, is_active=None, ) -> dict: row = await db.get(FinanceFluxAccountMap, uuid.UUID(str(entry_id))) if row is None: raise NotFoundError("account-map entry not found") if line_key is not None: row.line_key = line_key or None if sign is not None: row.sign = sign if is_active is not None: row.is_active = is_active await db.flush() return _serialize_map(row)