xiaoxia-saas/packages/application/video_share/use_cases.py

"""视频分享 Use cases."""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import List, Optional

from packages.application.video_share.commands import (
    CreateShareCommand,
    UpdateShareCommand,
)
from packages.domain.exceptions import NotFoundError
from packages.domain.generated_video import GeneratedVideo
from packages.domain.video_share import VideoShare
from packages.ports.generated_video_repository import GeneratedVideoRepository
from packages.ports.video_share_repository import VideoShareRepositoryPort


class VideoNotFoundError(Exception):
    """视频不存在."""

    pass


class ShareExpiredError(Exception):
    """分享已过期或已撤销."""

    pass


class PasswordRequiredError(Exception):
    """需要访问密码."""

    pass


class InvalidPasswordError(Exception):
    """密码错误."""

    pass


@dataclass
class ShareAccessResult:
    """分享访问结果（验证通过后返回视频信息+分享记录）."""

    share: VideoShare
    video: GeneratedVideo
    password_verified: bool = True


class CreateShareUseCase:
    """创建视频分享链接."""

    def __init__(
        self,
        share_repository: VideoShareRepositoryPort,
        video_repository: GeneratedVideoRepository,
    ) -> None:
        self.share_repo = share_repository
        self.video_repo = video_repository

    def execute(self, command: CreateShareCommand) -> VideoShare:
        # 校验视频存在且属于该用户
        video = self.video_repo.get(command.video_id)
        if video is None:
            raise VideoNotFoundError(f"Video {command.video_id} not found")

        # 用 user_id 校验（视频的user_id需要匹配）
        if hasattr(video, "user_id") and video.user_id and video.user_id != command.user_id:
            raise VideoNotFoundError("Video not found")

        share = VideoShare.create(
            video_id=command.video_id,
            user_id=command.user_id,
            password=command.password,
            expires_at=command.expires_at,
        )
        return self.share_repo.create(share)


class GetShareByTokenUseCase:
    """通过token获取分享信息（不带视频内容，仅元信息）。

    用于分享页加载前判断：是否需要密码、是否过期等。
    """

    def __init__(self, share_repository: VideoShareRepositoryPort) -> None:
        self.share_repo = share_repository

    def execute(self, token: str) -> VideoShare:
        share = self.share_repo.get_by_token(token)
        if share is None:
            raise NotFoundError(f"Share not found: {token}")
        if not share.is_accessible:
            raise ShareExpiredError("Share is not accessible")
        return share


class AccessShareUseCase:
    """访问分享内容（验证密码+返回视频信息+计数浏览量）。"""

    def __init__(
        self,
        share_repository: VideoShareRepositoryPort,
        video_repository: GeneratedVideoRepository,
    ) -> None:
        self.share_repo = share_repository
        self.video_repo = video_repository

    def execute(self, token: str, password: Optional[str] = None) -> ShareAccessResult:
        share = self.share_repo.get_by_token(token)
        if share is None:
            raise NotFoundError(f"Share not found: {token}")
        if not share.is_accessible:
            raise ShareExpiredError("Share is not accessible")

        # 密码校验
        password_verified = True
        if share.has_password:
            if not password:
                raise PasswordRequiredError("Password required")
            if not share.verify_password(password):
                raise InvalidPasswordError("Invalid password")
            password_verified = True

        # 获取视频信息
        video = self.video_repo.get(share.video_id)
        if video is None:
            raise VideoNotFoundError("Video not found")

        # 浏览量+1
        self.share_repo.increment_view(share.id)
        share.view_count += 1

        return ShareAccessResult(share=share, video=video, password_verified=password_verified)


class ListSharesByVideoUseCase:
    """列出某个视频的所有分享记录."""

    def __init__(self, share_repository: VideoShareRepositoryPort) -> None:
        self.share_repo = share_repository

    def execute(self, video_id: str, user_id: str) -> List[VideoShare]:
        return self.share_repo.list_by_video(video_id, user_id)


class ListSharesByUserUseCase:
    """列出用户创建的所有分享记录."""

    def __init__(self, share_repository: VideoShareRepositoryPort) -> None:
        self.share_repo = share_repository

    def execute(self, user_id: str, skip: int = 0, limit: int = 20) -> tuple[List[VideoShare], int]:
        items = self.share_repo.list_by_user(user_id, skip=skip, limit=limit)
        total = self.share_repo.count_by_user(user_id)
        return items, total


class UpdateShareUseCase:
    """更新分享配置（密码、有效期等）."""

    def __init__(self, share_repository: VideoShareRepositoryPort) -> None:
        self.share_repo = share_repository

    def execute(self, command: UpdateShareCommand) -> VideoShare:
        share = self.share_repo.get_by_id(command.share_id, command.user_id)
        if share is None:
            raise NotFoundError(f"Share {command.share_id} not found")

        # password=None表示不修改；空字符串表示清除密码
        if command.password is not None:
            from packages.domain.video_share import _hash_password

            if command.password == "":
                share.password_hash = None
            else:
                share.password_hash = _hash_password(command.password)

        # expires_at=None表示不修改
        if command.expires_at is not None:
            if command.expires_at < datetime.now(timezone.utc):
                raise ValueError("expires_at cannot be in the past")
            share.expires_at = command.expires_at

        return self.share_repo.update(share)


class RevokeShareUseCase:
    """撤销/删除分享."""

    def __init__(self, share_repository: VideoShareRepositoryPort) -> None:
        self.share_repo = share_repository

    def execute(self, share_id: str, user_id: str) -> bool:
        share = self.share_repo.get_by_id(share_id, user_id)
        if share is None:
            raise NotFoundError(f"Share {share_id} not found")
        return self.share_repo.delete(share_id, user_id)


class RecordShareDownloadUseCase:
    """记录分享下载（下载量+1）."""

    def __init__(self, share_repository: VideoShareRepositoryPort) -> None:
        self.share_repo = share_repository

    def execute(self, token: str, password: Optional[str] = None) -> None:
        share = self.share_repo.get_by_token(token)
        if share is None:
            raise NotFoundError(f"Share not found: {token}")
        if not share.is_accessible:
            raise ShareExpiredError("Share is not accessible")

        # 密码校验
        if share.has_password:
            if not password or not share.verify_password(password):
                raise InvalidPasswordError("Invalid password")

        self.share_repo.increment_download(share.id)