203671a7bf
Add the experience service with sites through analytics/AI hooks, migrations through 0011, and phase docs/manifests marking the track complete. Co-authored-by: Cursor <cursoragent@cursor.com>
240 lines
8.4 KiB
Python
240 lines
8.4 KiB
Python
"""Localization & media-ref query specifications — Phase 11.5."""
|
|
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.localization import (
|
|
ExperienceLocalization,
|
|
ExperienceMediaRef,
|
|
SiteLocaleBinding,
|
|
)
|
|
from app.models.types import (
|
|
LocalizationStatus,
|
|
LocalizationTarget,
|
|
MediaKind,
|
|
MediaRefStatus,
|
|
TextDirection,
|
|
)
|
|
|
|
SITE_LOCALE_BINDING_SORT_FIELDS = frozenset(
|
|
{
|
|
"created_at",
|
|
"updated_at",
|
|
"locale_code",
|
|
"sort_order",
|
|
"is_default",
|
|
"text_direction",
|
|
}
|
|
)
|
|
LOCALIZATION_SORT_FIELDS = frozenset(
|
|
{
|
|
"created_at",
|
|
"updated_at",
|
|
"locale",
|
|
"status",
|
|
"title",
|
|
"text_direction",
|
|
"target_type",
|
|
}
|
|
)
|
|
MEDIA_REF_SORT_FIELDS = frozenset(
|
|
{
|
|
"created_at",
|
|
"updated_at",
|
|
"kind",
|
|
"sort_order",
|
|
"locale",
|
|
"status",
|
|
"storage_file_ref",
|
|
}
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class SiteLocaleBindingListSpec:
|
|
workspace_id: UUID | None = None
|
|
site_id: UUID | None = None
|
|
locale_profile_id: UUID | None = None
|
|
locale_code: str | None = None
|
|
text_direction: TextDirection | str | None = None
|
|
is_default: bool | None = None
|
|
is_active: bool | None = None
|
|
q: str | None = None
|
|
sort_by: str = "sort_order"
|
|
sort_dir: str = "asc"
|
|
|
|
def filter_clauses(self) -> list[ColumnElement]:
|
|
clauses: list[ColumnElement] = []
|
|
if self.workspace_id is not None:
|
|
clauses.append(SiteLocaleBinding.workspace_id == self.workspace_id)
|
|
if self.site_id is not None:
|
|
clauses.append(SiteLocaleBinding.site_id == self.site_id)
|
|
if self.locale_profile_id is not None:
|
|
clauses.append(
|
|
SiteLocaleBinding.locale_profile_id == self.locale_profile_id
|
|
)
|
|
if self.locale_code:
|
|
clauses.append(SiteLocaleBinding.locale_code == self.locale_code)
|
|
if self.text_direction is not None:
|
|
direction = (
|
|
self.text_direction
|
|
if isinstance(self.text_direction, TextDirection)
|
|
else TextDirection(str(self.text_direction))
|
|
)
|
|
clauses.append(SiteLocaleBinding.text_direction == direction)
|
|
if self.is_default is not None:
|
|
clauses.append(SiteLocaleBinding.is_default.is_(self.is_default))
|
|
if self.is_active is not None:
|
|
clauses.append(SiteLocaleBinding.is_active.is_(self.is_active))
|
|
if self.q:
|
|
term = f"%{self.q.strip()}%"
|
|
clauses.append(SiteLocaleBinding.locale_code.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 SITE_LOCALE_BINDING_SORT_FIELDS
|
|
else "sort_order"
|
|
)
|
|
column = getattr(SiteLocaleBinding, 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 LocalizationListSpec:
|
|
workspace_id: UUID | None = None
|
|
target_type: LocalizationTarget | str | None = None
|
|
target_id: UUID | None = None
|
|
locale: str | None = None
|
|
text_direction: TextDirection | str | None = None
|
|
status: LocalizationStatus | 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(ExperienceLocalization.workspace_id == self.workspace_id)
|
|
if self.target_type is not None:
|
|
target = (
|
|
self.target_type
|
|
if isinstance(self.target_type, LocalizationTarget)
|
|
else LocalizationTarget(str(self.target_type))
|
|
)
|
|
clauses.append(ExperienceLocalization.target_type == target)
|
|
if self.target_id is not None:
|
|
clauses.append(ExperienceLocalization.target_id == self.target_id)
|
|
if self.locale:
|
|
clauses.append(ExperienceLocalization.locale == self.locale)
|
|
if self.text_direction is not None:
|
|
direction = (
|
|
self.text_direction
|
|
if isinstance(self.text_direction, TextDirection)
|
|
else TextDirection(str(self.text_direction))
|
|
)
|
|
clauses.append(ExperienceLocalization.text_direction == direction)
|
|
if self.status is not None:
|
|
status = (
|
|
self.status
|
|
if isinstance(self.status, LocalizationStatus)
|
|
else LocalizationStatus(str(self.status))
|
|
)
|
|
clauses.append(ExperienceLocalization.status == status)
|
|
if self.q:
|
|
term = f"%{self.q.strip()}%"
|
|
clauses.append(
|
|
or_(
|
|
ExperienceLocalization.locale.ilike(term),
|
|
ExperienceLocalization.title.ilike(term),
|
|
ExperienceLocalization.summary.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 LOCALIZATION_SORT_FIELDS else "created_at"
|
|
)
|
|
column = getattr(ExperienceLocalization, 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 MediaRefListSpec:
|
|
workspace_id: UUID | None = None
|
|
site_id: UUID | None = None
|
|
page_id: UUID | None = None
|
|
owner_type: str | None = None
|
|
owner_id: UUID | None = None
|
|
kind: MediaKind | str | None = None
|
|
locale: str | None = None
|
|
status: MediaRefStatus | str | None = None
|
|
q: str | None = None
|
|
sort_by: str = "sort_order"
|
|
sort_dir: str = "asc"
|
|
|
|
def filter_clauses(self) -> list[ColumnElement]:
|
|
clauses: list[ColumnElement] = []
|
|
if self.workspace_id is not None:
|
|
clauses.append(ExperienceMediaRef.workspace_id == self.workspace_id)
|
|
if self.site_id is not None:
|
|
clauses.append(ExperienceMediaRef.site_id == self.site_id)
|
|
if self.page_id is not None:
|
|
clauses.append(ExperienceMediaRef.page_id == self.page_id)
|
|
if self.owner_type:
|
|
clauses.append(ExperienceMediaRef.owner_type == self.owner_type)
|
|
if self.owner_id is not None:
|
|
clauses.append(ExperienceMediaRef.owner_id == self.owner_id)
|
|
if self.kind is not None:
|
|
kind = (
|
|
self.kind
|
|
if isinstance(self.kind, MediaKind)
|
|
else MediaKind(str(self.kind))
|
|
)
|
|
clauses.append(ExperienceMediaRef.kind == kind)
|
|
if self.locale:
|
|
clauses.append(ExperienceMediaRef.locale == self.locale)
|
|
if self.status is not None:
|
|
status = (
|
|
self.status
|
|
if isinstance(self.status, MediaRefStatus)
|
|
else MediaRefStatus(str(self.status))
|
|
)
|
|
clauses.append(ExperienceMediaRef.status == status)
|
|
if self.q:
|
|
term = f"%{self.q.strip()}%"
|
|
clauses.append(
|
|
or_(
|
|
ExperienceMediaRef.storage_file_ref.ilike(term),
|
|
ExperienceMediaRef.alt_text.ilike(term),
|
|
ExperienceMediaRef.caption.ilike(term),
|
|
ExperienceMediaRef.locale.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 MEDIA_REF_SORT_FIELDS else "sort_order"
|
|
)
|
|
column = getattr(ExperienceMediaRef, sort_key)
|
|
order = desc(column) if self.sort_dir.lower() != "asc" else asc(column)
|
|
return stmt.order_by(order)
|