platform/harness/canonical.py

"""Canonical financial data model — the SHARED vocabulary of the harness.

Skills speak ONLY in these terms; connectors (Odoo / QuickBooks / Amazon / …) translate them to
their own source. This is the seam that decouples the dashboards from any single system: add a new
source by writing a connector that maps its API onto these entities/fields — every skill that needs
only the capabilities that source provides then works, unchanged.

Design rules:
- Entities and canonical field names are source-agnostic (no `amount_untaxed`, no `sale.order`).
- A skill builds a `Query` (what business question), a connector answers it (how, per source).
- Capabilities let a skill declare what it needs and a tenant show only the dashboards its
  connected sources can actually serve.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any


class Capability(str, Enum):
    """What a connector can serve. Skills require a set of these; a tenant's nav is the union of
    what its connectors provide. (A pure-accounting source like QuickBooks has LEDGER+INVOICING+
    CUSTOMERS but not SKU-level SALES velocity; Amazon has SALES but no LEDGER; Odoo has all.)"""
    SALES = 'sales'            # orders / order lines (SKU-level revenue, units)
    INVOICING = 'invoicing'   # customer invoices + credit notes (AR, returns)
    LEDGER = 'ledger'         # GL accounts + journal lines (P&L, balance sheet)
    PAYABLES = 'payables'     # vendor bills / AP
    INVENTORY = 'inventory'   # on-hand stock levels + valuation
    PRODUCTS = 'products'     # product master (SKU, category, cost)
    CUSTOMERS = 'customers'   # customer master (name, geo, assigned agent)


class Entity(str, Enum):
    """Canonical business objects. Each connector declares which it can materialize."""
    SALES_ORDER = 'sales_order'
    SALES_ORDER_LINE = 'sales_order_line'
    INVOICE = 'invoice'                 # customer invoice / credit note
    JOURNAL_LINE = 'journal_line'       # GL posting
    GL_ACCOUNT = 'gl_account'
    VENDOR_BILL = 'vendor_bill'
    STOCK_LEVEL = 'stock_level'
    PRODUCT = 'product'
    CUSTOMER = 'customer'


# Canonical field names per entity — the ONLY field identifiers a skill may use. Connectors own the
# mapping to their native fields. `date` is the entity's primary time axis (used by Query windows).
FIELDS: dict[Entity, set[str]] = {
    Entity.SALES_ORDER:      {'date', 'amount', 'customer', 'salesperson', 'agent', 'status', 'business_unit', 'channel'},
    Entity.SALES_ORDER_LINE: {'date', 'amount', 'qty', 'margin', 'product', 'category', 'customer', 'status', 'business_unit'},
    Entity.INVOICE:          {'date', 'amount', 'customer', 'doc_type', 'status', 'salesperson', 'business_unit'},
    Entity.JOURNAL_LINE:     {'date', 'amount', 'account', 'account_type', 'business_unit', 'status'},
    Entity.GL_ACCOUNT:       {'code', 'name', 'type'},
    Entity.VENDOR_BILL:      {'date', 'amount', 'vendor', 'status'},
    Entity.STOCK_LEVEL:      {'product', 'qty', 'value', 'location'},
    Entity.PRODUCT:          {'id', 'code', 'name', 'category', 'cost', 'active'},
    Entity.CUSTOMER:         {'id', 'name', 'city', 'state', 'agent', 'active'},
}

# Allowed filter/aggregate vocabulary (connectors translate these; they never see source operators).
OPS = frozenset({'eq', 'ne', 'in', 'not_in', 'gte', 'lte', 'gt', 'lt', 'like'})
AGGS = frozenset({'sum', 'count', 'avg', 'min', 'max'})
GRAINS = frozenset({'day', 'week', 'month', 'quarter', 'year'})   # time-bucket for a trend


@dataclass(frozen=True)
class Filter:
    """A canonical predicate: (canonical_field, op, value). op ∈ OPS."""
    fieldname: str
    op: str
    value: Any

    def __post_init__(self):
        if self.op not in OPS:
            raise ValueError(f'unknown op {self.op!r}; allowed {sorted(OPS)}')


@dataclass(frozen=True)
class Query:
    """A source-agnostic analytical request. `measures` are canonical fields aggregated by `aggregate`,
    optionally grouped by canonical `group_by` fields, filtered by `filters`, over [date_from, date_to]
    on the entity's `date` axis. A connector returns a list of plain dicts keyed by the canonical
    measure/group names (+ 'count'). Skills post-process those dicts — never source rows."""
    entity: Entity
    measures: tuple[str, ...] = ()
    aggregate: str = 'sum'
    group_by: tuple[str, ...] = ()
    filters: tuple[Filter, ...] = ()
    date_from: str | None = None
    date_to: str | None = None
    grain: str | None = None       # bucket the date axis into a trend; result rows carry 'period' (e.g. '2026-05')

    def __post_init__(self):
        if self.aggregate not in AGGS:
            raise ValueError(f'unknown aggregate {self.aggregate!r}')
        if self.grain is not None and self.grain not in GRAINS:
            raise ValueError(f'unknown grain {self.grain!r}; allowed {sorted(GRAINS)}')
        valid = FIELDS.get(self.entity, set())
        for f in list(self.measures) + list(self.group_by) + [x.fieldname for x in self.filters]:
            if f not in valid:
                raise ValueError(f'{f!r} is not a canonical field of {self.entity.value} (have {sorted(valid)})')