"""Bounded owner reports derived from one read-only tenant snapshot. No synthetic history, contact matching, accounting inference or reporting writes. See docs/REPORTING-DEFINITIONS.md for the metric contract. """ import csv import hashlib import io import json from datetime import date, datetime from typing import Literal from zoneinfo import ZoneInfo from fastapi import HTTPException, Request, Response from pydantic import Field, ValidationError, model_validator from .models import Strict from .store import many, now from .customer_history import csv_text from .current_positions import calendar_date, open_job, open_quote_state LONDON = ZoneInfo('Europe/London') VERSION = 'report-1' MAX_INPUT = 20_000 MAX_JOBS = 1_000 MAX_FACTS = 12_000 MAX_EXPORT = 5_000 STAGES = ('new', 'site_visit', 'accepted', 'scheduled', 'in_progress', 'completed', 'on_hold', 'cancelled') def fail(message, status=422): raise HTTPException(status, message) def day_only(value): return calendar_date(value) def instant(value): try: parsed = datetime.fromisoformat(value.replace('Z', '+00:00')) return parsed if parsed.tzinfo is not None else None except (ValueError, TypeError, AttributeError): return None def uk_day(value): parsed = instant(value) return parsed.astimezone(LONDON).date().isoformat() if parsed else None def service_key(value): return hashlib.sha256(value.encode()).hexdigest() if value and value.strip() else 'unclassified' class Filters(Strict): start: str end: str service: str = Field(default='', max_length=64) customer_id: str = Field(default='', pattern=r'^(?:[a-f0-9]{32})?$') stage: str = Field(default='', max_length=30) tag: str = Field(default='', max_length=40) archive: Literal['all', 'active', 'archived'] = 'all' @model_validator(mode='after') def valid_period(self): if not day_only(self.start) or not day_only(self.end) or not 0 <= (date.fromisoformat(self.end) - date.fromisoformat(self.start)).days <= 365: raise ValueError('Choose an inclusive period of at most 366 UK dates.') if self.stage not in ('', 'unknown', *STAGES): raise ValueError('Choose a saved job stage.') return self class ExportFilters(Filters): dimension: Literal['all', 'service', 'customer', 'team'] = 'all' group_id: str = Field(default='', max_length=64) metric: str = Field(default='', max_length=80) kind: Literal['summary', 'records'] = 'summary' snapshot: str = Field(pattern=r'^[a-f0-9]{64}$') def metric(label, scope, unit, definition): return dict(label=label, scope=scope, unit=unit, definition=definition) METRICS = { 'jobs_created': metric('Jobs created', 'period', 'jobs', 'One exact job by its UK creation date. Archived jobs remain unless filtered out.'), 'completed_jobs': metric('Completed jobs with a saved date', 'period', 'jobs', 'Current completed status and saved completion date in the range. No invoice/payment-date inference.'), 'quote_revisions_created': metric('Quote revisions created', 'period', 'revisions', 'All stored quote revisions by UK creation date, including superseded revisions. Not a count of unique jobs.'), 'issued_invoice_value_pence': metric('Issued invoice value', 'period', 'pence', 'Currently issued gross invoice pence by actual UK issue date. Draft/void/cancelled invoices and payment requests add nothing. Not recognised revenue.'), 'recorded_receipts_pence': metric('Recorded net receipts', 'period', 'pence', 'Original receipts by paid-on date, excluding retained error reversals. Allocations are not extra receipts. Not bank-cleared income.'), 'person_minutes': metric('Recorded person time', 'period', 'person_minutes', 'Active saved time entries by work date. Two people working one hour record 120 person-minutes. Not elapsed job time.'), 'manual_person_minutes': metric('Manual person time', 'period', 'person_minutes', 'Active manual time entries by work date. Saved minutes, no estimated hours.'), 'field_person_minutes': metric('Converted field person time', 'period', 'person_minutes', 'Active time entries from reviewed field conversion by work date. Raw timers and earlier archived generations are not counted again.'), 'internal_labour_cost_pence': metric('Internal labour cost', 'period', 'pence', 'Active saved entry costs by work date. Retained historical rates, half-up rounding per entry. Not payroll or profit.'), 'received_material_cost_pence': metric('Received material cost', 'period', 'pence', 'Active received material rows by received date, including converted field materials once. No document selling prices or unreceived costs.'), 'field_logs_reviewed': metric('Reviewed field logs', 'period', 'logs', 'Active reviewed logs by work date. Authors are accounts, not inferred scheduling resources; no private log bodies.'), 'field_logs_unreviewed': metric('Unreviewed field logs', 'period', 'logs', 'Active unreviewed logs by work date. These do not establish time, costs or staff performance.'), 'field_sessions_unconverted': metric('Unconverted field sessions', 'period', 'sessions', 'Sessions created in the period whose current state is not converted. No extrapolated elapsed time or labour cost.'), 'open_jobs': metric('Open jobs now', 'current', 'jobs', 'Current open, unarchived jobs; independent of the selected period.'), 'open_quote_estimate_pence': metric('Open quote estimate now', 'current', 'pence', 'Each chain’s current unexpired draft/shared revision on an open unarchived job. Separate chains may be alternative offers. Potential work, not expected income; independent of the period.'), 'issued_invoice_balance_pence': metric('Invoice balance now', 'current', 'pence', 'Currently issued invoice pence less all unreversed receipts; independent of the period. No additional balance for stages or payment requests.'), 'void_invoice_value_pence': metric('Excluded void invoice value', 'exclusions', 'pence', 'Currently void invoices by original UK issue date. Excluded from issued value; not a count of voids occurring in the period or a credit-note ledger.'), 'reversed_receipt_pence': metric('Excluded reversed receipts', 'exclusions', 'pence', 'Reversed recording errors by original paid-on date. Reversal timestamp retained. Restates original period; not a cash refund.'), 'receipt_reversals_recorded': metric('Receipt reversals recorded', 'period', 'reversals', 'Recording-error reversals by UK reversal timestamp, separate from the original paid-on period. Not refunds.'), 'archived_person_minutes': metric('Excluded archived person time', 'exclusions', 'person_minutes', 'Archived saved time entries by work date, including replaced field-conversion generations. Not active recorded time.'), 'archived_labour_cost_pence': metric('Excluded archived labour cost', 'exclusions', 'pence', 'Archived saved labour cost by work date; not active internal labour cost.'), 'archived_material_cost_pence': metric('Excluded archived received materials', 'exclusions', 'pence', 'Archived received material costs by received date; not active material cost.'), 'unknown_dates': metric('Unknown relevant dates', 'data_quality', 'date_fields', 'Missing or invalid relevant date fields on matching records. One record may have several. Not placed in any period; each date role is shown.'), 'cohort_created': metric('1 · Created jobs', 'cohort', 'jobs', 'Filtered jobs created in the period, observed at generation. One eligible population for every funnel step.'), 'cohort_shared': metric('2 · With a shared quote', 'cohort', 'jobs', 'Eligible jobs with retained sharing evidence on a quote revision by the observation point. Count each job once.'), 'cohort_accepted': metric('3 · Shared and accepted quote', 'cohort', 'jobs', 'Subset with sharing and acceptance evidence on the same quote revision by observation. No inferred acceptance from job stage.'), 'cohort_issued': metric('4 · Linked issued invoice', 'cohort', 'jobs', 'Subset with a currently issued invoice linked to an eligible shared/accepted quote revision. Void and direct invoices do not satisfy this step.'), 'cohort_receipt': metric('5 · Recorded receipt', 'cohort', 'jobs', 'Subset with a positive unreversed receipt on an eligible linked invoice by observation. Not necessarily fully paid.'), 'cohort_direct_invoice': metric('Cohort jobs with direct invoices', 'data_quality', 'jobs', 'Cohort jobs with a currently issued invoice without a quote link. Disclosed as a separate path, never bridged with synthetic jobs.'), } FUNNEL = ('cohort_created', 'cohort_shared', 'cohort_accepted', 'cohort_issued', 'cohort_receipt') TEAM_METRICS = ('person_minutes', 'manual_person_minutes', 'field_person_minutes', 'internal_labour_cost_pence', 'archived_person_minutes', 'archived_labour_cost_pence') def bounded(db, table, sql): # Fixed internal table names only. Reject rather than silently drop input rows. if db.execute('SELECT COUNT(*) FROM ' + table).fetchone()[0] > MAX_INPUT: fail(f'This business exceeds the local reporting limit of {MAX_INPUT:,} {table} records. No partial totals or file were produced.') return many(db, sql) def build_report(db, filters, *, generated_at=None): generated = generated_at or now() observed = instant(generated) if observed is None: raise ValueError('Reporting needs an offset-aware observation timestamp.') current_day = uk_day(generated) inside = lambda day: bool(day) and filters.start <= day <= filters.end occurred = lambda stamp: instant(stamp) is not None and instant(stamp) <= observed customers = {c['id']: c for c in bounded(db, 'customers', "SELECT c.id,c.name,COALESCE(r.state,'active') AS state,r.canonical_id FROM customers c LEFT JOIN customer_records r ON r.customer_id=c.id ORDER BY c.id")} resources = {r['id']: r for r in bounded(db, 'team_resources', 'SELECT id,name,archived FROM team_resources ORDER BY id')} jobs = bounded(db, 'jobs', "SELECT j.id,j.customer_id,j.service,j.status,j.created_at,COALESCE(json_extract(w.content,'$.tags'),'[]') AS tags,COALESCE(json_extract(w.content,'$.archived'),0) AS archived,COALESCE(json_extract(w.content,'$.completed_date'),'') AS completed_date FROM jobs j LEFT JOIN job_workflow w ON w.job_id=j.id ORDER BY j.id") services = {} tags = set() for j in jobs: j['tags'] = sorted(set(json.loads(j['tags']))) j['service_key'] = service_key(j['service']) j['service_label'] = j['service'] if j['service_key'] != 'unclassified' else 'Unclassified' j['stage'] = j['status'] if j['status'] in STAGES else 'unknown' services[j['service_key']] = j['service_label'] tags.update(j['tags']) if filters.customer_id and filters.customer_id not in customers: fail('Customer not found in this business.', 404) if filters.service and filters.service not in services: fail('Saved service not found in this business.', 404) if filters.tag and filters.tag not in tags: fail('Saved tag not found in this business.', 404) jobs = [j for j in jobs if (not filters.customer_id or j['customer_id'] == filters.customer_id) and (not filters.service or j['service_key'] == filters.service) and (not filters.tag or filters.tag in j['tags']) and (not filters.stage or j['stage'] == filters.stage) and (filters.archive == 'all' or bool(j['archived']) == (filters.archive == 'archived'))] if len(jobs) > MAX_JOBS: fail(f'Narrow the job filters to at most {MAX_JOBS:,} matching jobs. Current positions also need these jobs; changing dates alone does not narrow them. No partial report was produced.') selected = {j['id']: j for j in jobs} records = [] def add(key, row, value=1, *, kind='job', date_role='', event_date='', timestamp='', **extra): j = selected.get(row.get('job_id', row['id'])) if j is None: return c = customers[j['customer_id']] records.append(dict(metric=key, kind=kind, id=row['id'], job_id=j['id'], service_key=j['service_key'], service_label=j['service_label'], customer_id=c['id'], customer_name=c['name'], customer_state=c['state'], stage=j['stage'], job_archived=bool(j['archived']), tags=j['tags'], value=value, event_date=event_date or '', timestamp=timestamp or '', date_role=date_role, status=row.get('status', row.get('state', '')), **extra)) if len(records) > MAX_FACTS: fail(f'Narrow the filters or period: this report exceeds {MAX_FACTS:,} supporting metric rows. No partial totals were produced.') def period_day(row, value, role, kind, timestamp=False): day = uk_day(value) if timestamp else day_only(value) if day is None: add('unknown_dates', row, kind=kind, date_role=role) return day for j in jobs: created = period_day(j, j['created_at'], 'job creation', 'job', True) if inside(created): add('jobs_created', j, event_date=created, timestamp=j['created_at']) add('cohort_created', j, event_date=created, timestamp=j['created_at']) if j['status'] == 'completed': completed = period_day(j, j['completed_date'], 'saved completion', 'job') if inside(completed): add('completed_jobs', j, event_date=completed) if open_job(j['status'], j['archived']): add('open_jobs', j) projection = "json_extract(snapshot,'$.customer.id') AS original_customer_id,json_extract(snapshot,'$.customer.name') AS original_customer_name,json_extract(snapshot,'$.service') AS original_service" quotes = bounded(db, 'quotes', 'SELECT q.id,q.job_id,q.status,q.reference,q.total_pence,q.created_at,' + projection + ",l.current,l.shared_at,l.accepted_at,json_extract(snapshot,'$.valid_until') AS valid_until FROM quotes q JOIN quote_links l ON l.quote_id=q.id ORDER BY q.id") invoices = bounded(db, 'invoices', 'SELECT id,quote_id,job_id,status,reference,total_pence,created_at,issued_at,' + projection + ' FROM invoices ORDER BY id') payments = bounded(db, 'payments', 'SELECT p.id,p.invoice_id,p.amount_pence,p.created_at,d.paid_on,r.created_at AS reversed_at FROM payments p JOIN payment_details d ON d.payment_id=p.id LEFT JOIN payment_reversals r ON r.payment_id=p.id ORDER BY p.id') by_invoice = {i['id']: i for i in invoices if i['job_id'] in selected} net = {i: 0 for i in by_invoice} receipt_evidence = set() identity = lambda row: {k: row[k] or '' for k in ('original_customer_id', 'original_customer_name', 'original_service')} for p in payments: i = by_invoice.get(p['invoice_id']) if not i: continue p['job_id'] = i['job_id'] paid_day = period_day(p, p['paid_on'], 'receipt paid-on', 'receipt') extra = dict(reference=i['reference'], invoice_id=i['id'], reversed_at=p['reversed_at'] or '', **identity(i)) if not p['reversed_at']: net[i['id']] += p['amount_pence'] if paid_day and paid_day <= current_day and occurred(p['created_at']): receipt_evidence.add(i['id']) if inside(paid_day): add('reversed_receipt_pence' if p['reversed_at'] else 'recorded_receipts_pence', p, p['amount_pence'], kind='receipt', event_date=paid_day, timestamp=p['created_at'], **extra) if p['reversed_at']: reversed_day = period_day(p, p['reversed_at'], 'receipt reversal', 'receipt', True) if inside(reversed_day): add('receipt_reversals_recorded', p, kind='receipt', event_date=reversed_day, timestamp=p['reversed_at'], **extra) shared, accepted, issued, received, direct = set(), set(), set(), set(), set() accepted_ids = set() for q in quotes: j = selected.get(q['job_id']) if j is None: continue extra = dict(reference=q['reference'], current_revision=bool(q['current']), **identity(q)) created = period_day(q, q['created_at'], 'quote revision creation', 'quote', True) if inside(created): add('quote_revisions_created', q, kind='quote', event_date=created, timestamp=q['created_at'], **extra) estimate_state = open_quote_state(q, j['status'], j['archived'], current_day) if estimate_state == 'unknown_expiry': add('unknown_dates', q, kind='quote', date_role='quote expiry', **extra) elif estimate_state == 'included': add('open_quote_estimate_pence', q, q['total_pence'], kind='quote', **extra) if occurred(q['shared_at']): shared.add(j['id']) if occurred(q['accepted_at']): accepted.add(j['id']) accepted_ids.add(q['id']) for i in by_invoice.values(): extra = dict(reference=i['reference'], **identity(i)) if i['status'] == 'issued': add('issued_invoice_balance_pence', i, i['total_pence'] - net[i['id']], kind='invoice', **extra) if i['status'] in ('issued', 'void'): issue_day = period_day(i, i['issued_at'], 'invoice issue', 'invoice', True) if inside(issue_day): add('issued_invoice_value_pence' if i['status'] == 'issued' else 'void_invoice_value_pence', i, i['total_pence'], kind='invoice', event_date=issue_day, timestamp=i['issued_at'], **extra) if i['status'] == 'issued' and occurred(i['issued_at']): if i['quote_id'] is None: direct.add(i['job_id']) if i['quote_id'] in accepted_ids: issued.add(i['job_id']) if i['id'] in receipt_evidence and net[i['id']] > 0: received.add(i['job_id']) cohorts = {key: set() for key in FUNNEL} for j in jobs: if not inside(uk_day(j['created_at'])): continue cohorts['cohort_created'].add(j['id']) for key, group in zip(FUNNEL[1:], (shared, accepted, issued, received)): if j['id'] in group: cohorts[key].add(j['id']) add(key, j, event_date=uk_day(j['created_at']), timestamp=j['created_at']) if j['id'] in direct: add('cohort_direct_invoice', j, event_date=uk_day(j['created_at'])) times = bounded(db, 'time_entries', "SELECT t.id,t.job_id,t.resource_id,t.work_date,t.duration_minutes,t.cost_pence,t.archived,json_extract(t.resource_snapshot,'$.name') AS resource_name,l.session_id,l.active AS link_active FROM time_entries t LEFT JOIN field_time_links l ON l.time_id=t.id ORDER BY t.id") for t in times: if t['job_id'] not in selected: continue work_day = period_day(t, t['work_date'], 'recorded work', 'time') if not inside(work_day): continue resource = resources[t['resource_id']] extra = dict(resource_id=t['resource_id'], resource_name=t['resource_name'] or 'Unknown recorded name', resource_current_name=resource['name'], resource_archived=bool(resource['archived']), source_id=t['session_id'] or '', origin='field' if t['session_id'] else 'manual') if t['archived']: add('archived_person_minutes', t, t['duration_minutes'], kind='time', event_date=work_day, **extra) add('archived_labour_cost_pence', t, t['cost_pence'], kind='time', event_date=work_day, **extra) else: add('person_minutes', t, t['duration_minutes'], kind='time', event_date=work_day, **extra) add('field_person_minutes' if t['session_id'] else 'manual_person_minutes', t, t['duration_minutes'], kind='time', event_date=work_day, **extra) add('internal_labour_cost_pence', t, t['cost_pence'], kind='time', event_date=work_day, **extra) materials = bounded(db, 'job_materials', 'SELECT id,job_id,status,received_date,total_pence,archived FROM job_materials ORDER BY id') for m in materials: if m['job_id'] not in selected or m['status'] != 'received': continue received_day = period_day(m, m['received_date'], 'material received', 'material') if inside(received_day): add('archived_material_cost_pence' if m['archived'] else 'received_material_cost_pence', m, m['total_pence'], kind='material', event_date=received_day) logs = bounded(db, 'field_logs', 'SELECT id,job_id,kind,work_date,reviewed,archived,author_id,membership_id FROM field_logs ORDER BY id') for log in logs: if log['job_id'] not in selected or log['archived']: continue work_day = period_day(log, log['work_date'], 'field log work', 'field_log') if inside(work_day): add('field_logs_reviewed' if log['reviewed'] else 'field_logs_unreviewed', log, kind='field_log', event_date=work_day, author_id=log['author_id'], membership_id=log['membership_id']) sessions = bounded(db, 'field_sessions', 'SELECT id,job_id,state,created_at FROM field_sessions ORDER BY id') for s in sessions: if s['job_id'] not in selected or s['state'] == 'converted': continue created = period_day(s, s['created_at'], 'field session creation', 'field_session', True) if inside(created): add('field_sessions_unconverted', s, kind='field_session', event_date=created, timestamp=s['created_at']) records.sort(key=lambda r: (r['metric'], r['event_date'], r['kind'], r['id'], r['date_role'])) totals = {key: 0 for key in METRICS} groups = {key: {} for key in ('service', 'customer', 'team')} def group(dimension, gid, label, state=''): return groups[dimension].setdefault(gid, dict(id=gid, label=label, state=state, totals={key: 0 for key in (TEAM_METRICS if dimension == 'team' else METRICS)}, worked_job_ids=[])) for j in jobs: group('service', j['service_key'], j['service_label']) c = customers[j['customer_id']] group('customer', c['id'], c['name'], c['state']) for r in records: totals[r['metric']] += r['value'] group('service', r['service_key'], r['service_label'])['totals'][r['metric']] += r['value'] group('customer', r['customer_id'], r['customer_name'], r['customer_state'])['totals'][r['metric']] += r['value'] if r['metric'] in TEAM_METRICS: g = group('team', r['resource_id'], r['resource_current_name'], 'archived' if r['resource_archived'] else 'active') g['totals'][r['metric']] += r['value'] if r['metric'] == 'person_minutes' and r['job_id'] not in g['worked_job_ids']: g['worked_job_ids'].append(r['job_id']) denominator = len(cohorts['cohort_created']) funnel = [dict(metric=k, count=len(cohorts[k]), population=denominator, percent_tenths=(len(cohorts[k]) * 1000 + denominator // 2) // denominator if denominator else None) for k in FUNNEL] stage_counts, tag_counts = {}, {} for jid in sorted(cohorts['cohort_created']): j = selected[jid] stage_counts.setdefault(j['stage'], []).append(jid) for tag in j['tags']: tag_counts.setdefault(tag, []).append(jid) result = dict(definition_version=VERSION, generated_at=generated, observation_day=current_day, period=dict(start=filters.start, end=filters.end, timezone='Europe/London', inclusive=True), filters=filters.model_dump(), matched_jobs=len(jobs), definitions=METRICS, totals=totals, records=records, groups={d: sorted(g.values(), key=lambda x: (x['label'], x['id'])) for d, g in groups.items()}, funnel=funnel, cohort_stages=stage_counts, cohort_tags=tag_counts, options=dict(services=[dict(id=k, label=v) for k, v in sorted(services.items(), key=lambda x: (x[1], x[0]))], customers=list(customers.values()), team=list(resources.values()), tags=sorted(tags), stages=[*STAGES, 'unknown']), limits=dict(input_rows_per_table=MAX_INPUT, matching_jobs=MAX_JOBS, supporting_metric_rows=MAX_FACTS, export_rows=MAX_EXPORT)) result['snapshot'] = hashlib.sha256(json.dumps({k: v for k, v in result.items() if k != 'generated_at'}, sort_keys=True, ensure_ascii=False, separators=(',', ':')).encode()).hexdigest() return result def query(request, model): keys = [k for k, _ in request.query_params.multi_items()] if len(keys) != len(set(keys)): fail('Use each reporting filter once.') try: return model.model_validate(dict(request.query_params)) except ValidationError: fail('Check the reporting dates and filters. Use valid exact IDs, saved filter values and an inclusive range of at most 366 UK dates.') def select_records(report, dimension, group_id, key): if key not in METRICS: fail('Choose a defined reporting metric.') if dimension == 'team' and key not in TEAM_METRICS: fail('Team reporting does not attribute invoices or receipts to people.') field = {'service': 'service_key', 'customer': 'customer_id', 'team': 'resource_id'}.get(dimension) return [r for r in report['records'] if r['metric'] == key and (dimension != 'team' or 'resource_id' in r) and (not group_id or r.get(field) == group_id)] CSV_FIELDS = ('row_type', 'business_id', 'definition_version', 'generated_at_utc', 'period_start_uk', 'period_end_uk', 'filters_json', 'snapshot', 'metric', 'metric_label', 'scope', 'unit', 'definition', 'group_type', 'group_id', 'group_label', 'value', 'kind', 'id', 'job_id', 'service_label', 'customer_id', 'customer_name', 'customer_state', 'stage', 'job_archived', 'tags', 'event_date', 'timestamp', 'date_role', 'status', 'reference', 'invoice_id', 'original_customer_id', 'original_customer_name', 'original_service', 'resource_id', 'resource_name', 'resource_current_name', 'resource_archived', 'origin', 'source_id', 'reversed_at', 'author_id', 'membership_id') def export_csv(report, choice, business_id): if choice.snapshot != report['snapshot']: fail('Records or reporting context changed. Apply the filters again before exporting this screen.', 409) options = {'service': report['options']['services'], 'customer': report['options']['customers'], 'team': report['options']['team']} if choice.group_id and (choice.dimension == 'all' or not any(r['id'] == choice.group_id for r in options[choice.dimension])): fail('Reporting group not found in this business.', 404) if choice.kind == 'records' and not choice.metric: fail('Select one metric for a supporting-record export.') if choice.metric and choice.metric not in METRICS: fail('Choose a defined reporting metric.') if choice.dimension == 'team' and choice.metric and choice.metric not in TEAM_METRICS: fail('Team reporting does not attribute invoice value or receipts to people.') output = [] base = dict(row_type=choice.kind, business_id=business_id, definition_version=VERSION, generated_at_utc=report['generated_at'], period_start_uk=choice.start, period_end_uk=choice.end, filters_json=json.dumps(report['filters'], ensure_ascii=False, sort_keys=True), snapshot=report['snapshot'], group_type=choice.dimension, group_id=choice.group_id) if choice.kind == 'records': output = [{**base, **r} for r in select_records(report, choice.dimension, choice.group_id, choice.metric)] else: groups = report['groups'].get(choice.dimension, [dict(id='', label='All matching jobs', totals=report['totals'])]) for g in groups: if choice.group_id and g['id'] != choice.group_id: continue for key, value in g['totals'].items(): if not choice.metric or key == choice.metric: output.append({**base, 'group_id': g['id'], 'group_label': g['label'], 'metric': key, 'value': value}) if len(output) > MAX_EXPORT: fail(f'Narrow this export to at most {MAX_EXPORT:,} rows. No partial file was produced.') # An empty result still carries the selected context and explicit metric definition. if not output: output = [{**base, 'row_type': 'empty', 'metric': choice.metric}] stream = io.StringIO(newline='') writer = csv.DictWriter(stream, fieldnames=CSV_FIELDS, quoting=csv.QUOTE_ALL, lineterminator='\r\n', extrasaction='ignore') writer.writeheader() for row in output: definition = METRICS.get(row.get('metric'), {}) row.update(metric_label=definition.get('label', ''), scope=definition.get('scope', ''), unit=definition.get('unit', ''), definition=definition.get('definition', '')) writer.writerow({k: csv_text(json.dumps(v, ensure_ascii=False) if isinstance(v, (list, dict)) else str(v).lower() if isinstance(v, bool) else v) for k, v in row.items()}) return stream.getvalue().encode('utf-8-sig') def mount_reporting(app, store, session): @app.get('/api/reports/activity') def activity(request: Request): s = session(request) filters = query(request, Filters) with store.tenant(s['business_id']) as db: db.execute('PRAGMA query_only=ON') db.execute('BEGIN') return build_report(db, filters) @app.get('/api/reports/activity/export') def export(request: Request): s = session(request) choice = query(request, ExportFilters) filters = Filters.model_validate(choice.model_dump(include=set(Filters.model_fields))) with store.tenant(s['business_id']) as db: db.execute('PRAGMA query_only=ON') db.execute('BEGIN') body = export_csv(build_report(db, filters), choice, s['business_id']) return Response(body, media_type='text/csv; charset=utf-8', headers={'Content-Disposition': 'attachment; filename="activity-report.csv"'})