xiaoxia-saas/packages/domain/video_share.py

"""视频分享领域实体."""

from __future__ import annotations

import secrets
from dataclasses import dataclass, field
from datetime import datetime, timezone
from hashlib import sha256
from typing import Optional
from uuid import uuid4


def _hash_password(password: str) -> str:
    """简单密码哈希（SHA-256 + salt）。

    分享链接的密码保护安全级别要求不高，
    使用简单的加盐哈希即可，避免引入bcrypt等重依赖。
    """
    if not password:
        return ""
    salt = "xiaoxia_share_salt"
    return sha256(f"{salt}:{password}".encode()).hexdigest()


def generate_share_token(length: int = 12) -> str:
    """生成URL友好的分享token."""
    # 使用urlsafe的base64，但去掉可能引起歧义的字符
    alphabet = "abcdefghjkmnpqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789"
    return "".join(secrets.choice(alphabet) for _ in range(length))


@dataclass(slots=True)
class VideoShare:
    """视频分享记录."""

    id: str
    video_id: str
    user_id: str
    share_token: str
    password_hash: Optional[str] = None
    expires_at: Optional[datetime] = None
    view_count: int = 0
    download_count: int = 0
    is_active: bool = True
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    @classmethod
    def create(
        cls,
        video_id: str,
        user_id: str,
        *,
        password: Optional[str] = None,
        expires_at: Optional[datetime] = None,
    ) -> "VideoShare":
        if not video_id.strip():
            raise ValueError("video_id cannot be empty")
        if not user_id.strip():
            raise ValueError("user_id cannot be empty")
        if expires_at and expires_at < datetime.now(timezone.utc):
            raise ValueError("expires_at cannot be in the past")

        return cls(
            id=uuid4().hex,
            video_id=video_id.strip(),
            user_id=user_id.strip(),
            share_token=generate_share_token(),
            password_hash=_hash_password(password) if password else None,
            expires_at=expires_at,
        )

    @property
    def has_password(self) -> bool:
        """是否设置了访问密码."""
        return bool(self.password_hash)

    @property
    def is_expired(self) -> bool:
        """是否已过期."""
        if not self.expires_at:
            return False
        return datetime.now(timezone.utc) > self.expires_at

    @property
    def is_accessible(self) -> bool:
        """是否可以访问（活跃且未过期）."""
        return self.is_active and not self.is_expired

    def verify_password(self, password: str) -> bool:
        """验证访问密码."""
        if not self.password_hash:
            return True  # 没有密码直接通过
        if not password:
            return False
        return _hash_password(password) == self.password_hash

    def increment_view_count(self) -> None:
        """浏览次数+1."""
        self.view_count += 1

    def increment_download_count(self) -> None:
        """下载次数+1."""
        self.download_count += 1

    def revoke(self) -> None:
        """撤销分享."""
        self.is_active = False