75 KiB

Raw Permalink Blame History

项目管理服务

文档版本：v2.0
最后更新：2026-01-31

服务概述

项目管理服务负责处理项目的创建、查询、更新、删除等核心业务逻辑。

职责

项目 CRUD 操作
项目权限管理
项目成员管理
项目移动到文件夹
项目搜索和筛选
子项目管理（父子项目关系）

核心功能

1. 项目创建

支持两种项目类型：mine（个人项目）、collab（协作项目）
可指定所属文件夹
自动设置项目所有者
初始化项目配置

2. 项目查询

按用户查询项目列表
按文件夹查询项目
按项目类型筛选
支持分页

3. 项目更新

更新项目基本信息
更新项目配置
更新项目缩略图

4. 项目删除（回收站机制）

移至回收站（status=2，设置 trashed_at）
回收站恢复（status 恢复为 active）
永久删除（status=3，设置 deleted_at，用户不可见）
物理删除（仅管理员，真实删除数据库记录）
自动清理（30天后自动从回收站转为软删除）

5. 项目移动

移动项目到指定文件夹
移动到根目录（folder_id = null）
权限检查

6. 项目克隆

完整复制项目基本信息
复制所有关联资源（分镜、视频、附件等）
自动生成唯一名称（避免重名）
异步处理，支持进度追踪

7. 项目导出

打包项目为 ZIP 文件
包含项目配置、分镜、资源文件
异步任务处理
进度通知和文件下载

8. 项目分享

生成公开访问链接
可选密码保护
邀请特定用户并设置权限
管理分享权限（查看/编辑/管理）

9. 项目搜索和筛选

按名称、描述全文搜索
按内容类型筛选
按创建时间、更新时间、名称排序
支持分页

10. 项目排序

自定义项目显示顺序
拖拽排序支持

11. 子项目管理

支持父子项目关系（项目层级结构）
子项目关联剧本（一对一）
子项目继承父项目权限
子项目独立管理分镜和资源
自动创建子项目（上传剧本时）

服务实现

📊 流程图与关系图
详细的表关系图和业务流程图请参考：项目服务流程图

包含内容：

数据库表关系 ER 图

项目创建流程图

项目克隆流程图

权限检查流程图

项目生命周期状态图

子项目关系图

服务实现

ProjectService 类

# app/services/project_service.py
from typing import List, Optional, Dict, Any
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.project import Project
from app.repositories.project_repository import ProjectRepository
from app.repositories.folder_repository import FolderRepository
from app.schemas.project import ProjectCreate, ProjectUpdate, ProjectMove
from app.core.exceptions import NotFoundError, ValidationError, PermissionError
from app.core.logging import get_logger

# 获取模块级 logger
logger = get_logger(__name__)

