59 KiB

Raw Permalink Blame History

项目素材服务

文档版本：v3.7
最后更新：2026-02-03
符合规范：jointo-tech-stack v1.0

服务概述

项目素材服务（ProjectResourceService）负责管理项目专属的角色、场景、道具素材。每个素材归属一个项目，支持用户上传和 AI 生成。

职责

项目素材上传（角色、场景、道具图片）
项目素材管理（查询、更新、删除）
AI 生成素材管理
素材分类和搜索
素材与分镜的关联

设计原则

项目隔离：每个素材归属一个项目，项目间素材隔离
类型分类：角色（character）、场景（scene）、道具（prop）、实拍（footage）
去重支持：使用 FileStorageService 实现文件去重
AI 集成：支持 AI 生成素材的管理
UUID v7 主键：应用层生成 UUID v7，非数据库默认值
无物理外键：禁止数据库外键约束，应用层保证引用完整性
枚举使用 SMALLINT：使用 SMALLINT + Python IntEnum，不使用 PostgreSQL ENUM
异步优先：所有数据库操作使用 async/await

与资源库的区别

特性	project_resources（项目素材）	resources（资源库，后期扩展）
归属	归属单个项目	公共资源，可被多个项目使用
访问权限	项目成员可访问	公开或付费访问
用途	项目制作使用	素材市场、系统资源库
数据量	每项目 50-300 个	全局 1-10 万个

核心功能

1. 素材上传

支持四种类型的素材：

1.1 角色素材（character）

主角、配角、群演的形象图
支持多角度、多表情
支持 AI 生成

1.2 场景素材（scene）

室内场景、室外场景背景图
支持不同时间、天气的场景
支持 AI 生成

1.3 道具素材（prop）

物品、工具、装饰等道具图片
支持 AI 生成

1.4 实拍素材（footage）

实际拍摄的航拍图、瞰景图
实际拍摄的视频素材
支持图片和视频格式
不支持 AI 生成（真实拍摄）

通用功能：

自动计算文件校验和（SHA256）
文件去重（通过 FileStorageService）
上传到对象存储（MinIO/S3）
自动生成缩略图

2. 素材管理

按类型查询素材
按项目查询素材
素材搜索（名称模糊搜索）
素材更新（名称、描述）
素材删除（软删除）

3. AI 生成素材

关联 AI 任务（ai_job_id）
记录生成参数（meta_data）
支持重新生成

4. 素材与分镜关联

将素材添加到分镜
从分镜移除素材
查询分镜使用的素材

服务实现

ProjectResourceService 类

# app/services/project_resource_service.py
from typing import List, Optional, Dict, Any, TYPE_CHECKING
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import UploadFile
from app.models.project_resource import ProjectResource
from app.repositories.project_resource_repository import ProjectResourceRepository
from app.services.file_storage_service import FileStorageService
from app.core.exceptions import NotFoundError, ValidationError, PermissionError
from app.core.logging import get_logger
from app.utils.uuid import generate_uuid_v7
from PIL import Image
from io import BytesIO
import shutil

if TYPE_CHECKING:
    from app.services.project_service import ProjectService

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

