"""CFO-OS shared Odoo data layer (READ-ONLY). Reuses the existing write-guarded client (ffs_dashboard/odoo_client.py) so there is a single source of truth for the connection. Adds: - cached singleton client - read_group / search_read passthroughs with light cleanup - business constants (teams, excluded partners) - the "sales universe" base domain (confirmed orders, RI+FFS only) NOTHING here writes to Odoo — the underlying client hard-blocks writes. """ import os import sys import queue as _queue import threading from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from functools import lru_cache from pathlib import Path # Self-contained read-only client lives at the app root (platform/odoo_client.py) with its # own .env — so this whole folder can be moved/hosted anywhere without external paths. _ROOT = Path(__file__).resolve().parents[1] # platform/ if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) from odoo_client import OdooClient # noqa: E402 (read-only, write-guarded) # ---- business constants (TENANT #0's scope, kept as module attributes for its modules) ---- # These are Royal Imports' config — the default a request gets when nothing binds a tenant. # The canonical per-tenant home is `harness/tenants.py` Tenant.config; a BOUND tenant (see # `tenant_scope` below) carries its own team ids / exclusions and never reads these. TEAM_IDS = [5, 6] TEAM_NAMES = {5: 'Fisch', 6: 'Royal'} # Partners to exclude from analytics (not part of RI/FFS). Sourced from env so no # customer name is ever hardcoded in the (potentially public) repo; set as a Space secret. EXCLUDE_PARTNER_NAMES = {n.strip() for n in os.environ.get('EXCLUDE_PARTNER_NAMES', '').split(',') if n.strip()} # Confirmed sale.order.line scope for RI+FFS (the analytics "sales universe"). # ⚠ Documentation of tenant #0's base scope — `sale_line_domain()` below BUILDS its domain from # `active_team_ids()` (binding-aware) rather than reading this list; edit both or neither. SALE_LINE_BASE = [ ('state', 'in', ['sale', 'done']), ('order_id.team_id', 'in', TEAM_IDS), ('product_id', '!=', False), ] # Posted customer invoices/credit notes (AR + revenue recognised). INVOICE_BASE = [ ('move_type', 'in', ['out_invoice', 'out_refund']), ('state', '=', 'posted'), ] @lru_cache(maxsize=1) def _global_client(): """TENANT #0's read-only client (env/.env credentials) — the default for every unbound call.""" return OdooClient() # ---- per-tenant connections (R3 keychain cutover, 2026-08-04) -------------------------------- # One client per tenant slug, keyed by a fingerprint of its credentials so a rotated key # reconnects on the next call with no restart. The DEFAULT slug maps to the env client above. # ⛔ FAIL-CLOSED: a bound NON-default tenant with no credentials raises — the env fallback # would silently answer tenant B's question over tenant A's connection, which is the exact # failure this section exists to remove. _DEFAULT_SLUG = 'royal-imports' _TENANT_CLIENTS = {} # slug -> (creds fingerprint, OdooClient) _TC_LOCK = threading.Lock() class TenantOdooUnavailable(RuntimeError): """A bound tenant has no Odoo credentials. Callers surface 'connect a source', never another tenant's data.""" def _client_for(slug, creds): """The client for a BOUND tenant. creds None: default slug -> env client; anyone else -> raise.""" slug = str(slug or _DEFAULT_SLUG) if not creds: if slug == _DEFAULT_SLUG: return _global_client() raise TenantOdooUnavailable( f"tenant {slug!r} has no Odoo credentials — store them under Settings → Keychains. " f"(Refusing the environment fallback: that is another tenant's connection.)") fp = (str(creds.get('url') or ''), str(creds.get('db') or ''), str(creds.get('user') or ''), str(creds.get('api_key') or '')) with _TC_LOCK: cur = _TENANT_CLIENTS.get(slug) if cur and cur[0] == fp: return cur[1] cli = OdooClient(creds=creds) # built OUTSIDE the lock: it authenticates over the wire with _TC_LOCK: cur = _TENANT_CLIENTS.get(slug) if cur and cur[0] == fp: return cur[1] # another thread won the race — keep ONE client _TENANT_CLIENTS[slug] = (fp, cli) return cli @contextmanager def tenant_scope(slug, creds=None, team_ids=None, team_names=None, exclude_partner_names=None): """Bind THIS THREAD's Odoo context to a tenant: its connection + its scope constants. Unbound (the default) = tenant #0 via env, which is every existing Streamlit/API caller. The connector layer (harness/connectors/odoo.py) wraps each query in this, so a second tenant's keychain credentials reach the wire without any module knowing. Scope constants are PUSHED here from Tenant.config because core/ may not import harness/ (layering).""" prev = getattr(_tlocal, 'tenant_ctx', None) _tlocal.tenant_ctx = { 'slug': str(slug or _DEFAULT_SLUG), 'creds': creds, 'team_ids': list(team_ids) if team_ids else None, 'team_names': dict(team_names) if team_names else None, 'exclude_partner_names': set(exclude_partner_names) if exclude_partner_names else None, } try: yield finally: _tlocal.tenant_ctx = prev def _ctx(): return getattr(_tlocal, 'tenant_ctx', None) def active_team_ids(): """The bound tenant's team ids; unbound or default slug -> tenant #0's TEAM_IDS.""" c = _ctx() if c is not None and c['slug'] != _DEFAULT_SLUG: return list(c['team_ids'] or []) return list(TEAM_IDS) def active_team_names(): c = _ctx() if c is not None and c['slug'] != _DEFAULT_SLUG: return dict(c['team_names'] or {}) return dict(TEAM_NAMES) # ---- concurrent read pool ------------------------------------------------------------------- # Each Odoo XML-RPC call is a round-trip; a heavy drawer fires a dozen of them and they add up to # many seconds when run one-by-one. `parallel()` runs independent READ tasks across a small pool of # extra connections, with each task transparently using its own connection via a thread-local — so # the existing read_group/search_read/sum_field helpers need no change. READ-ONLY only. _POOL_MAX = 8 _tlocal = threading.local() _pool_q = _queue.Queue() _pool_n = 0 _pool_lock = threading.Lock() def _checkout(): global _pool_n try: return _pool_q.get_nowait() except _queue.Empty: with _pool_lock: grow = _pool_n < _POOL_MAX if grow: _pool_n += 1 return OdooClient() if grow else _pool_q.get() # else block until one frees def _checkin(c): _pool_q.put(c) def get_odoo(): """The active read-only client, in priority order: 1. a NON-default tenant binding (tenant_scope) — its own client, ALWAYS. The parallel() pool below holds tenant-#0 connections, so a pooled client must never answer a bound tenant's call (that is the cross-tenant leak R3's cutover removes); 2. this task's pooled connection (parallel(), and the app's dedicated-bg-thread pattern); 3. an explicit default-slug binding — the env client; 4. unbound: the shared env singleton. Transparent to every read_group/... caller.""" c = _ctx() if c is not None and c['slug'] != _DEFAULT_SLUG: return _client_for(c['slug'], c['creds']) task_client = getattr(_tlocal, 'client', None) if task_client is not None: return task_client if c is not None: return _client_for(c['slug'], c['creds']) return _global_client() def set_doc_mode(mode): """Recognition basis for THIS thread's domain builds: 'order' (all confirmed orders) or 'invoice' (fully-invoiced orders only). Cached bundles call this before pulling; parallel() propagates it to each worker. Thread-local, so concurrent user sessions never race on each other's basis.""" _tlocal.doc_mode = 'invoice' if mode == 'invoice' else 'order' def doc_mode(): return getattr(_tlocal, 'doc_mode', 'order') def parallel(tasks): """Run independent zero-arg READ callables concurrently across the connection pool; returns results in order. Inside each task, get_odoo() (hence read_group/search_read/sum_field/ search_count) uses that task's own connection. Never use for writes.""" tasks = list(tasks) if not tasks: return [] if len(tasks) == 1: return [tasks[0]()] caller_mode = doc_mode() # propagate the submitter's basis to each worker caller_ctx = _ctx() # and the tenant binding — a worker thread must # answer for the SAME tenant as its submitter def _run(fn): c = _checkout() _tlocal.client = c _tlocal.doc_mode = caller_mode _tlocal.tenant_ctx = caller_ctx try: return fn() finally: _tlocal.client = None _tlocal.doc_mode = 'order' _tlocal.tenant_ctx = None _checkin(c) with ThreadPoolExecutor(max_workers=min(len(tasks), _POOL_MAX)) as ex: return list(ex.map(_run, tasks)) def parallel_map(task_dict): """parallel() over a {name: callable} mapping -> {name: result}.""" keys = list(task_dict) return dict(zip(keys, parallel([task_dict[k] for k in keys]))) def m2o_id(v): return v[0] if isinstance(v, list) and v else None def m2o_name(v): return v[1] if isinstance(v, list) and len(v) > 1 else '' _EXCL_IDS = {} # slug -> tuple of excluded partner ids (resolved once per process) _EXCL_LOCK = threading.Lock() def excluded_partner_ids(): """Resolve the ACTIVE tenant's excluded partner names to ids, for fast domain filtering. Cached per tenant slug — the old lru_cache(1) would have handed tenant B whatever tenant A resolved first. Unbound/default = tenant #0's env-sourced names; a bound tenant's names come from its scope config (none configured -> nothing excluded).""" c = _ctx() bound_other = c is not None and c['slug'] != _DEFAULT_SLUG slug = c['slug'] if bound_other else _DEFAULT_SLUG names = (c['exclude_partner_names'] or set()) if bound_other else EXCLUDE_PARTNER_NAMES with _EXCL_LOCK: if slug in _EXCL_IDS: return _EXCL_IDS[slug] if not names: res = tuple() else: rows = get_odoo().search_read('res.partner', [('name', 'in', list(names))], ['id']) res = tuple(r['id'] for r in rows) with _EXCL_LOCK: _EXCL_IDS[slug] = res return res def sale_line_domain(date_from=None, date_to=None, team_id=None, extra=None, partner_ids=None): """Build a sale.order.line domain in the RI+FFS confirmed-order scope, excluding the configured house accounts, with optional date window / single team / extra clauses. partner_ids (a collection, possibly empty) restricts to those customers — the carrier for the Customer module's Agent filter; None = no partner restriction (an empty set matches none).""" tids = active_team_ids() dom = [('state', 'in', ['sale', 'done']), ('product_id', '!=', False)] if tids: dom.insert(1, ('order_id.team_id', 'in', tids)) if team_id is not None: dom = [d for d in dom if not (isinstance(d, tuple) and d[0] == 'order_id.team_id')] dom.append(('order_id.team_id', '=', team_id)) if date_from: dom.append(('order_id.date_order', '>=', f'{date_from} 00:00:00')) if date_to: dom.append(('order_id.date_order', '<=', f'{date_to} 23:59:59')) ex = excluded_partner_ids() if ex: dom.append(('order_partner_id', 'not in', list(ex))) if partner_ids is not None: dom.append(('order_partner_id', 'in', list(partner_ids))) if doc_mode() == 'invoice': dom.append(('order_id.invoice_status', '=', 'invoiced')) if extra: dom.extend(extra) return dom def read_group(model, domain, fields, groupby, **kw): return get_odoo().read_group(model, domain=domain, fields=fields, groupby=groupby, **kw) def search_read(model, domain=None, fields=None, **kw): return get_odoo().search_read(model, domain=domain or [], fields=fields or [], **kw) def sum_field(model, domain, field): """One-line aggregate sum of `field` over a domain (uses read_group, no row fetch). Robust to the empty match set: a groupby=[] read_group over zero rows makes Odoo return a None aggregate, which the server's XML-RPC layer cannot marshal ("cannot marshal None"). An empty set simply means the sum is 0.0, so we treat that specific Fault as zero.""" try: g = get_odoo().read_group(model, domain=domain, fields=[f'{field}:sum'], groupby=[], lazy=False) except Exception as e: # noqa: BLE001 — narrow check below if 'cannot marshal None' in str(e): return 0.0 raise return (g[0].get(field) or 0.0) if g else 0.0 def distinct_count(model, domain, field): """Distinct count of a (m2o) field over a domain via grouped read_group.""" g = get_odoo().read_group(model, domain=domain, fields=[field], groupby=[field], lazy=False) return sum(1 for r in g if r.get(field))