class ProjectService:
    def __init__(self, db: AsyncSession):
        self.repository = ProjectRepository(db)
        self.folder_repository = FolderRepository(db)
        self.db = db

    async def get_projects(
        self,
        user_id: str,
        project_type: Optional[str] = None,
        folder_id: Optional[str] = None,
        content_type: Optional[str] = None,
        search: Optional[str] = None,
        sort_by: str = 'updated_at',
        sort_order: str = 'desc',
        page: int = 1,
        page_size: int = 20
    ) -> Dict[str, Any]:
        """获取项目列表（支持搜索、筛选、排序）"""
        # 如果指定了文件夹，检查文件夹权限
        if folder_id:
            has_permission = await self.folder_repository.check_user_permission(
                user_id, folder_id, 'viewer'
            )
            if not has_permission:
                raise PermissionError("没有权限访问此文件夹")

        # 验证排序字段
        valid_sort_fields = ['created_at', 'updated_at', 'name', 'display_order']
        if sort_by not in valid_sort_fields:
            raise ValidationError("无效的排序字段: %s" % sort_by)

        # 验证排序方向
        if sort_order not in ['asc', 'desc']:
            raise ValidationError("无效的排序方向: %s" % sort_order)

        projects = await self.repository.get_by_user(
            user_id=user_id,
            project_type=project_type,
            folder_id=folder_id,
            content_type=content_type,
            search=search,
            sort_by=sort_by,
            sort_order=sort_order,
            page=page,
            page_size=page_size
        )

        total = await self.repository.count_by_user(
            user_id=user_id,
            project_type=project_type,
            folder_id=folder_id,
            content_type=content_type,
            search=search
        )

        return {
            'items': projects,
            'total': total,
            'page': page,
            'page_size': page_size,
            'total_pages': (total + page_size - 1) // page_size
        }

    async def get_project(
        self,
        user_id: str,
        project_id: str
    ) -> Project:
        """获取项目详情"""
        project = await self.repository.get_by_id(project_id)
        if not project:
            raise NotFoundError("项目不存在")

        # 检查权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'viewer'
        ):
            raise PermissionError("没有权限访问此项目")

        return project

    async def create_project(
        self,
        user_id: str,
        project_data: ProjectCreate
    ) -> Project:
        """创建项目"""
        # 校验用户存在性
        await self._validate_user_exists(user_id)
        
        # 如果指定了文件夹，检查存在性和权限
        if project_data.folder_id:
            await self._validate_folder_exists(project_data.folder_id)
            has_permission = await self.folder_repository.check_user_permission(
                user_id, project_data.folder_id, 'editor'
            )
            if not has_permission:
                raise PermissionError("没有权限在此文件夹中创建项目")

        # 校验封面图片存在性和所属权
        if hasattr(project_data, 'cover_image_id') and project_data.cover_image_id:
            await self._validate_attachment_exists_and_owned(
                project_data.cover_image_id, user_id
            )

        # 验证项目类型
        if project_data.type not in ['mine', 'collab']:
            raise ValidationError("无效的项目类型")

        project = Project(
            name=project_data.name,
            description=project_data.description,
            type=project_data.type,
            owner_id=user_id,
            folder_id=project_data.folder_id,
            settings=project_data.settings.dict() if project_data.settings else {}
        )

        created_project = await self.repository.create(project)
        logger.info(
            "用户创建项目: user_id=%s, project_id=%s, name=%s",
            user_id,
            str(created_project.id),
            created_project.name
        )

        # 如果是协作项目，添加创建者为成员
        if project_data.type == 'collab':
            await self.repository.add_member(
                created_project.id,
                user_id,
                'owner'
            )

        return created_project

    async def update_project(
        self,
        user_id: str,
        project_id: str,
        project_data: ProjectUpdate
    ) -> Project:
        """更新项目"""
        project = await self.repository.get_by_id(project_id)
        if not project:
            raise NotFoundError("项目不存在")

        # 检查权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'editor'
        ):
            raise PermissionError("没有权限编辑此项目")

        update_data = project_data.dict(exclude_unset=True)
        return await self.repository.update(project_id, update_data)

    async def delete_project(
        self,
        user_id: str,
        project_id: str
    ) -> None:
        """删除项目（软删除）"""
        project = await self.repository.get_by_id(project_id)
        if not project:
            raise NotFoundError("项目不存在")

        # 检查权限（只有 owner 可以删除）
        if not await self.repository.check_user_permission(
            user_id, project_id, 'owner'
        ):
            raise PermissionError("只有项目所有者可以删除项目")

        await self.repository.soft_delete(project_id)
        logger.info("用户删除项目: user_id=%s, project_id=%s", user_id, project_id)

    async def move_project(
        self,
        user_id: str,
        project_id: str,
        move_data: ProjectMove
    ) -> Project:
        """移动项目到文件夹"""
        # 检查项目权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'editor'
        ):
            raise PermissionError("没有权限移动此项目")

        # 检查目标文件夹权限
        if move_data.folder_id:
            has_permission = await self.folder_repository.check_user_permission(
                user_id, move_data.folder_id, 'editor'
            )
            if not has_permission:
                raise PermissionError("没有权限访问目标文件夹")

        return await self.repository.move_to_folder(
            project_id,
            move_data.folder_id
        )

    async def get_project_members(
        self,
        user_id: str,
        project_id: str
    ) -> List[Dict[str, Any]]:
        """获取项目成员列表"""
        # 检查权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'viewer'
        ):
            raise PermissionError("没有权限查看项目成员")

        return await self.repository.get_members(project_id)

    async def add_project_member(
        self,
        user_id: str,
        project_id: str,
        member_user_id: str,
        role: str
    ) -> None:
        """添加项目成员"""
        # 检查权限（只有 owner 可以添加成员）
        if not await self.repository.check_user_permission(
            user_id, project_id, 'owner'
        ):
            raise PermissionError("只有项目所有者可以添加成员")

        # 验证角色
        if role not in ['owner', 'editor', 'viewer']:
            raise ValidationError("无效的角色")

        await self.repository.add_member(project_id, member_user_id, role)

    async def remove_project_member(
        self,
        user_id: str,
        project_id: str,
        member_user_id: str
    ) -> None:
        """移除项目成员"""
        # 检查权限（只有 owner 可以移除成员）
        if not await self.repository.check_user_permission(
            user_id, project_id, 'owner'
        ):
            raise PermissionError("只有项目所有者可以移除成员")

        # 不能移除自己
        if user_id == member_user_id:
            raise ValidationError("不能移除自己")

        await self.repository.remove_member(project_id, member_user_id)

    async def clone_project(
        self,
        user_id: str,
        project_id: str
    ) -> Dict[str, Any]:
        """克隆项目（异步任务）"""
        # 检查源项目权限
        source_project = await self.repository.get_by_id(project_id)
        if not source_project:
            raise NotFoundError("项目不存在")

        if not await self.repository.check_user_permission(
            user_id, project_id, 'viewer'
        ):
            raise PermissionError("没有权限克隆此项目")

        # 生成唯一的副本名称
        base_name = f"{source_project.name}-副本"
        new_name = await self._generate_unique_project_name(
            base_name, 
            source_project.folder_id, 
            user_id
        )

        # 创建异步任务
        task_id = await self._create_clone_task(
            user_id,
            project_id,
            new_name
        )

        return {
            'task_id': task_id,
            'status': 'pending',
            'message': '克隆任务已创建'
        }

    async def export_project(
        self,
        user_id: str,
        project_id: str
    ) -> Dict[str, Any]:
        """导出项目为 ZIP 文件（异步任务）"""
        # 检查权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'viewer'
        ):
            raise PermissionError("没有权限导出此项目")

        # 创建导出任务
        task_id = await self._create_export_task(user_id, project_id)

        return {
            'task_id': task_id,
            'status': 'pending',
            'message': '导出任务已创建'
        }

    async def get_export_status(
        self,
        user_id: str,
        project_id: str,
        task_id: str
    ) -> Dict[str, Any]:
        """获取导出任务状态"""
        # 检查权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'viewer'
        ):
            raise PermissionError("没有权限访问此项目")

        # 查询任务状态（从 Redis 或任务队列）
        task_status = await self._get_task_status(task_id)
        
        return task_status

    async def create_share(
        self,
        user_id: str,
        project_id: str,
        share_data: Dict[str, Any]
    ) -> Dict[str, Any]:
        """创建项目分享"""
        # 检查权限（只有 owner 可以创建分享）
        if not await self.repository.check_user_permission(
            user_id, project_id, 'owner'
        ):
            raise PermissionError("只有项目所有者可以创建分享")

        # 生成分享链接
        share_token = await self._generate_share_token()
        
        share = await self.repository.create_share(
            project_id=project_id,
            created_by=user_id,
            share_token=share_token,
            password=share_data.get('password'),
            permission=share_data.get('permission', 'viewer'),
            expires_at=share_data.get('expires_at')
        )

        return {
            'share_id': share.id,
            'share_url': f"/share/p/{share_token}",
            'permission': share.permission,
            'expires_at': share.expires_at
        }

    async def get_shares(
        self,
        user_id: str,
        project_id: str
    ) -> List[Dict[str, Any]]:
        """获取项目分享列表"""
        # 检查权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'viewer'
        ):
            raise PermissionError("没有权限查看分享列表")

        return await self.repository.get_shares(project_id)

    async def revoke_share(
        self,
        user_id: str,
        project_id: str,
        share_id: str
    ) -> None:
        """撤销项目分享"""
        # 检查权限（只有 owner 可以撤销分享）
        if not await self.repository.check_user_permission(
            user_id, project_id, 'owner'
        ):
            raise PermissionError("只有项目所有者可以撤销分享")

        await self.repository.delete_share(share_id)

    async def update_project_order(
        self,
        user_id: str,
        project_id: str,
        new_order: int
    ) -> Project:
        """更新项目显示顺序"""
        # 检查权限
        if not await self.repository.check_user_permission(
            user_id, project_id, 'editor'
        ):
            raise PermissionError("没有权限修改项目顺序")

        return await self.repository.update(
            project_id,
            {'display_order': new_order}
        )

    async def create_subproject(
        self,
        user_id: str,
        parent_project_id: str,
        screenplay_id: str,
        name: str,
        description: Optional[str] = None
    ) -> Project:
        """创建子项目（由剧本服务调用）"""
        # 检查父项目权限
        parent_project = await self.repository.get_by_id(parent_project_id)
        if not parent_project:
            raise NotFoundError("父项目不存在")

        if not await self.repository.check_user_permission(
            user_id, parent_project_id, 'editor'
        ):
            raise PermissionError("没有权限在此项目中创建子项目")

        # 创建子项目
        subproject = Project(
            name=name,
            description=description,
            type=parent_project.type,  # 继承父项目类型
            owner_id=parent_project.owner_id,  # 继承父项目所有者
            owner_type=parent_project.owner_type,
            folder_id=parent_project.folder_id,  # 继承父项目文件夹
            parent_project_id=parent_project_id,
            screenplay_id=screenplay_id,
            settings=parent_project.settings.copy() if parent_project.settings else {}
        )

        created_subproject = await self.repository.create(subproject)

        # 如果父项目是协作项目，复制成员权限
        if parent_project.type == 'collab':
            members = await self.repository.get_members(parent_project_id)
            for member in members:
                await self.repository.add_member(
                    created_subproject.id,
                    member['user_id'],
                    member['role']
                )

        return created_subproject

    async def get_subprojects(
        self,
        user_id: str,
        parent_project_id: str,
        page: int = 1,
        page_size: int = 20
    ) -> Dict[str, Any]:
        """获取子项目列表"""
        # 检查父项目权限
        if not await self.repository.check_user_permission(
            user_id, parent_project_id, 'viewer'
        ):
            raise PermissionError("没有权限访问此项目")

        subprojects = await self.repository.get_subprojects(
            parent_project_id, page, page_size
        )

        total = await self.repository.count_subprojects(parent_project_id)

        return {
            'items': subprojects,
            'total': total,
            'page': page,
            'page_size': page_size,
            'total_pages': (total + page_size - 1) // page_size
        }

    async def _generate_unique_project_name(
        self,
        base_name: str,
        folder_id: str,
        user_id: str
    ) -> str:
        """生成唯一的项目名称"""
        # 获取同文件夹下的所有项目名称
        projects = await self.repository.get_by_folder(folder_id, user_id)
        existing_names = {p.name for p in projects}

        # 如果基础名称不存在，直接返回
        if base_name not in existing_names:
            return base_name

        # 否则添加数字后缀
        counter = 1
        while f"{base_name}{counter}" in existing_names:
            counter += 1
        
        return f"{base_name}{counter}"

    async def _create_clone_task(
        self,
        user_id: str,
        project_id: str,
        new_name: str
    ) -> str:
        """创建克隆任务（提交到任务队列）"""
        # TODO: 实现异步任务队列（Celery/ARQ）
        pass

    async def _create_export_task(
        self,
        user_id: str,
        project_id: str
    ) -> str:
        """创建导出任务（提交到任务队列）"""
        # TODO: 实现异步任务队列
        pass

    async def _get_task_status(self, task_id: str) -> Dict[str, Any]:
        """获取任务状态"""
        # TODO: 从 Redis 或任务队列查询状态
        pass

    async def _generate_share_token(self) -> str:
        """生成分享令牌"""
        import secrets
        return secrets.token_urlsafe(32)

    # ==================== 引用完整性校验方法 ====================

    async def _validate_user_exists(self, user_id: str) -> None:
        """校验用户存在性"""
        from app.repositories.user_repository import UserRepository
        user_repo = UserRepository(self.db)
        user = await user_repo.get_by_id(user_id)
        if not user:
            raise ValidationError(f"用户不存在: {user_id}")

    async def _validate_folder_exists(self, folder_id: str) -> None:
        """校验文件夹存在性"""
        folder = await self.folder_repository.get_by_id(folder_id)
        if not folder:
            raise ValidationError(f"文件夹不存在: {folder_id}")

    async def _validate_attachment_exists_and_owned(
        self,
        attachment_id: str,
        user_id: str
    ) -> None:
        """校验附件存在性和所属权"""
        from app.repositories.attachment_repository import AttachmentRepository
        attachment_repo = AttachmentRepository(self.db)
        attachment = await attachment_repo.get_by_id(attachment_id)
        
        if not attachment:
            raise ValidationError(f"附件不存在: {attachment_id}")
        
        if attachment.uploaded_by != user_id:
            raise PermissionError("无权使用此附件")

仓储层实现

ProjectRepository 类

# app/repositories/project_repository.py
from typing import List, Optional, Dict, Any
from sqlalchemy import select, update, delete, func, or_, and_
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.project import Project, ProjectStatus, ProjectType
from app.utils.uuid import generate_uuid_v7
from app.core.logging import get_logger
from datetime import datetime, UTC

logger = get_logger(__name__)

class ProjectRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, project_id: str) -> Optional[Project]:
        """根据ID获取项目"""
        stmt = select(Project).where(
            Project.id == project_id,
            Project.status.in_([ProjectStatus.ACTIVE, ProjectStatus.ARCHIVED])
        )
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def create(self, project: Project) -> Project:
        """创建项目"""
        if not project.id:
            project.id = generate_uuid_v7()
        
        self.db.add(project)
        await self.db.commit()
        await self.db.refresh(project)
        return project

    async def update(self, project_id: str, update_data: Dict[str, Any]) -> Project:
        """更新项目"""
        stmt = (
            update(Project)
            .where(Project.id == project_id)
            .values(**update_data, updated_at=datetime.now(UTC))
            .returning(Project)
        )
        result = await self.db.execute(stmt)
        await self.db.commit()
        return result.scalar_one()

    async def soft_delete(self, project_id: str) -> None:
        """软删除项目（移至回收站）"""
        stmt = (
            update(Project)
            .where(Project.id == project_id)
            .values(
                status=ProjectStatus.TRASHED,
                trashed_at=datetime.now(UTC),
                updated_at=datetime.now(UTC)
            )
        )
        await self.db.execute(stmt)
        await self.db.commit()

    async def get_by_user(
        self,
        user_id: str,
        project_type: Optional[str] = None,
        folder_id: Optional[str] = None,
        content_type: Optional[str] = None,
        search: Optional[str] = None,
        sort_by: str = 'updated_at',
        sort_order: str = 'desc',
        page: int = 1,
        page_size: int = 20
    ) -> List[Project]:
        """获取用户的项目列表（支持筛选、搜索、排序）"""
        stmt = select(Project).where(
            Project.owner_id == user_id,
            Project.status.in_([ProjectStatus.ACTIVE, ProjectStatus.ARCHIVED])
        )

        # 项目类型筛选
        if project_type:
            type_value = ProjectType.from_string(project_type)
            stmt = stmt.where(Project.type == type_value)

        # 文件夹筛选
        if folder_id:
            stmt = stmt.where(Project.folder_id == folder_id)

        # 内容类型筛选
        if content_type:
            from app.models.project import ProjectContentType
            content_type_value = ProjectContentType.from_string(content_type)
            stmt = stmt.where(Project.content_type == content_type_value)

        # 全文搜索
        if search:
            search_pattern = f"%{search}%"
            stmt = stmt.where(
                or_(
                    Project.name.ilike(search_pattern),
                    Project.description.ilike(search_pattern)
                )
            )

        # 排序
        sort_column = getattr(Project, sort_by)
        if sort_order == 'desc':
            stmt = stmt.order_by(sort_column.desc())
        else:
            stmt = stmt.order_by(sort_column.asc())

        # 分页
        offset = (page - 1) * page_size
        stmt = stmt.offset(offset).limit(page_size)

        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def count_by_user(
        self,
        user_id: str,
        project_type: Optional[str] = None,
        folder_id: Optional[str] = None,
        content_type: Optional[str] = None,
        search: Optional[str] = None
    ) -> int:
        """统计用户的项目数量"""
        stmt = select(func.count(Project.id)).where(
            Project.owner_id == user_id,
            Project.status.in_([ProjectStatus.ACTIVE, ProjectStatus.ARCHIVED])
        )

        if project_type:
            type_value = ProjectType.from_string(project_type)
            stmt = stmt.where(Project.type == type_value)

        if folder_id:
            stmt = stmt.where(Project.folder_id == folder_id)

        if content_type:
            from app.models.project import ProjectContentType
            content_type_value = ProjectContentType.from_string(content_type)
            stmt = stmt.where(Project.content_type == content_type_value)

        if search:
            search_pattern = f"%{search}%"
            stmt = stmt.where(
                or_(
                    Project.name.ilike(search_pattern),
                    Project.description.ilike(search_pattern)
                )
            )

        result = await self.db.execute(stmt)
        return result.scalar_one()

    async def move_to_folder(
        self,
        project_id: str,
        folder_id: Optional[str]
    ) -> Project:
        """移动项目到文件夹"""
        return await self.update(project_id, {'folder_id': folder_id})

    async def check_user_permission(
        self,
        user_id: str,
        project_id: str,
        required_role: str
    ) -> bool:
        """检查用户对项目的权限"""
        from app.models.project import MemberRole
        
        # 获取项目
        project = await self.get_by_id(project_id)
        if not project:
            return False

        # 检查是否为所有者
        if project.owner_id == user_id:
            return True

        # 检查成员权限
        if project.type == ProjectType.COLLAB:
            from app.models.project import ProjectMember
            stmt = select(ProjectMember).where(
                ProjectMember.project_id == project_id,
                ProjectMember.user_id == user_id
            )
            result = await self.db.execute(stmt)
            member = result.scalar_one_or_none()
            
            if not member:
                return False

            # 权限层级：owner > editor > viewer
            role_hierarchy = {
                'owner': 3,
                'editor': 2,
                'viewer': 1
            }
            
            member_role_str = MemberRole.to_string(member.role)
            return role_hierarchy.get(member_role_str, 0) >= role_hierarchy.get(required_role, 0)

        return False

    async def get_members(self, project_id: str) -> List[Dict[str, Any]]:
        """获取项目成员列表"""
        from app.models.project import ProjectMember
        stmt = select(ProjectMember).where(ProjectMember.project_id == project_id)
        result = await self.db.execute(stmt)
        members = result.scalars().all()
        
        return [
            {
                'user_id': m.user_id,
                'role': MemberRole.to_string(m.role),
                'invited_by': m.invited_by,
                'created_at': m.created_at.isoformat()
            }
            for m in members
        ]

    async def add_member(
        self,
        project_id: str,
        user_id: str,
        role: str
    ) -> None:
        """添加项目成员"""
        from app.models.project import ProjectMember, MemberRole
        
        member = ProjectMember(
            id=generate_uuid_v7(),
            project_id=project_id,
            user_id=user_id,
            role=MemberRole.from_string(role)
        )
        self.db.add(member)
        await self.db.commit()

    async def remove_member(
        self,
        project_id: str,
        user_id: str
    ) -> None:
        """移除项目成员"""
        from app.models.project import ProjectMember
        stmt = delete(ProjectMember).where(
            ProjectMember.project_id == project_id,
            ProjectMember.user_id == user_id
        )
        await self.db.execute(stmt)
        await self.db.commit()

    async def get_subprojects(
        self,
        parent_project_id: str,
        page: int = 1,
        page_size: int = 20
    ) -> List[Project]:
        """获取子项目列表"""
        stmt = select(Project).where(
            Project.parent_project_id == parent_project_id,
            Project.status.in_([ProjectStatus.ACTIVE, ProjectStatus.ARCHIVED])
        ).order_by(Project.created_at.desc())
        
        offset = (page - 1) * page_size
        stmt = stmt.offset(offset).limit(page_size)
        
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def count_subprojects(self, parent_project_id: str) -> int:
        """统计子项目数量"""
        stmt = select(func.count(Project.id)).where(
            Project.parent_project_id == parent_project_id,
            Project.status.in_([ProjectStatus.ACTIVE, ProjectStatus.ARCHIVED])
        )
        result = await self.db.execute(stmt)
        return result.scalar_one()

    async def get_by_folder(
        self,
        folder_id: Optional[str],
        user_id: str
    ) -> List[Project]:
        """获取文件夹下的所有项目（用于名称唯一性检查）"""
        stmt = select(Project).where(
            Project.folder_id == folder_id,
            Project.owner_id == user_id,
            Project.status.in_([ProjectStatus.ACTIVE, ProjectStatus.ARCHIVED])
        )
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def create_share(
        self,
        project_id: str,
        created_by: str,
        share_token: str,
        password: Optional[str],
        permission: str,
        expires_at: Optional[datetime]
    ) -> Any:
        """创建项目分享"""
        from app.models.project import ProjectShare, SharePermission
        
        share = ProjectShare(
            id=generate_uuid_v7(),
            project_id=project_id,
            share_token=share_token,
            password_hash=password,  # TODO: 实际应该哈希处理
            permission=SharePermission.from_string(permission),
            created_by=created_by,
            expires_at=expires_at
        )
        self.db.add(share)
        await self.db.commit()
        await self.db.refresh(share)
        return share

    async def get_shares(self, project_id: str) -> List[Dict[str, Any]]:
        """获取项目分享列表"""
        from app.models.project import ProjectShare, SharePermission
        stmt = select(ProjectShare).where(ProjectShare.project_id == project_id)
        result = await self.db.execute(stmt)
        shares = result.scalars().all()
        
        return [
            {
                'share_id': s.id,
                'share_url': f"/share/p/{s.share_token}",
                'permission': SharePermission.to_string(s.permission),
                'created_by': s.created_by,
                'created_at': s.created_at.isoformat(),
                'expires_at': s.expires_at.isoformat() if s.expires_at else None,
                'has_password': bool(s.password_hash),
                'access_count': s.access_count
            }
            for s in shares
        ]

    async def delete_share(self, share_id: str) -> None:
        """删除项目分享"""
        from app.models.project import ProjectShare
        stmt = delete(ProjectShare).where(ProjectShare.id == share_id)
        await self.db.execute(stmt)
        await self.db.commit()