class ProjectResourceService:
    def __init__(
        self,
        db: AsyncSession,
        repository: ProjectResourceRepository,
        file_storage: FileStorageService,
        project_service: 'ProjectService'
    ):
        self.db = db
        self.repository = repository
        self.file_storage = file_storage
        self.project_service = project_service
        
        # 检查 ffmpeg 是否可用（用于视频处理）
        if not shutil.which('ffmpeg'):
            logger.warning("ffmpeg 未安装，视频素材功能将不可用")

    # 允许的图片类型
    ALLOWED_IMAGE_TYPES = {
        'image/jpeg',
        'image/png',
        'image/gif',
        'image/webp',
        'image/bmp'
    }

    # 允许的视频类型（仅用于实拍素材）
    ALLOWED_VIDEO_TYPES = {
        'video/mp4',
        'video/quicktime',  # .mov
        'video/x-msvideo',  # .avi
        'video/webm'
    }

    # 文件大小限制
    MAX_IMAGE_SIZE = 20 * 1024 * 1024  # 20MB
    MAX_VIDEO_SIZE = 500 * 1024 * 1024  # 500MB

    async def upload_resource(
        self,
        user_id: str,
        project_id: str,
        file: UploadFile,
        name: str,
        resource_type: str,
        description: Optional[str] = None,
        element_tag_id: Optional[str] = None  # 新增：关联标签ID
    ) -> ProjectResource:
        """上传项目素材"""
        logger.info(
            "上传项目素材: user_id=%s, project_id=%s, resource_type=%s, filename=%s",
            user_id,
            project_id,
            resource_type,
            file.filename
        )
        
        # 检查项目存在性和权限（应用层引用完整性校验）
        project = await self.project_service.get_project(user_id, project_id)
        if not project:
            raise NotFoundError("项目不存在")
        
        # 检查项目权限
        await self._check_project_permission(user_id, project_id, 'editor')

        # 读取文件内容
        content = await file.read()
        file_size = len(content)

        # 根据素材类型验证文件
        is_video = False
        width = None
        height = None
        duration = None

        if resource_type == 'footage':
            # 实拍素材支持图片和视频
            if file.content_type in self.ALLOWED_IMAGE_TYPES:
                # 验证图片大小
                if file_size > self.MAX_IMAGE_SIZE:
                    logger.warning(
                        "图片大小超过限制: file_size=%d, max_size=%d, filename=%s",
                        file_size,
                        self.MAX_IMAGE_SIZE,
                        file.filename
                    )
                    raise ValidationError(
                        "图片大小超过限制: %.2fMB > %.2fMB" % (
                            file_size / 1024 / 1024,
                            self.MAX_IMAGE_SIZE / 1024 / 1024
                        )
                    )
                # 获取图片尺寸
                image = Image.open(BytesIO(content))
                width, height = image.size
            elif file.content_type in self.ALLOWED_VIDEO_TYPES:
                # 检查 ffmpeg 是否可用
                if not shutil.which('ffmpeg'):
                    raise ValidationError("视频处理功能不可用，请联系管理员")
                
                # 验证视频大小
                if file_size > self.MAX_VIDEO_SIZE:
                    logger.warning(
                        "视频大小超过限制: file_size=%d, max_size=%d, filename=%s",
                        file_size,
                        self.MAX_VIDEO_SIZE,
                        file.filename
                    )
                    raise ValidationError(
                        "视频大小超过限制: %.2fMB > %.2fMB" % (
                            file_size / 1024 / 1024,
                            self.MAX_VIDEO_SIZE / 1024 / 1024
                        )
                    )
                is_video = True
                # 获取视频元数据（需要 ffmpeg）
                video_meta = await self._get_video_meta_data(content)
                width = video_meta.get('width')
                height = video_meta.get('height')
                duration = video_meta.get('duration')
            else:
                raise ValidationError("实拍素材不支持的文件类型: %s" % file.content_type)
        else:
            # 其他素材类型仅支持图片
            if file.content_type not in self.ALLOWED_IMAGE_TYPES:
                raise ValidationError("不支持的图片类型: %s" % file.content_type)

            # 验证文件大小
            if file_size > self.MAX_IMAGE_SIZE:
                logger.warning(
                    "图片大小超过限制: file_size=%d, max_size=%d, filename=%s",
                    file_size,
                    self.MAX_IMAGE_SIZE,
                    file.filename
                )
                raise ValidationError(
                    "文件大小超过限制: %.2fMB > %.2fMB" % (
                        file_size / 1024 / 1024,
                        self.MAX_IMAGE_SIZE / 1024 / 1024
                    )
                )

            # 获取图片尺寸
            image = Image.open(BytesIO(content))
            width, height = image.size

        # 使用 FileStorageService 上传
        file_meta = await self.file_storage.upload_file(
            file_content=content,
            filename=file.filename,
            content_type=file.content_type,
            category=f'project_resource_{resource_type}',
            user_id=user_id
        )

        # 生成缩略图（视频和图片都生成）
        thumbnail_meta = None
        if is_video:
            # 从视频提取第一帧作为缩略图
            thumbnail_meta = await self._generate_video_thumbnail(
                content, file_meta.checksum, resource_type, user_id
            )
        else:
            thumbnail_meta = await self._generate_thumbnail(
                image, file_meta.checksum, resource_type, user_id
            )

        # 如果提供了 element_tag_id，自动填充冗余字段
        element_name = None
        tag_label = None
        
        if element_tag_id:
            # 验证标签存在性并获取冗余字段
            from app.repositories.screenplay_tag_repository import ScreenplayTagRepository
            tag_repo = ScreenplayTagRepository(self.db)
            tag = await tag_repo.get_by_id(element_tag_id)
            
            if not tag:
                raise NotFoundError("标签不存在")
            
            # 验证标签归属的剧本是否属于当前项目
            from app.repositories.screenplay_repository import ScreenplayRepository
            screenplay_repo = ScreenplayRepository(self.db)
            screenplay = await screenplay_repo.get_by_id(tag.screenplay_id)
            
            if not screenplay or screenplay.project_id != project_id:
                raise ValidationError("标签不属于当前项目")
            
            element_name = tag.element_name
            tag_label = tag.tag_label
            
            logger.info(
                "素材关联标签: element_name=%s, tag_label=%s",
                element_name,
                tag_label
            )

        # 创建项目素材记录（应用层生成 UUID v7）
        meta_data = {}
        if duration:
            meta_data['duration'] = duration

        resource = ProjectResource(
            project_resource_id=generate_uuid_v7(),  # 应用层生成 UUID v7
            project_id=project_id,
            name=name,
            type=resource_type,
            description=description,
            file_url=file_meta.file_url,
            thumbnail_url=thumbnail_meta.file_url if thumbnail_meta else None,
            file_size=file_meta.file_size,
            mime_type=file.content_type,
            width=width,
            height=height,
            checksum=file_meta.checksum,
            element_tag_id=element_tag_id,
            element_name=element_name,  # 冗余字段
            tag_label=tag_label,        # 冗余字段
            meta_data=meta_data,
            created_by=user_id
        )

        created_resource = await self.repository.create(resource)
        
        logger.info(
            "项目素材上传成功: resource_id=%s, project_id=%s, resource_type=%s",
            str(created_resource.project_resource_id),
            str(project_id),
            resource_type
        )
        
        return created_resource

    async def create_ai_generated_resource(
        self,
        user_id: str,
        project_id: str,
        name: str,
        resource_type: str,
        file_content: bytes,
        ai_job_id: str,
        meta_data: Dict[str, Any],
        element_tag_id: Optional[str] = None  # 新增：关联标签ID
    ) -> ProjectResource:
        """创建 AI 生成的素材"""
        logger.info(
            "创建 AI 生成素材: user_id=%s, project_id=%s, resource_type=%s, ai_job_id=%s",
            user_id,
            project_id,
            resource_type,
            ai_job_id
        )
        
        # 检查项目存在性和权限（应用层引用完整性校验）
        project = await self.project_service.get_project(user_id, project_id)
        if not project:
            raise NotFoundError("项目不存在")
        
        # 注意：AI 任务的引用完整性应该由 AI Service 保证
        # 这里假设 ai_job_id 已经过验证
        
        # 检查项目权限
        await self._check_project_permission(user_id, project_id, 'editor')

        # 获取图片尺寸
        image = Image.open(BytesIO(file_content))
        width, height = image.size
        content_type = f"image/{image.format.lower()}"

        # 使用 FileStorageService 上传
        file_meta = await self.file_storage.upload_file(
            file_content=file_content,
            filename=f"ai_generated_{resource_type}.{image.format.lower()}",
            content_type=content_type,
            category=f'project_resource_{resource_type}_ai',
            user_id=user_id
        )

        # 生成缩略图
        thumbnail_meta = await self._generate_thumbnail(
            image, file_meta.checksum, resource_type, user_id
        )

        # 如果提供了 element_tag_id，自动填充冗余字段
        element_name = None
        tag_label = None
        
        if element_tag_id:
            # 验证标签存在性并获取冗余字段
            from app.repositories.screenplay_tag_repository import ScreenplayTagRepository
            tag_repo = ScreenplayTagRepository(self.db)
            tag = await tag_repo.get_by_id(element_tag_id)
            
            if not tag:
                raise NotFoundError("标签不存在")
            
            # 验证标签归属的剧本是否属于当前项目
            from app.repositories.screenplay_repository import ScreenplayRepository
            screenplay_repo = ScreenplayRepository(self.db)
            screenplay = await screenplay_repo.get_by_id(tag.screenplay_id)
            
            if not screenplay or screenplay.project_id != project_id:
                raise ValidationError("标签不属于当前项目")
            
            element_name = tag.element_name
            tag_label = tag.tag_label
            
            logger.info(
                "AI生成素材关联标签: element_name=%s, tag_label=%s",
                element_name,
                tag_label
            )

        # 创建项目素材记录（应用层生成 UUID v7）
        resource = ProjectResource(
            project_resource_id=generate_uuid_v7(),  # 应用层生成 UUID v7
            project_id=project_id,
            name=name,
            type=resource_type,
            file_url=file_meta.file_url,
            thumbnail_url=thumbnail_meta.file_url if thumbnail_meta else None,
            file_size=file_meta.file_size,
            mime_type=content_type,
            width=width,
            height=height,
            checksum=file_meta.checksum,
            ai_job_id=ai_job_id,
            element_tag_id=element_tag_id,
            element_name=element_name,  # 冗余字段
            tag_label=tag_label,        # 冗余字段
            meta_data=meta_data,
            created_by=user_id
        )

        created_resource = await self.repository.create(resource)
        
        logger.info(
            "AI 生成素材创建成功: resource_id=%s, project_id=%s, ai_job_id=%s",
            str(created_resource.project_resource_id),
            str(project_id),
            str(ai_job_id)
        )
        
        return created_resource

    async def get_resource(
        self,
        user_id: str,
        resource_id: str
    ) -> ProjectResource:
        """获取素材详情"""
        resource = await self.repository.get_by_id(resource_id)
        if not resource:
            raise NotFoundError("素材不存在")

        # 检查项目权限
        await self._check_project_permission(user_id, resource.project_id, 'viewer')

        return resource

    async def update_resource(
        self,
        user_id: str,
        resource_id: str,
        name: Optional[str] = None,
        description: Optional[str] = None
    ) -> ProjectResource:
        """更新素材信息"""
        resource = await self.repository.get_by_id(resource_id)
        if not resource:
            raise NotFoundError("素材不存在")

        # 检查项目权限
        await self._check_project_permission(user_id, resource.project_id, 'editor')

        if name is not None:
            resource.name = name
        if description is not None:
            resource.description = description

        return await self.repository.update(resource)

    async def check_resource_usage(
        self,
        user_id: str,
        resource_id: str
    ) -> Dict[str, Any]:
        """检查素材使用情况"""
        resource = await self.repository.get_by_id(resource_id)
        if not resource:
            raise NotFoundError("素材不存在")

        # 检查项目权限
        await self._check_project_permission(user_id, resource.project_id, 'viewer')

        return {
            'resource_id': resource_id,
            'usage_count': resource.usage_count,
            'can_delete': resource.usage_count == 0
        }

    async def delete_resource(
        self,
        user_id: str,
        resource_id: str,
        force: bool = False
    ) -> None:
        """
        删除素材
        
        Args:
            user_id: 用户ID
            resource_id: 素材ID
            force: 是否强制删除（即使被使用）
        
        Raises:
            ValidationError: 素材正在被使用且未强制删除
        """
        resource = await self.repository.get_by_id(resource_id)
        if not resource:
            raise NotFoundError("素材不存在")

        # 检查项目权限
        await self._check_project_permission(user_id, resource.project_id, 'editor')

        # 检查使用情况
        if resource.usage_count > 0 and not force:
            logger.warning(
                "素材正在被使用，无法删除: resource_id=%s, usage_count=%d",
                str(resource_id),
                resource.usage_count
            )
            raise ValidationError(
                "该素材正在被 %d 个分镜使用，无法删除。请先从分镜中移除该素材，或使用强制删除。" % resource.usage_count
            )

        # 软删除
        await self.repository.soft_delete(resource_id)

        # 减少文件引用计数
        await self.file_storage.decrease_reference_count(resource.checksum)
        
        logger.info(
            "素材删除成功: resource_id=%s, force=%s",
            str(resource_id),
            force
        )

    async def get_resources_by_project(
        self,
        user_id: str,
        project_id: str,
        resource_type: Optional[str] = None,
        search: Optional[str] = None,
        page: int = 1,
        page_size: int = 20
    ) -> Dict[str, Any]:
        """获取项目的素材列表"""
        # 检查项目权限
        await self._check_project_permission(user_id, project_id, 'viewer')

        resources = await self.repository.get_by_project(
            project_id, resource_type, search, page, page_size
        )

        total = await self.repository.count_by_project(project_id, resource_type, search)

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

    async def _generate_thumbnail(
        self,
        image: Image.Image,
        checksum: str,
        resource_type: str,
        user_id: str
    ) -> Optional[Any]:
        """生成图片缩略图"""
        try:
            # 生成 300x300 缩略图
            thumbnail = image.copy()
            thumbnail.thumbnail((300, 300), Image.Resampling.LANCZOS)

            # 转换为字节
            buffer = BytesIO()
            thumbnail.save(buffer, format=image.format)
            thumbnail_content = buffer.getvalue()

            # 上传缩略图
            thumbnail_meta = await self.file_storage.upload_file(
                file_content=thumbnail_content,
                filename=f"thumbnail_{checksum}.{image.format.lower()}",
                content_type=f"image/{image.format.lower()}",
                category=f'thumbnail_{resource_type}',
                user_id=user_id
            )

            return thumbnail_meta
        except Exception as e:
            logger.error(
                "生成缩略图失败: %s",
                str(e),
                exc_info=True
            )
            return None

    async def _generate_video_thumbnail(
        self,
        video_content: bytes,
        checksum: str,
        resource_type: str,
        user_id: str
    ) -> Optional[Any]:
        """生成视频缩略图（提取第一帧）"""
        try:
            import subprocess
            import tempfile
            import os

            # 创建临时文件
            with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as video_file:
                video_file.write(video_content)
                video_path = video_file.name

            with tempfile.NamedTemporaryFile(suffix='.jpg', delete=False) as thumb_file:
                thumb_path = thumb_file.name

            try:
                # 使用 ffmpeg 提取第一帧
                subprocess.run([
                    'ffmpeg',
                    '-i', video_path,
                    '-ss', '00:00:01',  # 提取第1秒的帧
                    '-vframes', '1',
                    '-vf', 'scale=300:300:force_original_aspect_ratio=decrease',
                    thumb_path
                ], check=True, capture_output=True)

                # 读取缩略图
                with open(thumb_path, 'rb') as f:
                    thumbnail_content = f.read()

                # 上传缩略图
                thumbnail_meta = await self.file_storage.upload_file(
                    file_content=thumbnail_content,
                    filename=f"thumbnail_{checksum}.jpg",
                    content_type="image/jpeg",
                    category=f'thumbnail_{resource_type}',
                    user_id=user_id
                )

                return thumbnail_meta
            finally:
                # 清理临时文件
                os.unlink(video_path)
                os.unlink(thumb_path)
        except Exception as e:
            logger.error(
                "生成视频缩略图失败: %s",
                str(e),
                exc_info=True
            )
            return None

    async def _get_video_meta_data(self, video_content: bytes) -> Dict[str, Any]:
        """获取视频元数据"""
        try:
            import subprocess
            import tempfile
            import json
            import os

            # 创建临时文件
            with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as video_file:
                video_file.write(video_content)
                video_path = video_file.name

            try:
                # 使用 ffprobe 获取视频信息
                result = subprocess.run([
                    'ffprobe',
                    '-v', 'quiet',
                    '-print_format', 'json',
                    '-show_format',
                    '-show_streams',
                    video_path
                ], capture_output=True, text=True, check=True)

                data = json.loads(result.stdout)

                # 提取视频流信息
                video_stream = next(
                    (s for s in data.get('streams', []) if s.get('codec_type') == 'video'),
                    None
                )

                if video_stream:
                    return {
                        'width': video_stream.get('width'),
                        'height': video_stream.get('height'),
                        'duration': float(data.get('format', {}).get('duration', 0))
                    }

                return {}
            finally:
                os.unlink(video_path)
        except Exception as e:
            logger.error(
                "获取视频元数据失败: %s",
                str(e),
                exc_info=True
            )
            return {}

    async def _check_project_permission(
        self,
        user_id: str,
        project_id: str,
        required_role: str = 'viewer'
    ) -> None:
        """检查项目权限（应用层引用完整性校验）"""
        has_permission = await self.project_service.check_user_permission(
            user_id, project_id, required_role
        )
        if not has_permission:
            raise PermissionError("没有权限访问此项目")

