"""视频分享领域实体.""" 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