说明：

Repository 层负责所有数据库操作
使用 SQLAlchemy 2.0 风格的查询语句
所有方法都是异步的（async/await）
枚举转换在 Repository 层完成
权限检查逻辑封装在 check_user_permission() 方法中

API 接口

1. 获取项目列表

GET /api/v1/projects

查询参数：

type: 项目类型（mine | collab）
folder_id: 文件夹 ID
content_type: 内容类型（ad | movie | series | anime | short | concept）
search: 搜索关键词（按名称、描述搜索）
sort_by: 排序字段（created_at | updated_at | name | display_order）
sort_order: 排序方向（asc | desc，默认 desc）
page: 页码（默认 1）
page_size: 每页数量（默认 20）

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "items": [
      {
        "id": "project-123",
        "name": "我的项目",
        "type": "mine",
        "owner_id": "user-456",
        "folder_id": "folder-789",
        "created_at": "2025-01-27T10:00:00Z"
      }
    ],
    "total": 100,
    "page": 1,
    "page_size": 20,
    "total_pages": 5
  }
}

2. 创建项目

POST /api/v1/projects

请求体：

{
  "name": "我的项目",
  "description": "项目描述",
  "type": "mine",
  "folder_id": "folder-123",
  "settings": {
    "resolution": "1920x1080",
    "fps": 30
  }
}

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "id": "project-123",
    "name": "我的项目",
    "type": "mine",
    "owner_id": "user-456",
    "folder_id": "folder-123",
    "created_at": "2026-02-02T10:00:00Z",
    "updated_at": "2026-02-02T10:00:00Z"
  },
  "message": "项目创建成功"
}

3. 获取项目详情

GET /api/v1/projects/{project_id}

4. 更新项目

PUT /api/v1/projects/{project_id}

5. 移至回收站

DELETE /api/v1/projects/{project_id}

说明：将项目移至回收站（status=2），30天内可恢复

6. 从回收站恢复

POST /api/v1/projects/{project_id}/restore

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "id": "project-123",
    "name": "我的项目",
    "status": "active"
  },
  "message": "项目已恢复"
}

7. 永久删除（从回收站）

DELETE /api/v1/projects/{project_id}/permanent

说明：用户在回收站中立即永久删除项目（status=3），不可恢复

响应（符合统一响应格式）：

{
  "code": 200,
  "message": "项目已永久删除",
  "data": null
}

8. 查看回收站

GET /api/v1/projects/trash

查询参数：

page: 页码（默认 1）
page_size: 每页数量（默认 20）

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "items": [
      {
        "id": "project-123",
        "name": "我的项目",
        "status": "trashed",
        "trashed_at": "2026-01-21T10:00:00Z",
        "days_remaining": 25
      }
    ],
    "total": 10,
    "page": 1,
    "page_size": 20
  }
}

9. 移动项目

POST /api/v1/projects/{project_id}/move

请求体：

{
  "folder_id": "folder-456"
}

10. 克隆项目

POST /api/v1/projects/{project_id}/clone

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "task_id": "task-123",
    "status": "pending"
  },
  "message": "克隆任务已创建"
}

11. 导出项目

POST /api/v1/projects/{project_id}/export

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "task_id": "task-456",
    "status": "pending"
  },
  "message": "导出任务已创建"
}

12. 查询导出进度

GET /api/v1/projects/{project_id}/export/{task_id}

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "task_id": "task-456",
    "status": "completed",
    "progress": 100,
    "download_url": "/api/v1/downloads/project-123.zip",
    "expires_at": "2025-01-28T10:00:00Z"
  }
}

13. 创建分享

POST /api/v1/projects/{project_id}/shares

请求体：

{
  "password": "optional-password",
  "permission": "viewer",
  "expires_at": "2025-02-01T00:00:00Z"
}

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "share_id": "share-123",
    "share_url": "/share/p/abc123token",
    "permission": "viewer",
    "expires_at": "2025-02-01T00:00:00Z"
  }
}

14. 获取分享列表

GET /api/v1/projects/{project_id}/shares

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "shares": [
      {
        "share_id": "share-123",
        "share_url": "/share/p/abc123token",
        "permission": "viewer",
        "created_by": "user-456",
        "created_at": "2025-01-27T10:00:00Z",
        "expires_at": "2025-02-01T00:00:00Z",
        "has_password": true
      }
    ]
  }
}

15. 撤销分享

DELETE /api/v1/projects/{project_id}/shares/{share_id}

16. 更新项目顺序

PUT /api/v1/projects/{project_id}/order

请求体：

{
  "display_order": 10
}

17. 获取子项目列表

GET /api/v1/projects/{parent_project_id}/subprojects

查询参数：

page: 页码（默认 1）
page_size: 每页数量（默认 20）

响应（符合统一响应格式）：

{
  "success": true,
  "data": {
    "items": [
      {
        "project_id": "subproject-123",
        "name": "第一集剧本",
        "parent_project_id": "parent-project-456",
        "screenplay_id": "screenplay-789",
        "type": "mine",
        "created_at": "2025-01-27T10:00:00Z"
      }
    ],
    "total": 5,
    "page": 1,
    "page_size": 20,
    "total_pages": 1
  }
}

说明：

子项目通常由上传剧本时自动创建，不需要手动创建
此接口用于查询父项目下的所有子项目
子项目继承父项目的权限设置

数据库设计

projects 表结构

CREATE TABLE projects (
    project_id UUID PRIMARY KEY,  -- UUID v7 主键，应用层生成

    -- 基本信息
    name TEXT NOT NULL,  -- 项目名称，最大长度 255
    type SMALLINT NOT NULL DEFAULT 1,  -- 项目类型：1=mine(个人项目), 2=collab(协作项目)
    description TEXT,  -- 项目描述，详细描述项目内容和目标

    -- 所有者信息（V1: 仅支持个人用户，V2: 支持企业）
    owner_type TEXT NOT NULL DEFAULT 'user' CHECK (owner_type IN ('user', 'organization')),  -- 所有者类型：user(用户) | organization(组织)
    owner_id UUID NOT NULL,  -- 所有者ID（逻辑外键，无物理约束）

    -- 文件夹归属
    folder_id UUID,  -- 所属文件夹ID（逻辑外键，无物理约束），NULL 表示根目录

    -- 父子项目关系（支持项目层级结构）
    parent_project_id UUID,  -- 父项目ID（逻辑外键，无物理约束），NULL 表示根项目
    screenplay_id UUID,  -- 关联的剧本ID（逻辑外键，无物理约束），仅子项目使用

    -- 显示顺序
    display_order INTEGER NOT NULL DEFAULT 0,  -- 自定义排序顺序，数值越小越靠前

    -- 封面图片
    thumbnail_url TEXT,  -- 缩略图 URL（已废弃，保留兼容性）
    cover_image_id UUID,  -- 封面图片ID（逻辑外键，无物理约束）

    -- 项目预算积分（V1: 暂不使用，V2: 企业为项目充值预算）
    ai_credits_budget INTEGER NOT NULL DEFAULT 0,  -- V1: 保持为 0, V2: 企业为项目充值的专属积分预算
    budget_consumed INTEGER NOT NULL DEFAULT 0,  -- V1: 保持为 0, V2: 已消耗的项目预算积分

    -- 影视项目元数据
    content_type SMALLINT,  -- 项目内容类型：1=ad(广告片), 2=movie(电影), 3=series(剧集), 4=anime(动画), 5=short(短视频), 6=concept(概念片)
    aspect_ratio SMALLINT,  -- 画幅比例：1=16:9, 2=9:16, 3=4:3, 4=21:9, 5=1:1, 6=2.35:1, 7=2.39:1
    planned_duration INTEGER CHECK (planned_duration > 0),  -- 计划总时长（秒），用于项目规划
    actual_duration INTEGER CHECK (actual_duration > 0),  -- 实际总时长（秒），根据分镜/视频自动计算
    visual_style TEXT,  -- 视觉风格（英文名称，如 "cyberpunk", "minimalist", "anime"），用于 AI 生成时的风格参考
    style_and_characters TEXT,  -- 风格和角色描述，记录视觉风格、主要角色等信息

    -- 项目设置
    settings JSONB NOT NULL DEFAULT '{}' CHECK (jsonb_typeof(settings) = 'object'),  -- 项目配置（分辨率、帧率等）

    -- 状态（支持回收站机制）
    status SMALLINT NOT NULL DEFAULT 0 CHECK (status IN (0, 1, 2, 3)),  -- 项目状态：0=active(活跃), 1=archived(归档), 2=trashed(回收站), 3=soft_deleted(软删除)

    -- 时间戳（所有时间戳使用 TIMESTAMPTZ 存储 UTC 时间）
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),  -- 创建时间
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),  -- 最后更新时间
    trashed_at TIMESTAMPTZ,  -- 进入回收站时间，用于计算30天自动清理期限
    deleted_at TIMESTAMPTZ,  -- 软删除时间，用户从回收站删除或自动过期后设置

    -- 唯一约束
    CONSTRAINT projects_name_unique UNIQUE (folder_id, name) NULLS NOT DISTINCT,
    
    -- 子项目约束：子项目必须关联剧本
    CONSTRAINT projects_subproject_screenplay_check CHECK (
        (parent_project_id IS NULL AND screenplay_id IS NULL) OR
        (parent_project_id IS NOT NULL AND screenplay_id IS NOT NULL)
    )
);

