fix: 移除share_token冗余索引（unique已自动创建唯一索引）

op.create_table中Column(index=True)会自动创建索引，
后面再op.create_index就重复了。按项目规范，单列索引
通过Column(index=True)声明，去掉重复的create_index。

alembic/versions/050_video_shares.py

@@ -0,0 +1,48 @@
+"""#P3-2 - 视频分享表 video_shares
+
+Revision ID: 050
+Revises: 049
+Create Date: 2026-07-22
+
+Changes:
+1. 新建 video_shares 表，支持视频匿名分享链接
+2. share_token 唯一索引，用于公开分享URL
+3. 支持密码保护、有效期、浏览/下载计数
+"""
+
+import sqlalchemy as sa
+
+from alembic import context, op
+
+revision = "050_video_shares"
+down_revision = "049_wechat_login_phone"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if context.get_context().dialect.name == "postgresql":
+        # 检查表是否已存在（幂等）
+        result = conn.execute(sa.text("SELECT to_regclass('public.video_shares')"))
+        if result.scalar() is not None:
+            return
+
+    op.create_table(
+        "video_shares",
+        sa.Column("id", sa.String(32), primary_key=True),
+        sa.Column("video_id", sa.String(32), nullable=False, index=True),
+        sa.Column("user_id", sa.String(32), nullable=False, index=True),
+        sa.Column("share_token", sa.String(16), nullable=False, unique=True),
+        sa.Column("password_hash", sa.String(255), nullable=True),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("view_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("download_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.true()),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("video_shares")

apps/api/app/api/router.py

@@ -11,6 +11,7 @@ from app.api.routes.health import router as health_check_router
 from app.api.routes.ingest_jobs import router as ingest_jobs_router
 from app.api.routes.internal_render import router as internal_render_router
 from app.api.routes.projects import router as projects_router
+from app.api.routes.share import router as share_router
 from app.api.routes.subscription import router as subscription_router
 from app.api.routes.tags import router as tags_router
 from app.api.routes.task_center import router as task_center_router
@@ -104,6 +105,10 @@ api_router.include_router(
     videos_router,
     tags=["VideoCenter"],
 )
+api_router.include_router(
+    share_router,
+    tags=["Share"],
+)
 api_router.include_router(
     duplication_router,
     prefix="/duplication",

apps/api/app/api/routes/share.py

@@ -0,0 +1,298 @@
+"""视频分享 API 路由."""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from app.api.routes._helpers import format_utc_datetime
+from app.auth import AuthenticatedUser, get_current_user
+from app.core.storage import OSSStorageService, get_storage_service
+from app.dependencies import get_db_session, get_generated_video_repository
+from app.schemas.video_share import (
+    CreateShareRequest,
+    ShareAccessResponse,
+    ShareListResponse,
+    ShareMetaResponse,
+    ShareResponse,
+    UpdateShareRequest,
+    VerifySharePasswordRequest,
+)
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy.orm import Session
+
+from packages.adapters.sqlalchemy_impl.video_share_repository import (
+    SQLAlchemyVideoShareRepository,
+)
+from packages.application.video_share.commands import (
+    CreateShareCommand,
+    UpdateShareCommand,
+)
+from packages.application.video_share.use_cases import (
+    AccessShareUseCase,
+    CreateShareUseCase,
+    GetShareByTokenUseCase,
+    InvalidPasswordError,
+    ListSharesByUserUseCase,
+    ListSharesByVideoUseCase,
+    NotFoundError,
+    PasswordRequiredError,
+    RecordShareDownloadUseCase,
+    RevokeShareUseCase,
+    ShareExpiredError,
+    UpdateShareUseCase,
+    VideoNotFoundError,
+)
+from packages.ports.generated_video_repository import GeneratedVideoRepository
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+def _get_share_repository(
+    session: Session = Depends(get_db_session),
+) -> SQLAlchemyVideoShareRepository:
+    return SQLAlchemyVideoShareRepository(session)
+
+
+def _to_share_response(share) -> ShareResponse:
+    return ShareResponse(
+        id=share.id,
+        video_id=share.video_id,
+        share_token=share.share_token,
+        has_password=share.has_password,
+        expires_at=share.expires_at,
+        view_count=share.view_count,
+        download_count=share.download_count,
+        is_active=share.is_active,
+        created_at=format_utc_datetime(share.created_at),
+        updated_at=format_utc_datetime(share.updated_at),
+    )
+
+
+# ── 用户侧：创建/管理分享 ──────────────────────────────────
+
+
+@router.post("/videos/{video_id}/share", response_model=ShareResponse)
+def create_share(
+    video_id: str,
+    request: CreateShareRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+    video_repo: GeneratedVideoRepository = Depends(get_generated_video_repository),
+) -> ShareResponse:
+    """为视频创建分享链接."""
+    use_case = CreateShareUseCase(share_repo, video_repo)
+    try:
+        share = use_case.execute(
+            CreateShareCommand(
+                video_id=video_id,
+                user_id=authenticated_user.user.id,
+                password=request.password,
+                expires_at=request.expires_at,
+            )
+        )
+    except VideoNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e)) from e
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+
+    logger.info(
+        "Share created: video_id=%s share_id=%s token=%s user=%s",
+        video_id,
+        share.id,
+        share.share_token,
+        authenticated_user.user.id,
+    )
+    return _to_share_response(share)
+
+
+@router.get("/videos/{video_id}/shares", response_model=ShareListResponse)
+def list_video_shares(
+    video_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+) -> ShareListResponse:
+    """获取某个视频的所有分享记录."""
+    use_case = ListSharesByVideoUseCase(share_repo)
+    items = use_case.execute(video_id, authenticated_user.user.id)
+    return ShareListResponse(
+        items=[_to_share_response(s) for s in items],
+        total=len(items),
+        skip=0,
+        limit=len(items),
+    )
+
+
+@router.get("/shares", response_model=ShareListResponse)
+def list_user_shares(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(20, ge=1, le=100),
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+) -> ShareListResponse:
+    """获取用户创建的所有分享记录."""
+    use_case = ListSharesByUserUseCase(share_repo)
+    items, total = use_case.execute(authenticated_user.user.id, skip=skip, limit=limit)
+    return ShareListResponse(
+        items=[_to_share_response(s) for s in items],
+        total=total,
+        skip=skip,
+        limit=limit,
+    )
+
+
+@router.patch("/shares/{share_id}", response_model=ShareResponse)
+def update_share(
+    share_id: str,
+    request: UpdateShareRequest,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+) -> ShareResponse:
+    """更新分享配置（密码、有效期等）."""
+    use_case = UpdateShareUseCase(share_repo)
+    try:
+        share = use_case.execute(
+            UpdateShareCommand(
+                share_id=share_id,
+                user_id=authenticated_user.user.id,
+                password=request.password,
+                expires_at=request.expires_at,
+            )
+        )
+    except NotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e)) from e
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    return _to_share_response(share)
+
+
+@router.delete(
+    "/shares/{share_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    response_model=None,
+    response_class=Response,
+)
+def revoke_share(
+    share_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+) -> Response:
+    """撤销/删除分享链接."""
+    use_case = RevokeShareUseCase(share_repo)
+    try:
+        use_case.execute(share_id, authenticated_user.user.id)
+    except NotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e)) from e
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+# ── 公开侧：访问分享内容（无需登录） ──────────────────────
+
+
+@router.get("/share/{token}/meta", response_model=ShareMetaResponse)
+def get_share_meta(
+    token: str,
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+    video_repo: GeneratedVideoRepository = Depends(get_generated_video_repository),
+    storage: OSSStorageService = Depends(get_storage_service),
+) -> ShareMetaResponse:
+    """获取分享元信息（不需要密码，用于分享页加载前判断）。"""
+    use_case = GetShareByTokenUseCase(share_repo)
+    try:
+        share = use_case.execute(token)
+    except (NotFoundError, ShareExpiredError) as e:
+        raise HTTPException(status_code=404, detail="分享链接不存在或已失效") from e
+
+    video = video_repo.get(share.video_id)
+    video_name = video.name if video else ""
+    video_duration = video.duration if video else 0.0
+    thumbnail_url = None
+    if video and video.thumbnail_url:
+        try:
+            thumbnail_url = storage.get_download_url(video.thumbnail_url)
+        except Exception:
+            thumbnail_url = video.thumbnail_url
+
+    return ShareMetaResponse(
+        share_token=share.share_token,
+        has_password=share.has_password,
+        is_expired=share.is_expired,
+        is_active=share.is_active,
+        video_name=video_name,
+        video_duration=video_duration,
+        thumbnail_url=thumbnail_url,
+        created_at=format_utc_datetime(share.created_at),
+    )
+
+
+@router.post("/share/{token}/access", response_model=ShareAccessResponse)
+def access_share(
+    token: str,
+    request: Optional[VerifySharePasswordRequest] = None,
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+    video_repo: GeneratedVideoRepository = Depends(get_generated_video_repository),
+    storage: OSSStorageService = Depends(get_storage_service),
+) -> ShareAccessResponse:
+    """访问分享内容（验证密码后返回视频信息+播放/下载地址）。"""
+    use_case = AccessShareUseCase(share_repo, video_repo)
+    password = request.password if request else None
+    try:
+        result = use_case.execute(token, password=password)
+    except NotFoundError as e:
+        raise HTTPException(status_code=404, detail="分享链接不存在或已失效") from e
+    except ShareExpiredError as e:
+        raise HTTPException(status_code=410, detail="分享链接已过期或已撤销") from e
+    except PasswordRequiredError as e:
+        raise HTTPException(status_code=403, detail="需要访问密码") from e
+    except InvalidPasswordError as e:
+        raise HTTPException(status_code=403, detail="密码错误") from e
+    except VideoNotFoundError as e:
+        raise HTTPException(status_code=404, detail="视频不存在") from e
+
+    # 生成下载URL
+    download_url = None
+    if result.video.file_url:
+        try:
+            download_url = storage.get_download_url(result.video.file_url)
+        except Exception:
+            download_url = result.video.file_url
+
+    # 缩略图URL
+    thumbnail_url = None
+    if result.video.thumbnail_url:
+        try:
+            thumbnail_url = storage.get_download_url(result.video.thumbnail_url)
+        except Exception:
+            thumbnail_url = result.video.thumbnail_url
+
+    return ShareAccessResponse(
+        share=_to_share_response(result.share),
+        video_name=result.video.name,
+        video_duration=result.video.duration,
+        video_size=result.video.file_size,
+        thumbnail_url=thumbnail_url,
+        download_url=download_url,
+        password_verified=result.password_verified,
+    )
+
+
+@router.post("/share/{token}/download")
+def record_share_download(
+    token: str,
+    request: Optional[VerifySharePasswordRequest] = None,
+    share_repo: SQLAlchemyVideoShareRepository = Depends(_get_share_repository),
+) -> dict:
+    """记录分享下载（下载计数+1）。"""
+    use_case = RecordShareDownloadUseCase(share_repo)
+    password = request.password if request else None
+    try:
+        use_case.execute(token, password=password)
+    except NotFoundError as e:
+        raise HTTPException(status_code=404, detail="分享链接不存在或已失效") from e
+    except ShareExpiredError as e:
+        raise HTTPException(status_code=410, detail="分享链接已过期或已撤销") from e
+    except InvalidPasswordError as e:
+        raise HTTPException(status_code=403, detail="密码错误") from e
+    return {"success": True}

