"""Author RAG Chatbot SaaS — Admin API Pydantic Schemas. R-029: ALL admin API endpoints MUST use Pydantic BaseModel schemas (NOT raw dict). R-010: Password change MUST validate via Pydantic schema. """ from typing import Literal from pydantic import BaseModel, EmailStr, Field, field_validator # ─── Authentication ──────────────────────────────────────────────────────────── class PasswordChangeRequest(BaseModel): """R-010: Password change via validated schema.""" current_password: str = Field(min_length=1) new_password: str = Field(min_length=8, max_length=128) @field_validator("new_password") @classmethod def validate_new_password(cls, v: str) -> str: """R-009: Enforce password complexity.""" if not any(c.isupper() for c in v): raise ValueError("Password must contain at least one uppercase letter") if not any(c.islower() for c in v): raise ValueError("Password must contain at least one lowercase letter") if not any(c.isdigit() for c in v): raise ValueError("Password must contain at least one digit") return v # ─── Widget Configuration ───────────────────────────────────────────────────── class WidgetConfigUpdate(BaseModel): """Widget appearance and behavior settings.""" bot_name: str | None = Field(None, max_length=100) welcome_message: str | None = Field(None, max_length=500) theme: str | None = Field(None, pattern=r"^(midnight|ocean|forest|sunset|minimal)$") position: str | None = Field(None, pattern=r"^(bottom-right|bottom-left)$") auto_open_delay: int | None = Field(None, ge=0, le=60) is_active: bool | None = None # ─── Author Profile ─────────────────────────────────────────────────────────── class ProfileUpdate(BaseModel): """Author profile update fields.""" full_name: str | None = Field(None, max_length=255) website: str | None = Field(None, max_length=500) bio: str | None = Field(None, max_length=2000) timezone: str | None = Field(None, max_length=100) class PersonalityUpdate(BaseModel): """Bot personality configuration.""" response_style: Literal["balanced", "formal", "casual", "enthusiastic"] | None = None fallback_message: str | None = Field(None, max_length=500) out_of_scope_message: str | None = Field(None, max_length=500) class NotificationUpdate(BaseModel): """Notification preferences.""" weekly_digest: bool | None = None token_alerts: bool | None = None new_conversation: bool | None = None subscription_expiry: bool | None = None # ─── Smart Links ────────────────────────────────────────────────────────────── class SmartLinkUpdate(BaseModel): """R-031: URL validation on smart link updates.""" buy_url: str | None = Field(None, max_length=2000) preview_url: str | None = Field(None, max_length=2000) @field_validator("buy_url", "preview_url") @classmethod def validate_url_scheme(cls, v: str | None) -> str | None: """R-031: Validate URL scheme to prevent SSRF.""" if v is None or v == "": return v if not v.startswith(("https://", "http://")): raise ValueError("URL must start with https:// or http://") return v # ─── Book URL Import ────────────────────────────────────────────────────────── class ImportURLRequest(BaseModel): """Request body for URL import preview endpoint.""" url: str force_refresh: bool = False class ImportURLConfirmRequest(BaseModel): """Confirm import with optionally overridden metadata.""" url: str title: str = "" author: str = "" description: str = "" cover_url: str = "" isbn: str = "" genre: str = "" class BookActiveRequest(BaseModel): """Activate or deactivate a book for widget/RAG.""" is_active: bool class BookReorderRequest(BaseModel): """Ordered list of book IDs for widget display.""" ordered_ids: list[str] = Field(min_length=1) # ─── Q&A ────────────────────────────────────────────────────────────────────── class QACreateRequest(BaseModel): """Custom Q&A training data entry.""" question: str = Field(min_length=1, max_length=500) answer: str = Field(min_length=1, max_length=2000) book_id: str | None = None priority: int = Field(default=0, ge=0, le=100) category: str | None = Field(None, max_length=50) match_threshold: float = Field(default=0.85, ge=0.5, le=1.0) class QAUpdateRequest(BaseModel): """Custom Q&A update (partial).""" question: str | None = Field(None, min_length=1, max_length=500) answer: str | None = Field(None, min_length=1, max_length=2000) priority: int | None = Field(None, ge=0, le=100) category: str | None = Field(None, max_length=50) match_threshold: float | None = Field(None, ge=0.5, le=1.0) # ─── Chat Annotations ───────────────────────────────────────────────────────── class AnnotateRequest(BaseModel): """Admin annotation on a chat message.""" annotation: str = Field(min_length=1, max_length=2000) class FlagRequest(BaseModel): """Flag a chat message for review.""" flag_type: Literal["spam", "quality", "escalation"] | None = None # ─── Visitor Interactions ───────────────────────────────────────────────────── class TrackClickRequest(BaseModel): """Track purchase/preview link click.""" session_id: str link_type: str = "purchase" book_id: str | None = None class FeedbackRequest(BaseModel): """Thumbs up/down on a bot message.""" session_id: str message_index: int = Field(ge=0) reaction: Literal["up", "down"] class RateSessionRequest(BaseModel): """Star rating for a chat session.""" session_id: str rating: int = Field(ge=1, le=5) # ─── Publishing Help ─────────────────────────────────────────────────────────── _VALID_PLATFORM_IDS = frozenset({ "amazon", "apple_books", "google_books", "barnes_noble", "kobo", "goodreads", "thriftbooks", "abebooks", "bookshop", "open_library", "bookshop", "walmart", "goodreads", "thriftbooks", "abebooks", }) class ContactRequest(BaseModel): """General support contact from the admin panel.""" category: Literal["general", "billing", "technical", "publishing"] = "general" message: str = Field(min_length=1, max_length=2000) class PublishHelpRequest(BaseModel): """Author request for help publishing on missing platforms.""" book_title: str = Field(min_length=1, max_length=500) book_id: str | None = None platform_ids: list[str] = Field(min_length=1, max_length=10) message: str = Field(default="", max_length=2000) source_url: str | None = Field(default=None, max_length=2000) @field_validator("book_id") @classmethod def validate_book_id_uuid(cls, v: str | None) -> str | None: """R-030: book_id must be a valid UUID when provided.""" if v is None: return v import uuid try: uuid.UUID(v) except ValueError as exc: raise ValueError("book_id must be a valid UUID") from exc return v @field_validator("platform_ids") @classmethod def validate_platform_ids(cls, v: list[str]) -> list[str]: """Ensure platform IDs are known retailer channels.""" invalid = [pid for pid in v if pid not in _VALID_PLATFORM_IDS] if invalid: raise ValueError(f"Unknown platform(s): {', '.join(invalid)}") return v # ─── Announcements ──────────────────────────────────────────────────────────── class AnnouncementRequest(BaseModel): """Author announcement message.""" message: str = Field(min_length=1, max_length=2000) # ─── SuperAdmin ─────────────────────────────────────────────────────────────── class GrantByEmailRequest(BaseModel): """SuperAdmin: grant access by email.""" author_email: EmailStr plan: str = "monthly" token_budget: int = Field(default=500000, ge=10000) notes: str = ""