TorbatYar/backend/services/experience/app/specifications/publishing.py

"""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)