"""Publishing / SEO / PWA list specifications — Phase 11.7.""" from __future__ import annotations from dataclasses import dataclass from uuid import UUID from sqlalchemy import Select, asc, desc, or_ from sqlalchemy.sql import ColumnElement from app.models.publishing import ( DomainEdgeBinding, PublishRelease, PwaManifest, RobotsShell, SeoMetadata, SitemapShell, ) from app.models.types import ( DomainEdgeStatus, PublishReleaseStatus, PublishTarget, PwaStatus, RobotsStatus, SeoStatus, SeoTarget, SitemapStatus, ) PUBLISH_SORT = frozenset({"created_at", "updated_at", "completed_at", "status"}) EDGE_SORT = frozenset({"created_at", "updated_at", "host", "status"}) SEO_SORT = frozenset({"created_at", "updated_at", "title", "status", "target_type"}) SITEMAP_SORT = frozenset({"created_at", "updated_at", "status"}) ROBOTS_SORT = frozenset({"created_at", "updated_at", "status"}) PWA_SORT = frozenset({"created_at", "updated_at", "name", "status"}) def _enum_clause(column, value, enum_cls): if value is None: return None parsed = value if isinstance(value, enum_cls) else enum_cls(str(value)) return column == parsed @dataclass(frozen=True, slots=True) class PublishReleaseListSpec: workspace_id: UUID | None = None target_type: PublishTarget | str | None = None target_id: UUID | None = None status: PublishReleaseStatus | str | None = None q: str | None = None sort_by: str = "created_at" sort_dir: str = "desc" def filter_clauses(self) -> list[ColumnElement]: clauses: list[ColumnElement] = [] if self.workspace_id is not None: clauses.append(PublishRelease.workspace_id == self.workspace_id) c = _enum_clause(PublishRelease.target_type, self.target_type, PublishTarget) if c is not None: clauses.append(c) if self.target_id is not None: clauses.append(PublishRelease.target_id == self.target_id) c = _enum_clause(PublishRelease.status, self.status, PublishReleaseStatus) if c is not None: clauses.append(c) if self.q: term = f"%{self.q.strip()}%" clauses.append( or_(PublishRelease.notes.ilike(term), PublishRelease.error_message.ilike(term)) ) return clauses def apply(self, stmt: Select) -> Select: clauses = self.filter_clauses() if clauses: stmt = stmt.where(*clauses) sort_key = self.sort_by if self.sort_by in PUBLISH_SORT else "created_at" column = getattr(PublishRelease, sort_key) order = desc(column) if self.sort_dir.lower() != "asc" else asc(column) return stmt.order_by(order) @dataclass(frozen=True, slots=True) class DomainEdgeBindingListSpec: workspace_id: UUID | None = None site_id: UUID | None = None site_domain_id: UUID | None = None status: DomainEdgeStatus | str | None = None q: str | None = None sort_by: str = "created_at" sort_dir: str = "desc" def filter_clauses(self) -> list[ColumnElement]: clauses: list[ColumnElement] = [] if self.workspace_id is not None: clauses.append(DomainEdgeBinding.workspace_id == self.workspace_id) if self.site_id is not None: clauses.append(DomainEdgeBinding.site_id == self.site_id) if self.site_domain_id is not None: clauses.append(DomainEdgeBinding.site_domain_id == self.site_domain_id) c = _enum_clause(DomainEdgeBinding.status, self.status, DomainEdgeStatus) if c is not None: clauses.append(c) if self.q: term = f"%{self.q.strip()}%" clauses.append( or_( DomainEdgeBinding.host.ilike(term), DomainEdgeBinding.dns_status_ref.ilike(term), DomainEdgeBinding.ssl_cert_ref.ilike(term), ) ) return clauses def apply(self, stmt: Select) -> Select: clauses = self.filter_clauses() if clauses: stmt = stmt.where(*clauses) sort_key = self.sort_by if self.sort_by in EDGE_SORT else "created_at" column = getattr(DomainEdgeBinding, sort_key) order = desc(column) if self.sort_dir.lower() != "asc" else asc(column) return stmt.order_by(order) @dataclass(frozen=True, slots=True) class SeoMetadataListSpec: workspace_id: UUID | None = None target_type: SeoTarget | str | None = None target_id: UUID | None = None status: SeoStatus | str | None = None q: str | None = None sort_by: str = "created_at" sort_dir: str = "desc" def filter_clauses(self) -> list[ColumnElement]: clauses: list[ColumnElement] = [] if self.workspace_id is not None: clauses.append(SeoMetadata.workspace_id == self.workspace_id) c = _enum_clause(SeoMetadata.target_type, self.target_type, SeoTarget) if c is not None: clauses.append(c) if self.target_id is not None: clauses.append(SeoMetadata.target_id == self.target_id) c = _enum_clause(SeoMetadata.status, self.status, SeoStatus) if c is not None: clauses.append(c) if self.q: term = f"%{self.q.strip()}%" clauses.append( or_( SeoMetadata.title.ilike(term), SeoMetadata.description.ilike(term), SeoMetadata.canonical_url.ilike(term), ) ) return clauses def apply(self, stmt: Select) -> Select: clauses = self.filter_clauses() if clauses: stmt = stmt.where(*clauses) sort_key = self.sort_by if self.sort_by in SEO_SORT else "created_at" column = getattr(SeoMetadata, sort_key) order = desc(column) if self.sort_dir.lower() != "asc" else asc(column) return stmt.order_by(order) @dataclass(frozen=True, slots=True) class SitemapShellListSpec: workspace_id: UUID | None = None site_id: UUID | None = None status: SitemapStatus | str | None = None q: str | None = None sort_by: str = "created_at" sort_dir: str = "desc" def filter_clauses(self) -> list[ColumnElement]: clauses: list[ColumnElement] = [] if self.workspace_id is not None: clauses.append(SitemapShell.workspace_id == self.workspace_id) if self.site_id is not None: clauses.append(SitemapShell.site_id == self.site_id) c = _enum_clause(SitemapShell.status, self.status, SitemapStatus) if c is not None: clauses.append(c) if self.q: term = f"%{self.q.strip()}%" clauses.append(SitemapShell.storage_file_ref.ilike(term)) return clauses def apply(self, stmt: Select) -> Select: clauses = self.filter_clauses() if clauses: stmt = stmt.where(*clauses) sort_key = self.sort_by if self.sort_by in SITEMAP_SORT else "created_at" column = getattr(SitemapShell, sort_key) order = desc(column) if self.sort_dir.lower() != "asc" else asc(column) return stmt.order_by(order) @dataclass(frozen=True, slots=True) class RobotsShellListSpec: workspace_id: UUID | None = None site_id: UUID | None = None status: RobotsStatus | str | None = None q: str | None = None sort_by: str = "created_at" sort_dir: str = "desc" def filter_clauses(self) -> list[ColumnElement]: clauses: list[ColumnElement] = [] if self.workspace_id is not None: clauses.append(RobotsShell.workspace_id == self.workspace_id) if self.site_id is not None: clauses.append(RobotsShell.site_id == self.site_id) c = _enum_clause(RobotsShell.status, self.status, RobotsStatus) if c is not None: clauses.append(c) if self.q: term = f"%{self.q.strip()}%" clauses.append(RobotsShell.content_text.ilike(term)) return clauses def apply(self, stmt: Select) -> Select: clauses = self.filter_clauses() if clauses: stmt = stmt.where(*clauses) sort_key = self.sort_by if self.sort_by in ROBOTS_SORT else "created_at" column = getattr(RobotsShell, sort_key) order = desc(column) if self.sort_dir.lower() != "asc" else asc(column) return stmt.order_by(order) @dataclass(frozen=True, slots=True) class PwaManifestListSpec: workspace_id: UUID | None = None site_id: UUID | None = None status: PwaStatus | str | None = None q: str | None = None sort_by: str = "created_at" sort_dir: str = "desc" def filter_clauses(self) -> list[ColumnElement]: clauses: list[ColumnElement] = [] if self.workspace_id is not None: clauses.append(PwaManifest.workspace_id == self.workspace_id) if self.site_id is not None: clauses.append(PwaManifest.site_id == self.site_id) c = _enum_clause(PwaManifest.status, self.status, PwaStatus) if c is not None: clauses.append(c) if self.q: term = f"%{self.q.strip()}%" clauses.append( or_( PwaManifest.name.ilike(term), PwaManifest.short_name.ilike(term), PwaManifest.start_url.ilike(term), ) ) return clauses def apply(self, stmt: Select) -> Select: clauses = self.filter_clauses() if clauses: stmt = stmt.where(*clauses) sort_key = self.sort_by if self.sort_by in PWA_SORT else "created_at" column = getattr(PwaManifest, sort_key) order = desc(column) if self.sort_dir.lower() != "asc" else asc(column) return stmt.order_by(order)