-- 索引（优化查询性能）
CREATE INDEX idx_projects_owner ON projects (owner_type, owner_id) WHERE status IN (0, 1);  -- 按所有者查询
CREATE INDEX idx_projects_folder_id ON projects (folder_id) WHERE status IN (0, 1) AND folder_id IS NOT NULL;  -- 按文件夹查询
CREATE INDEX idx_projects_type ON projects (type) WHERE status IN (0, 1);  -- 按项目类型筛选
CREATE INDEX idx_projects_status ON projects (status);  -- 按状态筛选
CREATE INDEX idx_projects_trashed_at ON projects (trashed_at) WHERE status = 2;  -- 回收站自动清理
CREATE INDEX idx_projects_created_at ON projects (created_at) WHERE status IN (0, 1);  -- 按创建时间排序
CREATE INDEX idx_projects_updated_at ON projects (updated_at) WHERE status IN (0, 1);  -- 按更新时间排序
CREATE INDEX idx_projects_display_order ON projects (folder_id, display_order) WHERE status IN (0, 1);  -- 自定义排序
CREATE INDEX idx_projects_cover_image_id ON projects (cover_image_id) WHERE cover_image_id IS NOT NULL;  -- 封面图片关联
CREATE INDEX idx_projects_settings_gin ON projects USING GIN (settings);  -- JSONB 字段查询
CREATE INDEX idx_projects_name_trgm ON projects USING GIN (name gin_trgm_ops) WHERE status IN (0, 1);  -- 项目名称全文搜索
CREATE INDEX idx_projects_description_trgm ON projects USING GIN (description gin_trgm_ops) WHERE status IN (0, 1) AND description IS NOT NULL;  -- 描述全文搜索
CREATE INDEX idx_projects_content_type ON projects (content_type) WHERE status IN (0, 1) AND content_type IS NOT NULL;  -- 按内容类型筛选
CREATE INDEX idx_projects_aspect_ratio ON projects (aspect_ratio) WHERE status IN (0, 1) AND aspect_ratio IS NOT NULL;  -- 按画幅比例筛选
CREATE INDEX idx_projects_parent_project_id ON projects (parent_project_id) WHERE parent_project_id IS NOT NULL AND status IN (0, 1);  -- 按父项目查询子项目
CREATE INDEX idx_projects_screenplay_id ON projects (screenplay_id) WHERE screenplay_id IS NOT NULL;  -- 按剧本查询子项目

-- 表和字段注释
COMMENT ON TABLE projects IS '项目表，存储视频制作项目的基本信息和元数据';
COMMENT ON COLUMN projects.project_id IS '项目ID (UUID v7)';
COMMENT ON COLUMN projects.name IS '项目名称';
COMMENT ON COLUMN projects.type IS '项目类型: 1=mine(个人项目), 2=collab(协作项目)';
COMMENT ON COLUMN projects.description IS '项目描述';
COMMENT ON COLUMN projects.owner_type IS '所有者类型: user(用户) | organization(组织)';
COMMENT ON COLUMN projects.owner_id IS '所有者ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN projects.folder_id IS '所属文件夹ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN projects.parent_project_id IS '父项目ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN projects.screenplay_id IS '关联的剧本ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN projects.display_order IS '自定义排序顺序';
COMMENT ON COLUMN projects.thumbnail_url IS '缩略图 URL（已废弃）';
COMMENT ON COLUMN projects.cover_image_id IS '封面图片ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN projects.ai_credits_budget IS 'V1: 保持为 0, V2: 企业为项目充值的专属积分预算';
COMMENT ON COLUMN projects.budget_consumed IS 'V1: 保持为 0, V2: 已消耗的项目预算积分';
COMMENT ON COLUMN projects.content_type IS '项目内容类型: 1=ad, 2=movie, 3=series, 4=anime, 5=short, 6=concept';
COMMENT ON COLUMN projects.aspect_ratio IS '画幅比例: 1=16:9, 2=9:16, 3=4:3, 4=21:9, 5=1:1, 6=2.35:1, 7=2.39:1';
COMMENT ON COLUMN projects.planned_duration IS '计划总时长（秒）';
COMMENT ON COLUMN projects.actual_duration IS '实际总时长（秒）';
COMMENT ON COLUMN projects.visual_style IS '视觉风格（英文名称，如 cyberpunk, minimalist）';
COMMENT ON COLUMN projects.style_and_characters IS '风格和角色描述';
COMMENT ON COLUMN projects.settings IS '项目配置（JSONB格式）';
COMMENT ON COLUMN projects.status IS '项目状态: 0=active, 1=archived, 2=trashed, 3=soft_deleted';
COMMENT ON COLUMN projects.created_at IS '创建时间（自动记录时区）';
COMMENT ON COLUMN projects.updated_at IS '更新时间（自动记录时区）';
COMMENT ON COLUMN projects.trashed_at IS '进入回收站时间（自动记录时区）';
COMMENT ON COLUMN projects.deleted_at IS '软删除时间（自动记录时区）';

-- 触发器
CREATE TRIGGER update_projects_updated_at
    BEFORE UPDATE ON projects
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at_column();

project_shares 表结构

CREATE TABLE project_shares (
    share_id UUID PRIMARY KEY,  -- UUID v7 主键
    project_id UUID NOT NULL,  -- 关联项目ID（逻辑外键，无物理约束）
    share_token TEXT NOT NULL UNIQUE,  -- 分享令牌，用于生成公开访问链接
    password_hash TEXT,  -- 访问密码的哈希值（可选）
    permission SMALLINT NOT NULL DEFAULT 1,  -- 分享权限：1=viewer(查看), 2=editor(编辑)
    created_by UUID NOT NULL,  -- 创建者用户ID（逻辑外键，无物理约束）
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),  -- 创建时间
    expires_at TIMESTAMPTZ,  -- 过期时间，NULL 表示永不过期
    last_accessed_at TIMESTAMPTZ,  -- 最后访问时间
    access_count INTEGER NOT NULL DEFAULT 0,  -- 访问次数统计
    CONSTRAINT project_shares_token_unique UNIQUE (share_token)
);

-- 索引
CREATE INDEX idx_project_shares_project_id ON project_shares (project_id);  -- 按项目查询分享
CREATE INDEX idx_project_shares_token ON project_shares (share_token);  -- 按令牌查询
CREATE INDEX idx_project_shares_created_by ON project_shares (created_by);  -- 按创建者查询
CREATE INDEX idx_project_shares_expires_at ON project_shares (expires_at) WHERE expires_at IS NOT NULL;  -- 过期清理

-- 表和字段注释
COMMENT ON TABLE project_shares IS '项目分享链接表，管理公开分享链接和权限';
COMMENT ON COLUMN project_shares.share_id IS '分享记录ID (UUID v7)';
COMMENT ON COLUMN project_shares.project_id IS '关联项目ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN project_shares.share_token IS '分享令牌';
COMMENT ON COLUMN project_shares.password_hash IS '访问密码哈希值';
COMMENT ON COLUMN project_shares.permission IS '分享权限: 1=viewer, 2=editor';
COMMENT ON COLUMN project_shares.created_by IS '创建者用户ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN project_shares.created_at IS '创建时间（自动记录时区）';
COMMENT ON COLUMN project_shares.expires_at IS '过期时间（自动记录时区）';
COMMENT ON COLUMN project_shares.last_accessed_at IS '最后访问时间（自动记录时区）';
COMMENT ON COLUMN project_shares.access_count IS '访问次数统计';

project_members 表结构

CREATE TABLE project_members (
    member_id UUID PRIMARY KEY,  -- UUID v7 主键
    project_id UUID NOT NULL,  -- 关联项目ID（逻辑外键，无物理约束）
    user_id UUID NOT NULL,  -- 成员用户ID（逻辑外键，无物理约束）
    role SMALLINT NOT NULL DEFAULT 3,  -- 成员角色：1=owner(所有者), 2=editor(编辑者), 3=viewer(查看者)
    invited_by UUID,  -- 邀请人用户ID（逻辑外键，无物理约束）
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),  -- 创建时间（邀请创建时间）
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),  -- 更新时间
    CONSTRAINT project_members_unique UNIQUE (project_id, user_id) NULLS NOT DISTINCT
);