apps/api/app/schemas/video_share.py

@@ -0,0 +1,92 @@
+"""视频分享相关 schemas."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+
+class CreateShareRequest(BaseModel):
+    """创建分享请求."""
+
+    password: Optional[str] = Field(
+        None,
+        description="访问密码（可选，不设置则无需密码）",
+        min_length=0,
+        max_length=50,
+    )
+    expires_at: Optional[datetime] = Field(
+        None,
+        description="过期时间（可选，不设置则永久有效）",
+    )
+
+
+class UpdateShareRequest(BaseModel):
+    """更新分享配置请求."""
+
+    password: Optional[str] = Field(
+        None,
+        description="新密码（传空字符串清除密码，不传则不修改）",
+        max_length=50,
+    )
+    expires_at: Optional[datetime] = Field(
+        None,
+        description="新的过期时间（不传则不修改）",
+    )
+
+
+class VerifySharePasswordRequest(BaseModel):
+    """验证分享密码请求."""
+
+    password: str = Field(..., description="访问密码")
+
+
+class ShareResponse(BaseModel):
+    """分享记录响应."""
+
+    id: str
+    video_id: str
+    share_token: str
+    has_password: bool = False
+    expires_at: Optional[datetime] = None
+    view_count: int = 0
+    download_count: int = 0
+    is_active: bool = True
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+
+
+class ShareListResponse(BaseModel):
+    """分享列表响应."""
+
+    items: List[ShareResponse]
+    total: int = 0
+    skip: int = 0
+    limit: int = 20
+
+
+class ShareAccessResponse(BaseModel):
+    """分享访问成功响应（含视频信息）."""
+
+    share: ShareResponse
+    video_name: str
+    video_duration: float = 0.0
+    video_size: int = 0
+    thumbnail_url: Optional[str] = None
+    download_url: Optional[str] = None
+    password_verified: bool = True
+
+
+class ShareMetaResponse(BaseModel):
+    """分享元信息响应（访问前获取，用于判断是否需要密码）。"""
+
+    share_token: str
+    has_password: bool = False
+    is_expired: bool = False
+    is_active: bool = True
+    video_name: str = ""
+    video_duration: float = 0.0
+    thumbnail_url: Optional[str] = None
+    created_at: Optional[datetime] = None

packages/adapters/sqlalchemy_impl/models.py

@@ -574,3 +574,21 @@ class VerificationCodeModel(Base):
     used_at = Column(DateTime, nullable=True)
     attempts = Column(Integer, nullable=False, default=0)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
+
+
+class VideoShareModel(Base):
+    """视频分享记录."""
+
+    __tablename__ = "video_shares"
+
+    id = Column(String(32), primary_key=True)
+    video_id = Column(String(32), nullable=False, index=True)
+    user_id = Column(String(32), nullable=False, index=True)
+    share_token = Column(String(16), nullable=False, unique=True)
+    password_hash = Column(String(255), nullable=True)
+    expires_at = Column(DateTime(timezone=True), nullable=True)
+    view_count = Column(Integer, nullable=False, default=0)
+    download_count = Column(Integer, nullable=False, default=0)
+    is_active = Column(Boolean, nullable=False, default=True)
+    created_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
+    updated_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))

packages/adapters/sqlalchemy_impl/video_share_repository.py

@@ -0,0 +1,147 @@
+"""视频分享 SQLAlchemy Repository 实现."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import List, 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(timezone.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(timezone.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(timezone.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(timezone.utc),
+            },
+            synchronize_session=False,
+        )
+        self.session.commit()

