443 lines
18 KiB
Markdown
443 lines
18 KiB
Markdown
**角色 (Persona):**
|
||
你是一位资深的Python后端开发专家，精通使用FastAPI进行快速开发，并且在领域驱动设计（DDD）和构建可扩展、可维护的AI应用方面拥有丰富的实践经验。
|
||
|
||
**任务 (Objective):**
|
||
根据以下详细规格说明，为我创建一个完整、可运行的Python后端项目。该项目是一个模块化的AI视频 storyboard（分镜）生成器，包含独立的**提示词管理模块**和**项目管理模块**。
|
||
|
||
-----
|
||
|
||
### **1. 核心架构与技术要求**
|
||
|
||
* **编程语言:** Python 3.10+
|
||
* **Web框架:** FastAPI
|
||
* **架构模式:** 严格遵循领域驱动设计（DDD）分层架构。
|
||
* **数据库与ORM:** PostgreSQL，使用SQLAlchemy 2.0。
|
||
* **数据库迁移:** 使用 Alembic。
|
||
* **依赖管理:** 使用 `requirements.txt` 文件。
|
||
* **配置管理:** 使用 `.env` 文件和 `pydantic-settings`。
|
||
* **代码质量:** PEP 8规范, 类型提示, 健壮的错误处理, 结构化日志。
|
||
|
||
### **2. 技术栈 (Technology Stack)**
|
||
|
||
请在 `requirements.txt` 文件中包含以下核心依赖：
|
||
|
||
```txt
|
||
fastapi
|
||
uvicorn[standard]
|
||
sqlalchemy
|
||
psycopg2-binary
|
||
alembic
|
||
pydantic
|
||
pydantic-settings
|
||
python-dotenv
|
||
google-generativeai
|
||
qiniu
|
||
openai # OpenRouter使用与OpenAI兼容的SDK
|
||
loguru
|
||
httpx # 用于异步下载图片
|
||
```
|
||
|
||
### **3. 环境变量配置**
|
||
|
||
在项目根目录创建 `.env.example` 文件：
|
||
|
||
```ini
|
||
# Database
|
||
DATABASE_URL="postgresql://user:password@host:port/dbname"
|
||
|
||
# Qiniu Cloud
|
||
QINIU_ACCESS_KEY="your_qiniu_access_key"
|
||
QINIU_SECRET_KEY="your_qiniu_secret_key"
|
||
QINIU_BUCKET_NAME="your_bucket_name"
|
||
QINIU_DOMAIN="your_qiniu_cdn_domain"
|
||
|
||
# AI Models
|
||
GOOGLE_API_KEY="your_google_ai_api_key"
|
||
OPENROUTER_API_KEY="your_openrouter_api_key"
|
||
OPENROUTER_BASE_URL="https://openrouter.ai/api/v1"
|
||
```
|
||
|
||
### **4. 项目结构**
|
||
|
||
请严格按照以下模块化的结构生成项目文件和目录：
|
||
|
||
```
|
||
.
|
||
├── alembic/
|
||
├── src/
|
||
│ ├── api/
|
||
│ │ ├── dependencies.py
|
||
│ │ └── routes/
|
||
│ │ ├── prompts.py # 提示词模块路由
|
||
│ │ ├── qiniu.py
|
||
│ │ └── projects.py # 项目、素材、分镜模块路由
|
||
│ ├── application/
|
||
│ │ ├── schemas/
|
||
│ │ │ ├── prompt.py
|
||
│ │ │ ├── project.py
|
||
│ │ │ ├── asset.py
|
||
│ │ │ └── storyboard.py
|
||
│ │ └── use_cases/
|
||
│ │ ├── prompt_use_cases.py
|
||
│ │ └── project_use_cases.py # 包含项目、素材、分镜的用例
|
||
│ ├── domain/
|
||
│ │ ├── entities/
|
||
│ │ │ ├── prompt.py
|
||
│ │ │ ├── project.py
|
||
│ │ │ ├── asset.py
|
||
│ │ │ └── storyboard.py
|
||
│ │ ├── repositories/
|
||
│ │ │ ├── prompt_repository.py
|
||
│ │ │ └── project_repository.py # 统一管理项目聚合根下的所有实体
|
||
│ │ └── services/
|
||
│ │ ├── ai_service.py
|
||
│ │ └── storage_service.py
|
||
│ ├── infrastructure/
|
||
│ │ ├── database/
|
||
│ │ │ ├── models.py
|
||
│ │ │ ├── repository_impl/
|
||
│ │ │ │ ├── prompt_repository_impl.py
|
||
│ │ │ │ └── project_repository_impl.py
|
||
│ │ │ └── session.py
|
||
│ │ ├── external/
|
||
│ │ │ ├── gemini_client.py
|
||
│ │ │ ├── openrouter_client.py
|
||
│ │ │ └── qiniu_client.py
|
||
│ │ └── config.py
|
||
│ └── main.py
|
||
├── .env
|
||
├── .env.example
|
||
├── .gitignore
|
||
├── alembic.ini
|
||
└── requirements.txt
|
||
```
|
||
|
||
### **5. 数据库表结构**
|
||
|
||
* **基础模型:** 所有模型继承一个基础类，自动包含 `id (int, primary_key)`, `created_at (datetime)`, `updated_at (datetime)` 字段。
|
||
* **`prompts` 表:**
|
||
* `step (str, index)`: 描述提示词在哪个阶段使用，例如 "create\_shots"。
|
||
* `name (str, index)`: 提示词的名称。
|
||
* `prompt (text)`: 提示词的具体内容。
|
||
* **约束:** `step` 和 `name` 构成联合唯一约束。
|
||
* **`projects` 表 (项目表):**
|
||
* `name (str)`: 项目名。
|
||
* `script (text)`: 剧本内容。
|
||
* `full_script (text)`: 完整剧本内容。
|
||
* `prompt_used (json)`: 生成分镜时使用的提示词快照。
|
||
* **`assets` 表 (素材表):**
|
||
* `project_id (int, ForeignKey("projects.id"))`: 关联的项目ID。
|
||
* `name (str)`: 素材名。
|
||
* `description (text)`: 素材描述。
|
||
* `tags (ARRAY(String))`: 素材标签数组。
|
||
* `original_url (str)`: 素材原始URL。
|
||
* **`storyboards` 表 (分镜表):**
|
||
* `project_id (int, ForeignKey("projects.id"))`: 关联的项目ID。
|
||
* `frame_prompt (text)`: 首帧提示词。
|
||
* `frame_asset_ids (ARRAY(Integer))`: 用于生成首帧图的素材ID数组。
|
||
* `frame_image_url (str, nullable=True)`: 生成的首帧图URL。
|
||
* `shot_prompt (text)`: 分镜动态内容提示词。
|
||
* `shot_video_url (str, nullable=True)`: 生成的分镜视频URL。
|
||
|
||
### **6. API 端点功能规格**
|
||
|
||
|
||
获取七牛云上传Token
|
||
路径: /api/v1/qiniu/upload-token
|
||
|
||
文件: src/api/routes/qiniu.py
|
||
|
||
方法: POST
|
||
|
||
成功响应 (200 OK): { "token": "...", "domain": "...", "bucket": "..." }
|
||
|
||
核心实现逻辑:
|
||
|
||
从环境变量中读取七牛云的 access_key, secret_key, bucket_name, domain。
|
||
|
||
使用七牛云SDK生成上传token。
|
||
|
||
返回token、domain和bucket name。
|
||
|
||
参考提供的 get_qiniu_upload_token 函数实现，并做好异常处理。
|
||
|
||
-----
|
||
|
||
#### **模块一：提示词管理 (Prompt Management)**
|
||
|
||
* **文件:** `src/api/routes/prompts.py`
|
||
|
||
<!-- end list -->
|
||
|
||
1. **查询所有提示词 (GET /api/v1/prompts)**
|
||
|
||
* **成功响应 (200 OK):** `[ { "step": "...", "name": "...", "prompt": "..." } ]`
|
||
* **核心实现逻辑:** 从数据库查询并返回所有提示词记录。
|
||
|
||
2. **新增或修改提示词 (POST /api/v1/prompts)**
|
||
|
||
* **请求体:** `{ "step": "create_shots", "name": "default_v1", "prompt": "..." }`
|
||
* **成功响应 (200 OK):** `{ "id": 1, "step": "...", "name": "...", "prompt": "..." }`
|
||
* **核心实现逻辑:** 根据 `step` 和 `name` 查找记录。如果存在，则更新 `prompt`；如果不存在，则创建新记录。
|
||
|
||
-----
|
||
|
||
#### **模块二：项目与视频生成 (Project & Video Generation)**
|
||
|
||
* **文件:** `src/api/routes/projects.py`
|
||
|
||
<!-- end list -->
|
||
|
||
1. **分页获取项目列表 (GET /api/v1/projects)**
|
||
|
||
* **查询参数:** `?page=1&size=20`
|
||
* **成功响应 (200 OK):** `{ "items": [ { "id": 1, "name": "...", "created_at": "...", "updated_at": "..." } ], "total": 100, "page": 1, "size": 20 }`
|
||
* **核心实现逻辑:** 实现数据库分页查询。
|
||
|
||
2. **新建项目 (POST /api/v1/projects)**
|
||
|
||
* **请求体:** `{ "name": "我的第一个项目" }`
|
||
* **成功响应 (201 OK):** `{ "id": 1, "name": "我的第一个项目" }`
|
||
* **核心实现逻辑:** 在 `projects` 表中创建一条新记录。
|
||
|
||
3. **生成剧本 (POST /api/v1/projects/{project\_id}/generate-script)**
|
||
|
||
* **请求体:** `{ "script_or_idea": "一个关于小猫冒险的故事..." }`
|
||
* **成功响应 (200 OK):** `{ "full_script": "...", "assets": [ { "id": 1, "name": "小猫", "description": "一只可爱的橘猫", "tags": ["动物", "主角"] } ] }`
|
||
* **核心实现逻辑:**
|
||
1. 接收 `project_id` 和剧本或创意内容。
|
||
2. 调用大模型生成完整剧本和所需素材信息（素材名、描述、标签）。
|
||
3. 将完整剧本更新到 `projects` 表的 `full_script` 字段。
|
||
4. 将素材信息批量存入 `assets` 表，关联 `project_id`。
|
||
5. 返回生成的完整剧本和素材列表。
|
||
|
||
4. **更新素材图 (PUT /api/v1/assets/{asset\_id}/image)**
|
||
|
||
* **请求体:** `{ "image_url": "https://example.com/image.jpg" }`
|
||
* **成功响应 (200 OK):** `{ "message": "素材图片更新成功" }`
|
||
* **核心实现逻辑:**
|
||
1. 接收 `asset_id` 和素材图片URL。
|
||
2. 更新 `assets` 表中对应记录的 `original_url` 字段。
|
||
3. 返回成功响应。
|
||
|
||
5. **创建素材图 (POST /api/v1/projects/{project\_id}/generate-asset-images)**
|
||
|
||
* **请求体:** 无
|
||
* **成功响应 (200 OK):** `[ { "id": 1, "name": "小猫", "description": "...", "tags": [], "original_url": "https://..." } ]`
|
||
* **核心实现逻辑:**
|
||
1. 接收 `project_id`。
|
||
2. 查询该项目下所有没有 `original_url` 的素材。
|
||
3. 对于每个无图片的素材，使用 `gemini-2.5-flash-image-preview` 生成图片。
|
||
4. 将生成的图片上传到七牛云，更新 `original_url` 字段。
|
||
5. 返回所有素材信息列表。
|
||
|
||
6. **上传并分析素材 (POST /api/v1/projects/{project\_id}/assets)** [保留但不推荐使用]
|
||
|
||
* **请求体:** `{ "asset_urls": ["url1.jpg", "url2.png"] }`
|
||
* **成功响应 (200 OK):** `[ { "id": 1, "name": "...", "description": "...", "tags": ["...", "..."] } ]`
|
||
* **核心实现逻辑:**
|
||
1. 接收 `project_id` 和素材URL列表。
|
||
2. 对于每个URL，异步下载图片。
|
||
3. 使用大模型（如Gemini）分析图片，要求返回结构化JSON：`{ "name": "...", "description": "...", "tags": ["..."] }`。
|
||
4. 将分析结果连同 `project_id` 和 `original_url` 存入 `assets` 表。
|
||
5. 返回新创建的素材记录列表。
|
||
|
||
7. **生成分镜 (POST /api/v1/projects/{project\_id}/generate-storyboards)**
|
||
|
||
* **请求体:** `{ "script": "...", "prompt_step": "create_shots", "prompt_name": "default_v1" }`
|
||
* **成功响应 (200 OK):** `[ { "id": 1, "frame_prompt": "...", "frame_asset_ids": [1, 3], "shot_prompt": "..." } ]`
|
||
* **核心实现逻辑:**
|
||
1. **检查所有素材是否都有图片**：确保该项目下所有素材的 `original_url` 字段都不为空。
|
||
2. 根据 `prompt_step` 和 `prompt_name` 从数据库获取提示词。
|
||
3. 获取该项目 (`project_id`) 下所有素材的JSON描述。
|
||
4. 将 **系统提示词**、**所有素材的JSON描述** 和 **用户剧本** 组合成一个完整的Prompt。
|
||
5. 调用OpenRouter大模型，要求返回结构化JSON数组，每项包含 `frame_prompt`, `frame_asset_ids`, `shot_prompt`。
|
||
6. 将返回的数组解析并批量存入 `storyboards` 表，关联 `project_id`。
|
||
7. 返回新创建的分镜记录列表。
|
||
|
||
8. **生成分镜首帧图 (POST /api/v1/storyboards/{storyboard\_id}/generate-frame)**
|
||
|
||
* **请求体:** `{ "frame_prompt": "...", "frame_asset_ids": [1, 2] }`
|
||
* **成功响应 (200 OK):** `{ "frame_image_url": "https://..." }`
|
||
* **核心实现逻辑:**
|
||
1. 接收 `storyboard_id` 和可编辑的 `frame_prompt`, `frame_asset_ids`。
|
||
2. 根据 `frame_asset_ids` 从数据库获取对应素材的原始URL并下载图片。
|
||
3. 调用 `gemini-2.5-flash-image-preview` 模型生成图片。
|
||
4. 上传生成的图片到七牛云。
|
||
5. 更新 `storyboards` 表中对应记录的 `frame_prompt`, `frame_asset_ids`, 和 `frame_image_url` 字段。
|
||
6. 返回生成的图片URL。
|
||
|
||
9. **首帧图指导修改 (POST /api/v1/storyboards/{storyboard\_id}/guide-frame)**
|
||
|
||
* **请求体:** `{ "guide_image_url": "https://example.com/guide.jpg" }`
|
||
* **成功响应 (200 OK):** `{ "frame_image_url": "https://..." }`
|
||
* **核心实现逻辑:**
|
||
1. 接收 `storyboard_id` 和指导图URL。
|
||
2. 从数据库获取该分镜的原首帧图URL并下载。
|
||
3. 下载指导图。
|
||
4. 使用 `gemini-2.5-flash-image-preview` 模型，结合原首帧图和指导图生成新的首帧图。
|
||
5. 上传生成的图片到七牛云。
|
||
6. 更新 `storyboards` 表中对应记录的 `frame_image_url` 字段（覆盖原首帧图）。
|
||
7. 返回新生成的图片URL。
|
||
|
||
10. **生成分镜视频 (POST /api/v1/storyboards/{storyboard\_id}/generate-video)**
|
||
|
||
* **请求体:** `{ "shot_prompt": "..." }`
|
||
* **成功响应 (200 OK):** `{ "shot_video_url": "https://..." }`
|
||
* **核心实现逻辑:**
|
||
1. 接收 `storyboard_id` 和可编辑的 `shot_prompt`。
|
||
2. 从数据库获取该分镜的 `frame_image_url` 并下载图片。
|
||
3. 调用 `veo-3.0-generate-preview` 模型生成视频（需要轮询等待）。
|
||
4. 上传生成的视频到七牛云。
|
||
5. 更新 `storyboards` 表中对应记录的 `shot_prompt` 和 `shot_video_url` 字段。
|
||
6. 返回生成的视频URL。
|
||
|
||
11. **获取项目所有信息 (GET /api/v1/projects/{project\_id})**
|
||
|
||
* **成功响应 (200 OK):**
|
||
```json
|
||
{
|
||
"project": { "id": 1, "name": "...", "created_at": "...", "updated_at": "..." },
|
||
"assets": [ { "id": 1, "name": "...", "description": "...", "tags": [] } ],
|
||
"storyboards": [ { "id": 1, "frame_prompt": "...", "frame_asset_ids": [], "frame_image_url": "...", "shot_prompt": "...", "shot_video_url": "..." } ]
|
||
}
|
||
```
|
||
* **核心实现逻辑:** 根据 `project_id`，联表查询或分别查询项目、素材、分镜的所有相关信息并组合返回。
|
||
|
||
-----
|
||
|
||
**执行指令 (Execution Instruction):**
|
||
请从 `requirements.txt` 文件开始，然后是 `.env.example` 和配置文件，接着按照 `infrastructure`, `domain`, `application`, `api` 的顺序，逐一生成每个文件的完整代码。确保代码是完整、可用且符合上述所有要求的。
|
||
|
||
|
||
|
||
相关技术参考:
|
||
1.对接gemini-2.5-flash-image-preview图片生成模型,API参考地址https://ai.google.dev/gemini-api/docs/image-generation?hl=zh-cn
|
||
官方示例:
|
||
from google import genai
|
||
from google.genai import types
|
||
from PIL import Image
|
||
from io import BytesIO
|
||
|
||
client = genai.Client()
|
||
|
||
# Base image prompts:
|
||
# 1. Dress: "A professionally shot photo of a blue floral summer dress on a plain white background, ghost mannequin style."
|
||
# 2. Model: "Full-body shot of a woman with her hair in a bun, smiling, standing against a neutral grey studio background."
|
||
dress_image = Image.open('/path/to/your/dress.png')
|
||
model_image = Image.open('/path/to/your/model.png')
|
||
|
||
text_input = """Create a professional e-commerce fashion photo. Take the blue floral dress from the first image and let the woman from the second image wear it. Generate a realistic, full-body shot of the woman wearing the dress, with the lighting and shadows adjusted to match the outdoor environment."""
|
||
|
||
# Generate an image from a text prompt
|
||
response = client.models.generate_content(
|
||
model="gemini-2.5-flash-image-preview",
|
||
contents=[dress_image, model_image, text_input],
|
||
)
|
||
|
||
image_parts = [
|
||
part.inline_data.data
|
||
for part in response.candidates[0].content.parts
|
||
if part.inline_data
|
||
]
|
||
|
||
if image_parts:
|
||
image = Image.open(BytesIO(image_parts[0]))
|
||
image.save('fashion_ecommerce_shot.png')
|
||
image.show()
|
||
|
||
2.对接veo3生成视频模型,API参考地址:https://ai.google.dev/gemini-api/docs/video?hl=zh-cn
|
||
参考实现:
|
||
import time
|
||
from google import genai
|
||
|
||
client = genai.Client()
|
||
|
||
prompt = "Panning wide shot of a calico kitten sleeping in the sunshine"
|
||
|
||
image = #base64
|
||
|
||
operation = client.models.generate_videos(
|
||
model="veo-3.0-generate-preview",
|
||
prompt=prompt,
|
||
image=image,
|
||
)
|
||
|
||
# Poll the operation status until the video is ready.
|
||
while not operation.done:
|
||
print("Waiting for video generation to complete...")
|
||
time.sleep(10)
|
||
operation = client.operations.get(operation)
|
||
|
||
# Download the video.
|
||
video = operation.response.generated_videos[0]
|
||
client.files.download(file=video.video)
|
||
video.video.save("veo3_with_image_input.mp4")
|
||
print("Generated video saved to veo3_with_image_input.mp4")
|
||
|
||
3.对接七牛云,向前端提供获取token的接口,参考:
|
||
@router.post("/upload-token")
|
||
def get_upload_token():
|
||
"""
|
||
获取七牛云上传token
|
||
Returns:
|
||
上传token和域名信息
|
||
"""
|
||
try:
|
||
# 获取上传token
|
||
token = get_qiniu_upload_token()
|
||
|
||
if not token:
|
||
raise HTTPException(status_code=500, detail="获取上传token失败")
|
||
|
||
# 返回token和域名信息
|
||
return {
|
||
"token": token,
|
||
"domain": f"https://{qiniu_domain}",
|
||
"bucket": qiniu_bucket_name
|
||
}
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f"获取上传token异常: {e}")
|
||
raise HTTPException(status_code=500, detail=f"获取上传token失败: {str(e)}")
|
||
|
||
|
||
def get_qiniu_upload_token() -> str:
|
||
"""
|
||
获取七牛云上传token
|
||
Returns:
|
||
上传token
|
||
"""
|
||
try:
|
||
q = qiniu.Auth(qiniu_access_key, qiniu_secret_key)
|
||
token = q.upload_token(qiniu_bucket_name)
|
||
return token
|
||
except Exception as e:
|
||
logger.error(f"获取七牛云上传token失败: {e}")
|
||
return None
|
||
|
||
def upload_file_to_qiniu(file_bytes: bytes, suffix: str) -> str:
|
||
"""
|
||
上传文件到七牛云
|
||
Args:
|
||
file_bytes: 文件二进制数据
|
||
filename: 文件名
|
||
Returns:
|
||
文件 URL
|
||
:param suffix: 文件扩展名
|
||
"""
|
||
try:
|
||
q = qiniu.Auth(qiniu_access_key, qiniu_secret_key)
|
||
token = q.upload_token(qiniu_bucket_name)
|
||
filename = f"{int(time.time())}.{suffix}"
|
||
ret, info = qiniu.put_data( up_token=token, key=filename, data=file_bytes)
|
||
if ret is None:
|
||
logger.error(f"上传到七牛云失败: {info}")
|
||
return None
|
||
return f"https://{qiniu_domain}/{filename}"
|
||
except Exception as e:
|
||
logger.error(f"上传到七牛云失败: {e}")
|
||
return None
|