-- 索引
CREATE INDEX idx_project_members_project_id ON project_members (project_id);  -- 按项目查询成员
CREATE INDEX idx_project_members_user_id ON project_members (user_id);  -- 按用户查询项目
CREATE INDEX idx_project_members_role ON project_members (role);  -- 按角色筛选

-- 表和字段注释
COMMENT ON TABLE project_members IS '项目成员表，管理协作项目的成员和权限';
COMMENT ON COLUMN project_members.member_id IS '成员记录ID (UUID v7)';
COMMENT ON COLUMN project_members.project_id IS '关联项目ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN project_members.user_id IS '成员用户ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN project_members.role IS '成员角色: 1=owner, 2=editor, 3=viewer';
COMMENT ON COLUMN project_members.invited_by IS '邀请人用户ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN project_members.created_at IS '创建时间（邀请创建时间，自动记录时区）';
COMMENT ON COLUMN project_members.updated_at IS '更新时间（自动记录时区）';

-- 触发器
CREATE TRIGGER update_project_members_updated_at
    BEFORE UPDATE ON project_members
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at_column();

project_versions 表结构

CREATE TABLE project_versions (
    version_id UUID PRIMARY KEY,  -- UUID v7 主键
    project_id UUID NOT NULL,  -- 关联项目ID（逻辑外键，无物理约束）
    version_number INTEGER NOT NULL,  -- 版本号，从 1 开始递增
    snapshot_data JSONB NOT NULL,  -- 项目快照数据（完整的项目状态）
    created_by UUID NOT NULL,  -- 创建者用户ID（逻辑外键，无物理约束）
    note TEXT,  -- 版本说明
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),  -- 创建时间
    CONSTRAINT project_versions_unique UNIQUE (project_id, version_number)
);

-- 索引
CREATE INDEX idx_project_versions_project_id ON project_versions (project_id);  -- 按项目查询版本
CREATE INDEX idx_project_versions_created_at ON project_versions (created_at);  -- 按时间排序
CREATE INDEX idx_project_versions_snapshot_gin ON project_versions USING GIN (snapshot_data);  -- JSONB 查询

-- 表和字段注释
COMMENT ON TABLE project_versions IS '项目版本表，存储项目快照用于版本控制和回滚';
COMMENT ON COLUMN project_versions.version_id IS '版本记录ID (UUID v7)';
COMMENT ON COLUMN project_versions.project_id IS '关联项目ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN project_versions.version_number IS '版本号（从 1 开始递增）';
COMMENT ON COLUMN project_versions.snapshot_data IS '项目快照数据（JSONB格式）';
COMMENT ON COLUMN project_versions.created_by IS '创建者用户ID（逻辑外键，无物理约束）';
COMMENT ON COLUMN project_versions.note IS '版本说明';
COMMENT ON COLUMN project_versions.created_at IS '创建时间（自动记录时区）';

设计说明

UUID v7 主键：所有表使用 UUID v7 作为主键，应用层生成（app.utils.uuid.generate_uuid_v7()）
项目类型：使用 SMALLINT 存储，后端枚举映射区分个人项目（mine）和协作项目（collab）
父子项目关系：
- parent_project_id：父项目 ID，NULL 表示根项目
- screenplay_id：关联的剧本 ID（仅子项目使用）
- 约束：子项目必须关联剧本（parent_project_id 和 screenplay_id 要么都为 NULL，要么都不为 NULL）
- 子项目继承父项目的权限设置
- 子项目可以独立管理分镜、资源、导出等功能
多态所有者（V2 预留）：
- owner_type：所有者类型（'user' | 'organization'）
- owner_id：所有者 ID，根据 owner_type 指向不同表
- V1 阶段：owner_type 固定为 'user'，owner_id 指向 users.user_id
- V2 阶段：支持 owner_type='organization'，owner_id 指向 organizations.organization_id
项目预算积分（V2 预留）：
- ai_credits_budget：项目专属积分预算（企业为项目充值）
- budget_consumed：已消耗的预算积分
- V1 阶段：两个字段保持为 0，积分从 users.ai_credits_balance 扣除
- V2 阶段：启用项目预算功能，优先从项目预算扣除
影视项目元数据：
- content_type：项目内容类型（广告片/电影/剧集/动画/短视频/概念片）
- aspect_ratio：画幅比例（16:9/9:16/4:3/21:9/1:1/2.35:1/2.39:1）
- planned_duration：计划总时长（秒），用于项目规划
- actual_duration：实际总时长（秒），根据分镜/视频自动计算
- visual_style：视觉风格（英文名称，如 "cyberpunk", "minimalist", "anime"），用于 AI 生成时的风格参考
- style_and_characters：风格和角色描述，记录视觉风格、主要角色等信息
文件夹归属：folder_id 支持项目分组管理
显示顺序：display_order 支持自定义排序，默认为 0
封面图片：cover_image_id 关联 attachments 表（一对一）
项目设置：settings 使用 JSONB 存储灵活配置（分辨率、帧率等）
名称唯一：同一文件夹下项目名称唯一
协作管理：project_members 表管理项目成员和权限
版本控制：project_versions 表存储项目快照，支持版本回滚
软删除：使用 deleted_at 字段
全文搜索：使用 pg_trgm 扩展支持项目名称、描述模糊搜索
分享管理：project_shares 表管理公开分享链接
- share_token：唯一的分享令牌
- password_hash：可选的访问密码
- permission：分享权限（viewer/editor）
- expires_at：过期时间（可选）
- access_count：访问统计
扩展性设计：
- V1 实现简单，只处理个人用户场景
- V2 升级时无需修改表结构，只需启用预留字段
- 业务逻辑通过 owner_type 判断，易于扩展

测试

单元测试

# 运行 ProjectService 单元测试
docker exec jointo-server-app pytest tests/unit/services/test_project_service.py -v

# 运行特定测试
docker exec jointo-server-app pytest tests/unit/services/test_project_service.py::test_create_project -v

集成测试

# 运行 ProjectService 集成测试
docker exec jointo-server-app pytest tests/integration/services/test_project_service.py -v

# 运行权限检查测试
docker exec jointo-server-app pytest tests/integration/services/test_project_service.py::test_project_permissions -v

测试覆盖率

# 生成测试覆盖率报告
docker exec jointo-server-app pytest tests/ --cov=app.services.project_service --cov-report=html

数据库迁移

创建迁移文件

# 1. 创建迁移文件
docker exec jointo-server-app alembic revision -m "create_projects_tables"

# 2. 编辑迁移文件
# 将本文档中的 SQL 表结构复制到迁移文件的 upgrade() 函数中

# 3. 检查迁移状态
docker exec jointo-server-app alembic current

# 4. 执行迁移
docker exec jointo-server-app python scripts/db_migrate.py upgrade

# 5. 验证迁移
docker exec jointo-server-app alembic history

回滚迁移

# 回滚到上一个版本
docker exec jointo-server-app python scripts/db_migrate.py downgrade -1

# 回滚到指定版本
docker exec jointo-server-app python scripts/db_migrate.py downgrade <revision_id>

数据模型

枚举值映射表

枚举类型	数值	字符串值	说明
project_type	1	mine	个人项目
	2	collab	协作项目
project_status	0	active	活跃项目
	1	archived	用户归档
	2	trashed	回收站（30天内可恢复）
	3	soft_deleted	软删除（用户不可见）
content_type	1	ad	广告片
	2	movie	电影
	3	series	剧集
	4	anime	动画
	5	short	短视频
	6	concept	概念片
aspect_ratio	1	16:9	标准宽屏
	2	9:16	竖屏
	3	4:3	传统电视
	4	21:9	超宽屏
	5	1:1	正方形
	6	2.35:1	电影宽屏
	7	2.39:1	电影宽屏
share_permission	1	viewer	查看权限
	2	editor	编辑权限
member_role	1	owner	所有者
	2	editor	编辑者
	3	viewer	查看者

Project 模型

# app/models/project.py
from sqlalchemy import Column, String, Text, ForeignKey, JSONB, Integer, SmallInteger
from sqlalchemy.dialects.postgresql import TIMESTAMP
from sqlalchemy.orm import relationship
from app.core.database import Base
from app.utils.uuid import generate_uuid_v7
from datetime import datetime, UTC
from enum import IntEnum

class ProjectStatus(IntEnum):
    """项目状态枚举"""
    ACTIVE = 0
    ARCHIVED = 1
    TRASHED = 2
    SOFT_DELETED = 3

    @classmethod
    def from_string(cls, value: str) -> int:
        """字符串转数字"""
        mapping = {
            'active': cls.ACTIVE,
            'archived': cls.ARCHIVED,
            'trashed': cls.TRASHED,
            'soft_deleted': cls.SOFT_DELETED
        }
        return mapping.get(value.lower())

    @classmethod
    def to_string(cls, value: int) -> str:
        """数字转字符串"""
        mapping = {
            cls.ACTIVE: 'active',
            cls.ARCHIVED: 'archived',
            cls.TRASHED: 'trashed',
            cls.SOFT_DELETED: 'soft_deleted'
        }
        return mapping.get(value)

