autoclip/docs/TECHNICAL_ROADMAP.md

🎬 AI自动切片工具 - 技术架构改造规划

📋 项目现状分析

当前架构特点

1. 双前端架构: Streamlit原型 + React生产界面
2. 多后端服务: FastAPI主服务 + 多个API文件
3. 6步流水线: 从大纲提取到视频切割的完整流程
4. 多项目支持: 独立的数据目录和配置管理

现存关键问题

1. 架构冗余与混乱

• 存在多个重复的API服务文件 (backend_server.py, src/api.py, simple_api.py)
• Streamlit和React双前端造成维护负担
• 缺乏统一的服务入口和路由管理

2. 技术债严重

• 依赖管理分散 (requirements.txt, backend_requirements.txt)
• 缺乏完整的错误处理和监控机制
• 文件结构不够清晰，模块间耦合度高

3. 性能与可扩展性问题

• 缺乏缓存机制和数据库支持
• 文件存储方式简单，不支持大文件处理
• 并发处理能力有限

4. 用户体验问题

• 缺乏进度反馈和错误恢复机制
• 配置管理不够友好
• 缺乏完整的日志和监控

🚀 阶段性技术演进规划

第一阶段：架构清理与基础重构 (2-3周)

目标

清理冗余代码，建立清晰的技术架构，为后续演进打下基础。

具体任务

1. 后端架构重构

backend/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI应用入口
│   ├── config.py            # 统一配置管理
│   ├── dependencies.py      # 依赖注入
│   └── middleware.py        # 中间件
├── api/
│   ├── __init__.py
│   ├── v1/
│   │   ├── __init__.py
│   │   ├── projects.py      # 项目相关API
│   │   ├── processing.py    # 处理相关API
│   │   ├── files.py         # 文件上传API
│   │   └── settings.py      # 设置相关API
│   └── deps.py              # API依赖
├── core/
│   ├── __init__.py
│   ├── config.py            # 核心配置
│   ├── security.py          # 安全相关
│   └── exceptions.py        # 异常处理
├── models/
│   ├── __init__.py
│   ├── project.py           # 项目模型
│   ├── clip.py              # 切片模型
│   └── collection.py        # 合集模型
├── services/
│   ├── __init__.py
│   ├── project_service.py   # 项目服务
│   ├── processing_service.py # 处理服务
│   ├── file_service.py      # 文件服务
│   └── llm_service.py       # LLM服务
├── pipeline/
│   ├── __init__.py
│   ├── base.py              # 流水线基类
│   ├── steps/               # 处理步骤
│   └── orchestrator.py      # 流水线编排
└── utils/
    ├── __init__.py
    ├── file_utils.py        # 文件工具
    ├── video_utils.py       # 视频工具
    └── text_utils.py        # 文本工具

2. 前端架构优化

frontend/
├── src/
│   ├── components/
│   │   ├── common/          # 通用组件
│   │   ├── forms/           # 表单组件
│   │   ├── layout/          # 布局组件
│   │   └── features/        # 功能组件
│   ├── hooks/
│   │   ├── useApi.ts        # API调用钩子
│   │   ├── useProject.ts    # 项目管理钩子
│   │   └── useProcessing.ts # 处理状态钩子
│   ├── services/
│   │   ├── api.ts           # API客户端
│   │   ├── project.ts       # 项目服务
│   │   └── processing.ts    # 处理服务
│   ├── store/
│   │   ├── index.ts         # 状态管理入口
│   │   ├── project.ts       # 项目状态
│   │   └── settings.ts      # 设置状态
│   ├── types/
│   │   ├── api.ts           # API类型定义
│   │   ├── project.ts       # 项目类型
│   │   └── common.ts        # 通用类型
│   └── utils/
│       ├── constants.ts     # 常量定义
│       ├── helpers.ts       # 工具函数
│       └── validation.ts    # 验证函数

3. 依赖管理统一

# pyproject.toml - 统一Python依赖管理
[tool.poetry]
name = "auto-clip"
version = "1.0.0"
description = "AI自动切片工具"

[tool.poetry.dependencies]
python = "^3.9"
fastapi = "^0.104.1"
uvicorn = {extras = ["standard"], version = "^0.24.0"}
pydantic = "^2.11.7"
dashscope = "^1.23.5"
pydub = "^0.25.1"
pysrt = "^1.1.2"
aiofiles = "^23.2.1"
python-multipart = "^0.0.6"
cryptography = "^42.0.5"
redis = "^5.0.1"
celery = "^5.3.4"

