hs-video-api/README.md

视频生成API服务

基于Flask的视频生成API服务，集成火山引擎豆包视频生成功能，提供异步视频生成服务。

功能特性

• 🎬 集成火山引擎豆包视频生成API
• 🔄 异步视频生成任务处理与队列管理
• 📊 实时任务状态查询
• 🎯 支持文本+图片的多模态输入
• 📁 支持图片文件上传和URL输入
• 🔧 RESTful API设计
• ⚡ 完整的错误处理和日志记录
• 💾 任务队列持久化和缓存管理
• 🔧 灵活的环境配置管理

快速开始

1. 安装依赖

pip install -r requirements.txt

2. 配置环境变量

根据部署环境选择对应的配置文件：

• 开发环境：.env.dev
• 测试环境：.env.dev-server
• 生产环境：.env.pro

或者复制示例文件进行自定义配置：

cp .env.example .env

设置环境变量：

# Windows
set APP_ENV=dev

# Linux/Mac
export APP_ENV=dev

3. 启动服务

python app.py

服务将在配置的地址和端口启动（默认 http://localhost:5000）。

API文档

视频生成API接口

创建视频生成任务

POST /api/video/create

支持两种请求方式：

方式1：JSON请求（图片URL）

请求头：

Content-Type: application/json

请求体：

{
    "prompt": "一只可爱的小猫在花园里玩耍",
    "image_url": "https://example.com/cat.jpg",
    "duration": 5,
    "callback_url": "https://your-callback-url.com/webhook"
}

方式2：文件上传（图片文件）

请求头：

Content-Type: multipart/form-data

表单数据：

• image_file: 图片文件
• prompt: 文本描述
• duration: 视频时长（可选）
• callback_url: 回调URL（可选）

响应：

{
    "success": true,
    "task_id": "task-uuid-here",
    "message": "任务创建成功"
}

查询任务状态

GET /api/video/status/<task_id>

响应：

{
    "success": true,
    "data": {
        "task_id": "task-uuid",
        "status": "running",
        "message": "任务正在处理中，请稍后查询"
    }
}

查询任务结果

GET /api/video/result/<task_id>

响应（成功）：

{
    "success": true,
    "data": {
        "task_id": "task-uuid",
        "status": "succeeded",
        "video_url": "https://example.com/generated_video.mp4",
        "created_at": "2024-01-01T00:00:00",
        "updated_at": "2024-01-01T00:05:00"
    }
}

获取任务列表

GET /api/video/tasks

响应：

{
    "success": true,
    "data": {
        "tasks": [
            {
                "task_id": "task-uuid-1",
                "status": "succeeded",
                "created_at": "2024-01-01T00:00:00"
            }
        ],
        "total": 1
    }
}

取消/删除任务

DELETE /api/video/cancel/<task_id>

响应：

{
    "success": true,
    "message": "任务已取消"
}

查询队列状态

GET /api/video/queue/status

响应：

{
    "success": true,
    "data": {
        "running_tasks_count": 2,
        "completed_tasks_count": 10,
        "waiting_queue_count": 0,
        "total_cache_count": 12
    }
}

使用示例

cURL示例

创建视频生成任务（JSON方式）

curl -X POST http://localhost:5000/api/video/create \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "一只可爱的小猫在花园里玩耍",
    "image_url": "https://example.com/cat.jpg",
    "duration": 5,
    "callback_url": "https://your-callback-url.com/webhook"
  }'

创建视频生成任务（文件上传方式）

curl -X POST http://localhost:5000/api/video/create \
  -F "image_file=@/path/to/your/image.jpg" \
  -F "prompt=一只可爱的小猫在花园里玩耍" \
  -F "duration=5"

查询任务状态

curl -X GET http://localhost:5000/api/video/status/task-uuid

查询任务结果

curl -X GET http://localhost:5000/api/video/result/task-uuid

获取任务列表

curl -X GET http://localhost:5000/api/video/tasks

查询队列状态

curl -X GET http://localhost:5000/api/video/queue/status

Python示例

import requests
import json
import time

# 配置
API_BASE_URL = "http://localhost:5000"

# 创建视频生成任务（JSON方式）
def create_video_task_json(prompt, image_url=None, duration=5, callback_url=None):
    url = f"{API_BASE_URL}/api/video/create"
    
    data = {
        "prompt": prompt,
        "duration": duration
    }
    
    if image_url:
        data["image_url"] = image_url
    
    if callback_url:
        data["callback_url"] = callback_url
    
    response = requests.post(url, json=data)
    return response.json()

# 创建视频生成任务（文件上传方式）
def create_video_task_file(prompt, image_file_path, duration=5):
    url = f"{API_BASE_URL}/api/video/create"
    
    with open(image_file_path, 'rb') as f:
        files = {'image_file': f}
        data = {
            'prompt': prompt,
            'duration': str(duration)
        }
        response = requests.post(url, files=files, data=data)
    
    return response.json()

# 查询任务状态
def get_task_status(task_id):
    url = f"{API_BASE_URL}/api/video/status/{task_id}"
    response = requests.get(url)
    return response.json()

# 查询任务结果
def get_task_result(task_id):
    url = f"{API_BASE_URL}/api/video/result/{task_id}"
    response = requests.get(url)
    return response.json()

# 获取任务列表
def get_task_list():
    url = f"{API_BASE_URL}/api/video/tasks"
    response = requests.get(url)
    return response.json()