class ProjectType(IntEnum):
    """项目类型枚举"""
    MINE = 1
    COLLAB = 2

    @classmethod
    def from_string(cls, value: str) -> int:
        """字符串转数字"""
        mapping = {'mine': cls.MINE, 'collab': cls.COLLAB}
        return mapping.get(value.lower())

    @classmethod
    def to_string(cls, value: int) -> str:
        """数字转字符串"""
        mapping = {cls.MINE: 'mine', cls.COLLAB: 'collab'}
        return mapping.get(value)

class ProjectContentType(IntEnum):
    """项目内容类型枚举"""
    AD = 1
    MOVIE = 2
    SERIES = 3
    ANIME = 4
    SHORT = 5
    CONCEPT = 6

    @classmethod
    def from_string(cls, value: str) -> int:
        mapping = {
            'ad': cls.AD,
            'movie': cls.MOVIE,
            'series': cls.SERIES,
            'anime': cls.ANIME,
            'short': cls.SHORT,
            'concept': cls.CONCEPT
        }
        return mapping.get(value.lower())

    @classmethod
    def to_string(cls, value: int) -> str:
        mapping = {
            cls.AD: 'ad',
            cls.MOVIE: 'movie',
            cls.SERIES: 'series',
            cls.ANIME: 'anime',
            cls.SHORT: 'short',
            cls.CONCEPT: 'concept'
        }
        return mapping.get(value)

class AspectRatioType(IntEnum):
    """画幅比例枚举"""
    RATIO_16_9 = 1
    RATIO_9_16 = 2
    RATIO_4_3 = 3
    RATIO_21_9 = 4
    RATIO_1_1 = 5
    RATIO_2_35_1 = 6
    RATIO_2_39_1 = 7

    @classmethod
    def from_string(cls, value: str) -> int:
        mapping = {
            '16:9': cls.RATIO_16_9,
            '9:16': cls.RATIO_9_16,
            '4:3': cls.RATIO_4_3,
            '21:9': cls.RATIO_21_9,
            '1:1': cls.RATIO_1_1,
            '2.35:1': cls.RATIO_2_35_1,
            '2.39:1': cls.RATIO_2_39_1
        }
        return mapping.get(value)

    @classmethod
    def to_string(cls, value: int) -> str:
        mapping = {
            cls.RATIO_16_9: '16:9',
            cls.RATIO_9_16: '9:16',
            cls.RATIO_4_3: '4:3',
            cls.RATIO_21_9: '21:9',
            cls.RATIO_1_1: '1:1',
            cls.RATIO_2_35_1: '2.35:1',
            cls.RATIO_2_39_1: '2.39:1'
        }
        return mapping.get(value)

class SharePermission(IntEnum):
    """分享权限枚举"""
    VIEWER = 1
    EDITOR = 2

    @classmethod
    def from_string(cls, value: str) -> int:
        mapping = {'viewer': cls.VIEWER, 'editor': cls.EDITOR}
        return mapping.get(value.lower())

    @classmethod
    def to_string(cls, value: int) -> str:
        mapping = {cls.VIEWER: 'viewer', cls.EDITOR: 'editor'}
        return mapping.get(value)

class MemberRole(IntEnum):
    """成员角色枚举"""
    OWNER = 1
    EDITOR = 2
    VIEWER = 3

    @classmethod
    def from_string(cls, value: str) -> int:
        mapping = {'owner': cls.OWNER, 'editor': cls.EDITOR, 'viewer': cls.VIEWER}
        return mapping.get(value.lower())

    @classmethod
    def to_string(cls, value: int) -> str:
        mapping = {cls.OWNER: 'owner', cls.EDITOR: 'editor', cls.VIEWER: 'viewer'}
        return mapping.get(value)

class Project(Base):
    __tablename__ = "projects"

    # 使用 UUID v7 主键（应用层生成）
    id = Column('project_id', String, primary_key=True, default=generate_uuid_v7)
    
    def __repr__(self):
        return f"<Project(id={self.id}, name={self.name}, type={self.type_str})>"
    name = Column(String(255), nullable=False)
    description = Column(Text)
    type = Column(SmallInteger, nullable=False, default=ProjectType.MINE)
    owner_type = Column(String, nullable=False, default='user')
    owner_id = Column(String, nullable=False)
    folder_id = Column(String)
    parent_project_id = Column(String)  # 父项目ID
    screenplay_id = Column(String)  # 关联的剧本ID（仅子项目）
    display_order = Column(Integer, default=0, nullable=False)
    thumbnail_url = Column(String(500))
    cover_image_id = Column(String)

    # 项目预算积分（V2 预留）
    ai_credits_budget = Column(Integer, nullable=False, default=0)
    budget_consumed = Column(Integer, nullable=False, default=0)

    # 影视项目元数据
    content_type = Column(SmallInteger)
    aspect_ratio = Column(SmallInteger)
    planned_duration = Column(Integer)
    actual_duration = Column(Integer)
    visual_style = Column(Text)  # 视觉风格（英文名称）
    style_and_characters = Column(Text)

    # 项目设置
    settings = Column(JSONB, default={})

    # 状态
    status = Column(SmallInteger, nullable=False, default=ProjectStatus.ACTIVE)

    # 时间戳（使用 TIMESTAMPTZ 存储 UTC 时间）
    created_at = Column(TIMESTAMP(timezone=True), default=lambda: datetime.now(UTC))
    updated_at = Column(TIMESTAMP(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
    trashed_at = Column(TIMESTAMP(timezone=True))
    deleted_at = Column(TIMESTAMP(timezone=True))

    # 关系（注意：数据库层不使用外键约束，仅应用层维护引用完整性）
    # owner = relationship("User", back_populates="projects")
    # folder = relationship("Folder", back_populates="projects")
    # storyboards = relationship("Storyboard", back_populates="project")
    # members = relationship("ProjectMember", back_populates="project")

    @property
    def type_str(self) -> str:
        """获取项目类型字符串"""
        return ProjectType.to_string(self.type)

    @property
    def status_str(self) -> str:
        """获取项目状态字符串"""
        return ProjectStatus.to_string(self.status)

    @property
    def content_type_str(self) -> str:
        """获取内容类型字符串"""
        return ProjectContentType.to_string(self.content_type) if self.content_type else None

    @property
    def aspect_ratio_str(self) -> str:
        """获取画幅比例字符串"""
        return AspectRatioType.to_string(self.aspect_ratio) if self.aspect_ratio else None

    @property
    def is_subproject(self) -> bool:
        """判断是否为子项目"""
        return self.parent_project_id is not None

ProjectCreate Schema

# app/schemas/project.py
from pydantic import BaseModel, Field, field_validator, ConfigDict
from typing import Optional, Dict, Any
from enum import Enum
from app.models.project import ProjectType, ProjectContentType, AspectRatioType, SharePermission, MemberRole

class ProjectSettings(BaseModel):
    resolution: str = "1920x1080"
    fps: int = 30
    duration: Optional[int] = None

class ProjectCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=255)
    description: Optional[str] = None
    type: str = Field(..., pattern="^(mine|collab)$")
    folder_id: Optional[str] = None
    content_type: Optional[str] = None  # 'ad', 'movie', 'series', 'anime', 'short', 'concept'
    aspect_ratio: Optional[str] = None  # '16:9', '9:16', '4:3', '21:9', '1:1', '2.35:1', '2.39:1'
    planned_duration: Optional[int] = Field(None, gt=0)
    visual_style: Optional[str] = None  # 视觉风格（英文名称，如 "cyberpunk", "minimalist"）
    style_and_characters: Optional[str] = None
    settings: Optional[ProjectSettings] = None

    @field_validator('content_type')
    @classmethod
    def validate_content_type(cls, v: Optional[str]) -> Optional[str]:
        if v is not None:
            valid_types = ['ad', 'movie', 'series', 'anime', 'short', 'concept']
            if v.lower() not in valid_types:
                raise ValueError(f'content_type must be one of {valid_types}')
        return v

    @field_validator('aspect_ratio')
    @classmethod
    def validate_aspect_ratio(cls, v: Optional[str]) -> Optional[str]:
        if v is not None:
            valid_ratios = ['16:9', '9:16', '4:3', '21:9', '1:1', '2.35:1', '2.39:1']
            if v not in valid_ratios:
                raise ValueError(f'aspect_ratio must be one of {valid_ratios}')
        return v

class ProjectUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=255)
    description: Optional[str] = None
    thumbnail_url: Optional[str] = None
    content_type: Optional[str] = None
    aspect_ratio: Optional[str] = None
    planned_duration: Optional[int] = Field(None, gt=0)
    actual_duration: Optional[int] = Field(None, gt=0)
    visual_style: Optional[str] = None  # 视觉风格（英文名称）
    style_and_characters: Optional[str] = None
    settings: Optional[ProjectSettings] = None

    @field_validator('content_type')
    @classmethod
    def validate_content_type(cls, v: Optional[str]) -> Optional[str]:
        if v is not None:
            valid_types = ['ad', 'movie', 'series', 'anime', 'short', 'concept']
            if v.lower() not in valid_types:
                raise ValueError(f'content_type must be one of {valid_types}')
        return v

    @field_validator('aspect_ratio')
    @classmethod
    def validate_aspect_ratio(cls, v: Optional[str]) -> Optional[str]:
        if v is not None:
            valid_ratios = ['16:9', '9:16', '4:3', '21:9', '1:1', '2.35:1', '2.39:1']
            if v not in valid_ratios:
                raise ValueError(f'aspect_ratio must be one of {valid_ratios}')
        return v

