"""Filtering, search and sorting specifications — Phase 11.10 AI hooks.""" from dataclasses import dataclass from uuid import UUID from sqlalchemy import asc, desc, or_ from app.models.ai_hooks import ( ExperienceAiContentDispatch, ExperienceAiContentHookRegistration, ) HOOK_REGISTRATION_SORT = frozenset( {"created_at", "updated_at", "code", "name", "status", "task_kind"} ) DISPATCH_SORT = frozenset({"created_at", "updated_at", "status", "task"}) @dataclass(frozen=True, slots=True) class AiHooksListSpec: workspace_id: UUID | None = None site_id: UUID | None = None page_id: UUID | None = None registration_id: UUID | None = None status: str | None = None task_kind: 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", "page_id", "registration_id", "status", "task_kind", ): 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", "task", "context_ref") 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 AiContentHookRegistrationListSpec(AiHooksListSpec): model = ExperienceAiContentHookRegistration sort_fields = HOOK_REGISTRATION_SORT @dataclass(frozen=True, slots=True) class AiContentDispatchListSpec(AiHooksListSpec): model = ExperienceAiContentDispatch sort_fields = DISPATCH_SORT