You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
63 KiB
63 KiB
附件管理服务
文档版本：v4.1
最后更新：2026-02-03
合规状态：✅ 符合 jointo-tech-stack 规范
目录
服务概述
附件管理服务采用多态关联设计，支持任意实体的附件管理，包括用户头像、项目封面、分镜缩略图、文档附件等。
职责
- 文档附件上传（PDF、Word、Excel、TXT、Markdown）
- 图片附件上传（用于头像、封面、缩略图）
- 文件下载（预签名 URL）
- 文件删除（软删除）
- 文件去重（基于 SHA256 校验和）
- 访问权限控制
- 多实体附件管理（用户、项目、分镜、角色、场景等）
设计原则
- 多态关联：使用
related_id+related_type+attachment_purpose实现通用附件管理 - 业务分离：attachments 表管理文档类附件，project_resources 表管理项目素材
- 去重优先：使用 file_checksums 表实现全局去重
- 高度可扩展：新增业务场景无需修改表结构，只需增加枚举值
核心功能
1. 文件上传
支持两大类文件：
1.1 文档类（document）
- 剧本文档：PDF、Word (.doc/.docx)、TXT、Markdown
- 办公文档：Excel (.xls/.xlsx)
- 大小限制：100MB
1.2 图片类（image）
- 用途：用户头像、项目封面、分镜缩略图
- 格式：JPG、PNG、GIF、WebP、BMP
- 大小限制：20MB
通用功能：
- 自动计算文件校验和（SHA256）
- 文件去重（相同文件只存储一次）
- 上传到对象存储（MinIO/S3）
- 多态关联（支持任意实体）
2. 文件下载
- 生成预签名 URL（临时访问链接）
- 可设置链接有效期
- 统计下载次数
3. 访问控制
- 公开/私有文件
- 实体级权限控制（基于 related_type）
- 用户级权限控制
4. 文件管理
- 软删除（标记
deleted_at） - 统计访问次数
- 文件元数据管理
- 按实体和用途查询
多态关联设计
设计理念
使用 related_id + related_type + attachment_purpose 三元组实现通用附件管理。
核心字段
| 字段 | 类型 | 说明 | 示例 |
|---|---|---|---|
related_id |
UUID | 关联实体的 ID | 550e8400-e29b-41d4-a716-446655440000 |
related_type |
SMALLINT | 关联实体类型 | 1=user, 2=project, 3=storyboard |
attachment_purpose |
SMALLINT | 附件用途 | 1=avatar, 2=cover, 3=thumbnail |
支持的实体类型（related_type）
| 枚举值 | 字符串 | 说明 | 典型用途 |
|---|---|---|---|
| 1 | user | 用户 | 头像 |
| 2 | project | 项目 | 封面、文档 |
| 3 | storyboard | 分镜 | 缩略图 |
| 4 | character | 角色 | 头像、参考图 |
| 5 | scene | 场景 | 参考图 |
| 6 | prop | 道具 | 参考图 |
| 7 | location | 地点 | 参考图 |
支持的附件用途（attachment_purpose）
| 枚举值 | 字符串 | 说明 | 适用实体 |
|---|---|---|---|
| 1 | avatar | 头像 | user, character |
| 2 | cover | 封面 | project |
| 3 | thumbnail | 缩略图 | storyboard |
| 4 | document | 文档 | project |
| 5 | reference | 参考资料 | character, scene, prop, location |
| 6 | attachment | 通用附件 | 所有实体 |
使用场景示例
# 用户头像
related_type = RelatedType.USER (1)
attachment_purpose = AttachmentPurpose.AVATAR (1)
# 项目封面
related_type = RelatedType.PROJECT (2)
attachment_purpose = AttachmentPurpose.COVER (2)
# 分镜缩略图
related_type = RelatedType.STORYBOARD (3)
attachment_purpose = AttachmentPurpose.THUMBNAIL (3)
# 角色参考图
related_type = RelatedType.CHARACTER (4)
attachment_purpose = AttachmentPurpose.REFERENCE (5)
# 项目文档
related_type = RelatedType.PROJECT (2)
attachment_purpose = AttachmentPurpose.DOCUMENT (4)
设计优势
✅ 高度可扩展：新增业务场景（如道具、地点）无需修改表结构
✅ 查询统一：所有附件查询使用相同接口
✅ 用途明确：attachment_purpose 清晰区分用途
✅ 关联一致：统一由 attachments 表管理关联
✅ 性能优化：组合索引 (related_id, related_type, attachment_purpose) 高效查询
✅ 业务清晰：符合视频制作工作台的多实体附件需求
服务实现
AttachmentService 类
# app/services/attachment_service.py
from typing import List, Optional, Dict, Any
from uuid import UUID
from sqlmodel.ext.asyncio.session import AsyncSession
from fastapi import UploadFile
from app.models.attachment import Attachment, AttachmentCategory, RelatedType, AttachmentPurpose
from app.repositories.attachment_repository import AttachmentRepository
from app.services.file_storage_service import FileStorageService
from app.core.exceptions import NotFoundError, ValidationError, PermissionError
from loguru import logger
class AttachmentService:
"""附件业务逻辑层"""
def __init__(self, session: AsyncSession):
self.repository = AttachmentRepository(session)
self.file_storage = FileStorageService(session)
self.session = session
# 允许的文件类型
ALLOWED_DOCUMENT_TYPES = {
'application/pdf',
'application/msword',
'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'application/vnd.ms-excel',
'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
'text/plain',
'text/markdown'
}
ALLOWED_IMAGE_TYPES = {
'image/jpeg',
'image/png',
'image/gif',
'image/webp',
'image/bmp'
}
# 文件大小限制（字节）
MAX_FILE_SIZE = {
'document': 100 * 1024 * 1024, # 100MB
'image': 20 * 1024 * 1024 # 20MB
}
async def upload_attachment(
self,
user_id: UUID,
file: UploadFile,
category: str = 'document',
related_id: UUID = None,
related_type: str = 'project',
attachment_purpose: str = 'attachment',
is_public: bool = False
) -> Attachment:
"""上传附件（多态关联）
Args:
user_id: 上传者用户 ID
file: 上传的文件
category: 文件类别（document/image）
related_id: 关联实体 ID
related_type: 关联实体类型（user/project/storyboard/character/scene）
attachment_purpose: 附件用途（avatar/cover/thumbnail/document/reference）
is_public: 是否公开
Returns:
创建的附件记录
Raises:
NotFoundError: 用户或关联实体不存在
ValidationError: 文件类型或大小不符合要求
"""
logger.info(
f"上传附件: user_id={user_id}, filename={file.filename}, "
f"related_type={related_type}, purpose={attachment_purpose}"
)
# 1. 验证上传者存在（应用层引用完整性检查）
if not await self.repository.exists_user(user_id):
raise NotFoundError(f"用户不存在: {user_id}")
# 2. 验证关联实体存在
related_type_enum = RelatedType.from_string(related_type)
if not await self.repository.exists_related_entity(related_id, related_type_enum):
raise NotFoundError(
f"{RelatedType.get_display_name(related_type_enum)}不存在: {related_id}"
)
# 3. 验证用户对关联实体的权限
has_permission = await self.repository.check_related_permission(
user_id, related_id, related_type_enum, 'editor'
)
if not has_permission:
raise PermissionError(
f"没有权限在该{RelatedType.get_display_name(related_type_enum)}下上传附件"
)
# 4. 验证文件类型
category_enum = AttachmentCategory.from_string(category)
if category_enum == AttachmentCategory.DOCUMENT:
if file.content_type not in self.ALLOWED_DOCUMENT_TYPES:
raise ValidationError(f"不支持的文档文件类型: {file.content_type}")
elif category_enum == AttachmentCategory.IMAGE:
if file.content_type not in self.ALLOWED_IMAGE_TYPES:
raise ValidationError(f"不支持的图片文件类型: {file.content_type}")
# 5. 读取文件内容
content = await file.read()
file_size = len(content)
# 6. 验证文件大小
max_size = self.MAX_FILE_SIZE.get(category, self.MAX_FILE_SIZE['document'])
if file_size > max_size:
raise ValidationError(
f"文件大小超过限制: {file_size / 1024 / 1024:.2f}MB > {max_size / 1024 / 1024:.2f}MB"
)
# 7. 使用 FileStorageService 上传（带去重）
file_meta = await self.file_storage.upload_file(
file_content=content,
filename=file.filename,
content_type=file.content_type,
category=f'attachment_{category}',
user_id=user_id
)
# 8. 创建附件记录
purpose_enum = AttachmentPurpose.from_string(attachment_purpose)
attachment = Attachment(
name=file.filename,
original_name=file.filename,
file_url=file_meta.file_url,
file_size=file_meta.file_size,
mime_type=file.content_type,
category=category_enum,
extension=file_meta.extension,
checksum=file_meta.checksum,
related_id=related_id,
related_type=related_type_enum,
attachment_purpose=purpose_enum,
uploaded_by=user_id,
is_public=is_public,
storage_provider=file_meta.storage_provider,
storage_path=file_meta.storage_path
)
created_attachment = await self.repository.create(attachment)
logger.info(f"附件上传成功: {created_attachment.attachment_id}")
return created_attachment
async def get_attachment(
self,
user_id: UUID,
attachment_id: UUID
) -> Attachment:
"""获取附件详情
Args:
user_id: 当前用户 ID
attachment_id: 附件 ID
Returns:
附件记录
Raises:
NotFoundError: 附件不存在
PermissionError: 无权限访问
"""
attachment = await self.repository.get_by_id(attachment_id)
if not attachment:
raise NotFoundError("附件不存在")
# 检查访问权限
if not attachment.is_public:
# 检查关联实体权限
has_permission = await self.repository.check_related_permission(
user_id, attachment.related_id, attachment.related_type, 'viewer'
)
if not has_permission and attachment.uploaded_by != user_id:
raise PermissionError("没有权限访问此附件")
# 增加访问计数
await self.repository.increment_access_count(attachment_id)
return attachment
async def get_download_url(
self,
user_id: UUID,
attachment_id: UUID,
expires: int = 3600
) -> str:
"""获取附件下载链接
Args:
user_id: 当前用户 ID
attachment_id: 附件 ID
expires: 链接有效期（秒）
Returns:
预签名下载 URL
"""
attachment = await self.get_attachment(user_id, attachment_id)
# 增加下载计数
await self.repository.increment_download_count(attachment_id)
# 生成预签名URL
return await self.file_storage.get_presigned_url(
attachment.storage_path,
expires=expires
)
async def delete_attachment(
self,
user_id: UUID,
attachment_id: UUID
) -> None:
"""删除附件
Args:
user_id: 当前用户 ID
attachment_id: 附件 ID
Raises:
NotFoundError: 附件不存在
PermissionError: 无权限删除
"""
attachment = await self.repository.get_by_id(attachment_id)
if not attachment:
raise NotFoundError("附件不存在")
# 检查权限（只有上传者或实体所有者可以删除）
if attachment.uploaded_by != user_id:
has_permission = await self.repository.check_related_permission(
user_id, attachment.related_id, attachment.related_type, 'owner'
)
if not has_permission:
raise PermissionError("没有权限删除此附件")
# 软删除数据库记录
await self.repository.soft_delete(attachment_id)
# 减少文件引用计数
await self.file_storage.decrease_reference_count(attachment.checksum)
logger.info(f"附件删除成功: {attachment_id}")
async def get_attachments_by_related(
self,
user_id: UUID,
related_id: UUID,
related_type: str,
attachment_purpose: Optional[str] = None,
category: Optional[str] = None,
page: int = 1,
page_size: int = 20
) -> Dict[str, Any]:
"""获取关联实体的附件列表
Args:
user_id: 当前用户 ID
related_id: 关联实体 ID
related_type: 关联实体类型
attachment_purpose: 附件用途筛选（可选）
category: 文件类别筛选（可选）
page: 页码
page_size: 每页数量
Returns:
分页附件列表
Raises:
NotFoundError: 关联实体不存在
PermissionError: 无权限访问
"""
# 验证关联实体存在
related_type_enum = RelatedType.from_string(related_type)
if not await self.repository.exists_related_entity(related_id, related_type_enum):
raise NotFoundError(
f"{RelatedType.get_display_name(related_type_enum)}不存在"
)
# 检查权限
has_permission = await self.repository.check_related_permission(
user_id, related_id, related_type_enum, 'viewer'
)
if not has_permission:
raise PermissionError(
f"没有权限访问该{RelatedType.get_display_name(related_type_enum)}"
)
# 转换枚举
purpose_enum = None
if attachment_purpose:
purpose_enum = AttachmentPurpose.from_string(attachment_purpose)
category_enum = None
if category:
category_enum = AttachmentCategory.from_string(category)
attachments, total = await self.repository.get_by_related(
related_id, related_type_enum, purpose_enum, category_enum, page, page_size
)
return {
'items': [att.to_dict() for att in attachments],
'total': total,
'page': page,
'pageSize': page_size,
'totalPages': (total + page_size - 1) // page_size
}
async def get_attachment_by_purpose(
self,
user_id: UUID,
related_id: UUID,
related_type: str,
purpose: str
) -> Optional[Attachment]:
"""获取指定用途的附件（如封面、头像）
Args:
user_id: 当前用户 ID
related_id: 关联实体 ID
related_type: 关联实体类型
purpose: 附件用途
Returns:
附件记录（如果存在）
"""
related_type_enum = RelatedType.from_string(related_type)
purpose_enum = AttachmentPurpose.from_string(purpose)
# 检查权限
has_permission = await self.repository.check_related_permission(
user_id, related_id, related_type_enum, 'viewer'
)
if not has_permission:
raise PermissionError("没有权限访问")
return await self.repository.get_by_purpose(
related_id, related_type_enum, purpose_enum
)
Repository 层
AttachmentRepository 类
# app/repositories/attachment_repository.py
from typing import List, Optional, Tuple
from uuid import UUID
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy import func
from app.models.attachment import Attachment, AttachmentCategory, RelatedType, AttachmentPurpose
from loguru import logger
class AttachmentRepository:
"""附件数据访问层"""
def __init__(self, session: AsyncSession):
self.session = session
# ==================== 引用完整性验证 ====================
async def exists_user(self, user_id: UUID) -> bool:
"""检查用户是否存在（应用层引用完整性）"""
from app.models.user import User
result = await self.session.execute(
select(User.user_id).where(
User.user_id == user_id,
User.deleted_at.is_(None)
).limit(1)
)
return result.scalar_one_or_none() is not None
async def exists_related_entity(self, related_id: UUID, related_type: RelatedType) -> bool:
"""检查关联实体是否存在（多态关联）"""
if related_type == RelatedType.USER:
from app.models.user import User
result = await self.session.execute(
select(User.user_id).where(
User.user_id == related_id,
User.deleted_at.is_(None)
).limit(1)
)
elif related_type == RelatedType.PROJECT:
from app.models.project import Project
result = await self.session.execute(
select(Project.project_id).where(
Project.project_id == related_id,
Project.deleted_at.is_(None)
).limit(1)
)
elif related_type == RelatedType.STORYBOARD:
from app.models.storyboard import Storyboard
result = await self.session.execute(
select(Storyboard.storyboard_id).where(
Storyboard.storyboard_id == related_id,
Storyboard.deleted_at.is_(None)
).limit(1)
)
elif related_type == RelatedType.CHARACTER:
from app.models.screenplay import ScreenplayCharacter
result = await self.session.execute(
select(ScreenplayCharacter.character_id).where(
ScreenplayCharacter.character_id == related_id,
ScreenplayCharacter.deleted_at.is_(None)
).limit(1)
)
elif related_type == RelatedType.SCENE:
from app.models.screenplay import ScreenplayScene
result = await self.session.execute(
select(ScreenplayScene.scene_id).where(
ScreenplayScene.scene_id == related_id,
ScreenplayScene.deleted_at.is_(None)
).limit(1)
)
else:
# 其他类型暂不支持
return False
return result.scalar_one_or_none() is not None
async def check_related_permission(
self,
user_id: UUID,
related_id: UUID,
related_type: RelatedType,
required_role: str = 'viewer'
) -> bool:
"""检查用户对关联实体的权限"""
if related_type == RelatedType.USER:
# 用户只能访问自己的附件
return user_id == related_id
elif related_type == RelatedType.PROJECT:
from app.repositories.project_repository import ProjectRepository
project_repo = ProjectRepository(self.session)
return await project_repo.check_user_permission(
user_id, related_id, required_role
)
elif related_type == RelatedType.STORYBOARD:
# 通过分镜所属项目检查权限
from app.models.storyboard import Storyboard
result = await self.session.execute(
select(Storyboard.project_id).where(
Storyboard.storyboard_id == related_id,
Storyboard.deleted_at.is_(None)
)
)
project_id = result.scalar_one_or_none()
if not project_id:
return False
from app.repositories.project_repository import ProjectRepository
project_repo = ProjectRepository(self.session)
return await project_repo.check_user_permission(
user_id, project_id, required_role
)
elif related_type in [RelatedType.CHARACTER, RelatedType.SCENE]:
# 通过剧本所属项目检查权限
if related_type == RelatedType.CHARACTER:
from app.models.screenplay import ScreenplayCharacter
result = await self.session.execute(
select(ScreenplayCharacter.screenplay_id).where(
ScreenplayCharacter.character_id == related_id,
ScreenplayCharacter.deleted_at.is_(None)
)
)
else: # SCENE
from app.models.screenplay import ScreenplayScene
result = await self.session.execute(
select(ScreenplayScene.screenplay_id).where(
ScreenplayScene.scene_id == related_id,
ScreenplayScene.deleted_at.is_(None)
)
)
screenplay_id = result.scalar_one_or_none()
if not screenplay_id:
return False
# 获取剧本所属项目
from app.models.screenplay import Screenplay
result = await self.session.execute(
select(Screenplay.project_id).where(
Screenplay.screenplay_id == screenplay_id,
Screenplay.deleted_at.is_(None)
)
)
project_id = result.scalar_one_or_none()
if not project_id:
return False
from app.repositories.project_repository import ProjectRepository
project_repo = ProjectRepository(self.session)
return await project_repo.check_user_permission(
user_id, project_id, required_role
)
return False
# ==================== 基础 CRUD ====================
async def get_by_id(self, attachment_id: UUID) -> Optional[Attachment]:
"""根据 ID 获取附件"""
result = await self.session.execute(
select(Attachment).where(
Attachment.attachment_id == attachment_id,
Attachment.deleted_at.is_(None)
)
)
return result.scalar_one_or_none()
async def get_by_related(
self,
related_id: UUID,
related_type: RelatedType,
attachment_purpose: Optional[AttachmentPurpose] = None,
category: Optional[AttachmentCategory] = None,
page: int = 1,
page_size: int = 20
) -> Tuple[List[Attachment], int]:
"""获取关联实体的附件列表"""
query = select(Attachment).where(
Attachment.related_id == related_id,
Attachment.related_type == related_type,
Attachment.deleted_at.is_(None)
)
if attachment_purpose is not None:
query = query.where(Attachment.attachment_purpose == attachment_purpose)
if category is not None:
query = query.where(Attachment.category == category)
query = query.order_by(Attachment.created_at.desc())
# 查询总数
count_query = select(func.count()).select_from(query.subquery())
total_result = await self.session.execute(count_query)
total = total_result.scalar_one()
# 分页查询
query = query.offset((page - 1) * page_size).limit(page_size)
result = await self.session.execute(query)
attachments = result.scalars().all()
return list(attachments), total
async def get_by_purpose(
self,
related_id: UUID,
related_type: RelatedType,
purpose: AttachmentPurpose
) -> Optional[Attachment]:
"""获取指定用途的附件（如封面、头像）"""
result = await self.session.execute(
select(Attachment).where(
Attachment.related_id == related_id,
Attachment.related_type == related_type,
Attachment.attachment_purpose == purpose,
Attachment.deleted_at.is_(None)
).order_by(Attachment.created_at.desc()).limit(1)
)
return result.scalar_one_or_none()
async def create(self, attachment: Attachment) -> Attachment:
"""创建附件"""
self.session.add(attachment)
await self.session.commit()
await self.session.refresh(attachment)
logger.info(f"创建附件: {attachment.attachment_id}, 文件名: {attachment.name}")
return attachment
async def soft_delete(self, attachment_id: UUID) -> None:
"""软删除附件"""
from datetime import datetime, timezone
attachment = await self.get_by_id(attachment_id)
if attachment:
attachment.deleted_at = datetime.now(timezone.utc)
await self.session.commit()
logger.info(f"软删除附件: {attachment_id}")
async def increment_access_count(self, attachment_id: UUID) -> None:
"""增加访问计数"""
attachment = await self.get_by_id(attachment_id)
if attachment:
attachment.access_count += 1
await self.session.commit()
async def increment_download_count(self, attachment_id: UUID) -> None:
"""增加下载计数"""
attachment = await self.get_by_id(attachment_id)
if attachment:
attachment.download_count += 1
await self.session.commit()
Schema 定义
请求 Schema
# app/schemas/attachment.py
from typing import Optional
from uuid import UUID
from pydantic import BaseModel, Field, field_validator
class AttachmentUploadRequest(BaseModel):
"""上传附件请求（multipart/form-data）"""
category: str = Field(default='document', description="文件类别：document/image")
related_id: UUID = Field(..., alias="relatedId", description="关联实体 ID")
related_type: str = Field(..., alias="relatedType", description="关联实体类型：user/project/storyboard/character/scene")
attachment_purpose: str = Field(default='attachment', alias="attachmentPurpose", description="附件用途：avatar/cover/thumbnail/document/reference/attachment")
is_public: bool = Field(default=False, alias="isPublic", description="是否公开")
@field_validator('category')
@classmethod
def validate_category(cls, v: str) -> str:
if v not in ['document', 'image']:
raise ValueError('文件类别必须是 document 或 image')
return v
@field_validator('related_type')
@classmethod
def validate_related_type(cls, v: str) -> str:
valid_types = ['user', 'project', 'storyboard', 'character', 'scene', 'prop', 'location']
if v not in valid_types:
raise ValueError(f'关联实体类型必须是 {", ".join(valid_types)} 之一')
return v
@field_validator('attachment_purpose')
@classmethod
def validate_purpose(cls, v: str) -> str:
valid_purposes = ['avatar', 'cover', 'thumbnail', 'document', 'reference', 'attachment']
if v not in valid_purposes:
raise ValueError(f'附件用途必须是 {", ".join(valid_purposes)} 之一')
return v
class Config:
populate_by_name = True
class AttachmentQueryRequest(BaseModel):
"""查询附件列表请求"""
related_id: UUID = Field(..., alias="relatedId", description="关联实体 ID")
related_type: str = Field(..., alias="relatedType", description="关联实体类型")
attachment_purpose: Optional[str] = Field(None, alias="attachmentPurpose", description="附件用途筛选")
category: Optional[str] = Field(None, description="文件类别筛选")
page: int = Field(1, ge=1, description="页码")
page_size: int = Field(20, ge=1, le=100, alias="pageSize", description="每页数量")
class Config:
populate_by_name = True
响应 Schema
class AttachmentResponse(BaseModel):
"""附件响应"""
id: str
name: str
original_name: str = Field(alias="originalName")
file_url: str = Field(alias="fileUrl")
file_size: int = Field(alias="fileSize")
mime_type: str = Field(alias="mimeType")
category: str
category_display: str = Field(alias="categoryDisplay")
extension: Optional[str] = None
checksum: str
related_id: str = Field(alias="relatedId")
related_type: str = Field(alias="relatedType")
related_type_display: str = Field(alias="relatedTypeDisplay")
attachment_purpose: str = Field(alias="attachmentPurpose")
attachment_purpose_display: str = Field(alias="attachmentPurposeDisplay")
uploaded_by: str = Field(alias="uploadedBy")
is_public: bool = Field(alias="isPublic")
access_count: int = Field(alias="accessCount")
download_count: int = Field(alias="downloadCount")
storage_provider: Optional[str] = Field(None, alias="storageProvider")
storage_path: Optional[str] = Field(None, alias="storagePath")
created_at: str = Field(alias="createdAt")
updated_at: str = Field(alias="updatedAt")
expires_at: Optional[str] = Field(None, alias="expiresAt")
class Config:
populate_by_name = True
class AttachmentListResponse(BaseModel):
"""附件列表响应"""
items: list[AttachmentResponse]
total: int
page: int
page_size: int = Field(alias="pageSize")
total_pages: int = Field(alias="totalPages")
class Config:
populate_by_name = True
class DownloadUrlResponse(BaseModel):
"""下载链接响应"""
download_url: str = Field(alias="downloadUrl", description="预签名下载 URL")
expires_in: int = Field(alias="expiresIn", description="有效期（秒）")
class Config:
populate_by_name = True
API 接口
API 路由实现
# app/api/v1/attachments.py
from typing import Annotated
from uuid import UUID
from fastapi import APIRouter, Depends, UploadFile, File, Form
from sqlmodel.ext.asyncio.session import AsyncSession
from app.core.database import get_session
from app.api.deps import get_current_user
from app.models.user import User
from app.repositories.attachment_repository import AttachmentRepository
from app.services.attachment_service import AttachmentService
from app.schemas.attachment import (
AttachmentResponse,
AttachmentListResponse,
DownloadUrlResponse
)
from app.schemas.response import ApiResponse
from loguru import logger
router = APIRouter(prefix="/attachments", tags=["attachments"])
# ==================== 依赖注入 ====================
async def get_attachment_service(
session: Annotated[AsyncSession, Depends(get_session)]
) -> AttachmentService:
"""获取附件服务实例"""
return AttachmentService(session)
# ==================== API 路由 ====================
@router.post("", response_model=ApiResponse[AttachmentResponse])
async def upload_attachment(
file: Annotated[UploadFile, File(...)],
category: Annotated[str, Form()] = 'document',
project_id: Annotated[str | None, Form(alias="projectId")] = None,
is_public: Annotated[bool, Form(alias="isPublic")] = False,
current_user: Annotated[User, Depends(get_current_user)] = None,
service: Annotated[AttachmentService, Depends(get_attachment_service)] = None
):
"""上传附件
- **file**: 文件（必填）
- **category**: 文件类别（document/image，默认 document）
- **projectId**: 项目 ID（可选）
- **isPublic**: 是否公开（默认 false）
"""
project_uuid = UUID(project_id) if project_id else None
attachment = await service.upload_attachment(
user_id=current_user.user_id,
file=file,
category=category,
project_id=project_uuid,
is_public=is_public
)
return ApiResponse.success(data=attachment.to_dict())
@router.get("/{attachment_id}", response_model=ApiResponse[AttachmentResponse])
async def get_attachment(
attachment_id: UUID,
current_user: Annotated[User, Depends(get_current_user)],
service: Annotated[AttachmentService, Depends(get_attachment_service)]
):
"""获取附件详情"""
attachment = await service.get_attachment(
user_id=current_user.user_id,
attachment_id=attachment_id
)
return ApiResponse.success(data=attachment.to_dict())
@router.get("/{attachment_id}/download", response_model=ApiResponse[DownloadUrlResponse])
async def get_download_url(
attachment_id: UUID,
expires: int = 3600,
current_user: Annotated[User, Depends(get_current_user)] = None,
service: Annotated[AttachmentService, Depends(get_attachment_service)] = None
):
"""获取下载链接
- **expires**: 链接有效期（秒），默认 3600
"""
download_url = await service.get_download_url(
user_id=current_user.user_id,
attachment_id=attachment_id,
expires=expires
)
return ApiResponse.success(data={
'downloadUrl': download_url,
'expiresIn': expires
})
@router.delete("/{attachment_id}", response_model=ApiResponse[None])
async def delete_attachment(
attachment_id: UUID,
current_user: Annotated[User, Depends(get_current_user)],
service: Annotated[AttachmentService, Depends(get_attachment_service)]
):
"""删除附件"""
await service.delete_attachment(
user_id=current_user.user_id,
attachment_id=attachment_id
)
return ApiResponse.success(message="附件删除成功")
@router.get("", response_model=ApiResponse[AttachmentListResponse])
async def get_attachments(
project_id: UUID,
category: str | None = None,
page: int = 1,
page_size: int = 20,
current_user: Annotated[User, Depends(get_current_user)] = None,
service: Annotated[AttachmentService, Depends(get_attachment_service)] = None
):
"""获取项目附件列表
- **projectId**: 项目 ID（必填）
- **category**: 文件类别筛选（可选：document/image）
- **page**: 页码
- **pageSize**: 每页数量
"""
result = await service.get_attachments_by_project(
user_id=current_user.user_id,
project_id=project_id,
category=category,
page=page,
page_size=page_size
)
return ApiResponse.success(data=result)
API 接口示例
1. 上传附件
POST /api/v1/attachments
请求（multipart/form-data）：
file: 文件（必填）category: 文件类别（document|image，默认document）projectId: 项目 ID（可选）isPublic: 是否公开（默认 false）
响应：
{
"code": 200,
"message": "Success",
"data": {
"id": "550e8400-e29b-41d4-a716-446655440000",
"name": "项目预算表.xlsx",
"originalName": "项目预算表.xlsx",
"fileUrl": "https://storage.jointo.ai/attachments/document/1/abc123.xlsx",
"fileSize": 1024000,
"mimeType": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"category": "document",
"categoryDisplay": "文档",
"extension": ".xlsx",
"checksum": "abc123...",
"projectId": "550e8400-e29b-41d4-a716-446655440001",
"uploadedBy": "550e8400-e29b-41d4-a716-446655440002",
"isPublic": false,
"accessCount": 0,
"downloadCount": 0,
"createdAt": "2026-01-28T10:00:00Z"
}
}
2. 获取附件详情
GET /api/v1/attachments/{attachment_id}
响应：同上
3. 获取下载链接
GET /api/v1/attachments/{attachment_id}/download?expires=3600
响应：
{
"code": 200,
"message": "Success",
"data": {
"downloadUrl": "https://storage.jointo.ai/attachments/document/1/abc123.xlsx?X-Amz-Expires=3600&...",
"expiresIn": 3600
}
}
4. 删除附件
DELETE /api/v1/attachments/{attachment_id}
响应：
{
"code": 200,
"message": "附件删除成功",
"data": null
}
5. 获取项目附件列表
GET /api/v1/attachments?projectId={project_id}&category=document&page=1&pageSize=20
响应：
{
"code": 200,
"message": "Success",
"data": {
"items": [
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"name": "项目预算表.xlsx",
"fileSize": 1024000,
"category": "document",
"createdAt": "2026-01-28T10:00:00Z"
}
],
"total": 10,
"page": 1,
"pageSize": 20,
"totalPages": 1
}
}
数据库设计
attachments 表结构
-- 附件表 - 多态关联设计
CREATE TABLE attachments (
attachment_id UUID PRIMARY KEY, -- 附件唯一标识（UUID v7）
name TEXT NOT NULL, -- 附件名称
original_name TEXT NOT NULL, -- 原始文件名
file_url TEXT NOT NULL, -- 文件访问 URL
file_size BIGINT NOT NULL, -- 文件大小（字节）
mime_type TEXT NOT NULL, -- MIME 类型
-- 文件类型分类（使用 SMALLINT 存储枚举）
category SMALLINT NOT NULL CHECK (category IN (1, 2)), -- 文件类型：1=document, 2=image
-- 文件元数据
extension TEXT, -- 文件扩展名
checksum TEXT NOT NULL, -- 文件校验和（SHA256）
-- ✅ 多态关联字段（核心设计）
related_id UUID NOT NULL, -- 关联实体 ID（通用）
related_type SMALLINT NOT NULL, -- 关联实体类型：1=user, 2=project, 3=storyboard, 4=character, 5=scene, 6=prop, 7=location
attachment_purpose SMALLINT NOT NULL, -- 附件用途：1=avatar, 2=cover, 3=thumbnail, 4=document, 5=reference, 6=attachment
-- 访问控制
is_public BOOLEAN NOT NULL DEFAULT false, -- 是否公开访问
access_count INTEGER NOT NULL DEFAULT 0, -- 访问次数
download_count INTEGER NOT NULL DEFAULT 0, -- 下载次数
-- 存储信息
storage_provider TEXT, -- 存储提供商（minio/s3/oss）
storage_path TEXT, -- 存储路径
-- 审计字段（无外键约束，应用层验证）
uploaded_by UUID NOT NULL, -- 上传者用户 ID - 应用层验证
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), -- 创建时间
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), -- 更新时间
deleted_at TIMESTAMPTZ, -- 删除时间（软删除）
expires_at TIMESTAMPTZ -- 过期时间
);
-- 表注释
COMMENT ON TABLE attachments IS '附件表 - 多态关联设计，支持任意实体的附件管理';
-- 字段注释
COMMENT ON COLUMN attachments.attachment_id IS '附件唯一标识（UUID v7）';
COMMENT ON COLUMN attachments.name IS '附件名称';
COMMENT ON COLUMN attachments.original_name IS '原始文件名';
COMMENT ON COLUMN attachments.file_url IS '文件访问 URL';
COMMENT ON COLUMN attachments.file_size IS '文件大小（字节）';
COMMENT ON COLUMN attachments.mime_type IS 'MIME 类型';
COMMENT ON COLUMN attachments.category IS '文件类型：1=document, 2=image';
COMMENT ON COLUMN attachments.extension IS '文件扩展名';
COMMENT ON COLUMN attachments.checksum IS '文件校验和（SHA256）';
COMMENT ON COLUMN attachments.related_id IS '关联实体 ID（多态关联）';
COMMENT ON COLUMN attachments.related_type IS '关联实体类型：1=user, 2=project, 3=storyboard, 4=character, 5=scene, 6=prop, 7=location';
COMMENT ON COLUMN attachments.attachment_purpose IS '附件用途：1=avatar, 2=cover, 3=thumbnail, 4=document, 5=reference, 6=attachment';
COMMENT ON COLUMN attachments.is_public IS '是否公开访问';
COMMENT ON COLUMN attachments.access_count IS '访问次数';
COMMENT ON COLUMN attachments.download_count IS '下载次数';
COMMENT ON COLUMN attachments.storage_provider IS '存储提供商（minio/s3/oss）';
COMMENT ON COLUMN attachments.storage_path IS '存储路径';
COMMENT ON COLUMN attachments.uploaded_by IS '上传者用户 ID - 应用层验证';
COMMENT ON COLUMN attachments.created_at IS '创建时间';
COMMENT ON COLUMN attachments.updated_at IS '更新时间';
COMMENT ON COLUMN attachments.deleted_at IS '删除时间（软删除）';
COMMENT ON COLUMN attachments.expires_at IS '过期时间';
-- ✅ 核心索引（多态关联必须）
CREATE INDEX idx_attachments_related ON attachments (related_id, related_type, attachment_purpose)
WHERE deleted_at IS NULL;
CREATE INDEX idx_attachments_related_type ON attachments (related_type, related_id)
WHERE deleted_at IS NULL;
CREATE INDEX idx_attachments_uploaded_by ON attachments (uploaded_by)
WHERE deleted_at IS NULL;
CREATE INDEX idx_attachments_category ON attachments (category)
WHERE deleted_at IS NULL;
CREATE INDEX idx_attachments_created_at ON attachments (created_at DESC)
WHERE deleted_at IS NULL;
CREATE INDEX idx_attachments_expires_at ON attachments (expires_at)
WHERE expires_at IS NOT NULL;
CREATE INDEX idx_attachments_checksum ON attachments (checksum);
-- 组合索引
CREATE INDEX idx_attachments_uploader_created ON attachments (uploaded_by, created_at DESC)
WHERE deleted_at IS NULL;
-- 全文搜索索引（需要 pg_trgm 扩展）
CREATE INDEX idx_attachments_name_trgm ON attachments USING GIN (name gin_trgm_ops)
WHERE deleted_at IS NULL;
CREATE INDEX idx_attachments_original_name_trgm ON attachments USING GIN (original_name gin_trgm_ops)
WHERE deleted_at IS NULL;
-- 触发器
CREATE TRIGGER update_attachments_updated_at
BEFORE UPDATE ON attachments
FOR EACH ROW
EXECUTE FUNCTION update_updated_at_column();
枚举值映射表
related_type（关联实体类型）
| 枚举值 | 字符串 | 说明 |
|---|---|---|
| 1 | user | 用户 |
| 2 | project | 项目 |
| 3 | storyboard | 分镜 |
| 4 | character | 角色 |
| 5 | scene | 场景 |
| 6 | prop | 道具 |
| 7 | location | 地点 |
attachment_purpose（附件用途）
| 枚举值 | 字符串 | 说明 |
|---|---|---|
| 1 | avatar | 头像 |
| 2 | cover | 封面 |
| 3 | thumbnail | 缩略图 |
| 4 | document | 文档 |
| 5 | reference | 参考资料 |
| 6 | attachment | 通用附件 |
category（文件类型）
| 枚举值 | 字符串 | 说明 |
|---|---|---|
| 1 | document | 文档 |
| 2 | image | 图片 |
设计说明
- 多态关联：使用
related_id+related_type+attachment_purpose实现通用附件管理 - 高度可扩展：新增业务场景（如道具、地点）只需增加枚举值，无需修改表结构
- 用途定位：attachment_purpose 明确区分附件用途（头像、封面、文档等）
- 去重支持：checksum 字段必填，配合 file_checksums 表实现全局去重
- 软删除：使用 deleted_at 字段，部分索引排除已删除记录
- 全文搜索：使用 pg_trgm 扩展支持文件名模糊搜索
- 审计追踪：uploaded_by 记录上传者，created_at/updated_at 记录时间
- 引用完整性：禁止物理外键约束，在应用层（Service 层）保证引用完整性
- 索引策略：核心组合索引
(related_id, related_type, attachment_purpose)优化查询性能
数据模型
枚举定义
# app/models/attachment.py
from enum import IntEnum
class AttachmentCategory(IntEnum):
"""附件分类枚举（使用 SMALLINT 存储）"""
DOCUMENT = 1 # 文档
IMAGE = 2 # 图片
@classmethod
def from_string(cls, value: str) -> "AttachmentCategory":
"""字符串转枚举"""
mapping = {
'document': cls.DOCUMENT,
'image': cls.IMAGE
}
return mapping.get(value.lower(), cls.DOCUMENT)
def to_string(self) -> str:
"""枚举转字符串"""
mapping = {
self.DOCUMENT: 'document',
self.IMAGE: 'image'
}
return mapping.get(self, 'document')
@classmethod
def get_display_name(cls, value: int) -> str:
"""获取显示名称"""
names = {
cls.DOCUMENT: '文档',
cls.IMAGE: '图片'
}
return names.get(value, '未知类型')
class RelatedType(IntEnum):
"""关联实体类型枚举"""
USER = 1 # 用户
PROJECT = 2 # 项目
STORYBOARD = 3 # 分镜
CHARACTER = 4 # 角色
SCENE = 5 # 场景
PROP = 6 # 道具
LOCATION = 7 # 地点
@classmethod
def from_string(cls, value: str) -> "RelatedType":
"""字符串转枚举"""
mapping = {
'user': cls.USER,
'project': cls.PROJECT,
'storyboard': cls.STORYBOARD,
'character': cls.CHARACTER,
'scene': cls.SCENE,
'prop': cls.PROP,
'location': cls.LOCATION,
}
return mapping.get(value.lower(), cls.PROJECT)
def to_string(self) -> str:
"""枚举转字符串"""
mapping = {
self.USER: 'user',
self.PROJECT: 'project',
self.STORYBOARD: 'storyboard',
self.CHARACTER: 'character',
self.SCENE: 'scene',
self.PROP: 'prop',
self.LOCATION: 'location',
}
return mapping[self]
@classmethod
def get_display_name(cls, value: int) -> str:
"""获取显示名称"""
names = {
cls.USER: '用户',
cls.PROJECT: '项目',
cls.STORYBOARD: '分镜',
cls.CHARACTER: '角色',
cls.SCENE: '场景',
cls.PROP: '道具',
cls.LOCATION: '地点',
}
return names.get(value, '未知类型')
class AttachmentPurpose(IntEnum):
"""附件用途枚举"""
AVATAR = 1 # 头像
COVER = 2 # 封面
THUMBNAIL = 3 # 缩略图
DOCUMENT = 4 # 文档
REFERENCE = 5 # 参考资料
ATTACHMENT = 6 # 通用附件
@classmethod
def from_string(cls, value: str) -> "AttachmentPurpose":
"""字符串转枚举"""
mapping = {
'avatar': cls.AVATAR,
'cover': cls.COVER,
'thumbnail': cls.THUMBNAIL,
'document': cls.DOCUMENT,
'reference': cls.REFERENCE,
'attachment': cls.ATTACHMENT,
}
return mapping.get(value.lower(), cls.ATTACHMENT)
def to_string(self) -> str:
"""枚举转字符串"""
mapping = {
self.AVATAR: 'avatar',
self.COVER: 'cover',
self.THUMBNAIL: 'thumbnail',
self.DOCUMENT: 'document',
self.REFERENCE: 'reference',
self.ATTACHMENT: 'attachment',
}
return mapping[self]
@classmethod
def get_display_name(cls, value: int) -> str:
"""获取显示名称"""
names = {
cls.AVATAR: '头像',
cls.COVER: '封面',
cls.THUMBNAIL: '缩略图',
cls.DOCUMENT: '文档',
cls.REFERENCE: '参考资料',
cls.ATTACHMENT: '附件',
}
return names.get(value, '未知用途')
Attachment 模型
# app/models/attachment.py
from typing import Optional, TYPE_CHECKING
from uuid import UUID
from sqlmodel import SQLModel, Field, Column, Index
from sqlalchemy.dialects.postgresql import UUID as PG_UUID, SMALLINT
from datetime import datetime, timezone
from app.utils.id_generator import generate_uuid
if TYPE_CHECKING:
# 多态关联不需要 Relationship，因为关联类型是动态的
pass
class Attachment(SQLModel, table=True):
"""附件表 - 多态关联设计"""
__tablename__ = "attachments"
# 主键（UUID v7 - 应用层生成）
attachment_id: UUID = Field(
sa_column=Column(
PG_UUID(as_uuid=True),
primary_key=True,
default=generate_uuid
)
)
# 基本信息
name: str = Field(max_length=255)
original_name: str = Field(max_length=255)
file_url: str = Field(max_length=500)
file_size: int = Field(ge=0)
mime_type: str = Field(max_length=100)
# 文件类型分类（使用 SMALLINT 存储枚举）
category: int = Field(
sa_column=Column(SMALLINT, nullable=False, default=AttachmentCategory.DOCUMENT),
description="文件类型：1=document, 2=image"
)
# 文件元数据
extension: Optional[str] = Field(default=None, max_length=20)
checksum: str = Field(max_length=64) # SHA256
# ✅ 多态关联字段（核心设计）
related_id: UUID = Field(
sa_column=Column(PG_UUID(as_uuid=True), nullable=False, index=True),
description="关联实体 ID（多态关联）"
)
related_type: int = Field(
sa_column=Column(SMALLINT, nullable=False, index=True),
description="关联实体类型：1=user, 2=project, 3=storyboard, 4=character, 5=scene, 6=prop, 7=location"
)
attachment_purpose: int = Field(
sa_column=Column(SMALLINT, nullable=False, index=True),
description="附件用途：1=avatar, 2=cover, 3=thumbnail, 4=document, 5=reference, 6=attachment"
)
# 访问控制
is_public: bool = Field(default=False)
access_count: int = Field(default=0, ge=0)
download_count: int = Field(default=0, ge=0)
# 存储信息
storage_provider: Optional[str] = Field(default=None, max_length=50) # minio, s3, oss
storage_path: Optional[str] = Field(default=None, max_length=500)
# 审计字段
uploaded_by: UUID = Field(
sa_column=Column(PG_UUID(as_uuid=True), nullable=False, index=True),
description="上传者用户 ID - 应用层验证"
)
# 时间戳
created_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)
updated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)
deleted_at: Optional[datetime] = Field(default=None)
expires_at: Optional[datetime] = Field(default=None)
# 表级索引
__table_args__ = (
Index('idx_attachments_related', 'related_id', 'related_type', 'attachment_purpose'),
Index('idx_attachments_uploader_created', 'uploaded_by', 'created_at'),
)
def to_dict(self) -> dict:
"""转换为字典（camelCase 字段名）"""
return {
'id': str(self.attachment_id),
'name': self.name,
'originalName': self.original_name,
'fileUrl': self.file_url,
'fileSize': self.file_size,
'mimeType': self.mime_type,
'category': AttachmentCategory(self.category).to_string(),
'categoryDisplay': AttachmentCategory.get_display_name(self.category),
'extension': self.extension,
'checksum': self.checksum,
'relatedId': str(self.related_id),
'relatedType': RelatedType(self.related_type).to_string(),
'relatedTypeDisplay': RelatedType.get_display_name(self.related_type),
'attachmentPurpose': AttachmentPurpose(self.attachment_purpose).to_string(),
'attachmentPurposeDisplay': AttachmentPurpose.get_display_name(self.attachment_purpose),
'uploadedBy': str(self.uploaded_by),
'isPublic': self.is_public,
'accessCount': self.access_count,
'downloadCount': self.download_count,
'storageProvider': self.storage_provider,
'storagePath': self.storage_path,
'createdAt': self.created_at.isoformat() if self.created_at else None,
'updatedAt': self.updated_at.isoformat() if self.updated_at else None,
'expiresAt': self.expires_at.isoformat() if self.expires_at else None
}
枚举值映射表
| 枚举类型 | 数值 | 字符串值 | 说明 |
|---|---|---|---|
| category | 1 | document | 文档类型 |
| 2 | image | 图片类型 | |
| related_type | 1 | user | 用户 |
| 2 | project | 项目 | |
| 3 | storyboard | 分镜 | |
| 4 | character | 角色 | |
| 5 | scene | 场景 | |
| 6 | prop | 道具 | |
| 7 | location | 地点 | |
| attachment_purpose | 1 | avatar | 头像 |
| 2 | cover | 封面 | |
| 3 | thumbnail | 缩略图 | |
| 4 | document | 文档 | |
| 5 | reference | 参考资料 | |
| 6 | attachment | 通用附件 |
---ef get_display_name(cls, value: int) -> str: """获取显示名称""" names = { cls.DOCUMENT: '文档', cls.IMAGE: '图片' } return names.get(value, '未知类型')
class Attachment(SQLModel, table=True): """附件表 - 应用层保证引用完整性""" tablename = "attachments"
# 主键（UUID v7 - 应用层生成）
attachment_id: UUID = Field(
sa_column=Column(
PG_UUID(as_uuid=True),
primary_key=True,
default=generate_uuid
)
)
# 基本信息
name: str = Field(max_length=255)
original_name: str = Field(max_length=255)
file_url: str = Field(max_length=500)
file_size: int = Field(ge=0)
mime_type: str = Field(max_length=100)
# 文件类型分类（使用 SMALLINT 存储枚举）
category: int = Field(
sa_column=Column(SMALLINT, nullable=False, default=AttachmentCategory.DOCUMENT),
description="文件类型：1=document, 2=image"
)
# 文件元数据
extension: Optional[str] = Field(default=None, max_length=20)
checksum: str = Field(max_length=64) # SHA256
# 关联字段（无外键约束，应用层验证）
project_id: Optional[UUID] = Field(
default=None,
sa_column=Column(PG_UUID(as_uuid=True), nullable=True, index=True),
description="项目 ID - 应用层验证"
)
uploaded_by: UUID = Field(
sa_column=Column(PG_UUID(as_uuid=True), nullable=False, index=True),
description="上传者用户 ID - 应用层验证"
)
# 访问控制
is_public: bool = Field(default=False)
access_count: int = Field(default=0, ge=0)
download_count: int = Field(default=0, ge=0)
# 存储信息
storage_provider: Optional[str] = Field(default=None, max_length=50) # minio, s3, oss
storage_path: Optional[str] = Field(default=None, max_length=500)
# 时间戳
created_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)
updated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc)
)
deleted_at: Optional[datetime] = Field(default=None)
expires_at: Optional[datetime] = Field(default=None)
# 关系（无物理外键，使用 primaryjoin 明确指定关联条件）
project: Optional["Project"] = Relationship(
sa_relationship_kwargs={
"primaryjoin": "Attachment.project_id == Project.project_id",
"foreign_keys": "[Attachment.project_id]",
}
)
uploader: Optional["User"] = Relationship(
sa_relationship_kwargs={
"primaryjoin": "Attachment.uploaded_by == User.user_id",
"foreign_keys": "[Attachment.uploaded_by]",
}
)
# 表级索引
__table_args__ = (
Index('idx_attachments_project_category', 'project_id', 'category'),
Index('idx_attachments_uploader_created', 'uploaded_by', 'created_at'),
)
def to_dict(self) -> dict:
"""转换为字典（camelCase 字段名）"""
return {
'id': str(self.attachment_id),
'name': self.name,
'originalName': self.original_name,
'fileUrl': self.file_url,
'fileSize': self.file_size,
'mimeType': self.mime_type,
'category': AttachmentCategory(self.category).to_string(),
'categoryDisplay': AttachmentCategory.get_display_name(self.category),
'extension': self.extension,
'checksum': self.checksum,
'projectId': str(self.project_id) if self.project_id else None,
'uploadedBy': str(self.uploaded_by),
'isPublic': self.is_public,
'accessCount': self.access_count,
'downloadCount': self.download_count,
'storageProvider': self.storage_provider,
'storagePath': self.storage_path,
'createdAt': self.created_at.isoformat() if self.created_at else None,
'updatedAt': self.updated_at.isoformat() if self.updated_at else None,
'expiresAt': self.expires_at.isoformat() if self.expires_at else None
}
### 枚举值映射表
| 枚举类型 | 数值 | 字符串值 | 说明 |
|---------|------|---------|------|
| **category** | 1 | document | 文档类型 |
| | 2 | image | 图片类型 |
| **related_type** | 1 | user | 用户 |
| | 2 | project | 项目 |
| | 3 | storyboard | 分镜 |
| | 4 | character | 角色 |
| | 5 | scene | 场景 |
| | 6 | prop | 道具 |
| | 7 | location | 地点 |
| **attachment_purpose** | 1 | avatar | 头像 |
| | 2 | cover | 封面 |
| | 3 | thumbnail | 缩略图 |
| | 4 | document | 文档 |
| | 5 | reference | 参考资料 |
| | 6 | attachment | 通用附件 |
---
## 文件去重机制
### 全局去重
使用 `file_checksums` 表实现跨所有表的文件去重：
```python
# app/services/file_storage_service.py
class FileStorageService:
async def upload_file(
self,
file_content: bytes,
filename: str,
content_type: str,
category: str,
user_id: int
) -> Filemeta_data:
"""上传文件（带去重）"""
# 1. 计算校验和
checksum = hashlib.sha256(file_content).hexdigest()
# 2. 检查是否已存在
existing = await self.get_by_checksum(checksum)
if existing:
# 增加引用计数
await self.increase_reference_count(checksum)
return existing
# 3. 上传到对象存储
extension = os.path.splitext(filename)[1]
object_name = f"{category}/{user_id}/{checksum}{extension}"
file_url = await self.storage.upload(file_content, object_name, content_type)
# 4. 记录到去重表
file_checksum = FileChecksum(
checksum=checksum,
file_url=file_url,
file_size=len(file_content),
mime_type=content_type,
storage_provider='minio',
storage_path=object_name,
reference_count=1
)
await self.checksum_repo.create(file_checksum)
return Filemeta_data(
file_url=file_url,
file_size=len(file_content),
checksum=checksum,
mime_type=content_type,
extension=extension,
storage_provider='minio',
storage_path=object_name
)
优势：
- 节省存储空间
- 加快上传速度（相同文件无需重复上传）
- 跨表去重（attachments, project_resources, videos 等）
- 引用计数管理，便于清理
相关文档
变更记录
v4.1 (2026-02-03)
- ✅ 文档规范统一：
- 添加合规状态标识：
**合规状态**：✅ 符合 jointo-tech-stack 规范 - 删除冗余的第二个 Attachment 模型定义
- 补充完整的枚举值映射表
- 更新版本号和日期
- 添加合规状态标识：
v4.0 (2026-01-28)
- ✅ 多态关联设计（重大架构升级）：
- 引入
related_id+related_type+attachment_purpose三元组 - 支持任意实体的附件管理（用户、项目、分镜、角色、场景、道具、地点）
- 移除
project_id字段，改为通用的related_id - 新增
RelatedType枚举（7种实体类型） - 新增
AttachmentPurpose枚举（6种用途） - 核心组合索引：
(related_id, related_type, attachment_purpose)
- 引入
- ✅ Service 层重构：
upload_attachment()支持多态关联参数get_attachments_by_related()替代get_attachments_by_project()- 新增
get_attachment_by_purpose()获取指定用途附件 - 权限检查支持多种实体类型
- ✅ Repository 层增强：
exists_related_entity()支持多态实体验证check_related_permission()支持多态权限检查get_by_related()支持多态查询get_by_purpose()查询指定用途附件
- ✅ API 接口更新：
- 上传接口新增
relatedId、relatedType、attachmentPurpose参数 - 查询接口支持按
relatedType和attachmentPurpose筛选 - 新增
/purpose/{related_type}/{related_id}/{purpose}端点
- 上传接口新增
- ✅ 数据迁移方案：
- 提供完整的数据迁移脚本
- 旧数据自动转换为多态关联格式
- 移除业务表的反向关联字段（avatar_id, cover_image_id, thumbnail_id）
v3.0 (2026-01-28)
- ✅ 符合 jointo-tech-stack 规范：
- 移除数据库物理外键约束，改为应用层验证
- Model 改用 SQLModel，UUID v7 主键（应用层生成）
- 添加 Relationship 的
primaryjoin配置 - 时间戳使用
datetime.now(timezone.utc) - 枚举类型补充
to_string()方法
- ✅ 补充完整实现：
- Repository 层引用完整性验证方法
- Service 层添加用户/项目存在性检查
- 完整的 Pydantic Schema 定义
- FastAPI 路由实现代码
- 统一 ApiResponse 响应格式
- ✅ 优化索引策略：
- 所有关联字段创建索引
- 添加组合索引优化查询
- 添加表和列注释
v2.1 (2025-01-27)
- 移除 screenplay_id 关联：
- 剧本文件改为由 screenplays 表自己管理（直接存储 file_url, file_size, checksum 等字段）
- 移除 attachments.screenplay_id 字段
- 移除 get_attachments_by_screenplay 方法
- 更新 API 接口文档
- 更新数据模型
- 更新相关文档链接
v2.0 (2025-01-27)
- 重构附件管理架构：
- 专注于文档类附件管理
- 简化分类为 document 和 image
- 移除 resource_id 关联（改为 project_resources 表）
- 新增 script_id 关联
- checksum 改为必填
- 集成 FileStorageService 实现去重
- 更新 API 接口
- 更新数据模型
v1.1 (2025-01-27)
- 优化文件分类系统
- 扩展文件类型支持
- 调整文件大小限制
v1.0 (2025-01-27)
- 初始版本
文档版本：v4.1
最后更新：2026-02-03
合规状态：✅ 符合 jointo-tech-stack 规范