[tool.poetry.dev-dependencies]
pytest = "^8.0.0"
pytest-asyncio = "^0.21.1"
black = "^23.12.1"
isort = "^5.13.2"
mypy = "^1.8.0"

第二阶段：核心功能增强 (3-4周)

目标

增强核心处理能力，提升用户体验和系统稳定性。

具体任务

1. 数据库集成

# 使用SQLAlchemy + PostgreSQL
from sqlalchemy import create_engine, Column, String, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class Project(Base):
    __tablename__ = "projects"
    
    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    status = Column(String, default="created")
    video_category = Column(String, default="default")
    metadata = Column(JSON)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

2. 缓存系统

# Redis缓存集成
import redis
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_result(expire_time=3600):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
            cached_result = redis_client.get(cache_key)
            
            if cached_result:
                return json.loads(cached_result)
            
            result = await func(*args, **kwargs)
            redis_client.setex(cache_key, expire_time, json.dumps(result))
            return result
        return wrapper
    return decorator

3. 异步任务队列

# Celery任务队列
from celery import Celery
from celery.utils.log import get_task_logger

celery_app = Celery('auto_clips', broker='redis://localhost:6379/1')

@celery_app.task(bind=True)
def process_video_pipeline(self, project_id: str, start_step: int = 1):
    """异步处理视频流水线"""
    try:
        processor = AutoClipsProcessor(project_id)
        
        # 更新任务状态
        self.update_state(
            state='PROGRESS',
            meta={'current_step': start_step, 'total_steps': 6}
        )
        
        if start_step == 1:
            result = processor.run_full_pipeline()
        else:
            result = processor.run_from_step(start_step)
            
        return {'status': 'SUCCESS', 'result': result}
    except Exception as e:
        return {'status': 'FAILURE', 'error': str(e)}

4. 文件存储优化

# 支持多种存储后端
from abc import ABC, abstractmethod
import boto3
from pathlib import Path

class StorageBackend(ABC):
    @abstractmethod
    async def upload_file(self, file_path: Path, destination: str) -> str:
        pass
    
    @abstractmethod
    async def download_file(self, source: str, destination: Path) -> None:
        pass

class LocalStorageBackend(StorageBackend):
    async def upload_file(self, file_path: Path, destination: str) -> str:
        # 本地文件存储逻辑
        pass

class S3StorageBackend(StorageBackend):
    def __init__(self, bucket_name: str):
        self.s3_client = boto3.client('s3')
        self.bucket_name = bucket_name
    
    async def upload_file(self, file_path: Path, destination: str) -> str:
        # S3上传逻辑
        pass

第三阶段：性能优化与监控 (2-3周)

目标

提升系统性能，建立完善的监控和日志体系。

具体任务

1. 性能监控

# Prometheus + Grafana监控
from prometheus_client import Counter, Histogram, Gauge
import time

# 定义监控指标
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint'])
REQUEST_DURATION = Histogram('http_request_duration_seconds', 'HTTP request duration')
ACTIVE_PROCESSING = Gauge('active_processing_tasks', 'Number of active processing tasks')

# 监控中间件
@app.middleware("http")
async def monitor_requests(request: Request, call_next):
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path).inc()
    REQUEST_DURATION.observe(duration)
    
    return response

2. 日志系统

# 结构化日志
import structlog
from structlog.stdlib import LoggerFactory

structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

3. 错误处理增强

# 全局错误处理
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    logger.error(
        "Unhandled exception",
        exc_info=exc,
        path=request.url.path,
        method=request.method
    )
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal server error",
            "message": "An unexpected error occurred",
            "request_id": request.headers.get("X-Request-ID", "unknown")
        }
    )

第四阶段：用户体验优化 (2-3周)

目标

优化用户界面和交互体验，提供更直观的操作流程。

具体任务

1. 实时进度反馈

// WebSocket实时通信
import { io, Socket } from 'socket.io-client';

class ProcessingSocket {
  private socket: Socket;
  
  constructor(projectId: string) {
    this.socket = io('ws://localhost:8000', {
      query: { project_id: projectId }
    });
    
    this.socket.on('processing_progress', (data) => {
      this.updateProgress(data);
    });
    
    this.socket.on('processing_complete', (data) => {
      this.handleComplete(data);
    });
  }
  
  private updateProgress(data: ProcessingProgress) {
    // 更新进度UI
  }
}

2. 拖拽上传优化

// 增强的文件上传组件
import { useDropzone } from 'react-dropzone';

