"""Report data builders for all 8 INDL-45 report types. Each builder returns a normalised :class:`ReportResult` — a title, tabular ``columns``/``rows`` used by the PDF/Excel/CSV exporters, and an ``extra`` dict of report-specific structure (summaries, aging buckets) surfaced by the JSON endpoints. Every query is hard-filtered by ``tenant_id`` (from the request's tenant context, never user input) and respects ``deleted_at IS NULL`` on soft-delete models. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone from typing import Any, Optional from uuid import UUID from sqlalchemy import and_, case, func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased from src.apps.auth.models.user import User from src.apps.billing.models.invoice import Invoice from src.apps.billing.models.invoice_payment import InvoicePayment from src.apps.crew_members.models.crew_member import CrewMember from src.apps.memorials.models.memorial import Memorial from src.apps.plots.models.plot import Plot from src.apps.plots.models.plot_type import PlotType from src.apps.records.models.burial_info import BurialInfo from src.apps.records.models.record import Record from src.apps.sales.models.contract import Contract from src.apps.scheduling.models.service_crew_assignment import ServiceCrewAssignment from src.apps.scheduling.models.service_event import ServiceEvent from src.apps.sections.models.section import Section from src.apps.site_admin.models.audit_log import AuditLog from ..constants import ( AUDIT_LOG, BURIAL_REGISTER, CAPACITY, MAX_EXPORT_ROWS, MEMORIAL_STATUS, MONTHLY_SALES, OUTSTANDING_BALANCES, REPORT_TITLES, REVENUE_BY_SECTION, SERVICE_SCHEDULE, ) # Invoices in these statuses are never "owed" even if a stale balance lingers. _CLOSED_INVOICE_STATUSES = ("void", "draft", "paid") _SOLD_PLOT_STATUSES = ("reserved", "occupied") @dataclass class ReportResult: """Normalised report output consumed by both JSON and export endpoints.""" title: str columns: list[str] rows: list[list[Any]] extra: dict[str, Any] = field(default_factory=dict) # Optional multi-table payload: [{"title": str, "columns": [...], "rows": [...]}, ...] # When set, the export renderers emit one section/sheet per entry in addition to # (not instead of) the primary `columns`/`rows` above. tables: Optional[list[dict]] = None def as_json(self) -> dict[str, Any]: data = { "title": self.title, "columns": self.columns, "rows": self.rows, "row_count": len(self.rows), **self.extra, } if self.tables is not None: data["tables"] = self.tables return data # ── helpers ─────────────────────────────────────────────────────────────────── def _today() -> date: return datetime.now(timezone.utc).date() def _dt_range(col, from_date: Optional[date], to_date: Optional[date]) -> list: """Range filters for a *datetime* column, compared on its date part.""" f = [] if from_date is not None: f.append(func.date(col) >= from_date) if to_date is not None: f.append(func.date(col) <= to_date) return f def _d_range(col, from_date: Optional[date], to_date: Optional[date]) -> list: """Range filters for a plain *date* column.""" f = [] if from_date is not None: f.append(col >= from_date) if to_date is not None: f.append(col <= to_date) return f def _money(value) -> float: return round(float(value or 0), 2) def _months_adjacent(a: str, b: str) -> bool: """True when two 'YYYY-MM' labels are consecutive calendar months.""" try: ya, ma = (int(x) for x in a.split("-")) yb, mb = (int(x) for x in b.split("-")) except (ValueError, AttributeError): return False return (yb * 12 + mb) - (ya * 12 + ma) == 1 def _aging_bucket(days_overdue: int) -> str: if days_overdue <= 30: return "0_30_days" if days_overdue <= 60: return "31_60_days" if days_overdue <= 90: return "61_90_days" return "over_90_days" class ReportDataService: """Builds each report's data. All methods are tenant-scoped and read-only.""" async def build( self, db: AsyncSession, report_type: str, *, tenant_id: str, from_date: Optional[date] = None, to_date: Optional[date] = None, section_id: Optional[UUID] = None, ) -> ReportResult: builders = { MONTHLY_SALES: self.monthly_sales, BURIAL_REGISTER: self.burial_register, CAPACITY: self.capacity, REVENUE_BY_SECTION: self.revenue_by_section, OUTSTANDING_BALANCES: self.outstanding_balances, SERVICE_SCHEDULE: self.service_schedule, AUDIT_LOG: self.audit_log, MEMORIAL_STATUS: self.memorial_status, } return await builders[report_type]( db, tenant_id=tenant_id, from_date=from_date, to_date=to_date, section_id=section_id, ) # ── 1. Monthly Sales Summary ──────────────────────────────────────────── async def monthly_sales( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: filters = [Invoice.tenant_id == tenant_id] filters += _dt_range(Invoice.created_at, from_date, to_date) month = func.to_char(Invoice.created_at, "YYYY-MM").label("month") stmt = ( select( month, func.count(Invoice.id).label("count"), func.coalesce(func.sum(Invoice.total_amount), 0).label("total"), func.coalesce(func.sum(Invoice.paid_amount), 0).label("paid"), func.coalesce(func.sum(Invoice.balance_due), 0).label("balance"), ) .where(*filters) .group_by(month) .order_by(month) .limit(MAX_EXPORT_ROWS) ) result = await db.execute(stmt) records = result.all() rows = [ [r.month, r.count, _money(r.total), _money(r.paid), _money(r.balance)] for r in records ] # Plots sold (reserved/occupied), optionally scoped to the requested section. plot_filters = [ Plot.tenant_id == tenant_id, Plot.status.in_(_SOLD_PLOT_STATUSES), ] if section_id is not None: plot_filters.append(Plot.section_id == section_id) plot_filters += _dt_range(Plot.reserved_at, from_date, to_date) plots_sold = ( await db.execute(select(func.count(Plot.id)).where(*plot_filters)) ).scalar_one() total_invoiced = sum(r[2] for r in rows) total_paid = sum(r[3] for r in rows) # Month-over-month growth — only meaningful when the last two rows are # calendar-adjacent months (gaps with no invoices are skipped by grouping). mom_growth_pct = 0.0 if len(rows) >= 2 and _months_adjacent(rows[-2][0], rows[-1][0]): prev, curr = rows[-2][2], rows[-1][2] if prev: mom_growth_pct = round((curr - prev) / prev * 100, 1) # ── Revenue by Plot Type — same sold-plot pattern as revenue_by_section() # used to use, just grouped by PlotType instead of Section. price = func.coalesce(Plot.price_override, PlotType.default_price, 0) pt_filters = list(plot_filters) # tenant + sold status + section_id + date range pt_stmt = ( select( PlotType.name, func.count(Plot.id).label("plots"), func.coalesce(func.sum(price), 0).label("revenue"), ) .select_from(Plot) .join( PlotType, and_(PlotType.id == Plot.plot_type_id, PlotType.tenant_id == tenant_id), isouter=True, ) .where(*pt_filters) .group_by(PlotType.name) .order_by(PlotType.name) ) pt_rows = [ [pt_name or "—", plots, _money(revenue)] for pt_name, plots, revenue in (await db.execute(pt_stmt)).all() ] # ── Agent Performance — attributed via Contract.created_by (the user who # created the closed-sale contract record); rows with no contract or no # created_by roll up under "Unassigned". # `User.full_name` is a Python @property (first_name + " " + last_name), # not a mapped column, so it can't be used inside a SQL expression here. # Grouped by the actual user identity (Contract.created_by / User.id), # not by the computed display name — otherwise two different users who # happen to share a full name would be silently merged into one row. # User.first_name/last_name are also listed since Postgres requires # every column referenced in the SELECT list (agent_label's CASE reads # User.id, User.first_name, User.last_name) to be in the GROUP BY. agent_label = case( (User.id.isnot(None), func.concat(User.first_name, " ", User.last_name)), else_="Unassigned", ).label("agent") agent_filters = [Invoice.tenant_id == tenant_id] agent_filters += _dt_range(Invoice.created_at, from_date, to_date) if section_id is not None: agent_filters.append(Contract.section_id == section_id) agent_stmt = ( select( agent_label, func.count(func.distinct(Contract.id)).label("contracts"), func.coalesce(func.sum(Invoice.total_amount), 0).label("revenue"), func.count(func.distinct(Contract.plot_id)).label("plots_sold"), ) .select_from(Invoice) .join( Contract, and_( Contract.id == Invoice.contract_id, Contract.tenant_id == tenant_id, Contract.deleted_at.is_(None), ), isouter=True, ) .join( User, and_(User.id == Contract.created_by, User.tenant_id == tenant_id), isouter=True, ) .where(*agent_filters) .group_by(Contract.created_by, User.id, User.first_name, User.last_name) .order_by(agent_label) ) agent_rows = [ [agent, contracts, _money(revenue), plots_sold] for agent, contracts, revenue, plots_sold in ( await db.execute(agent_stmt) ).all() ] # ── Payment Method Breakdown — grouped by InvoicePayment.method, joined to # Invoice for tenant scoping and (optionally) Contract for the section filter. method_label = func.coalesce( func.nullif(InvoicePayment.method, ""), "Unspecified" ).label("method") pm_filters = [Invoice.tenant_id == tenant_id] pm_filters += _d_range(InvoicePayment.received_on, from_date, to_date) if section_id is not None: pm_filters.append(Contract.section_id == section_id) pm_stmt = ( select( method_label, func.count(InvoicePayment.id).label("payments"), func.coalesce(func.sum(InvoicePayment.amount), 0).label("amount"), ) .select_from(InvoicePayment) .join( Invoice, and_(Invoice.id == InvoicePayment.invoice_id, Invoice.tenant_id == tenant_id), ) .join( Contract, and_( Contract.id == Invoice.contract_id, Contract.tenant_id == tenant_id, Contract.deleted_at.is_(None), ), isouter=True, ) .where(*pm_filters) .group_by(method_label) .order_by(method_label) ) pm_rows = [ [method, payments, _money(amount)] for method, payments, amount in (await db.execute(pm_stmt)).all() ] return ReportResult( title=REPORT_TITLES[MONTHLY_SALES], columns=["Month", "Invoices", "Total Invoiced", "Paid", "Balance Due"], rows=rows, tables=[ { "title": "Revenue by Plot Type", "columns": ["Plot Type", "Plots Sold", "Revenue"], "rows": pt_rows, }, { "title": "Agent Performance", "columns": ["Agent", "Contracts", "Total Revenue", "Plots Sold"], "rows": agent_rows, }, { "title": "Payment Method Breakdown", "columns": ["Payment Method", "Payments", "Amount Collected"], "rows": pm_rows, }, ], extra={ "summary": { "months": len(rows), "total_invoiced": round(total_invoiced, 2), "total_paid": round(total_paid, 2), "plots_sold": int(plots_sold or 0), "mom_growth_pct": mom_growth_pct, } }, ) # ── 2. Burial Register ────────────────────────────────────────────────── async def burial_register( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: filters = [ Record.tenant_id == tenant_id, Record.deleted_at.is_(None), BurialInfo.interment_date.isnot(None), ] filters += _d_range(BurialInfo.interment_date, from_date, to_date) if section_id is not None: filters.append(Plot.section_id == section_id) stmt = ( select(Record, BurialInfo, Plot, Section) .join(BurialInfo, BurialInfo.record_id == Record.id) .join( Plot, and_(Plot.id == Record.plot_id, Plot.tenant_id == tenant_id), isouter=True, ) .join( Section, and_(Section.id == Plot.section_id, Section.tenant_id == tenant_id), isouter=True, ) .where(*filters) .order_by(BurialInfo.interment_date.asc()) .limit(MAX_EXPORT_ROWS) ) result = await db.execute(stmt) rows = [] for record, burial, plot, section in result.all(): name = " ".join( p for p in [record.first_name, record.last_name] if p ) rows.append([ name, record.date_of_birth.isoformat() if record.date_of_birth else "", record.date_of_death.isoformat() if record.date_of_death else "", record.gender or "", plot.plot_ref if plot else "", section.name if section else "", burial.interment_type or "", burial.interment_date.isoformat() if burial.interment_date else "", burial.interment_time.strftime("%H:%M") if burial.interment_time else "", burial.officiant or "", ]) return ReportResult( title=REPORT_TITLES[BURIAL_REGISTER], columns=[ "Full Name", "Date of Birth", "Date of Death", "Gender", "Plot", "Section", "Interment Type", "Interment Date", "Interment Time", "Officiant", ], rows=rows, extra={"summary": {"total_interments": len(rows)}}, ) # ── 3. Capacity Report ────────────────────────────────────────────────── async def capacity( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: filters = [Plot.tenant_id == tenant_id] if section_id is not None: filters.append(Plot.section_id == section_id) stmt = ( select( Section.id, Section.name, Plot.status, func.count(Plot.id).label("count"), ) .select_from(Plot) .join( Section, and_(Section.id == Plot.section_id, Section.tenant_id == tenant_id), isouter=True, ) .where(*filters) .group_by(Section.id, Section.name, Plot.status) ) result = await db.execute(stmt) # Pivot into per-section status counts. Any status other than the four # knowns still contributes to `total` (and thus to "used" via total-vacant). sections: dict[str, dict[str, Any]] = {} for sec_id, sec_name, status, count in result.all(): key = str(sec_id) if sec_id else "unassigned" sec = sections.setdefault( key, {"name": sec_name or "Unassigned", "vacant": 0, "reserved": 0, "occupied": 0, "hold": 0, "total": 0}, ) if status in ("vacant", "reserved", "occupied", "hold"): sec[status] += count sec["total"] += count # Projected fill: use plots sold in the trailing 365 days as an annual rate. year_ago = _today() - timedelta(days=365) rate_stmt = ( select(Plot.section_id, func.count(Plot.id)) .where( Plot.tenant_id == tenant_id, Plot.status.in_(_SOLD_PLOT_STATUSES), func.date(Plot.reserved_at) >= year_ago, ) .group_by(Plot.section_id) ) annual_rate = { str(sid) if sid else "unassigned": cnt for sid, cnt in (await db.execute(rate_stmt)).all() } rows = [] grand_total = grand_vacant = 0 for key, sec in sorted(sections.items(), key=lambda kv: kv[1]["name"]): total = sec["total"] vacant = sec["vacant"] used = total - vacant # any non-vacant plot counts as used util_pct = round(used / total * 100, 1) if total else 0.0 rate = annual_rate.get(key, 0) if vacant > 0 and rate > 0: years = vacant / rate projected_full = str(_today().year + int(years) + 1) elif vacant == 0 and total > 0: projected_full = "Full" else: projected_full = "N/A" rows.append([ sec["name"], total, vacant, sec["reserved"], sec["occupied"], sec["hold"], util_pct, projected_full, ]) grand_total += total grand_vacant += vacant overall_used = grand_total - grand_vacant overall_util = ( round(overall_used / grand_total * 100, 1) if grand_total else 0.0 ) # ── Plot Detail — per-plot rows, same grain as the Cemetery Map list view. # `Record` is deduplicated to at most one row per plot (Postgres # `DISTINCT ON`) *before* the join, so a plot never appears twice even if # it somehow ends up with more than one live Record — `Plot.record` is # modeled elsewhere as a singular `Optional[Record]` relationship, but # nothing in the schema enforces that 1:1 at the DB level. price = func.coalesce(Plot.price_override, PlotType.default_price, 0) record_dedup_sq = ( select(Record) .where(Record.tenant_id == tenant_id, Record.deleted_at.is_(None)) .distinct(Record.plot_id) .order_by(Record.plot_id, Record.id) ).subquery() RecordDedup = aliased(Record, record_dedup_sq) detail_stmt = ( select(Plot, Section, PlotType, RecordDedup, price.label("price")) .select_from(Plot) .join( Section, and_(Section.id == Plot.section_id, Section.tenant_id == tenant_id), isouter=True, ) .join( PlotType, and_(PlotType.id == Plot.plot_type_id, PlotType.tenant_id == tenant_id), isouter=True, ) .join( RecordDedup, RecordDedup.plot_id == Plot.id, isouter=True, ) .where(*filters) .order_by(Section.name, Plot.plot_ref) .limit(MAX_EXPORT_ROWS) ) detail_rows = [] for plot, section, plot_type, record, plot_price in ( await db.execute(detail_stmt) ).all(): plot_id = plot.label_manual or plot.plot_ref status_display = (plot.status or "").capitalize() if plot.status == "occupied" and record is not None: held_by = " ".join( p for p in [record.first_name, record.last_name] if p ) or "—" elif plot.status in ("reserved", "hold"): held_by = plot.reserved_by or "—" else: held_by = "—" detail_rows.append([ plot_id, section.name if section else "", status_display, plot_type.name if plot_type else "—", _money(plot_price), held_by, ]) return ReportResult( title=REPORT_TITLES[CAPACITY], columns=[ "Section", "Total", "Vacant", "Reserved", "Occupied", "Hold", "Utilisation %", "Projected Full", ], rows=rows, tables=[ { "title": "Plot Detail", "columns": [ "Plot ID", "Section", "Status", "Type", "Price", "Reserved / Interred By", ], "rows": detail_rows, }, ], extra={ "summary": { "total_plots": grand_total, "available": grand_vacant, "utilisation_pct": overall_util, } }, ) # ── 4. Revenue by Section ─────────────────────────────────────────────── async def revenue_by_section( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: """Real invoice-level revenue, grouped/ordered by the contract's section. Section is resolved from the contract's own ``section_id`` when set, falling back to the contract's linked plot's section otherwise — ``Contract.section_id`` is populated on essentially no real contracts in practice (data entry always goes through picking a plot, not a section field), while ``Contract.plot_id`` -> ``Plot.section_id`` is reliably populated whenever a contract is tied to a sold plot. Invoices with no linked contract, or whose contract has neither a section nor a plot with a section, are grouped under "Unassigned" rather than omitted. """ resolved_section_id = func.coalesce(Contract.section_id, Plot.section_id) filters = [Invoice.tenant_id == tenant_id] filters += _dt_range(Invoice.created_at, from_date, to_date) if section_id is not None: filters.append(resolved_section_id == section_id) stmt = ( select(Invoice, Section) .select_from(Invoice) .join( Contract, and_( Contract.id == Invoice.contract_id, Contract.tenant_id == tenant_id, Contract.deleted_at.is_(None), ), isouter=True, ) .join( Plot, and_(Plot.id == Contract.plot_id, Plot.tenant_id == tenant_id), isouter=True, ) .join( Section, and_(Section.id == resolved_section_id, Section.tenant_id == tenant_id), isouter=True, ) .where(*filters) .order_by(Section.name.asc().nullslast(), Invoice.due_date.asc()) .limit(MAX_EXPORT_ROWS) ) result = await db.execute(stmt) rows: list[list[Any]] = [] section_totals: dict[str, dict[str, Any]] = {} total_invoices = 0 total_revenue = 0.0 for invoice, section in result.all(): section_name = section.name if section else "Unassigned" total = _money(invoice.total_amount) paid = _money(invoice.paid_amount) due = _money(invoice.balance_due) rows.append([ invoice.invoice_number, invoice.purchaser_name or "", total, paid, due, invoice.due_date.isoformat() if invoice.due_date else "", invoice.status or "", section_name, ]) total_invoices += 1 total_revenue += total st = section_totals.setdefault( section_name, {"invoice_count": 0, "total": 0.0, "paid": 0.0, "due": 0.0}, ) st["invoice_count"] += 1 st["total"] = round(st["total"] + total, 2) st["paid"] = round(st["paid"] + paid, 2) st["due"] = round(st["due"] + due, 2) return ReportResult( title=REPORT_TITLES[REVENUE_BY_SECTION], columns=[ "Invoice #", "Purchaser", "Total", "Paid", "Due Amount", "Due Date", "Status", "Section", ], rows=rows, extra={ "summary": { "total_invoices": total_invoices, "total_revenue": round(total_revenue, 2), }, "section_totals": section_totals, }, ) # ── 5. Outstanding Balances ───────────────────────────────────────────── async def outstanding_balances( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: filters = [ Invoice.tenant_id == tenant_id, Invoice.balance_due > 0, Invoice.status.notin_(_CLOSED_INVOICE_STATUSES), ] filters += _dt_range(Invoice.created_at, from_date, to_date) stmt = ( select(Invoice) .where(*filters) .order_by(Invoice.due_date.asc().nullslast()) .limit(MAX_EXPORT_ROWS) ) invoices = list((await db.execute(stmt)).scalars().all()) today = _today() buckets = { "0_30_days": {"count": 0, "total_amount": 0.0, "total_due": 0.0}, "31_60_days": {"count": 0, "total_amount": 0.0, "total_due": 0.0}, "61_90_days": {"count": 0, "total_amount": 0.0, "total_due": 0.0}, "over_90_days": {"count": 0, "total_amount": 0.0, "total_due": 0.0}, } invoice_dicts, rows = [], [] total_amount = total_paid = total_due = 0.0 for inv in invoices: total = _money(inv.total_amount) paid = _money(inv.paid_amount) due = _money(inv.balance_due) days_overdue = (today - inv.due_date).days if inv.due_date else 0 days_overdue = max(days_overdue, 0) bucket = _aging_bucket(days_overdue) paid_pct = round(paid / total * 100, 1) if total else 0.0 due_pct = round(due / total * 100, 1) if total else 0.0 buckets[bucket]["count"] += 1 buckets[bucket]["total_amount"] = round( buckets[bucket]["total_amount"] + total, 2 ) buckets[bucket]["total_due"] = round( buckets[bucket]["total_due"] + due, 2 ) total_amount += total total_paid += paid total_due += due invoice_dicts.append({ "invoice_number": inv.invoice_number, "purchaser_name": inv.purchaser_name, "due_date": inv.due_date.isoformat() if inv.due_date else None, "total_amount": total, "paid_amount": paid, "balance_due": due, "paid_pct": paid_pct, "due_pct": due_pct, "aging_bucket": bucket, "days_overdue": days_overdue, }) rows.append([ inv.invoice_number, inv.purchaser_name or "", inv.due_date.isoformat() if inv.due_date else "", total, paid, due, paid_pct, due_pct, bucket.replace("_", " "), days_overdue, ]) collection_rate = ( round(total_paid / total_amount * 100, 1) if total_amount else 0.0 ) return ReportResult( title=REPORT_TITLES[OUTSTANDING_BALANCES], columns=[ "Invoice #", "Purchaser", "Due Date", "Total", "Paid", "Due", "Paid %", "Due %", "Aging Bucket", "Days Overdue", ], rows=rows, extra={ "summary": { "total_invoices": len(invoices), "total_amount": round(total_amount, 2), "total_paid": round(total_paid, 2), "total_due": round(total_due, 2), "collection_rate_pct": collection_rate, }, "aging_buckets": buckets, "invoices": invoice_dicts, }, ) # ── 6. Service Schedule ───────────────────────────────────────────────── async def service_schedule( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: filters = [ ServiceEvent.tenant_id == tenant_id, ServiceEvent.deleted_at.is_(None), ] if from_date is not None or to_date is not None: filters += _d_range(ServiceEvent.scheduled_date, from_date, to_date) else: filters.append(ServiceEvent.scheduled_date >= _today()) if section_id is not None: filters.append(ServiceEvent.section_id == section_id) stmt = ( select(ServiceEvent, Section, Plot) .join( Section, and_(Section.id == ServiceEvent.section_id, Section.tenant_id == tenant_id), isouter=True, ) .join( Plot, and_(Plot.id == ServiceEvent.plot_id, Plot.tenant_id == tenant_id), isouter=True, ) .where(*filters) .order_by( ServiceEvent.scheduled_date.asc(), ServiceEvent.scheduled_time.asc().nullsfirst(), ) .limit(MAX_EXPORT_ROWS) ) result = await db.execute(stmt) service_rows = result.all() # Assigned Crew — a separate lookup query (simpler and lower-risk than # restructuring the join above into a GROUP BY/string_agg). service_ids = [svc.id for svc, _section, _plot in service_rows] crew_by_service: dict[Any, list[str]] = {} if service_ids: crew_stmt = ( select(ServiceCrewAssignment.service_id, CrewMember.name) .select_from(ServiceCrewAssignment) .join( CrewMember, and_( CrewMember.id == ServiceCrewAssignment.crew_member_id, CrewMember.tenant_id == tenant_id, CrewMember.deleted_at.is_(None), ), ) .where( ServiceCrewAssignment.tenant_id == tenant_id, ServiceCrewAssignment.service_id.in_(service_ids), ) ) for service_id, crew_name in (await db.execute(crew_stmt)).all(): crew_by_service.setdefault(service_id, []).append(crew_name) rows = [] for svc, section, plot in service_rows: assigned_crew = ", ".join(crew_by_service.get(svc.id, [])) rows.append([ svc.scheduled_date.isoformat() if svc.scheduled_date else "", svc.scheduled_time.strftime("%H:%M") if svc.scheduled_time else "", svc.service_type or "", svc.decedent_name or "", section.name if section else "", plot.plot_ref if plot else "", svc.family_contact_name or "", svc.family_contact_phone or "", svc.officiant or "", assigned_crew, svc.equipment_needed or "", svc.status or "", ]) return ReportResult( title=REPORT_TITLES[SERVICE_SCHEDULE], columns=[ "Date", "Time", "Service Type", "Decedent", "Section", "Plot", "Family Contact", "Phone", "Officiant", "Assigned Crew", "Equipment", "Status", ], rows=rows, extra={"summary": {"total_services": len(rows)}}, ) # ── 7. Audit Activity Log ─────────────────────────────────────────────── async def audit_log( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: filters = [AuditLog.tenant_id == tenant_id] filters += _dt_range(AuditLog.created_at, from_date, to_date) stmt = ( select(AuditLog, User) .join( User, and_(User.id == AuditLog.user_id, User.tenant_id == tenant_id), isouter=True, ) .where(*filters) .order_by(AuditLog.created_at.desc()) .limit(MAX_EXPORT_ROWS) ) result = await db.execute(stmt) rows = [] for entry, user in result.all(): user_name = user.full_name if user else "System" changed_keys = set() if entry.new_value: changed_keys.update(entry.new_value.keys()) if entry.old_value: changed_keys.update(entry.old_value.keys()) changed = ", ".join(sorted(changed_keys)) rows.append([ entry.created_at.isoformat() if entry.created_at else "", user_name, entry.action, entry.entity_type, str(entry.entity_id) if entry.entity_id else "", changed, ]) return ReportResult( title=REPORT_TITLES[AUDIT_LOG], columns=[ "Timestamp", "User", "Action", "Entity Type", "Entity ID", "Changed Fields", ], rows=rows, extra={"summary": {"total_events": len(rows)}}, ) # ── 8. Memorial Status ────────────────────────────────────────────────── async def memorial_status( self, db, *, tenant_id, from_date=None, to_date=None, section_id=None ) -> ReportResult: # NOTE: deliberately does NOT filter out memorials whose linked Record # has been soft-deleted — the Memorials moderation page (memorial_service.py) # lists memorials the same way, and Memorial.display_name is already # denormalized independent of Record, so excluding these here would only # make this report silently diverge from what staff already see on # /portal/memorials, not prevent any actual PII exposure. filters = [Memorial.tenant_id == tenant_id] filters += _dt_range(Memorial.created_at, from_date, to_date) if section_id is not None: filters.append(Plot.section_id == section_id) stmt = ( select(Memorial, Record, Section) .join( Record, and_(Record.id == Memorial.record_id, Record.tenant_id == tenant_id), isouter=True, ) .join( Plot, and_(Plot.id == Record.plot_id, Plot.tenant_id == tenant_id), isouter=True, ) .join( Section, and_(Section.id == Plot.section_id, Section.tenant_id == tenant_id), isouter=True, ) .where(*filters) .order_by(Memorial.created_at.desc()) .limit(MAX_EXPORT_ROWS) ) result = await db.execute(stmt) rows, published, drafts = [], 0, 0 for memorial, record, section in result.all(): name = memorial.display_name if not name and record: name = " ".join( p for p in [record.first_name, record.last_name] if p ) status = "Published" if memorial.is_published else "Draft" if memorial.is_published: published += 1 else: drafts += 1 rows.append([ name or memorial.slug, status, section.name if section else "", memorial.published_at.isoformat() if memorial.published_at else "", memorial.created_at.isoformat() if memorial.created_at else "", ]) return ReportResult( title=REPORT_TITLES[MEMORIAL_STATUS], columns=["Memorial", "Status", "Section", "Approved At", "Submitted At"], rows=rows, extra={ "summary": { "total": len(rows), "published": published, "draft": drafts, } }, )