xiaoxia-saas/packages/adapters/sqlalchemy_impl/title_library_repository.py

"""SQLAlchemy implementation of TitleLibraryRepository."""

from __future__ import annotations

from typing import Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import TitleLibraryModel
from packages.domain.title_library import TitleLibraryItem


class SQLAlchemyTitleLibraryRepository:
    """SQLAlchemy 标题库仓储"""

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

    def list_by_user(
        self,
        user_id: str,
        *,
        category: Optional[str] = None,
        is_active: bool = True,
        skip: int = 0,
        limit: int = 50,
    ) -> list[TitleLibraryItem]:
        query = self.session.query(TitleLibraryModel).filter(
            TitleLibraryModel.user_id == user_id,
            TitleLibraryModel.is_active == is_active,
        )
        if category:
            query = query.filter(TitleLibraryModel.category == category)
        query = query.order_by(TitleLibraryModel.created_at.desc())
        models = query.offset(skip).limit(limit).all()
        return [self._model_to_entity(m) for m in models]

    def get(self, title_id: str, user_id: str) -> Optional[TitleLibraryItem]:
        model = (
            self.session.query(TitleLibraryModel)
            .filter(
                TitleLibraryModel.id == title_id,
                TitleLibraryModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return None
        return self._model_to_entity(model)

    def create(self, item: TitleLibraryItem) -> TitleLibraryItem:
        model = TitleLibraryModel(
            id=item.id,
            user_id=item.user_id,
            name=item.name,
            description=item.description,
            category=item.category,
            text=item.text,
            tags=item.tags,
            usage_count=item.usage_count,
            is_active=item.is_active,
            extra_meta=item.metadata_,
        )
        self.session.add(model)
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def update(self, item: TitleLibraryItem) -> TitleLibraryItem:
        model = (
            self.session.query(TitleLibraryModel)
            .filter(
                TitleLibraryModel.id == item.id,
                TitleLibraryModel.user_id == item.user_id,
            )
            .first()
        )
        if model is None:
            raise ValueError(f"TitleLibraryItem {item.id} not found")
        model.name = item.name
        model.description = item.description
        model.category = item.category
        model.text = item.text
        model.tags = item.tags
        model.is_active = item.is_active
        model.extra_meta = item.metadata_
        self.session.commit()
        self.session.refresh(model)
        return self._model_to_entity(model)

    def delete(self, title_id: str, user_id: str) -> bool:
        model = (
            self.session.query(TitleLibraryModel)
            .filter(
                TitleLibraryModel.id == title_id,
                TitleLibraryModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return False
        model.is_active = False
        self.session.commit()
        return True

    def increment_usage_count(self, title_id: str, user_id: str, increment: int = 1) -> bool:
        """递增标题使用次数。返回是否成功。"""
        from sqlalchemy import func

        model = (
            self.session.query(TitleLibraryModel)
            .filter(
                TitleLibraryModel.id == title_id,
                TitleLibraryModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return False
        model.usage_count = (model.usage_count or 0) + increment
        model.updated_at = func.now()
        self.session.commit()
        return True

    def count_by_user(self, user_id: str, is_active: bool = True) -> int:
        return (
            self.session.query(TitleLibraryModel)
            .filter(
                TitleLibraryModel.user_id == user_id,
                TitleLibraryModel.is_active == is_active,
            )
            .count()
        )

    @staticmethod
    def _model_to_entity(model: TitleLibraryModel) -> TitleLibraryItem:
        return TitleLibraryItem(
            id=model.id,
            user_id=model.user_id,
            name=model.name,
            description=model.description,
            category=model.category,
            text=model.text,
            tags=model.tags or [],
            usage_count=model.usage_count or 0,
            is_active=model.is_active,
            metadata_=model.extra_meta or {},
            created_at=model.created_at,
            updated_at=model.updated_at,
        )