API 接口

API 路由实现

# app/api/v1/project_resources.py
from fastapi import APIRouter, Depends, UploadFile, File, Form, Query
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional
from app.core.deps import get_current_user, get_db
from app.services.project_resource_service import ProjectResourceService
from app.services.file_storage_service import FileStorageService
from app.services.project_service import ProjectService
from app.repositories.project_resource_repository import ProjectResourceRepository
from app.repositories.project_repository import ProjectRepository
from app.schemas.project_resource import (
    ProjectResourceResponse,
    ProjectResourceListResponse,
    ProjectResourceUpdate
)
from app.schemas.common import SuccessResponse
from app.models.user import User

router = APIRouter()

def get_project_resource_service(
    db: AsyncSession = Depends(get_db)
) -> ProjectResourceService:
    """依赖注入：获取项目素材服务"""
    repository = ProjectResourceRepository(db)
    file_storage = FileStorageService(db)
    project_repo = ProjectRepository(db)
    project_service = ProjectService(db, project_repo)
    
    return ProjectResourceService(
        db=db,
        repository=repository,
        file_storage=file_storage,
        project_service=project_service
    )

@router.post(
    "/projects/{project_id}/resources",
    response_model=SuccessResponse[ProjectResourceResponse],
    summary="上传项目素材"
)
async def upload_resource(
    project_id: str,
    file: UploadFile = File(..., description="素材文件"),
    name: str = Form(..., description="素材名称"),
    type: str = Form(..., description="素材类型: character/scene/prop/footage"),
    description: Optional[str] = Form(None, description="素材描述"),
    element_tag_id: Optional[str] = Form(None, description="关联的标签ID"),
    current_user: User = Depends(get_current_user),
    service: ProjectResourceService = Depends(get_project_resource_service)
):
    """
    上传项目素材
    
    - **character/scene/prop**: 仅支持图片格式（JPEG、PNG、GIF、WebP、BMP），最大 20MB
    - **footage**: 支持图片和视频格式（MP4、MOV、AVI、WebM），图片最大 20MB，视频最大 500MB
    """
    resource = await service.upload_resource(
        user_id=current_user.user_id,
        project_id=project_id,
        file=file,
        name=name,
        resource_type=type,
        description=description,
        element_tag_id=element_tag_id
    )
    
    return SuccessResponse(data=ProjectResourceResponse.model_validate(resource))