# 查询队列状态
def get_queue_status():
    url = f"{API_BASE_URL}/api/video/queue/status"
    response = requests.get(url)
    return response.json()

# 使用示例
if __name__ == "__main__":
    # 创建任务（JSON方式）
    result = create_video_task_json(
        prompt="一只可爱的小猫在花园里玩耍",
        image_url="https://example.com/cat.jpg",
        duration=5
    )
    print("创建任务结果:", result)
    
    if result.get("success") and "task_id" in result:
        task_id = result["task_id"]
        
        # 轮询查询任务状态
        while True:
            status = get_task_status(task_id)
            print(f"任务状态: {status}")
            
            if status.get("data", {}).get("status") in ["succeeded", "failed"]:
                # 获取最终结果
                result = get_task_result(task_id)
                print(f"任务结果: {result}")
                break
            
            time.sleep(5)  # 等待5秒后再次查询
    
    # 查询队列状态
    queue_status = get_queue_status()
    print("队列状态:", queue_status)

测试

运行测试脚本：

# 设置API密钥
export ARK_API_KEY=your_api_key

# 运行测试
python test_doubao_api.py

任务状态说明

• pending: 任务已创建，等待处理
• running: 任务正在处理中
• succeeded: 任务处理成功，视频生成完成
• failed: 任务处理失败
• cancelled: 任务已被取消

配置说明

环境变量配置

项目支持多环境配置，通过 APP_ENV 环境变量指定：

• dev: 开发环境（使用 .env.dev）
• dev-server: 测试环境（使用 .env.dev-server）
• pro: 生产环境（使用 .env.pro）

队列管理配置

详细的队列配置说明请参考 QUEUE_CONFIG_README.md 文件。

主要配置参数：

• QUEUE_MAX_RUNNING_TASKS: 最大并发运行任务数
• QUEUE_UPDATE_INTERVAL: 任务状态更新间隔（秒）
• QUEUE_PERSISTENCE_FILE: 队列持久化文件路径
• QUEUE_MAX_COMPLETED_CACHE_SIZE: 最大已完成任务缓存数量
• QUEUE_COMPLETED_CACHE_TTL_HOURS: 已完成任务缓存保留时间（小时）

使用注意事项

• 该模型主要用于图生视频，需要提供图片URL作为输入
• 如需使用文生视频功能，可以将模型改为 doubao-seedance-1-0-lite-t2v-250428
• 视频生成为异步过程，通常需要等待较长时间

项目结构

hs-video-api/
├── app.py                    # Flask应用主文件
├── config.py                 # 配置管理
├── routes.py                 # API路由定义
├── video_service.py          # 视频生成服务
├── task_queue_manager.py     # 任务队列管理
├── requirements.txt          # 依赖包列表
├── .env.dev                  # 开发环境配置
├── .env.dev-server          # 测试环境配置
├── .env.pro                 # 生产环境配置
├── .env.example             # 环境变量示例
├── Dockerfile               # Docker配置
├── README.md                # 项目说明
├── QUEUE_CONFIG_README.md   # 队列配置说明
└── tests/                   # 测试文件目录
    ├── test_cache_persistence.py
    ├── test_robustness.py
    └── test_stress.py

Docker部署

构建镜像

docker build -t hs-video-api .

运行容器

docker run -d \
  --name hs-video-api \
  -p 5000:5000 \
  -e APP_ENV=pro \
  hs-video-api

使用docker-compose

创建 docker-compose.yml：

version: '3.8'
services:
  hs-video-api:
    build: .
    ports:
      - "5000:5000"
    environment:
      - APP_ENV=pro
    volumes:
      - ./queue_data:/app/queue_data  # 队列持久化数据
    restart: unless-stopped

启动服务：

docker-compose up -d

错误处理

常见错误码

• 400: 请求参数错误
• 404: 任务不存在
• 405: 请求方法不允许
• 413: 上传文件过大
• 500: 服务器内部错误

错误响应格式

{
  "success": false,
  "error": "参数验证失败",
  "message": "prompt字段为必填项"
}

成功响应格式

{
  "success": true,
  "data": {
    // 具体数据内容
  },
  "message": "操作成功"
}

开发和测试

运行测试

项目包含多种测试用例：

# 缓存持久化测试
python tests/test_cache_persistence.py

# 鲁棒性测试
python tests/test_robustness.py

# 压力测试
python tests/test_stress.py

日志查看

应用日志会输出到控制台和日志文件（如果配置了的话）。可以通过调整 LOG_LEVEL 环境变量来控制日志级别。

性能优化建议

1. 队列配置优化：根据服务器性能调整 QUEUE_MAX_RUNNING_TASKS
2. 缓存管理：合理设置 QUEUE_MAX_COMPLETED_CACHE_SIZE 和 QUEUE_COMPLETED_CACHE_TTL_HOURS
3. 文件上传：对于大文件上传，建议使用CDN或对象存储
4. 监控告警：建议监控队列状态和任务处理时间

注意事项

1. API密钥安全：确保火山引擎API密钥的安全性，不要在代码中硬编码
2. 文件存储：上传的图片文件会临时存储，建议定期清理
3. 队列持久化：队列数据会持久化到文件，确保有足够的磁盘空间
4. 并发限制：根据API配额和服务器性能合理设置并发数
5. 错误重试：建议在客户端实现适当的重试机制

许可证

MIT License

---

注意: 本项目仅供学习和研究使用，请遵守火山引擎API的使用条款和限制。