xiaoxia-saas/apps/api/app/services/script_service.py

"""ScriptService — Issue #1795 口播文案库 CRUD.

纯 Service 层封装，routes 直接调用。
"""

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import ScriptModel


class ScriptNotFoundError(Exception):
    """文案不存在或不属于当前用户."""


class ScriptService:
    """口播文案 CRUD."""

    def __init__(self, db: Session) -> None:
        self.db = db

    # ── list ──────────────────────────────────────────────────────────────

    def list_scripts(
        self,
        user_id: str,
        skip: int = 0,
        limit: int = 50,
        tag: Optional[str] = None,
    ) -> tuple[list[ScriptModel], int]:
        """返回 (items, total)."""
        q = self.db.query(ScriptModel).filter(ScriptModel.user_id == user_id)
        if tag:
            # JSON 数组包含查询
            q = q.filter(ScriptModel.tags.contains([tag]))
        total = q.count()
        items = q.order_by(ScriptModel.created_at.desc()).offset(skip).limit(limit).all()
        return items, total

    # ── create ────────────────────────────────────────────────────────────

    def create_script(
        self,
        user_id: str,
        title: str,
        content: str = "",
        segments: list | None = None,
        tags: list | None = None,
    ) -> ScriptModel:
        script = ScriptModel(
            id=str(uuid.uuid4()),
            user_id=user_id,
            title=title,
            content=content,
            segments=segments if segments is not None else [],
            tags=tags if tags is not None else [],
        )
        self.db.add(script)
        self.db.commit()
        self.db.refresh(script)
        return script

    # ── get ───────────────────────────────────────────────────────────────

    def get_script(self, script_id: str, user_id: str) -> ScriptModel:
        script = self.db.query(ScriptModel).filter(ScriptModel.id == script_id, ScriptModel.user_id == user_id).first()
        if script is None:
            raise ScriptNotFoundError(f"Script {script_id} not found")
        return script

    # ── update ────────────────────────────────────────────────────────────

    def update_script(
        self,
        script_id: str,
        user_id: str,
        title: Optional[str] = None,
        content: Optional[str] = None,
        segments: Optional[list] = None,
        tags: Optional[list] = None,
    ) -> ScriptModel:
        script = self.get_script(script_id, user_id)
        if title is not None:
            script.title = title
        if content is not None:
            script.content = content
        if segments is not None:
            script.segments = segments
        if tags is not None:
            script.tags = tags
        script.updated_at = datetime.now(timezone.utc)
        self.db.commit()
        self.db.refresh(script)
        return script

    # ── delete ────────────────────────────────────────────────────────────

    def delete_script(self, script_id: str, user_id: str) -> bool:
        script = self.db.query(ScriptModel).filter(ScriptModel.id == script_id, ScriptModel.user_id == user_id).first()
        if script is None:
            return False
        self.db.delete(script)
        self.db.commit()
        return True