@router.get(
    "/resources/{resource_id}",
    response_model=SuccessResponse[ProjectResourceResponse],
    summary="获取素材详情"
)
async def get_resource(
    resource_id: str,
    current_user: User = Depends(get_current_user),
    service: ProjectResourceService = Depends(get_project_resource_service)
):
    """获取素材详情"""
    resource = await service.get_resource(current_user.user_id, resource_id)
    return SuccessResponse(data=ProjectResourceResponse.model_validate(resource))

@router.patch(
    "/resources/{resource_id}",
    response_model=SuccessResponse[ProjectResourceResponse],
    summary="更新素材信息"
)
async def update_resource(
    resource_id: str,
    update_data: ProjectResourceUpdate,
    current_user: User = Depends(get_current_user),
    service: ProjectResourceService = Depends(get_project_resource_service)
):
    """更新素材信息（仅支持修改名称和描述）"""
    resource = await service.update_resource(
        user_id=current_user.user_id,
        resource_id=resource_id,
        name=update_data.name,
        description=update_data.description
    )
    return SuccessResponse(data=ProjectResourceResponse.model_validate(resource))

@router.get(
    "/resources/{resource_id}/usage",
    response_model=SuccessResponse[Dict[str, Any]],
    summary="检查素材使用情况"
)
async def check_resource_usage(
    resource_id: str,
    current_user: User = Depends(get_current_user),
    service: ProjectResourceService = Depends(get_project_resource_service)
):
    """检查素材使用情况（用于删除前确认）"""
    usage = await service.check_resource_usage(current_user.user_id, resource_id)
    return SuccessResponse(data=usage)

@router.delete(
    "/resources/{resource_id}",
    response_model=SuccessResponse[None],
    summary="删除素材"
)
async def delete_resource(
    resource_id: str,
    force: bool = Query(False, description="是否强制删除（即使被使用）"),
    current_user: User = Depends(get_current_user),
    service: ProjectResourceService = Depends(get_project_resource_service)
):
    """
    删除素材（软删除）
    
    - 如果素材正在被分镜使用（usage_count > 0），默认禁止删除
    - 可以使用 force=true 强制删除（会导致分镜失去素材引用）
    """
    await service.delete_resource(current_user.user_id, resource_id, force)
    return SuccessResponse(data=None, message="素材已删除")

@router.get(
    "/projects/{project_id}/resources",
    response_model=SuccessResponse[ProjectResourceListResponse],
    summary="获取项目素材列表"
)
async def get_project_resources(
    project_id: str,
    type: Optional[str] = Query(None, description="素材类型过滤"),
    element_tag_id: Optional[str] = Query(None, description="标签ID过滤"),
    search: Optional[str] = Query(None, description="搜索关键词"),
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
    current_user: User = Depends(get_current_user),
    service: ProjectResourceService = Depends(get_project_resource_service)
):
    """获取项目素材列表"""
    result = await service.get_resources_by_project(
        user_id=current_user.user_id,
        project_id=project_id,
        resource_type=type,
        search=search,
        page=page,
        page_size=page_size
    )
    
    return SuccessResponse(data=ProjectResourceListResponse(
        items=[ProjectResourceResponse.model_validate(r) for r in result['items']],
        total=result['total'],
        page=result['page'],
        page_size=result['page_size'],
        total_pages=result['total_pages']
    ))

