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

"""视频分享 SQLAlchemy Repository 实现."""

from __future__ import annotations

from datetime import UTC, datetime
from typing import Optional

from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.models import VideoShareModel
from packages.domain.video_share import VideoShare
from packages.ports.video_share_repository import VideoShareRepositoryPort


def _model_to_domain(model: VideoShareModel) -> VideoShare:
    return VideoShare(
        id=model.id,
        video_id=model.video_id,
        user_id=model.user_id,
        share_token=model.share_token,
        password_hash=model.password_hash,
        expires_at=model.expires_at,
        view_count=model.view_count or 0,
        download_count=model.download_count or 0,
        is_active=model.is_active if model.is_active is not None else True,
        created_at=model.created_at,
        updated_at=model.updated_at,
    )


class SQLAlchemyVideoShareRepository(VideoShareRepositoryPort):
    def __init__(self, session: Session):
        self.session = session

    def create(self, share: VideoShare) -> VideoShare:
        model = VideoShareModel(
            id=share.id,
            video_id=share.video_id,
            user_id=share.user_id,
            share_token=share.share_token,
            password_hash=share.password_hash,
            expires_at=share.expires_at,
            view_count=share.view_count,
            download_count=share.download_count,
            is_active=share.is_active,
            created_at=share.created_at,
            updated_at=share.updated_at,
        )
        self.session.add(model)
        self.session.commit()
        return share

    def get_by_token(self, token: str) -> Optional[VideoShare]:
        model = self.session.query(VideoShareModel).filter(VideoShareModel.share_token == token).first()
        if model is None:
            return None
        return _model_to_domain(model)

    def get_by_id(self, share_id: str, user_id: str) -> Optional[VideoShare]:
        model = (
            self.session.query(VideoShareModel)
            .filter(
                VideoShareModel.id == share_id,
                VideoShareModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return None
        return _model_to_domain(model)

    def list_by_video(self, video_id: str, user_id: str) -> list[VideoShare]:
        models = (
            self.session.query(VideoShareModel)
            .filter(
                VideoShareModel.video_id == video_id,
                VideoShareModel.user_id == user_id,
            )
            .order_by(VideoShareModel.created_at.desc())
            .all()
        )
        return [_model_to_domain(m) for m in models]

    def list_by_user(self, user_id: str, skip: int = 0, limit: int = 20) -> list[VideoShare]:
        models = (
            self.session.query(VideoShareModel)
            .filter(VideoShareModel.user_id == user_id)
            .order_by(VideoShareModel.created_at.desc())
            .offset(skip)
            .limit(limit)
            .all()
        )
        return [_model_to_domain(m) for m in models]

    def count_by_user(self, user_id: str) -> int:
        return self.session.query(VideoShareModel).filter(VideoShareModel.user_id == user_id).count()

    def update(self, share: VideoShare) -> VideoShare:
        model = self.session.query(VideoShareModel).filter(VideoShareModel.id == share.id).first()
        if model is None:
            return share
        model.password_hash = share.password_hash
        model.expires_at = share.expires_at
        model.is_active = share.is_active
        model.view_count = share.view_count
        model.download_count = share.download_count
        model.updated_at = datetime.now(UTC)
        self.session.add(model)
        self.session.commit()
        return share

    def delete(self, share_id: str, user_id: str) -> bool:
        model = (
            self.session.query(VideoShareModel)
            .filter(
                VideoShareModel.id == share_id,
                VideoShareModel.user_id == user_id,
            )
            .first()
        )
        if model is None:
            return False
        model.is_active = False
        model.updated_at = datetime.now(UTC)
        self.session.add(model)
        self.session.commit()
        return True

    def increment_view(self, share_id: str) -> None:
        self.session.query(VideoShareModel).filter(VideoShareModel.id == share_id).update(
            {
                "view_count": VideoShareModel.view_count + 1,
                "updated_at": datetime.now(UTC),
            },
            synchronize_session=False,
        )
        self.session.commit()

    def increment_download(self, share_id: str) -> None:
        self.session.query(VideoShareModel).filter(VideoShareModel.id == share_id).update(
            {
                "download_count": VideoShareModel.download_count + 1,
                "updated_at": datetime.now(UTC),
            },
            synchronize_session=False,
        )
        self.session.commit()