"""Filtering, search and sorting specifications — Phase 11.10 analytics.""" from dataclasses import dataclass from uuid import UUID from sqlalchemy import asc, desc, or_ from app.models.analytics import ( ExperienceAnalyticsReportDefinition, ExperienceAnalyticsSnapshot, ) REPORT_DEFINITION_SORT = frozenset( {"created_at", "updated_at", "code", "name", "status"} ) SNAPSHOT_SORT = frozenset( {"created_at", "updated_at", "status", "period_start", "period_end"} ) @dataclass(frozen=True, slots=True) class AnalyticsListSpec: workspace_id: UUID | None = None site_id: UUID | None = None report_id: UUID | None = None status: str | None = None q: str | None = None sort_by: str = "created_at" sort_dir: str = "desc" model = None sort_fields = frozenset() def filter_clauses(self): m = self.model clauses = [] for field in ("workspace_id", "site_id", "report_id", "status"): value = getattr(self, field) if value is not None and hasattr(m, field): clauses.append(getattr(m, field) == value) if self.q: term = f"%{self.q.strip()}%" searchable = [ getattr(m, name).ilike(term) for name in ("code", "name", "period_start", "period_end") if hasattr(m, name) ] if searchable: clauses.append(or_(*searchable)) return clauses def apply(self, stmt): for clause in self.filter_clauses(): stmt = stmt.where(clause) direction = desc if self.sort_dir.lower() == "desc" else asc sort_col = getattr(self.model, self.sort_by, self.model.created_at) if self.sort_by not in self.sort_fields: sort_col = self.model.created_at return stmt.order_by(direction(sort_col)) @dataclass(frozen=True, slots=True) class AnalyticsReportDefinitionListSpec(AnalyticsListSpec): model = ExperienceAnalyticsReportDefinition sort_fields = REPORT_DEFINITION_SORT @dataclass(frozen=True, slots=True) class AnalyticsSnapshotListSpec(AnalyticsListSpec): model = ExperienceAnalyticsSnapshot sort_fields = SNAPSHOT_SORT