@router.get(
    "/tags/{tag_id}/resources",
    response_model=SuccessResponse[ProjectResourceListResponse],
    summary="获取标签的素材列表"
)
async def get_tag_resources(
    tag_id: str,
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
    current_user: User = Depends(get_current_user),
    service: ProjectResourceService = Depends(get_project_resource_service)
):
    """
    获取标签的素材列表（快捷方式）
    
    这是 GET /api/v1/projects/{project_id}/resources?element_tag_id={tag_id} 的快捷方式
    """
    # 注意：需要先获取 tag 的 project_id
    # 这里简化处理，实际应该通过 tag_service 获取
    # result = await service.get_resources_by_tag(...)
    pass

Schema 定义

# app/schemas/project_resource.py
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional, List, Dict, Any
from datetime import datetime
import uuid

class ProjectResourceBase(BaseModel):
    """项目素材基础模型"""
    name: str = Field(..., description="素材名称")
    type: str = Field(..., description="素材类型: character/scene/prop/footage")
    description: Optional[str] = Field(None, description="素材描述")

class ProjectResourceUpdate(BaseModel):
    """项目素材更新模型"""
    name: Optional[str] = Field(None, description="素材名称")
    description: Optional[str] = Field(None, description="素材描述")

class ProjectResourceResponse(BaseModel):
    """项目素材响应模型"""
    project_resource_id: uuid.UUID = Field(..., description="素材ID")
    project_id: uuid.UUID = Field(..., description="项目ID")
    name: str = Field(..., description="素材名称")
    type: str = Field(..., description="素材类型")
    description: Optional[str] = Field(None, description="素材描述")
    
    # 文件信息
    file_url: str = Field(..., description="文件URL")
    thumbnail_url: Optional[str] = Field(None, description="缩略图URL")
    file_size: int = Field(..., description="文件大小（字节）")
    mime_type: str = Field(..., description="MIME类型")
    width: Optional[int] = Field(None, description="宽度（像素）")
    height: Optional[int] = Field(None, description="高度（像素）")
    checksum: str = Field(..., description="文件校验和")
    
    # 标签关联
    element_tag_id: Optional[uuid.UUID] = Field(None, description="关联的标签ID")
    tag_label: Optional[str] = Field(None, description="标签名称")
    element_name: Optional[str] = Field(None, description="元素名称")
    
    # AI 生成
    ai_job_id: Optional[uuid.UUID] = Field(None, description="AI任务ID")
    meta_data: Dict[str, Any] = Field(default_factory=dict, description="元数据")
    
    # 使用统计
    usage_count: int = Field(0, description="引用计数（被多少个分镜使用）")
    
    # 审计
    created_by: uuid.UUID = Field(..., description="创建者ID")
    created_at: datetime = Field(..., description="创建时间")
    updated_at: datetime = Field(..., description="更新时间")
    
    model_config = ConfigDict(from_attributes=True)

class ProjectResourceListResponse(BaseModel):
    """项目素材列表响应模型"""
    items: List[ProjectResourceResponse] = Field(..., description="素材列表")
    total: int = Field(..., description="总数")
    page: int = Field(..., description="当前页码")
    page_size: int = Field(..., description="每页数量")
    total_pages: int = Field(..., description="总页数")

# app/schemas/common.py
from pydantic import BaseModel, Field
from typing import Generic, TypeVar, Optional

T = TypeVar('T')

class SuccessResponse(BaseModel, Generic[T]):
    """统一成功响应格式"""
    code: int = Field(200, description="状态码")
    message: str = Field("success", description="响应消息")
    data: Optional[T] = Field(None, description="响应数据")

API 接口说明

1. 上传项目素材

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

请求（multipart/form-data）：

file: 图片或视频文件（必填）
name: 素材名称（必填）
type: 素材类型（character | scene | prop | footage，必填）
description: 素材描述（可选）
element_tag_id: 关联的标签 ID（可选，用于关联到角色/场景/道具的特定标签）

说明：

character、scene、prop 仅支持图片格式（JPEG、PNG、GIF、WebP、BMP），最大 20MB
footage 支持图片和视频格式（MP4、MOV、AVI、WebM），图片最大 20MB，视频最大 500MB
如果提供 element_tag_id，素材会关联到该标签（如"张三-少年"）
如果不提供 element_tag_id，素材为通用素材（不关联特定标签）

响应：

{
  "code": 200,
  "message": "success",
  "data": {
    "project_resource_id": "019d1234-5678-7abc-def0-123456789abc",
    "project_id": "019d1234-5678-7abc-def0-111111111111",
    "name": "主角-张三-少年形象",
    "type": "character",
    "file_url": "https://storage.jointo.ai/project_resource/character/1/abc123.png",
    "thumbnail_url": "https://storage.jointo.ai/thumbnail/character/1/thumb_abc123.png",
    "file_size": 1024000,
    "width": 1920,
    "height": 1080,
    "checksum": "abc123...",
    "element_tag_id": "019d1234-5678-7abc-def0-666666666666",
    "element_tag_label": "少年",
    "element_name": "张三",
    "created_by": "019d1234-5678-7abc-def0-999999999999",
    "created_at": "2025-01-27T10:00:00+00:00"
  }
}

2. 获取素材详情

GET /api/v1/resources/{resource_id}

3. 更新素材信息

PATCH /api/v1/resources/{resource_id}

请求体：

{
  "name": "主角-张三（正面）",
  "description": "主角正面形象，用于对话场景"
}

4. 检查素材使用情况

GET /api/v1/resources/{resource_id}/usage

响应：

{
  "code": 200,
  "message": "success",
  "data": {
    "resource_id": "019d1234-5678-7abc-def0-123456789abc",
    "usage_count": 3,
    "can_delete": false
  }
}

5. 删除素材

DELETE /api/v1/resources/{resource_id}?force=false

查询参数：

force: 是否强制删除（默认 false）

说明：

如果 usage_count > 0 且 force=false，返回 400 错误
如果 force=true，即使被使用也会删除（分镜会失去素材引用）

错误响应（usage_count > 0 且 force=false）：

{
  "code": 400,
  "message": "该素材正在被 3 个分镜使用，无法删除。请先从分镜中移除该素材，或使用强制删除。",
  "data": null
}

6. 获取项目素材列表

GET /api/v1/projects/{project_id}/resources?type=character&search=主角

查询参数：

type: 素材类型（可选：character | scene | prop | footage）
element_tag_id: 标签 ID 过滤（可选，用于获取特定标签的素材）
search: 搜索关键词（可选）
page: 页码
page_size: 每页数量

响应：