const FileUploadZone = () => {
  const { getRootProps, getInputProps, isDragActive } = useDropzone({
    accept: {
      'video/*': ['.mp4', '.avi', '.mov', '.mkv'],
      'text/plain': ['.srt']
    },
    multiple: true,
    onDrop: handleFileDrop
  });
  
  return (
    <div {...getRootProps()} className={isDragActive ? 'drag-active' : ''}>
      <input {...getInputProps()} />
      {isDragActive ? (
        <p>将文件拖拽到这里...</p>
      ) : (
        <p>点击或拖拽文件到此处上传</p>
      )}
    </div>
  );
};

3. 智能配置助手

// 配置向导组件
const ConfigurationWizard = () => {
  const [currentStep, setCurrentStep] = useState(1);
  const [config, setConfig] = useState({});
  
  const steps = [
    {
      title: 'API配置',
      component: <ApiConfigStep config={config} onChange={setConfig} />
    },
    {
      title: '处理参数',
      component: <ProcessingConfigStep config={config} onChange={setConfig} />
    },
    {
      title: '存储设置',
      component: <StorageConfigStep config={config} onChange={setConfig} />
    }
  ];
  
  return (
    <div className="config-wizard">
      <Steps current={currentStep} items={steps} />
      {steps[currentStep - 1].component}
    </div>
  );
};

🛠️ 关键技术栈选择

后端技术栈

核心框架

• FastAPI: 高性能异步Web框架，自动API文档生成
• SQLAlchemy: ORM框架，支持多种数据库
• Pydantic: 数据验证和序列化
• Celery: 分布式任务队列

数据存储

• PostgreSQL: 主数据库，支持JSON字段和复杂查询
• Redis: 缓存和会话存储
• MinIO/S3: 对象存储，支持大文件

监控和日志

• Prometheus: 指标收集
• Grafana: 监控面板
• ELK Stack: 日志分析

前端技术栈

核心框架

• React 18: 用户界面框架
• TypeScript: 类型安全
• Vite: 构建工具

状态管理

• Zustand: 轻量级状态管理
• React Query: 服务端状态管理

UI组件

• Ant Design: 企业级UI组件库
• Tailwind CSS: 原子化CSS框架

实时通信

• Socket.IO: WebSocket通信
• Server-Sent Events: 单向实时数据流

部署和运维

容器化

• Docker: 应用容器化
• Docker Compose: 多服务编排

CI/CD

• GitHub Actions: 自动化部署
• ArgoCD: GitOps部署

监控

• Prometheus: 指标监控
• Grafana: 可视化面板
• Jaeger: 分布式追踪

📅 实施时间线

第1-2周: 架构清理
├── 后端代码重构
├── 前端代码优化
└── 依赖管理统一

第3-5周: 核心功能增强
├── 数据库集成
├── 缓存系统
├── 异步任务队列
└── 文件存储优化

第6-7周: 性能优化与监控
├── 性能监控
├── 日志系统
└── 错误处理增强

第8-9周: 用户体验优化
├── 实时进度反馈
├── 拖拽上传优化
└── 智能配置助手

第10周: 测试与部署
├── 集成测试
├── 性能测试
└── 生产部署

🎯 预期收益

技术收益

1. 架构清晰: 模块化设计，易于维护和扩展
2. 性能提升: 缓存和异步处理提升响应速度
3. 稳定性增强: 完善的错误处理和监控机制
4. 可扩展性: 支持水平扩展和微服务架构

用户体验收益

1. 操作简化: 直观的界面和智能配置
2. 实时反馈: 处理进度实时更新
3. 错误恢复: 智能错误处理和恢复机制
4. 性能感知: 快速响应和流畅交互

📋 风险评估与应对

技术风险

1. 迁移风险: 现有功能在重构过程中可能受到影响

   • 应对: 采用渐进式重构，保持向后兼容

2. 性能风险: 新架构可能引入性能瓶颈

   • 应对: 建立性能基准，持续监控和优化

3. 依赖风险: 新依赖可能带来兼容性问题

   • 应对: 充分测试，制定回滚方案

项目风险

1. 时间风险: 开发周期可能超出预期

   • 应对: 设置里程碑检查点，及时调整计划

2. 资源风险: 开发资源可能不足

   • 应对: 优先实现核心功能，分阶段交付

🔄 持续改进计划

短期改进 (1-3个月)

• 用户反馈收集和分析
• 性能优化和bug修复
• 功能完善和用户体验提升

中期改进 (3-6个月)

• 新功能开发和集成
• 架构进一步优化
• 扩展性和稳定性提升

长期规划 (6-12个月)

• 微服务架构迁移
• AI能力增强
• 商业化功能开发

---

本文档将作为项目技术演进的主要指导文件，需要根据实际开发进度和用户反馈进行定期更新和调整。