**角色 (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` 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` 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