{
  "code": 200,
  "message": "success",
  "data": {
    "items": [
      {
        "project_resource_id": "019d1234-5678-7abc-def0-123456789abc",
        "name": "主角-张三",
        "type": "character",
        "thumbnail_url": "...",
        "element_tag_id": "019d1234-5678-7abc-def0-666666666666",
        "element_tag_label": "少年",
        "created_at": "2025-01-27T10:00:00+00:00"
      }
    ],
    "total": 50,
    "page": 1,
    "page_size": 20,
    "total_pages": 3
  }
}

7. 获取标签的素材列表（快捷方式）

GET /api/v1/tags/{tag_id}/resources

说明：这是 GET /api/v1/projects/{project_id}/resources?element_tag_id={tag_id} 的快捷方式

查询参数：

page: 页码
page_size: 每页数量

响应：同上

外部依赖

系统依赖

项目素材服务需要以下系统依赖：

ffmpeg - 用于视频处理（提取缩略图、获取元数据）
- 安装方式（Ubuntu/Debian）：apt-get install -y ffmpeg
- 安装方式（macOS）：brew install ffmpeg
- Docker 镜像需要在 Dockerfile 中添加：
```
RUN apt-get update && apt-get install -y ffmpeg && rm -rf /var/lib/apt/lists/*
```
Pillow - Python 图像处理库（已在 requirements.txt 中）

Python 依赖

确保 server/requirements.txt 包含以下依赖：

# 图像处理
Pillow==10.2.0

# UUID v7
uuid-utils==0.9.0

# 其他依赖（已有）
fastapi==0.109.0
sqlalchemy[asyncio]==2.0.25
asyncpg==0.29.0

依赖检查

Service 在初始化时会检查 ffmpeg 是否可用：

if not shutil.which('ffmpeg'):
    logger.warning("ffmpeg 未安装，视频素材功能将不可用")

如果 ffmpeg 不可用，上传视频素材时会抛出 ValidationError。

数据库设计

project_resources 表结构

-- 素材类型：1=角色, 2=场景, 3=道具, 4=实拍
-- 使用 SMALLINT 而非 ENUM，配合 Python IntEnum

CREATE TABLE project_resources (
    project_resource_id UUID PRIMARY KEY,  -- 应用层生成 UUID v7
    project_id UUID NOT NULL,  -- 无物理外键，应用层校验
    name TEXT NOT NULL,
    type SMALLINT NOT NULL,  -- 1=character, 2=scene, 3=prop, 4=footage
    description TEXT,

    -- 文件信息
    file_url TEXT NOT NULL,
    thumbnail_url TEXT,
    file_size BIGINT,
    mime_type TEXT,
    width INTEGER,
    height INTEGER,
    checksum TEXT NOT NULL,

    -- 标签关联（关联到剧本元素的具体标签）
    element_tag_id UUID,  -- 无物理外键，应用层校验
    element_name TEXT,    -- 冗余字段：元素名称（如"孙悟空"、"花果山"）
    tag_label TEXT,       -- 冗余字段：标签名称（如"少年"、"夜晚"）

    -- 来源（后期扩展）
    source_resource_id UUID,  -- 无物理外键，应用层校验

    -- AI 生成
    ai_job_id UUID,  -- 无物理外键，应用层校验
    meta_data JSONB NOT NULL DEFAULT '{}',

    -- 使用统计
    usage_count INTEGER NOT NULL DEFAULT 0,

    -- 审计
    created_by UUID NOT NULL,  -- 无物理外键，应用层校验
    created_at TIMESTAMPTZ NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL,
    deleted_at TIMESTAMPTZ,
    
    -- 约束
    CONSTRAINT project_resources_footage_no_tag_check CHECK (
        (type = 4 AND element_tag_id IS NULL AND element_name IS NULL AND tag_label IS NULL) OR
        (type IN (1, 2, 3))
    ),
    CONSTRAINT project_resources_usage_count_check CHECK (usage_count >= 0)
);

-- 表级注释
COMMENT ON TABLE project_resources IS '项目素材表 - 存储项目专属的角色、场景、道具、实拍素材';

-- 字段注释（使用 COMMENT ON 语法）
COMMENT ON COLUMN project_resources.project_resource_id IS '素材唯一标识（UUID v7，应用层生成）';
COMMENT ON COLUMN project_resources.project_id IS '所属项目ID（无物理外键，应用层校验）';
COMMENT ON COLUMN project_resources.name IS '素材名称';
COMMENT ON COLUMN project_resources.type IS '素材类型: 1=角色(character), 2=场景(scene), 3=道具(prop), 4=实拍(footage)';
COMMENT ON COLUMN project_resources.description IS '素材描述';
COMMENT ON COLUMN project_resources.file_url IS '文件URL';
COMMENT ON COLUMN project_resources.thumbnail_url IS '缩略图URL';
COMMENT ON COLUMN project_resources.file_size IS '文件大小（字节）';
COMMENT ON COLUMN project_resources.mime_type IS 'MIME类型';
COMMENT ON COLUMN project_resources.width IS '宽度（像素）';
COMMENT ON COLUMN project_resources.height IS '高度（像素）';
COMMENT ON COLUMN project_resources.checksum IS '文件校验和（SHA256）';
COMMENT ON COLUMN project_resources.element_tag_id IS '关联的剧本元素标签ID（无物理外键，应用层校验）';
COMMENT ON COLUMN project_resources.element_name IS '元素名称（冗余存储，便于查询，如"孙悟空"、"花果山"）';
COMMENT ON COLUMN project_resources.tag_label IS '标签名称（冗余存储，便于显示，如"少年"、"夜晚"）';
COMMENT ON COLUMN project_resources.source_resource_id IS '来源资源ID（后期扩展，无物理外键）';
COMMENT ON COLUMN project_resources.ai_job_id IS '关联的AI任务ID（无物理外键，应用层校验）';
COMMENT ON COLUMN project_resources.meta_data IS '扩展元数据（JSONB格式，如拍摄角度、光照等）';
COMMENT ON COLUMN project_resources.usage_count IS '引用计数（被多少个分镜使用，由 StoryboardResourceService 维护）';
COMMENT ON COLUMN project_resources.created_by IS '创建者用户ID（无物理外键，应用层校验）';
COMMENT ON COLUMN project_resources.created_at IS '创建时间（UTC，自动记录时区）';
COMMENT ON COLUMN project_resources.updated_at IS '更新时间（UTC，自动记录时区）';
COMMENT ON COLUMN project_resources.deleted_at IS '删除时间（软删除）';