packages/application/video_share/commands.py

@@ -0,0 +1,35 @@
+"""视频分享 Commands."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+
+@dataclass
+class CreateShareCommand:
+    """创建分享链接命令."""
+
+    video_id: str
+    user_id: str
+    password: Optional[str] = None
+    expires_at: Optional[datetime] = None  # None表示永久有效
+
+
+@dataclass
+class VerifySharePasswordCommand:
+    """验证分享密码命令."""
+
+    share_token: str
+    password: str
+
+
+@dataclass
+class UpdateShareCommand:
+    """更新分享配置命令."""
+
+    share_id: str
+    user_id: str
+    password: Optional[str] = None  # None表示不修改，空字符串表示清除密码
+    expires_at: Optional[datetime] = None  # None表示不修改

packages/application/video_share/use_cases.py

@@ -0,0 +1,227 @@
+"""视频分享 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.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 NotFoundError(Exception):
+    """分享记录不存在."""
+
+    pass
+
+
+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)

packages/domain/video_share.py

@@ -0,0 +1,108 @@
+"""视频分享领域实体."""
+
+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

packages/ports/video_share_repository.py

@@ -0,0 +1,62 @@
+"""视频分享 Repository 端口."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import List, Optional
+
+from packages.domain.video_share import VideoShare
+
+
+class VideoShareRepositoryPort(ABC):
+    """视频分享 Repository 接口."""
+
+    @abstractmethod
+    def create(self, share: VideoShare) -> VideoShare:
+        """创建分享记录."""
+        ...
+
+    @abstractmethod
+    def get_by_token(self, token: str) -> Optional[VideoShare]:
+        """通过分享token获取分享记录."""
+        ...
+
+    @abstractmethod
+    def get_by_id(self, share_id: str, user_id: str) -> Optional[VideoShare]:
+        """通过ID获取分享记录（带用户校验）."""
+        ...
+
+    @abstractmethod
+    def list_by_video(self, video_id: str, user_id: str) -> List[VideoShare]:
+        """列出某个视频的所有分享记录."""
+        ...
+
+    @abstractmethod
+    def list_by_user(self, user_id: str, skip: int = 0, limit: int = 20) -> List[VideoShare]:
+        """列出用户创建的所有分享记录."""
+        ...
+
+    @abstractmethod
+    def count_by_user(self, user_id: str) -> int:
+        """统计用户创建的分享数量."""
+        ...
+
+    @abstractmethod
+    def update(self, share: VideoShare) -> VideoShare:
+        """更新分享记录."""
+        ...
+
+    @abstractmethod
+    def delete(self, share_id: str, user_id: str) -> bool:
+        """删除分享记录（软删除：is_active=False）."""
+        ...
+
+    @abstractmethod
+    def increment_view(self, share_id: str) -> None:
+        """浏览次数+1."""
+        ...
+
+    @abstractmethod
+    def increment_download(self, share_id: str) -> None:
+        """下载次数+1."""
+        ...

tests/unit/test_video_share.py

@@ -0,0 +1,603 @@
+"""视频分享 - 领域实体 + Use cases 单元测试."""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from unittest.mock import MagicMock
+
+import pytest
+
+from packages.application.video_share.commands import (
+    CreateShareCommand,
+    UpdateShareCommand,
+)
+from packages.application.video_share.use_cases import (
+    AccessShareUseCase,
+    CreateShareUseCase,
+    GetShareByTokenUseCase,
+    InvalidPasswordError,
+    ListSharesByUserUseCase,
+    ListSharesByVideoUseCase,
+    NotFoundError,
+    PasswordRequiredError,
+    RecordShareDownloadUseCase,
+    RevokeShareUseCase,
+    ShareAccessResult,
+    ShareExpiredError,
+    UpdateShareUseCase,
+    VideoNotFoundError,
+)
+from packages.domain.generated_video import GeneratedVideo
+from packages.domain.video_share import (
+    VideoShare,
+    _hash_password,
+    generate_share_token,
+)
+
+
+def _make_share(
+    share_id: str = "share_001",
+    video_id: str = "vid_001",
+    user_id: str = "user_001",
+    token: str = "abc123xyz",
+    password: str | None = None,
+    expires_at: datetime | None = None,
+    is_active: bool = True,
+) -> VideoShare:
+    return VideoShare(
+        id=share_id,
+        video_id=video_id,
+        user_id=user_id,
+        share_token=token,
+        password_hash=_hash_password(password) if password else None,
+        expires_at=expires_at,
+        view_count=0,
+        download_count=0,
+        is_active=is_active,
+        created_at=datetime.now(timezone.utc),
+        updated_at=datetime.now(timezone.utc),
+    )
+
+
+def _make_video(video_id: str = "vid_001", user_id: str = "user_001") -> GeneratedVideo:
+    return GeneratedVideo(
+        id=video_id,
+        project_id="proj_001",
+        generation_task_id="task_001",
+        name="测试视频",
+        file_url="oss://bucket/video.mp4",
+        file_size=1024000,
+        duration=30.5,
+        width=1080,
+        height=1920,
+        fps=30.0,
+        user_id=user_id,
+    )
+
+
+class TestVideoShareDomain:
+    def test_generate_token_length(self) -> None:
+        token = generate_share_token(12)
+        assert len(token) == 12
+
+    def test_generate_token_url_safe(self) -> None:
+        token = generate_share_token(16)
+        # 只包含字母数字，没有特殊字符
+        assert token.isalnum()
+
+    def test_hash_password_consistent(self) -> None:
+        h1 = _hash_password("mypassword")
+        h2 = _hash_password("mypassword")
+        assert h1 == h2
+        assert len(h1) == 64  # sha256 hex
+
+    def test_hash_password_different_for_different_passwords(self) -> None:
+        h1 = _hash_password("password1")
+        h2 = _hash_password("password2")
+        assert h1 != h2
+
+    def test_hash_empty_password(self) -> None:
+        assert _hash_password("") == ""
+
+    def test_create_share_success(self) -> None:
+        share = VideoShare.create(
+            video_id="vid_001",
+            user_id="user_001",
+        )
+        assert share.video_id == "vid_001"
+        assert share.user_id == "user_001"
+        assert len(share.id) == 32
+        assert len(share.share_token) == 12
+        assert share.password_hash is None
+        assert share.expires_at is None
+        assert share.is_active is True
+        assert share.view_count == 0
+        assert share.download_count == 0
+
+    def test_create_share_with_password(self) -> None:
+        share = VideoShare.create(
+            video_id="vid_001",
+            user_id="user_001",
+            password="secret123",
+        )
+        assert share.has_password is True
+        assert share.verify_password("secret123") is True
+        assert share.verify_password("wrong") is False
+
+    def test_create_share_with_expiry(self) -> None:
+        future = datetime.now(timezone.utc) + timedelta(days=7)
+        share = VideoShare.create(
+            video_id="vid_001",
+            user_id="user_001",
+            expires_at=future,
+        )
+        assert share.expires_at == future
+        assert share.is_expired is False
+
+    def test_create_share_past_expiry_raises(self) -> None:
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        with pytest.raises(ValueError, match="past"):
+            VideoShare.create(
+                video_id="vid_001",
+                user_id="user_001",
+                expires_at=past,
+            )
+
+    def test_create_share_empty_video_id_raises(self) -> None:
+        with pytest.raises(ValueError, match="video_id"):
+            VideoShare.create(video_id="", user_id="user_001")
+
+    def test_create_share_empty_user_id_raises(self) -> None:
+        with pytest.raises(ValueError, match="user_id"):
+            VideoShare.create(video_id="vid_001", user_id="  ")
+
+    def test_is_expired_false_when_no_expiry(self) -> None:
+        share = _make_share()
+        assert share.is_expired is False
+
+    def test_is_expired_true_when_past(self) -> None:
+        past = datetime.now(timezone.utc) - timedelta(hours=1)
+        share = _make_share(expires_at=past)
+        assert share.is_expired is True
+
+    def test_is_accessible_active_not_expired(self) -> None:
+        share = _make_share()
+        assert share.is_accessible is True
+
+    def test_is_accessible_inactive(self) -> None:
+        share = _make_share(is_active=False)
+        assert share.is_accessible is False
+
+    def test_is_accessible_expired(self) -> None:
+        past = datetime.now(timezone.utc) - timedelta(hours=1)
+        share = _make_share(expires_at=past)
+        assert share.is_accessible is False
+
+    def test_has_password_false_when_no_password(self) -> None:
+        share = _make_share()
+        assert share.has_password is False
+
+    def test_has_password_true_when_password_set(self) -> None:
+        share = _make_share(password="pass123")
+        assert share.has_password is True
+
+    def test_verify_no_password_always_true(self) -> None:
+        share = _make_share()  # 没有密码
+        assert share.verify_password("") is True
+        assert share.verify_password("anything") is True
+
+    def test_verify_correct_password(self) -> None:
+        share = _make_share(password="mysecret")
+        assert share.verify_password("mysecret") is True
+
+    def test_verify_wrong_password(self) -> None:
+        share = _make_share(password="mysecret")
+        assert share.verify_password("wrong") is False
+
+    def test_verify_empty_password_with_password_set(self) -> None:
+        share = _make_share(password="mysecret")
+        assert share.verify_password("") is False
+
+    def test_increment_view_count(self) -> None:
+        share = _make_share()
+        share.increment_view_count()
+        assert share.view_count == 1
+        share.increment_view_count()
+        assert share.view_count == 2
+
+    def test_increment_download_count(self) -> None:
+        share = _make_share()
+        share.increment_download_count()
+        assert share.download_count == 1
+
+    def test_revoke_sets_inactive(self) -> None:
+        share = _make_share()
+        share.revoke()
+        assert share.is_active is False
+        assert share.is_accessible is False
+
+
+class TestCreateShareUseCase:
+    def test_create_success(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        video_repo.get.return_value = _make_video()
+        share_repo.create.side_effect = lambda s: s
+
+        use_case = CreateShareUseCase(share_repo, video_repo)
+        cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
+        result = use_case.execute(cmd)
+
+        assert result.video_id == "vid_001"
+        assert result.user_id == "user_001"
+        share_repo.create.assert_called_once()
+
+    def test_create_with_password(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        video_repo.get.return_value = _make_video()
+        share_repo.create.side_effect = lambda s: s
+
+        use_case = CreateShareUseCase(share_repo, video_repo)
+        cmd = CreateShareCommand(
+            video_id="vid_001",
+            user_id="user_001",
+            password="secret",
+        )
+        result = use_case.execute(cmd)
+
+        assert result.has_password is True
+
+    def test_create_with_expiry(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        video_repo.get.return_value = _make_video()
+        share_repo.create.side_effect = lambda s: s
+
+        use_case = CreateShareUseCase(share_repo, video_repo)
+        future = datetime.now(timezone.utc) + timedelta(days=1)
+        cmd = CreateShareCommand(
+            video_id="vid_001",
+            user_id="user_001",
+            expires_at=future,
+        )
+        result = use_case.execute(cmd)
+
+        assert result.expires_at == future
+
+    def test_video_not_found_raises(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        video_repo.get.return_value = None
+
+        use_case = CreateShareUseCase(share_repo, video_repo)
+        cmd = CreateShareCommand(video_id="nonexistent", user_id="user_001")
+
+        with pytest.raises(VideoNotFoundError):
+            use_case.execute(cmd)
+
+    def test_wrong_user_cannot_share(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        video_repo.get.return_value = _make_video(user_id="other_user")
+
+        use_case = CreateShareUseCase(share_repo, video_repo)
+        cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
+
+        with pytest.raises(VideoNotFoundError):
+            use_case.execute(cmd)
+
+
+class TestGetShareByTokenUseCase:
+    def test_found_active_share(self) -> None:
+        repo = MagicMock()
+        repo.get_by_token.return_value = _make_share()
+
+        use_case = GetShareByTokenUseCase(repo)
+        result = use_case.execute("abc123xyz")
+        assert result.share_token == "abc123xyz"
+
+    def test_not_found_raises(self) -> None:
+        repo = MagicMock()
+        repo.get_by_token.return_value = None
+
+        use_case = GetShareByTokenUseCase(repo)
+        with pytest.raises(NotFoundError):
+            use_case.execute("nonexistent")
+
+    def test_inactive_share_raises_expired(self) -> None:
+        repo = MagicMock()
+        repo.get_by_token.return_value = _make_share(is_active=False)
+
+        use_case = GetShareByTokenUseCase(repo)
+        with pytest.raises(ShareExpiredError):
+            use_case.execute("token")
+
+    def test_expired_share_raises(self) -> None:
+        repo = MagicMock()
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        repo.get_by_token.return_value = _make_share(expires_at=past)
+
+        use_case = GetShareByTokenUseCase(repo)
+        with pytest.raises(ShareExpiredError):
+            use_case.execute("token")
+
+
+class TestAccessShareUseCase:
+    def test_access_no_password(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        share_repo.get_by_token.return_value = _make_share()
+        video_repo.get.return_value = _make_video()
+        share_repo.increment_view.return_value = None
+
+        use_case = AccessShareUseCase(share_repo, video_repo)
+        result = use_case.execute("abc123xyz")
+
+        assert isinstance(result, ShareAccessResult)
+        assert result.video.id == "vid_001"
+        assert result.password_verified is True
+        assert result.share.view_count == 1  # 浏览量+1
+        share_repo.increment_view.assert_called_once()
+
+    def test_access_with_correct_password(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        share_repo.get_by_token.return_value = _make_share(password="mypass")
+        video_repo.get.return_value = _make_video()
+        share_repo.increment_view.return_value = None
+
+        use_case = AccessShareUseCase(share_repo, video_repo)
+        result = use_case.execute("token", password="mypass")
+
+        assert result.password_verified is True
+
+    def test_access_password_required_but_not_provided(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        share_repo.get_by_token.return_value = _make_share(password="secret")
+        video_repo.get.return_value = _make_video()
+
+        use_case = AccessShareUseCase(share_repo, video_repo)
+
+        with pytest.raises(PasswordRequiredError):
+            use_case.execute("token", password=None)
+
+    def test_access_wrong_password(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        share_repo.get_by_token.return_value = _make_share(password="correct")
+        video_repo.get.return_value = _make_video()
+
+        use_case = AccessShareUseCase(share_repo, video_repo)
+
+        with pytest.raises(InvalidPasswordError):
+            use_case.execute("token", password="wrong")
+
+    def test_access_share_not_found(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        share_repo.get_by_token.return_value = None
+
+        use_case = AccessShareUseCase(share_repo, video_repo)
+
+        with pytest.raises(NotFoundError):
+            use_case.execute("nonexistent")
+
+    def test_access_share_expired(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        share_repo.get_by_token.return_value = _make_share(expires_at=past)
+        video_repo.get.return_value = _make_video()
+
+        use_case = AccessShareUseCase(share_repo, video_repo)
+
+        with pytest.raises(ShareExpiredError):
+            use_case.execute("token")
+
+    def test_access_video_not_found(self) -> None:
+        share_repo = MagicMock()
+        video_repo = MagicMock()
+        share_repo.get_by_token.return_value = _make_share()
+        video_repo.get.return_value = None
+
+        use_case = AccessShareUseCase(share_repo, video_repo)
+
+        with pytest.raises(VideoNotFoundError):
+            use_case.execute("token")
+
+
+class TestListSharesByVideoUseCase:
+    def test_lists_shares(self) -> None:
+        repo = MagicMock()
+        expected = [_make_share(), _make_share(share_id="share_002", token="tok2")]
+        repo.list_by_video.return_value = expected
+
+        use_case = ListSharesByVideoUseCase(repo)
+        result = use_case.execute("vid_001", "user_001")
+
+        assert len(result) == 2
+        repo.list_by_video.assert_called_once_with("vid_001", "user_001")
+
+
+class TestListSharesByUserUseCase:
+    def test_lists_with_total(self) -> None:
+        repo = MagicMock()
+        items = [_make_share(), _make_share(share_id="s2", token="t2")]
+        repo.list_by_user.return_value = items
+        repo.count_by_user.return_value = 10
+
+        use_case = ListSharesByUserUseCase(repo)
+        result_items, total = use_case.execute("user_001", skip=0, limit=2)
+
+        assert len(result_items) == 2
+        assert total == 10
+        repo.list_by_user.assert_called_once_with("user_001", skip=0, limit=2)
+
+
+class TestUpdateShareUseCase:
+    def test_update_password(self) -> None:
+        repo = MagicMock()
+        share = _make_share()
+        repo.get_by_id.return_value = share
+        repo.update.side_effect = lambda s: s
+
+        use_case = UpdateShareUseCase(repo)
+        cmd = UpdateShareCommand(
+            share_id="share_001",
+            user_id="user_001",
+            password="newpass",
+        )
+        result = use_case.execute(cmd)
+
+        assert result.has_password is True
+        assert result.verify_password("newpass") is True
+        repo.update.assert_called_once()
+
+    def test_clear_password(self) -> None:
+        repo = MagicMock()
+        share = _make_share(password="oldpass")
+        repo.get_by_id.return_value = share
+        repo.update.side_effect = lambda s: s
+
+        use_case = UpdateShareUseCase(repo)
+        cmd = UpdateShareCommand(
+            share_id="share_001",
+            user_id="user_001",
+            password="",  # 空字符串=清除密码
+        )
+        result = use_case.execute(cmd)
+
+        assert result.has_password is False
+        assert result.password_hash is None
+
+    def test_password_none_does_not_change(self) -> None:
+        repo = MagicMock()
+        share = _make_share(password="existing")
+        repo.get_by_id.return_value = share
+        repo.update.side_effect = lambda s: s
+
+        use_case = UpdateShareUseCase(repo)
+        cmd = UpdateShareCommand(
+            share_id="share_001",
+            user_id="user_001",
+            password=None,  # None=不修改
+        )
+        result = use_case.execute(cmd)
+
+        assert result.verify_password("existing") is True
+
+    def test_update_expires_at(self) -> None:
+        repo = MagicMock()
+        share = _make_share()
+        repo.get_by_id.return_value = share
+        repo.update.side_effect = lambda s: s
+
+        use_case = UpdateShareUseCase(repo)
+        future = datetime.now(timezone.utc) + timedelta(days=3)
+        cmd = UpdateShareCommand(
+            share_id="share_001",
+            user_id="user_001",
+            expires_at=future,
+        )
+        result = use_case.execute(cmd)
+
+        assert result.expires_at == future
+
+    def test_not_found_raises(self) -> None:
+        repo = MagicMock()
+        repo.get_by_id.return_value = None
+
+        use_case = UpdateShareUseCase(repo)
+        cmd = UpdateShareCommand(share_id="no", user_id="u1", password="x")
+
+        with pytest.raises(NotFoundError):
+            use_case.execute(cmd)
+
+    def test_past_expiry_raises(self) -> None:
+        repo = MagicMock()
+        repo.get_by_id.return_value = _make_share()
+
+        use_case = UpdateShareUseCase(repo)
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        cmd = UpdateShareCommand(
+            share_id="share_001",
+            user_id="user_001",
+            expires_at=past,
+        )
+
+        with pytest.raises(ValueError, match="past"):
+            use_case.execute(cmd)
+
+
+class TestRevokeShareUseCase:
+    def test_revoke_success(self) -> None:
+        repo = MagicMock()
+        repo.get_by_id.return_value = _make_share()
+        repo.delete.return_value = True
+
+        use_case = RevokeShareUseCase(repo)
+        result = use_case.execute("share_001", "user_001")
+
+        assert result is True
+        repo.delete.assert_called_once_with("share_001", "user_001")
+
+    def test_revoke_not_found_raises(self) -> None:
+        repo = MagicMock()
+        repo.get_by_id.return_value = None
+
+        use_case = RevokeShareUseCase(repo)
+
+        with pytest.raises(NotFoundError):
+            use_case.execute("nonexistent", "user_001")
+
+
+class TestRecordShareDownloadUseCase:
+    def test_record_success(self) -> None:
+        repo = MagicMock()
+        repo.get_by_token.return_value = _make_share()
+        repo.increment_download.return_value = None
+
+        use_case = RecordShareDownloadUseCase(repo)
+        use_case.execute("token")
+
+        repo.increment_download.assert_called_once()
+
+    def test_record_with_password(self) -> None:
+        repo = MagicMock()
+        repo.get_by_token.return_value = _make_share(password="pass")
+        repo.increment_download.return_value = None
+
+        use_case = RecordShareDownloadUseCase(repo)
+        use_case.execute("token", password="pass")
+
+        repo.increment_download.assert_called_once()
+
+    def test_record_wrong_password_raises(self) -> None:
+        repo = MagicMock()
+        repo.get_by_token.return_value = _make_share(password="correct")
+
+        use_case = RecordShareDownloadUseCase(repo)
+
+        with pytest.raises(InvalidPasswordError):
+            use_case.execute("token", password="wrong")
+
+    def test_record_share_not_found(self) -> None:
+        repo = MagicMock()
+        repo.get_by_token.return_value = None
+
+        use_case = RecordShareDownloadUseCase(repo)
+
+        with pytest.raises(NotFoundError):
+            use_case.execute("nonexistent")
+
+    def test_record_expired_share(self) -> None:
+        repo = MagicMock()
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        repo.get_by_token.return_value = _make_share(expires_at=past)
+
+        use_case = RecordShareDownloadUseCase(repo)
+
+        with pytest.raises(ShareExpiredError):
+            use_case.execute("token")