class ProjectMove(BaseModel):
    folder_id: Optional[str] = None

class ProjectClone(BaseModel):
    """克隆项目请求"""
    pass  # 无需额外参数，使用默认行为

class ProjectExport(BaseModel):
    """导出项目请求"""
    pass  # 无需额外参数

class ProjectShareCreate(BaseModel):
    password: Optional[str] = None
    permission: str = Field(default='viewer', pattern='^(viewer|editor)$')
    expires_at: Optional[str] = None  # ISO 8601 格式

class ProjectShareResponse(BaseModel):
    share_id: str
    share_url: str
    permission: str
    created_by: str
    created_at: str
    expires_at: Optional[str]
    has_password: bool
    access_count: int

    model_config = ConfigDict(from_attributes=True)

class ProjectOrderUpdate(BaseModel):
    display_order: int = Field(..., ge=0)

class TaskStatusResponse(BaseModel):
    task_id: str
    status: str  # 'pending' | 'processing' | 'completed' | 'failed'
    progress: int = Field(..., ge=0, le=100)
    message: Optional[str] = None
    download_url: Optional[str] = None
    expires_at: Optional[str] = None

class ProjectResponse(BaseModel):
    id: str
    name: str
    description: Optional[str]
    type: str  # 返回字符串 'mine' 或 'collab'
    owner_id: str
    folder_id: Optional[str]
    parent_project_id: Optional[str]  # 父项目ID
    screenplay_id: Optional[str]  # 关联的剧本ID
    is_subproject: bool  # 是否为子项目
    display_order: int
    thumbnail_url: Optional[str]
    content_type: Optional[str]  # 返回字符串 'ad', 'movie' 等
    aspect_ratio: Optional[str]  # 返回字符串 '16:9', '9:16' 等
    planned_duration: Optional[int]
    actual_duration: Optional[int]
    visual_style: Optional[str]  # 视觉风格（英文名称）
    style_and_characters: Optional[str]
    settings: Dict[str, Any]
    created_at: str
    updated_at: str

    model_config = ConfigDict(from_attributes=True)

    @classmethod
    def from_orm(cls, obj):
        """从 ORM 对象转换，自动处理枚举转换"""
        return cls(
            id=obj.id,
            name=obj.name,
            description=obj.description,
            type=obj.type_str,  # 使用模型的 property 方法
            owner_id=obj.owner_id,
            folder_id=obj.folder_id,
            parent_project_id=obj.parent_project_id,
            screenplay_id=obj.screenplay_id,
            is_subproject=obj.is_subproject,
            display_order=obj.display_order,
            thumbnail_url=obj.thumbnail_url,
            content_type=obj.content_type_str,
            aspect_ratio=obj.aspect_ratio_str,
            planned_duration=obj.planned_duration,
            actual_duration=obj.actual_duration,
            visual_style=obj.visual_style,
            style_and_characters=obj.style_and_characters,
            settings=obj.settings,
            created_at=obj.created_at.isoformat(),
            updated_at=obj.updated_at.isoformat()
        )

说明：

API 层面保持使用字符串（'mine', 'collab', 'ad', '16:9' 等）
数据库层面使用 SMALLINT 存储
模型层提供 from_string() 和 to_string() 方法进行转换
Schema 层使用 field_validator 验证输入
Response 使用 from_orm() 自动转换枚举为字符串


---

## 相关文档

- [文件夹管理服务](./folder-service.md)
- [分镜管理服务](./storyboard-service.md)

---

## 变更记录

**v2.1 (2026-01-31)**

- 新增视觉风格字段：
  - 项目表新增 `visual_style` 字段（TEXT，可选）
  - 存储视觉风格的英文名称（如 "cyberpunk", "minimalist", "anime"）
  - 用于 AI 生成时的风格参考
  - 更新 `ProjectCreate`、`ProjectUpdate`、`ProjectResponse` Schema
  - 更新 `Project` 模型

**v2.0 (2026-01-31)**

- 新增子项目功能：
  - 项目表新增 `parent_project_id` 和 `screenplay_id` 字段
  - 支持父子项目关系（项目层级结构）
  - 子项目关联剧本（一对一）
  - 子项目继承父项目权限和设置
  - 新增 `create_subproject()` 方法
  - 新增 `get_subprojects()` 方法
  - 新增 `GET /api/v1/projects/{parent_project_id}/subprojects` 接口
  - 更新数据库设计（新增约束和索引）
  - 更新 `Project` 模型（新增 `is_subproject` 属性）
  - 更新 `ProjectResponse` Schema（新增子项目相关字段）

**v1.0 (2026-01-20)**

- 初始版本

---

**文档版本**：v2.1  
**最后更新**：2026-01-31

### ✅ 已完成（2026-01-20）

**核心功能**：
- 项目 CRUD 操作（创建、查询、更新、删除）
- 权限管理（owner/editor/viewer 三级权限）
- 成员管理（添加、移除、角色管理）
- 搜索和筛选（全文搜索 + 多条件筛选）
- 排序功能（时间/名称/自定义顺序）
- 分享功能（链接生成、密码保护、权限控制、过期时间）
- 项目移动（文件夹间移动）

**技术实现**：
- 数据模型层：`server/app/models/project.py`
- Schema 层：`server/app/schemas/project.py`
- 仓储层：`server/app/repositories/project_repository.py`
- 服务层：`server/app/services/project_service.py`
- API 层：`server/app/api/v1/projects.py`
- 数据库迁移：`server/app/migrations/003_project_tables.py`

**数据库特性**：
- 使用 UUID v7 主键（时间排序）
- pg_trgm 扩展支持全文搜索
- JSONB 存储灵活配置
- 软删除机制
- 自动更新时间戳

### ⚠️ 部分实现

**克隆功能**：
- 状态：同步版本已实现
- 限制：大项目可能超时
- 位置：`ProjectService.clone_project()`
- TODO：实现异步任务队列后改为异步处理

**导出功能**：
- 状态：接口框架已实现
- 限制：返回"开发中"状态
- 位置：`ProjectService.export_project()`
- TODO：实现异步任务队列和实际导出逻辑

### 📋 待实现

**异步任务系统**：
- 任务队列（Celery/ARQ）
- Redis 任务状态追踪
- 进度通知机制
- 文件下载管理

**版本控制功能**：
- 项目快照创建
- 版本回滚
- 版本对比
- 模型已预留：`ProjectVersion`

**企业功能（V2）**：
- 组织所有者支持（owner_type='organization'）
- 项目预算积分（ai_credits_budget）
- 预算消耗追踪（budget_consumed）

**资源关联**：
- 克隆时复制分镜
- 克隆时复制视频
- 克隆时复制附件
- 需要等待相关模型实现

### 🔧 技术债务

1. **全文搜索优化**：当前使用 ILIKE，可升级为 pg_trgm 的 similarity 函数
2. **权限继承**：项目权限可从文件夹继承，需要完善继承逻辑
3. **批量操作**：支持批量移动、批量删除等操作
4. **统计信息**：项目统计（分镜数、视频数、总时长等）

---

**实现完成时间**：2026-01-20

75 KiB Raw Permalink Blame History

项目管理服务

目录

服务概述

职责

核心功能

1. 项目创建

2. 项目查询

3. 项目更新

4. 项目删除（回收站机制）

5. 项目移动

6. 项目克隆

7. 项目导出

8. 项目分享

9. 项目搜索和筛选

10. 项目排序

11. 子项目管理

服务实现

服务实现

ProjectService 类

仓储层实现

ProjectRepository 类

API 接口

1. 获取项目列表

2. 创建项目

3. 获取项目详情

4. 更新项目

5. 移至回收站

6. 从回收站恢复

7. 永久删除（从回收站）

8. 查看回收站

9. 移动项目

10. 克隆项目

11. 导出项目

12. 查询导出进度

13. 创建分享

14. 获取分享列表

15. 撤销分享

16. 更新项目顺序

17. 获取子项目列表

数据库设计

projects 表结构

project_shares 表结构

project_members 表结构

project_versions 表结构

设计说明

测试

单元测试

集成测试

测试覆盖率

数据库迁移

创建迁移文件

回滚迁移

数据模型

枚举值映射表

Project 模型

ProjectCreate Schema

75 KiB

Raw Permalink Blame History