-- 索引（手动创建，确保引用完整性查询性能）
CREATE INDEX idx_project_resources_project_id ON project_resources (project_id) WHERE deleted_at IS NULL;
CREATE INDEX idx_project_resources_type ON project_resources (type) WHERE deleted_at IS NULL;
CREATE INDEX idx_project_resources_created_by ON project_resources (created_by) WHERE deleted_at IS NULL;
CREATE INDEX idx_project_resources_checksum ON project_resources (checksum);
CREATE INDEX idx_project_resources_element_tag_id ON project_resources (element_tag_id) WHERE element_tag_id IS NOT NULL;
CREATE INDEX idx_project_resources_ai_job_id ON project_resources (ai_job_id) WHERE ai_job_id IS NOT NULL;
CREATE INDEX idx_project_resources_meta_data_gin ON project_resources USING GIN (meta_data);
CREATE INDEX idx_project_resources_name_trgm ON project_resources USING GIN (name gin_trgm_ops) WHERE deleted_at IS NULL;
CREATE INDEX idx_project_resources_element_name_trgm ON project_resources USING GIN (element_name gin_trgm_ops) WHERE element_name IS NOT NULL;
CREATE INDEX idx_project_resources_tag_label_trgm ON project_resources USING GIN (tag_label gin_trgm_ops) WHERE tag_label IS NOT NULL;
CREATE INDEX idx_project_resources_usage_count ON project_resources (usage_count) WHERE deleted_at IS NULL;

-- 触发器（自动更新 updated_at）
CREATE TRIGGER update_project_resources_updated_at
    BEFORE UPDATE ON project_resources
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at_column();

storyboard_resources 关联表

CREATE TABLE storyboard_resources (
    storyboard_resource_id UUID PRIMARY KEY,  -- 应用层生成 UUID v7
    storyboard_id UUID NOT NULL,  -- 无物理外键，应用层校验
    project_resource_id UUID NOT NULL,  -- 无物理外键，应用层校验
    resource_type SMALLINT NOT NULL,  -- 1=character, 2=scene, 3=prop, 4=footage
    display_order INTEGER NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL,  -- 应用层传入 aware datetime
    CONSTRAINT storyboard_resources_unique UNIQUE (storyboard_id, project_resource_id) NULLS NOT DISTINCT
);

-- 字段注释
COMMENT ON COLUMN storyboard_resources.resource_type IS '素材类型: 1=角色(character), 2=场景(scene), 3=道具(prop), 4=实拍(footage)';
COMMENT ON COLUMN storyboard_resources.storyboard_id IS '分镜ID（无物理外键，应用层校验）';
COMMENT ON COLUMN storyboard_resources.project_resource_id IS '项目素材ID（无物理外键，应用层校验）';

-- 索引（手动创建，确保引用完整性查询性能）
CREATE INDEX idx_storyboard_resources_storyboard_id ON storyboard_resources (storyboard_id);
CREATE INDEX idx_storyboard_resources_project_resource_id ON storyboard_resources (project_resource_id);
CREATE INDEX idx_storyboard_resources_type ON storyboard_resources (resource_type);

项目隔离：每个素材通过 project_id 归属一个项目，项目间素材隔离
类型分类：使用 SMALLINT 定义四种类型（1=角色、2=场景、3=道具、4=实拍）
文件信息：存储完整的文件元数据（URL、大小、尺寸等）
去重支持：checksum 字段配合 file_checksums 表实现去重
标签关联：element_tag_id 关联到剧本元素的具体标签（如"张三-少年"），支持按标签分组展示素材
冗余字段优化：element_name 和 tag_label 冗余存储，避免多表 JOIN，大幅提升查询性能
数据完整性约束：实拍素材（type=4）不能有标签关联，通过 CHECK 约束保证
AI 集成：ai_job_id 关联 AI 生成任务，meta_data 存储生成参数（实拍素材不支持 AI 生成）
缩略图：thumbnail_url 存储自动生成的缩略图（图片和视频都支持）
视频支持：实拍素材支持视频格式，meta_data 存储视频时长等信息
后期扩展：source_resource_id 预留，供后期从资源库复制素材时使用
软删除：使用 deleted_at 字段，部分索引排除已删除记录
全文搜索：使用 pg_trgm 扩展支持素材名称、元素名称、标签名称的模糊搜索
手动索引：所有逻辑关联字段手动创建索引，确保查询性能
COMMENT ON 语法：使用 COMMENT ON 语法为表和字段添加注释，便于维护

冗余字段说明

为什么添加 element_name 和 tag_label？

性能优化：查询素材列表时无需 JOIN screenplay_element_tags 表，避免 3 表 JOIN
前端友好：API 响应直接包含"孙悟空-少年"、"花果山-夜晚"等完整信息
查询简化：支持直接按元素名称或标签名称搜索素材

数据一致性维护：

当标签的 tag_label 或 element_name 修改时，ScreenplayTagService.update_tag() 会自动同步更新所有关联的素材记录。这是低频操作，可以接受。

查询示例：

-- 查询"花果山-夜晚"的所有图片（无需 JOIN）
SELECT 
    project_resource_id,
    name,
    file_url,
    thumbnail_url,
    element_name,  -- "花果山"
    tag_label,     -- "夜晚"
    meta_data->>'angle' AS angle
FROM project_resources
WHERE element_tag_id = '019d1234-5678-7abc-def0-tag-005'
ORDER BY created_at;

-- 搜索所有"孙悟空"相关的素材
SELECT 
    project_resource_id,
    name,
    element_name,
    tag_label,
    thumbnail_url
FROM project_resources
WHERE element_name ILIKE '%孙悟空%'
  AND deleted_at IS NULL
ORDER BY created_at DESC;

数据模型

ProjectResource 模型

# app/models/project_resource.py
from sqlalchemy import Column, String, BigInteger, Integer, Text, JSON, SmallInteger
from sqlalchemy.dialects.postgresql import TIMESTAMP
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from app.core.database import Base
from datetime import datetime, timezone
from enum import IntEnum
import uuid_utils as uuid

class ResourceType(IntEnum):
    """素材类型枚举"""
    CHARACTER = 1  # 角色
    SCENE = 2      # 场景
    PROP = 3       # 道具
    FOOTAGE = 4    # 实拍

