lian/banana-video
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
banana-video/GUIDE_FOR_AI.md
lk-eternal 70d247fa4c feat: init
2025-08-31 18:38:41 +08:00

18 KiB
Raw Permalink Blame History

角色 (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 文件中包含以下核心依赖：

fastapi
uvicorn[standard]
sqlalchemy
psycopg2-binary
alembic
pydantic
pydantic-settings
python-dotenv
google-generativeai
qiniu
openai # OpenRouter使用与OpenAI兼容的SDK
loguru
httpx # 用于异步下载图片

3. 环境变量配置

在项目根目录创建 .env.example 文件：

# 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): 
      {
  "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