class ProjectResource(Base):
    __tablename__ = "project_resources"

    project_resource_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid7)
    project_id = Column(UUID(as_uuid=True), nullable=False)  # 无物理外键
    name = Column(String(255), nullable=False)
    type = Column(SmallInteger, nullable=False)  # 使用 SMALLINT，配合 ResourceType IntEnum
    description = Column(Text)

    # 文件信息
    file_url = Column(String(500), nullable=False)
    thumbnail_url = Column(String(500))
    file_size = Column(BigInteger)
    mime_type = Column(String(100))
    width = Column(Integer)
    height = Column(Integer)
    checksum = Column(String(64), nullable=False)

    # 标签关联
    element_tag_id = Column(UUID(as_uuid=True))  # 无物理外键
    element_name = Column(String(255))  # 冗余字段：元素名称
    tag_label = Column(String(100))     # 冗余字段：标签名称

    # 来源（后期扩展）
    source_resource_id = Column(UUID(as_uuid=True))  # 无物理外键

    # AI 生成
    ai_job_id = Column(UUID(as_uuid=True))  # 无物理外键
    meta_data = Column(JSON, default={})

    # 使用统计
    usage_count = Column(Integer, nullable=False, default=0)

    # 审计（使用 aware datetime）
    created_by = Column(UUID(as_uuid=True), nullable=False)  # 无物理外键
    created_at = Column(TIMESTAMP(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc))
    updated_at = Column(TIMESTAMP(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
    deleted_at = Column(TIMESTAMP(timezone=True))

    # 关系（仅用于 ORM 查询，无物理外键）
    # 引用完整性由 Service 层保证

测试

单元测试

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

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

集成测试

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

# 运行文件上传测试
docker exec jointo-server-app pytest tests/integration/services/test_project_resource_service.py::test_upload_image -v

测试覆盖率

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

数据库迁移

创建迁移文件

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

# 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>

变更记录

v3.7 (2026-02-03)

✅ 修复技术栈合规性：
- 修复日志系统：使用 app.core.logging.get_logger 替代 logging.getLogger
- 修复日志格式化：统一使用 %-formatting 替代 f-string
- 修复异常日志：所有异常日志添加 exc_info=True
- 修复 API 响应格式：统一添加 code 和 message 字段
- 添加测试指南：单元测试、集成测试、覆盖率测试
- 添加数据库迁移说明：创建迁移、执行迁移、回滚迁移
✅ 符合 jointo-tech-stack v1.0 规范：
- 日志系统规范（logging.md）
- 测试规范（testing.md）
- 迁移规范（migration.md）

v3.6 (2026-02-02)

将业务流程图和表关系图抽离到独立文件 project-resource-service-flows.md
主文档保持简洁，专注于服务实现和 API 设计
在目录中添加流程图文档的链接引用
优化文档结构，便于维护和查阅

v3.6 (2026-02-02)

将业务流程图和表关系图抽离到独立文件 project-resource-service-flows.md
主文档保持简洁，专注于服务实现和 API 设计
在目录中添加流程图文档的链接引用
优化文档结构，便于维护和查阅

v3.5 (2026-02-02)

添加完整的业务流程图（Mermaid 格式）
- 素材上传流程（含文件去重和缩略图生成）
- AI 生成素材流程（含异步任务处理）
- 素材删除流程（含使用检查和强制删除）
- 素材与分镜关联流程（含 usage_count 维护）
添加数据库表关系图（ER 图）
- 展示逻辑关联关系（无物理外键）
- 标注所有字段和约束
添加素材类型枚举图
- 展示 4 种素材类型及其特性
- 标注 AI 生成和标签关联支持情况
添加服务依赖关系图
- 展示 ProjectResourceService 与其他服务的依赖
- 展示数据流向和存储层
添加文件去重机制流程图
- 展示基于 SHA256 的去重逻辑
- 展示引用计数维护和清理机制

v3.3 (2026-02-02)

添加 usage_count 字段，记录素材被多少个分镜使用
添加 check_resource_usage() 方法，检查素材使用情况
delete_resource() 方法添加使用检查和强制删除参数
添加 usage_count 索引和 CHECK 约束
更新 API 接口，添加使用情况检查端点
更新 Schema，添加 usage_count 字段
完善删除保护逻辑（usage_count > 0 时禁止删除）

v3.2 (2026-01-29)

添加 element_name 和 tag_label 冗余字段，优化查询性能
添加 project_resources_footage_no_tag_check 约束，确保实拍素材不能有标签
添加 element_name 和 tag_label 的全文搜索索引
upload_resource() 和 create_ai_generated_resource() 支持 element_tag_id 参数
自动填充冗余字段（element_name、tag_label）
完善应用层引用完整性校验（验证标签归属项目）
更新文档说明冗余字段的设计原因和数据一致性维护策略

v3.1 (2026-01-29)

修复依赖注入模式，Service 通过构造函数注入依赖
统一日志系统，使用 app.core.logging.get_logger
添加 API 路由实现（app/api/v1/project_resources.py）
添加 Schema 定义（app/schemas/project_resource.py）
添加外部依赖说明（ffmpeg、Pillow）
修复模型导入语句（添加 Integer 类型）
改进错误处理和日志记录
添加 ffmpeg 可用性检查
完善应用层引用完整性校验（通过 ProjectService）

v3.0 (2026-01-29)

符合 jointo-tech-stack v1.0 规范
使用 UUID v7 主键（应用层生成）
移除所有物理外键约束，应用层保证引用完整性
使用 SMALLINT + IntEnum 替代 PostgreSQL ENUM
使用 AsyncSession 替代 Session
使用 aware datetime（timezone.utc）
使用 TIMESTAMP(timezone=True) 即 TIMESTAMPTZ 类型
统一 API 响应格式（code/message/data）
添加 COMMENT ON 语法注释
完善应用层引用完整性校验逻辑

v2.2 (2025-01-27)

新增 element_tag_id 字段，支持素材关联到剧本元素标签
新增按标签过滤素材的 API 接口
新增获取标签素材列表的快捷接口
完善 API 响应，包含标签信息（element_tag_label、element_name）

v2.1 (2025-01-27)

新增实拍素材类型（footage）
支持视频格式上传（MP4、MOV、AVI、WebM）
新增视频缩略图生成（提取第一帧）
新增视频元数据提取（时长、尺寸）

v2.0 (2025-01-27)

重构为项目素材服务（从 resource-service 拆分）
专注于项目专属素材管理
集成 FileStorageService 实现去重
新增缩略图自动生成
新增 AI 生成素材支持

v1.0 (2025-01-27)

初始版本（原 resource-service）

文档版本：v3.7
最后更新：2026-02-03

59 KiB Raw Permalink Blame History

项目素材服务

目录

服务概述

职责

设计原则

与资源库的区别

核心功能

1. 素材上传

1.1 角色素材（character）

1.2 场景素材（scene）

1.3 道具素材（prop）

1.4 实拍素材（footage）

2. 素材管理

3. AI 生成素材

4. 素材与分镜关联

服务实现

ProjectResourceService 类

API 接口

API 路由实现

Schema 定义

API 接口说明

1. 上传项目素材

2. 获取素材详情

3. 更新素材信息

4. 检查素材使用情况

5. 删除素材

6. 获取项目素材列表

7. 获取标签的素材列表（快捷方式）

外部依赖

系统依赖

Python 依赖

依赖检查

数据库设计

project_resources 表结构

storyboard_resources 关联表

冗余字段说明

数据模型

ProjectResource 模型

测试

单元测试

集成测试

测试覆盖率

数据库迁移

创建迁移文件

回滚迁移

相关文档

变更记录

59 KiB

Raw Permalink Blame History