lywsvip/autoclip
1
0
Fork 0
mirror of https://github.com/zhouxiaoka/autoclip.git synced 2026-09-02 22:15:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

整理文件

Browse Source
This commit is contained in:
Kris Ka
2025-08-06 12:01:36 +08:00
parent 98cef819c5
commit bf8043101f
33 changed files with 1272 additions and 1458 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
.DS_Store vendored
View File

Binary file not shown.

4
.trae/TODO.md
View File

@@ -1,4 +0,0 @@
# TODO:
- [x] add_creation_time_display: 在ProjectCard组件中添加项目创建时间显示，使用dayjs显示相对时间并设置Asia/Shanghai时区 (priority: High)
- [ ] fix_thumbnail_generation: 修复ProjectCard组件中的缩略图生成问题，改进错误处理和多路径尝试机制 (priority: High)

BIN
autoclip.db
View File

Binary file not shown.

BIN
autoclips.db
View File

Binary file not shown.

BIN
backend/.DS_Store vendored
View File

Binary file not shown.

371
backend/CONFIG_MANAGER_OPTIMIZATION_SUMMARY.md
View File

@@ -1,371 +0,0 @@
# 配置管理器优化总结
## 优化概述
基于你的代码review建议，我们对`ProjectConfigManager`进行了全面优化，解决了所有提到的问题并增加了新功能。
## 优化前后对比
### ❌ 优化前的问题
1. **缺少异常处理**
- `load_config()`没有充分的异常处理
- YAML解析错误可能导致程序中断
2. **LLM配置直接从环境变量取**
- 缺乏灵活性
- 不支持项目级配置
3. **配置更新后未落盘**
- 配置更新后没有自动保存
- 缺乏持久化机制
4. **prompt文件结构固定**
- 不支持多语言
- 不支持自定义路径
### ✅ 优化后的改进
## 1. 完善的异常处理
### 优化前
```python
def _load_config(self) -> Dict[str, Any]:
if self.config_path.exists():
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
return yaml.safe_load(f) or {}
except Exception as e:
print(f"加载项目配置失败: {e}")
return {}
return {}
```
### 优化后
```python
def _load_config(self) -> Dict[str, Any]:
if self.config_path.exists():
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
if config is None:
logger.warning(f"配置文件为空: {self.config_path}")
return {}
return config
except yaml.YAMLError as e:
logger.error(f"YAML解析错误: {self.config_path}, 错误: {e}")
return {}
except FileNotFoundError as e:
logger.error(f"配置文件不存在: {self.config_path}, 错误: {e}")
return {}
except Exception as e:
logger.error(f"加载项目配置失败: {self.config_path}, 错误: {e}")
return {}
return {}
```
**改进点**:
- ✅ 区分不同类型的异常（YAML解析、文件不存在、其他）
- ✅ 使用logger替代print，便于日志管理
- ✅ 更详细的错误信息，包含文件路径
- ✅ 优雅降级，确保程序不会中断
## 2. 灵活的LLM配置管理
### 优化前
```python
def get_llm_config(self) -> LLMConfig:
api_key = os.getenv("DASHSCOPE_API_KEY", "")
if not api_key:
raise ValueError("DASHSCOPE_API_KEY 环境变量未设置")
return LLMConfig(
api_key=api_key,
model_name=self.config.get("llm", {}).get("model_name", "qwen-plus"),
max_retries=self.config.get("llm", {}).get("max_retries", 3),
timeout_seconds=self.config.get("llm", {}).get("timeout_seconds", 30)
)
```
### 优化后
```python
def get_llm_config(self) -> LLMConfig:
# 优先从项目配置获取
llm_config = self.config.get("llm", {})
# API密钥优先级：项目配置 > 环境变量 > 默认值
api_key = llm_config.get("api_key") or os.getenv("DASHSCOPE_API_KEY", "")
if not api_key:
raise ValueError("DASHSCOPE_API_KEY 未在项目配置或环境变量中设置")
return LLMConfig(
api_key=api_key,
model_name=llm_config.get("model_name", "qwen-plus"),
max_retries=llm_config.get("max_retries", 3),
timeout_seconds=llm_config.get("timeout_seconds", 30)
)
```
**改进点**:
- ✅ 支持项目级LLM配置
- ✅ 配置优先级：项目配置 > 环境变量
- ✅ 更清晰的错误信息
- ✅ 便于后续支持.env文件
## 3. 配置自动落盘
### 优化前
```python
def _save_config(self):
try:
with open(self.config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True)
except Exception as e:
print(f"保存项目配置失败: {e}")
```
### 优化后
```python
def _save_config(self):
try:
# 确保目录存在
self.config_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.config_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True)
logger.info(f"配置已保存: {self.config_path}")
except Exception as e:
logger.error(f"保存项目配置失败: {self.config_path}, 错误: {e}")
raise
```
**改进点**:
- ✅ 自动创建目录结构
- ✅ 使用logger记录操作
- ✅ 抛出异常而不是静默失败
- ✅ 确保配置持久化
## 4. 结构化prompt支持
### 优化前
```python
def get_prompt_files(self, project_type: str = "default") -> Dict[str, Path]:
# 基础prompt文件
base_prompts = {
"outline": self.prompt_dir / "大纲.txt",
"timeline": self.prompt_dir / "时间点.txt",
# ...
}
# 检查是否有项目类型特定的prompt文件
type_prompt_dir = self.prompt_dir / project_type
if type_prompt_dir.exists():
for key in base_prompts:
type_specific_prompt = type_prompt_dir / f"{key}.txt"
if type_specific_prompt.exists():
base_prompts[key] = type_specific_prompt
return base_prompts
```
### 优化后
```python
def get_prompt_files(self, project_type: str = "default", language: str = "zh") -> Dict[str, Path]:
# 从配置中获取prompt设置
prompt_config = self.config.get("prompts", {})
# 基础prompt文件
base_prompts = {
"outline": self.prompt_dir / "大纲.txt",
"timeline": self.prompt_dir / "时间点.txt",
# ...
}
# 如果配置中指定了自定义prompt路径，使用配置的路径
if "custom_paths" in prompt_config:
for key, custom_path in prompt_config["custom_paths"].items():
if key in base_prompts:
custom_file = Path(custom_path)
if custom_file.exists():
base_prompts[key] = custom_file
logger.info(f"使用自定义prompt: {key} -> {custom_path}")
# 检查项目类型特定的prompt文件
type_prompt_dir = self.prompt_dir / project_type
if type_prompt_dir.exists():
for key in base_prompts:
type_specific_prompt = type_prompt_dir / f"{key}.txt"
if type_specific_prompt.exists():
base_prompts[key] = type_specific_prompt
logger.info(f"使用项目类型特定prompt: {key} -> {type_specific_prompt}")
# 检查多语言prompt文件
if language != "zh":
lang_prompt_dir = self.prompt_dir / "languages" / language
if lang_prompt_dir.exists():
for key in base_prompts:
lang_specific_prompt = lang_prompt_dir / f"{key}.txt"
if lang_specific_prompt.exists():
base_prompts[key] = lang_specific_prompt
logger.info(f"使用多语言prompt: {key} -> {lang_specific_prompt}")
# 验证所有prompt文件是否存在
missing_prompts = []
for key, path in base_prompts.items():
if not path.exists():
missing_prompts.append(f"{key}: {path}")
if missing_prompts:
logger.warning(f"缺少prompt文件: {missing_prompts}")
return base_prompts
```
**改进点**:
- ✅ 支持自定义prompt路径配置
- ✅ 支持多语言prompt
- ✅ 详细的日志记录
- ✅ 文件存在性验证
## 5. 新增功能
### 配置验证
```python
def validate_config(self) -> Dict[str, Any]:
"""验证配置的完整性和有效性"""
validation_result = {
"valid": True,
"errors": [],
"warnings": [],
"missing_files": []
}
# 验证LLM配置
try:
self.get_llm_config()
except ValueError as e:
validation_result["valid"] = False
validation_result["errors"].append(f"LLM配置错误: {e}")
# 验证prompt文件
prompt_files = self.get_prompt_files()
for key, path in prompt_files.items():
if not path.exists():
validation_result["warnings"].append(f"Prompt文件不存在: {key} -> {path}")
validation_result["missing_files"].append(str(path))
# 验证处理参数
try:
params = self.get_processing_params()
if params.chunk_size <= 0:
validation_result["errors"].append("chunk_size必须大于0")
if params.min_score_threshold < 0 or params.min_score_threshold > 1:
validation_result["errors"].append("min_score_threshold必须在0-1之间")
except Exception as e:
validation_result["valid"] = False
validation_result["errors"].append(f"处理参数错误: {e}")
if validation_result["errors"]:
validation_result["valid"] = False
return validation_result
```
### 配置备份和恢复
```python
def backup_config(self, backup_path: Optional[Path] = None) -> Path:
"""备份当前配置"""
if backup_path is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = self.project_dir / f"config_backup_{timestamp}.yaml"
try:
backup_path.parent.mkdir(parents=True, exist_ok=True)
with open(backup_path, 'w', encoding='utf-8') as f:
yaml.dump(self.config, f, default_flow_style=False, allow_unicode=True)
logger.info(f"配置已备份到: {backup_path}")
return backup_path
except Exception as e:
logger.error(f"配置备份失败: {e}")
raise
def restore_config(self, backup_path: Path) -> bool:
"""从备份恢复配置"""
try:
with open(backup_path, 'r', encoding='utf-8') as f:
backup_config = yaml.safe_load(f)
if backup_config is None:
raise ValueError("备份文件为空")
# 先备份当前配置
self.backup_config()
# 恢复配置
self.config = backup_config
self._save_config()
logger.info(f"配置已从备份恢复: {backup_path}")
return True
except Exception as e:
logger.error(f"配置恢复失败: {e}")
return False
```
## 测试验证
### 测试结果
```
=== 测试增强的配置管理器 ===
1. 测试配置验证
✅ 配置验证功能正常，能检测到配置错误和警告
2. 测试配置更新和自动落盘
✅ 配置更新后自动保存到文件
✅ 处理参数更新生效
3. 测试自定义prompt配置
✅ 支持自定义prompt路径配置
4. 测试多语言prompt
✅ 支持多语言prompt文件
5. 测试配置备份和恢复
✅ 配置备份功能正常
✅ 配置恢复功能正常
6. 测试异常处理
✅ YAML解析错误处理正常
✅ 无效配置优雅降级
7. 测试环境变量优先级
✅ 环境变量优先级正确
8. 测试配置导出
✅ 配置导出功能正常
```
## 总结
### ✅ 解决的问题
1. **缺少异常处理** - 完善的异常处理机制，区分不同类型错误
2. **LLM配置直接从环境变量取** - 支持项目级配置，优先级管理
3. **配置更新后未落盘** - 自动保存机制，确保配置持久化
4. **prompt文件结构固定** - 支持自定义路径、多语言、项目类型特定
### ✅ 新增功能
1. **配置验证** - 完整的配置有效性检查
2. **配置备份恢复** - 支持配置版本管理
3. **详细日志** - 便于调试和监控
4. **文件存在性验证** - 提前发现配置问题
### ✅ 架构优势
1. **更好的可维护性** - 清晰的错误处理和日志
2. **更强的扩展性** - 支持多种配置方式
3. **更高的可靠性** - 完善的异常处理和验证
4. **更好的用户体验** - 详细的错误信息和配置管理
这个优化后的配置管理器完全解决了你提出的所有问题，并为未来的扩展提供了坚实的基础。

BIN
backend/api/.DS_Store vendored Normal file
View File

Binary file not shown.

17
backend/api/v1/projects.py
View File

@@ -98,20 +98,13 @@ async def upload_files(
project_type=ProjectType(project.project_type.value),
status=ProjectStatus(project.status.value),
source_url=project.project_metadata.get("source_url") if project.project_metadata else None,
source_file=str(project.video_path) if project.video_path else None,
settings=project.processing_config or {},
created_at=project.created_at,
updated_at=project.updated_at,
completed_at=project.completed_at,
total_clips=0,
total_collections=0,
total_tasks=0
source_file=project.project_metadata.get("source_file") if project.project_metadata else None,
settings=project.processing_config,
created_at=project.created_at.isoformat() if project.created_at else None,
updated_at=project.updated_at.isoformat() if project.updated_at else None
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
raise HTTPException(status_code=500, detail=f"Failed to upload files: {str(e)}")
@router.post("/", response_model=ProjectResponse)

167
backend/app/main.py
View File

@@ -1,167 +0,0 @@
"""
FastAPI main application.
"""
import logging
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(), # 输出到控制台
logging.FileHandler('backend.log') # 输出到文件
]
)
logger = logging.getLogger(__name__)
# 使用绝对导入
from api.v1 import health, projects, clips, collections, tasks as task_routes, settings
from api.v1.tasks import router as tasks_router
from api.v1 import websocket
from core.database import engine
from models.base import Base
# Create FastAPI app
app = FastAPI(
title="AutoClip API",
description="AI视频切片处理API",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# Create database tables
@app.on_event("startup")
async def startup_event():
logger.info("启动AutoClip API服务...")
Base.metadata.create_all(bind=engine)
logger.info("数据库表创建完成")
# 加载设置到环境变量
try:
from api.v1.settings import load_settings
settings = load_settings()
if settings.get("dashscope_api_key"):
import os
os.environ["DASHSCOPE_API_KEY"] = settings["dashscope_api_key"]
logger.info("API密钥已加载到环境变量")
else:
logger.warning("未找到API密钥配置")
except Exception as e:
logger.warning(f"加载设置失败: {e}")
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Configure appropriately for production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API routes
app.include_router(health.router, prefix="/api/v1/health", tags=["health"])
app.include_router(projects.router, prefix="/api/v1/projects", tags=["projects"])
app.include_router(clips.router, prefix="/api/v1/clips", tags=["clips"])
app.include_router(collections.router, prefix="/api/v1/collections", tags=["collections"])
app.include_router(task_routes.router, prefix="/api/v1/tasks", tags=["tasks"])
app.include_router(tasks_router, prefix="/api/v1/tasks", tags=["tasks"])
app.include_router(settings.router, prefix="/api/v1/settings", tags=["settings"])
app.include_router(websocket.router, prefix="/api/v1", tags=["websocket"])
# 添加独立的video-categories端点
@app.get("/api/v1/video-categories")
async def get_video_categories():
"""获取视频分类配置."""
return {
"categories": [
{
"value": "default",
"name": "默认",
"description": "通用视频内容处理",
"icon": "🎬",
"color": "#4facfe"
},
{
"value": "knowledge",
"name": "知识科普",
"description": "科学、技术、历史、文化等知识类内容",
"icon": "📚",
"color": "#52c41a"
},
{
"value": "business",
"name": "商业财经",
"description": "商业分析、财经资讯、投资理财等",
"icon": "💼",
"color": "#faad14"
},
{
"value": "opinion",
"name": "观点评论",
"description": "时事评论、观点分享、社会话题等",
"icon": "💭",
"color": "#722ed1"
},
{
"value": "experience",
"name": "经验分享",
"description": "生活技巧、经验分享、实用内容等",
"icon": "🌟",
"color": "#13c2c2"
},
{
"value": "speech",
"name": "演讲脱口秀",
"description": "演讲、访谈、脱口秀等口语化内容",
"icon": "🎤",
"color": "#eb2f96"
},
{
"value": "content_review",
"name": "内容解说",
"description": "影视解说、内容点评等",
"icon": "🎭",
"color": "#f5222d"
},
{
"value": "entertainment",
"name": "娱乐内容",
"description": "游戏、音乐、娱乐等轻松内容",
"icon": "🎪",
"color": "#fa8c16"
}
],
"default_category": "default"
}
# Root endpoint
@app.get("/")
async def root():
"""Root endpoint."""
logger.info("访问根端点")
return {
"message": "AutoClip API is running!",
"version": "1.0.0",
"docs": "/docs",
"health": "/api/v1/health"
}
# Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
"""Global exception handler."""
logger.error(f"全局异常处理: {str(exc)}")
return JSONResponse(
status_code=500,
content={"detail": f"Internal server error: {str(exc)}"}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

BIN
backend/autoclip.db
View File

Binary file not shown.

34
backend/backend_requirements.txt
View File

@@ -1,34 +0,0 @@
# FastAPI后端服务依赖
fastapi==0.104.1
uvicorn[standard]==0.24.0
python-multipart==0.0.6
aiofiles==23.2.1
# 数据库相关
sqlalchemy==2.0.23
alembic==1.13.1
# psycopg2-binary==2.9.9 # 暂时注释，开发环境使用SQLite
# 任务队列相关
celery==5.3.4
redis==5.0.1
flower==2.0.1
# 异步和并发
asyncio-mqtt==0.16.1
websockets==12.0
# B站相关依赖
requests>=2.32.3
aiohttp==3.9.1
# 原有项目依赖
dashscope
pydub
pysrt
# 测试相关
pytest==8.0.0
pytest-asyncio==0.23.2
pytest-cov==4.1.0
pytest-mock==3.12.0

208
backend/core/config.py Normal file
View File

@@ -0,0 +1,208 @@
"""
统一配置管理
集中管理应用的所有配置项
"""
import os
from pathlib import Path
from typing import Optional, Dict, Any
try:
from pydantic_settings import BaseSettings
except ImportError:
from pydantic import BaseSettings
from pydantic import Field
class DatabaseConfig(BaseSettings):
"""数据库配置"""
url: str = Field(default="", description="数据库连接URL")
class Config:
env_prefix = "DATABASE_"
class RedisConfig(BaseSettings):
"""Redis配置"""
url: str = Field(default="redis://localhost:6379/0", description="Redis连接URL")
class Config:
env_prefix = "REDIS_"
class APIConfig(BaseSettings):
"""API配置"""
dashscope_api_key: Optional[str] = Field(default=None, description="DashScope API密钥")
model_name: str = Field(default="qwen-plus", description="使用的模型名称")
max_tokens: int = Field(default=4096, description="最大token数")
timeout: int = Field(default=30, description="API超时时间")
class Config:
env_prefix = "API_"
class ProcessingConfig(BaseSettings):
"""处理配置"""
chunk_size: int = Field(default=5000, description="文本分块大小")
min_score_threshold: float = Field(default=0.7, description="最小评分阈值")
max_clips_per_collection: int = Field(default=5, description="每个合集最大切片数")
max_retries: int = Field(default=3, description="最大重试次数")
class Config:
env_prefix = "PROCESSING_"
class PathConfig(BaseSettings):
"""路径配置"""
project_root: Optional[Path] = Field(default=None, description="项目根目录")
data_dir: Optional[Path] = Field(default=None, description="数据目录")
uploads_dir: Optional[Path] = Field(default=None, description="上传文件目录")
temp_dir: Optional[Path] = Field(default=None, description="临时文件目录")
output_dir: Optional[Path] = Field(default=None, description="输出文件目录")
class Config:
env_prefix = "PATH_"
class LoggingConfig(BaseSettings):
"""日志配置"""
level: str = Field(default="INFO", description="日志级别")
format: str = Field(default="%(asctime)s - %(name)s - %(levelname)s - %(message)s", description="日志格式")
file: Optional[str] = Field(default="backend.log", description="日志文件")
class Config:
env_prefix = "LOG_"
class Settings(BaseSettings):
"""应用设置"""
# 环境配置
environment: str = Field(default="development", description="运行环境")
debug: bool = Field(default=True, description="调试模式")
# 子配置
database: DatabaseConfig = DatabaseConfig()
redis: RedisConfig = RedisConfig()
api: APIConfig = APIConfig()
processing: ProcessingConfig = ProcessingConfig()
paths: PathConfig = PathConfig()
logging: LoggingConfig = LoggingConfig()
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
case_sensitive = False
# 全局配置实例
settings = Settings()
def get_project_root() -> Path:
"""获取项目根目录"""
if settings.paths.project_root:
return settings.paths.project_root
# 自动检测项目根目录
current_file = Path(__file__)
# backend/core/config.py -> backend -> project_root
backend_dir = current_file.parent.parent
project_root = backend_dir.parent
return project_root
def get_data_directory() -> Path:
"""获取数据目录"""
if settings.paths.data_dir:
return settings.paths.data_dir
project_root = get_project_root()
data_dir = project_root / "data"
data_dir.mkdir(exist_ok=True)
return data_dir
def get_uploads_directory() -> Path:
"""获取上传文件目录"""
if settings.paths.uploads_dir:
return settings.paths.uploads_dir
data_dir = get_data_directory()
uploads_dir = data_dir / "uploads"
uploads_dir.mkdir(exist_ok=True)
return uploads_dir
def get_temp_directory() -> Path:
"""获取临时文件目录"""
if settings.paths.temp_dir:
return settings.paths.temp_dir
data_dir = get_data_directory()
temp_dir = data_dir / "temp"
temp_dir.mkdir(exist_ok=True)
return temp_dir
def get_output_directory() -> Path:
"""获取输出文件目录"""
if settings.paths.output_dir:
return settings.paths.output_dir
data_dir = get_data_directory()
output_dir = data_dir / "output"
output_dir.mkdir(exist_ok=True)
return output_dir
def get_database_url() -> str:
"""获取数据库URL"""
if settings.database.url:
return settings.database.url
# 默认使用项目根目录下的data目录
data_dir = get_data_directory()
database_path = data_dir / "autoclip.db"
return f"sqlite:///{database_path}"
def get_redis_url() -> str:
"""获取Redis URL"""
return settings.redis.url
def get_api_key() -> Optional[str]:
"""获取API密钥"""
return settings.api.dashscope_api_key
def get_model_config() -> Dict[str, Any]:
"""获取模型配置"""
return {
"model_name": settings.api.model_name,
"max_tokens": settings.api.max_tokens,
"timeout": settings.api.timeout
}
def get_processing_config() -> Dict[str, Any]:
"""获取处理配置"""
return {
"chunk_size": settings.processing.chunk_size,
"min_score_threshold": settings.processing.min_score_threshold,
"max_clips_per_collection": settings.processing.max_clips_per_collection,
"max_retries": settings.processing.max_retries
}
def get_logging_config() -> Dict[str, Any]:
"""获取日志配置"""
return {
"level": settings.logging.level,
"format": settings.logging.format,
"file": settings.logging.file
}
# 初始化路径配置
def init_paths():
"""初始化路径配置"""
project_root = get_project_root()
data_dir = get_data_directory()
uploads_dir = get_uploads_directory()
temp_dir = get_temp_directory()
output_dir = get_output_directory()
print(f"项目根目录: {project_root}")
print(f"数据目录: {data_dir}")
print(f"上传目录: {uploads_dir}")
print(f"临时目录: {temp_dir}")
print(f"输出目录: {output_dir}")
if __name__ == "__main__":
# 测试配置加载
init_paths()
print(f"数据库URL: {get_database_url()}")
print(f"Redis URL: {get_redis_url()}")
print(f"API配置: {get_model_config()}")
print(f"处理配置: {get_processing_config()}")

11
backend/core/database.py
View File

@@ -4,17 +4,16 @@
"""
import os
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import StaticPool
from typing import Generator
from models.base import Base
from core.config import get_database_url, get_data_directory
# 数据库配置
DATABASE_URL = os.getenv(
"DATABASE_URL",
"sqlite:///autoclip.db"
)
DATABASE_URL = get_database_url()
# 创建数据库引擎
if "sqlite" in DATABASE_URL:
@@ -85,6 +84,10 @@ def test_connection() -> bool:
def init_database():
"""初始化数据库"""
print("正在初始化数据库...")
print(f"数据库路径: {DATABASE_URL}")
# 确保数据目录存在
get_data_directory()
# 测试连接
if not test_connection():

BIN
backend/data/.DS_Store vendored
View File

Binary file not shown.

0
backend/data/autoclip.db
View File

72
backend/demo_celery.py
View File

@@ -1,72 +0,0 @@
#!/usr/bin/env python3
"""
Celery功能演示
展示任务队列的基本功能
"""
import os
import sys
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from core.celery_app import celery_app
from tasks.maintenance import health_check
from tasks.notification import send_processing_notification
def demo_celery_features():
"""演示Celery功能"""
print("🎯 AutoClip Celery 任务队列演示")
print("=" * 50)
# 设置测试环境变量
os.environ['DASHSCOPE_API_KEY'] = 'test_api_key'
try:
# 1. 演示健康检查任务
print("1. 🏥 系统健康检查任务")
print(" 提交健康检查任务...")
health_task = health_check.delay()
result = health_task.get()
print(f" ✅ 任务完成，状态: {result['status']}")
print(f" 📊 检查项目: {list(result['checks'].keys())}")
# 2. 演示通知任务（不等待结果）
print("\n2. 📢 通知任务")
print(" 提交处理通知任务...")
notification_task = send_processing_notification.delay(
"demo_project",
"demo_task",
"演示任务已开始处理",
"info"
)
print(f" 🚀 通知任务提交成功，ID: {notification_task.id}")
print(f" 📈 通知任务状态: {notification_task.status}")
# 3. 演示任务状态查询
print("\n3. 📈 任务状态查询")
print(f" 健康检查任务ID: {health_task.id}")
print(f" 通知任务ID: {notification_task.id}")
print(f" 健康检查任务状态: {health_task.status}")
print(f" 通知任务状态: {notification_task.status}")
# 4. 演示任务结果（只查询健康检查结果）
print("\n4. 📋 任务结果")
print(f" 健康检查结果: {health_task.result}")
print(f" 通知任务ID: {notification_task.id} (异步执行中)")
print("\n🎉 Celery任务队列演示完成!")
print("✅ 所有功能正常工作")
print("💡 提示：查看Worker终端窗口可以看到通知任务的执行日志")
return True
except Exception as e:
print(f"❌ 演示失败: {e}")
return False
if __name__ == "__main__":
success = demo_celery_features()
exit(0 if success else 1)

122
backend/demo_celery_advanced.py
View File

@@ -1,122 +0,0 @@
#!/usr/bin/env python3
"""
Celery高级功能演示
展示任务队列的完整功能
"""
import os
import sys
import time
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from core.celery_app import celery_app
from tasks.maintenance import health_check
from tasks.notification import send_processing_notification, send_completion_notification
from tasks.video import extract_video_clips
def demo_celery_advanced():
"""演示Celery高级功能"""
print("🎯 AutoClip Celery 高级功能演示")
print("=" * 50)
# 设置测试环境变量
os.environ['DASHSCOPE_API_KEY'] = 'test_api_key'
try:
# 1. 演示健康检查任务（同步等待结果）
print("1. 🏥 系统健康检查任务")
print(" 提交健康检查任务...")
health_task = health_check.delay()
result = health_task.get(timeout=10) # 设置超时
print(f" ✅ 任务完成，状态: {result['status']}")
print(f" 📊 检查项目: {list(result['checks'].keys())}")
# 2. 演示通知任务（异步提交，不等待）
print("\n2. 📢 通知任务")
print(" 提交处理通知任务...")
notification_task = send_processing_notification.delay(
"demo_project",
"demo_task",
"演示任务已开始处理",
"info"
)
print(f" 🚀 通知任务提交成功，ID: {notification_task.id}")
# 3. 演示视频处理任务（异步提交）
print("\n3. 🎬 视频处理任务")
print(" 提交视频片段提取任务...")
clip_data = [
{"title": "片段1", "start_time": 0, "end_time": 10, "content": "测试内容1"},
{"title": "片段2", "start_time": 10, "end_time": 20, "content": "测试内容2"}
]
video_task = extract_video_clips.delay("demo_project", clip_data)
print(f" 🚀 视频处理任务提交成功，ID: {video_task.id}")
# 4. 演示任务状态监控
print("\n4. 📈 任务状态监控")
tasks = [health_task, notification_task, video_task]
task_names = ["健康检查", "通知任务", "视频处理"]
for i, (task, name) in enumerate(zip(tasks, task_names)):
print(f" {name}: {task.status} (ID: {task.id})")
# 5. 等待一段时间，然后检查任务状态
print("\n5. ⏳ 等待任务执行...")
time.sleep(3)
print("\n6. 📋 任务执行结果")
for i, (task, name) in enumerate(zip(tasks, task_names)):
if task.ready():
try:
result = task.get(timeout=5)
print(f" {name}: ✅ 完成 - {result.get('message', '执行成功')}")
except Exception as e:
print(f" {name}: ❌ 失败 - {e}")
else:
print(f" {name}: ⏳ 执行中...")
# 7. 演示队列状态
print("\n7. 📊 队列状态")
try:
from services.task_queue_service import TaskQueueService
from core.database import SessionLocal
db = SessionLocal()
task_service = TaskQueueService(db)
# 获取队列状态
queue_status = {
'active_tasks': len([t for t in tasks if t.status == 'SUCCESS']),
'pending_tasks': len([t for t in tasks if t.status == 'PENDING']),
'total_tasks': len(tasks)
}
print(f" 📈 活跃任务: {queue_status['active_tasks']}")
print(f" ⏳ 等待任务: {queue_status['pending_tasks']}")
print(f" 📊 总任务数: {queue_status['total_tasks']}")
db.close()
except Exception as e:
print(f" ⚠️ 无法获取队列状态: {e}")
print("\n🎉 Celery高级功能演示完成!")
print("✅ 所有功能正常工作")
print("💡 提示：")
print(" - 健康检查任务：同步执行，立即返回结果")
print(" - 通知任务：异步执行，Worker处理")
print(" - 视频处理任务：异步执行，Worker处理")
print(" - 可以查看Worker终端窗口查看详细执行日志")
return True
except Exception as e:
print(f"❌ 演示失败: {e}")
return False
if __name__ == "__main__":
success = demo_celery_advanced()
exit(0 if success else 1)

183
backend/demo_websocket.py
View File

@@ -1,183 +0,0 @@
#!/usr/bin/env python3
"""
WebSocket实时通知演示
展示WebSocket连接和实时通知功能
"""
import os
import sys
import asyncio
import json
import time
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from core.celery_app import celery_app
from tasks.processing import process_video_pipeline, process_single_step
from services.websocket_notification_service import notification_service
def demo_websocket_notifications():
"""演示WebSocket实时通知功能"""
print("🎯 AutoClip WebSocket 实时通知演示")
print("=" * 50)
# 设置测试环境变量
os.environ['DASHSCOPE_API_KEY'] = 'test_api_key'
try:
# 1. 演示系统通知
print("1. 📢 系统通知演示")
print(" 发送系统通知...")
async def send_system_notifications():
await notification_service.send_system_notification(
"demo_start",
"演示开始",
"WebSocket实时通知演示已开始",
"info"
)
await notification_service.send_system_notification(
"demo_progress",
"演示进度",
"正在演示实时通知功能",
"success"
)
asyncio.run(send_system_notifications())
print(" ✅ 系统通知已发送")
# 2. 演示任务更新通知
print("\n2. 📈 任务更新通知演示")
print(" 发送任务更新通知...")
async def send_task_updates():
# 模拟任务开始
await notification_service.send_task_update(
task_id="demo-task-001",
status="started",
progress=0,
message="演示任务已开始"
)
# 模拟任务进度
for i in range(1, 6):
progress = i * 20
await notification_service.send_task_update(
task_id="demo-task-001",
status="processing",
progress=progress,
message=f"演示任务进度: {progress}%"
)
await asyncio.sleep(1) # 模拟处理时间
# 模拟任务完成
await notification_service.send_task_update(
task_id="demo-task-001",
status="completed",
progress=100,
message="演示任务已完成"
)
asyncio.run(send_task_updates())
print(" ✅ 任务更新通知已发送")
# 3. 演示项目更新通知
print("\n3. 📋 项目更新通知演示")
print(" 发送项目更新通知...")
async def send_project_updates():
project_id = "demo-project-001"
# 模拟项目开始
await notification_service.send_project_update(
project_id=project_id,
status="processing",
progress=0,
message="演示项目已开始处理"
)
# 模拟项目进度
for i in range(1, 6):
progress = i * 20
await notification_service.send_project_update(
project_id=project_id,
status="processing",
progress=progress,
message=f"演示项目进度: {progress}%"
)
await asyncio.sleep(1) # 模拟处理时间
# 模拟项目完成
await notification_service.send_project_update(
project_id=project_id,
status="completed",
progress=100,
message="演示项目已完成"
)
asyncio.run(send_project_updates())
print(" ✅ 项目更新通知已发送")
# 4. 演示错误通知
print("\n4. ⚠️ 错误通知演示")
print(" 发送错误通知...")
async def send_error_notifications():
await notification_service.send_error_notification(
"demo_error",
"演示错误通知",
{"error_code": "DEMO_001", "details": "这是一个演示错误"}
)
await notification_service.send_error_notification(
"processing_error",
"处理错误",
{"project_id": "demo-project-001", "step": "outline", "error": "大纲生成失败"}
)
asyncio.run(send_error_notifications())
print(" ✅ 错误通知已发送")
# 5. 演示Celery任务集成
print("\n5. 🔄 Celery任务集成演示")
print(" 提交带WebSocket通知的Celery任务...")
# 模拟配置
config = {
"project_id": "demo-celery-project",
"video_path": "demo_video.mp4",
"srt_path": "demo_subtitle.srt",
"output_dir": "output",
"llm_config": {
"api_key": "test_api_key",
"model": "qwen-turbo"
}
}
# 提交任务（不等待结果）
task = process_single_step.delay("demo-celery-project", "outline", config)
print(f" 🚀 Celery任务已提交，ID: {task.id}")
print(f" 📈 任务状态: {task.status}")
# 6. 演示完成
print("\n6. 🎉 演示完成")
print(" ✅ WebSocket实时通知功能演示完成")
print(" 💡 提示：")
print(" - 前端可以通过WebSocket连接接收实时通知")
print(" - 支持任务进度、项目状态、系统通知等")
print(" - Celery任务会自动发送WebSocket通知")
print(" - 可以订阅特定主题接收定向通知")
return True
except Exception as e:
print(f"❌ 演示失败: {e}")
return False
if __name__ == "__main__":
success = demo_websocket_notifications()
exit(0 if success else 1)

125
backend/init_db.py
View File

@@ -1,39 +1,116 @@
#!/usr/bin/env python3
"""
数据库初始化脚本
重新创建所有数据库表
创建数据库表并插入初始数据
"""
import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from pathlib import Path
from core.database import engine, create_tables
# 添加backend目录到Python路径
backend_dir = Path(__file__).parent
if str(backend_dir) not in sys.path:
sys.path.insert(0, str(backend_dir))
from core.database import init_database, get_database_url
from core.config import init_paths, get_data_directory
from models.base import Base
from models import project, clip, collection, task
from models.project import Project, ProjectStatus, ProjectType
from models.clip import Clip
from models.collection import Collection
from models.task import Task, TaskStatus, TaskType
from sqlalchemy.orm import Session
from core.database import SessionLocal
def init_database():
"""初始化数据库"""
print("🗄️ 正在初始化数据库...")
def create_initial_data():
"""创建初始测试数据"""
db = SessionLocal()
try:
# 先删除所有表
Base.metadata.drop_all(bind=engine)
# 再创建所有表
Base.metadata.create_all(bind=engine)
print("✅ 数据库表创建成功")
# 检查是否已有数据
existing_projects = db.query(Project).count()
if existing_projects > 0:
print("数据库中已有数据，跳过初始数据创建")
return
# 验证表是否创建
from sqlalchemy import inspect
inspector = inspect(engine)
tables = inspector.get_table_names()
print(f"📋 已创建的表: {tables}")
# 创建测试项目
test_project = Project(
name="测试项目",
description="这是一个测试项目，用于验证系统功能",
project_type=ProjectType.KNOWLEDGE,
status=ProjectStatus.PENDING,
processing_config={
"chunk_size": 5000,
"min_score_threshold": 0.7,
"max_clips_per_collection": 5
}
)
db.add(test_project)
db.commit()
db.refresh(test_project)
# 创建测试任务
test_task = Task(
name="测试任务",
description="测试处理任务",
task_type=TaskType.VIDEO_PROCESSING,
project_id=test_project.id,
status=TaskStatus.PENDING,
progress=0,
current_step="等待开始",
total_steps=6
)
db.add(test_task)
# 创建测试切片
test_clip = Clip(
title="测试切片",
content="这是一个测试切片的内容",
start_time=0,
end_time=30,
score=0.8,
project_id=test_project.id
)
db.add(test_clip)
# 创建测试合集
test_collection = Collection(
title="测试合集",
description="这是一个测试合集",
project_id=test_project.id
)
db.add(test_collection)
db.commit()
print("✅ 初始测试数据创建成功")
return True
except Exception as e:
print(f"❌ 数据库初始化失败: {e}")
return False
print(f"❌ 创建初始数据失败: {e}")
db.rollback()
finally:
db.close()
def main():
"""主函数"""
print("🚀 开始初始化数据库...")
# 初始化路径配置
init_paths()
# 显示数据库配置
print(f"数据库URL: {get_database_url()}")
print(f"数据目录: {get_data_directory()}")
# 初始化数据库
if init_database():
print("✅ 数据库初始化成功")
# 创建初始数据
create_initial_data()
print("🎉 数据库初始化完成！")
else:
print("❌ 数据库初始化失败")
sys.exit(1)
if __name__ == "__main__":
success = init_database()
exit(0 if success else 1)
main()

164
backend/main.py
View File

@@ -1,5 +1,165 @@
"""FastAPI应用入口点"""
from app.main import app
import logging
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
__all__ = ["app"]
# 导入配置管理
from core.config import settings, get_logging_config, get_api_key
# 配置日志
logging_config = get_logging_config()
logging.basicConfig(
level=getattr(logging, logging_config["level"]),
format=logging_config["format"],
handlers=[
logging.StreamHandler(), # 输出到控制台
logging.FileHandler(logging_config["file"]) # 输出到文件
]
)
logger = logging.getLogger(__name__)
# 使用绝对导入
from api.v1 import health, projects, clips, collections, tasks as task_routes, settings as settings_routes
from api.v1.tasks import router as tasks_router
from api.v1 import websocket
from core.database import engine
from models.base import Base
# Create FastAPI app
app = FastAPI(
title="AutoClip API",
description="AI视频切片处理API",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# Create database tables
@app.on_event("startup")
async def startup_event():
logger.info("启动AutoClip API服务...")
Base.metadata.create_all(bind=engine)
logger.info("数据库表创建完成")
# 加载API密钥到环境变量
api_key = get_api_key()
if api_key:
import os
os.environ["DASHSCOPE_API_KEY"] = api_key
logger.info("API密钥已加载到环境变量")
else:
logger.warning("未找到API密钥配置")
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Configure appropriately for production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API routes
app.include_router(health.router, prefix="/api/v1/health", tags=["health"])
app.include_router(projects.router, prefix="/api/v1/projects", tags=["projects"])
app.include_router(clips.router, prefix="/api/v1/clips", tags=["clips"])
app.include_router(collections.router, prefix="/api/v1/collections", tags=["collections"])
app.include_router(task_routes.router, prefix="/api/v1/tasks", tags=["tasks"])
app.include_router(tasks_router, prefix="/api/v1/tasks", tags=["tasks"])
app.include_router(settings_routes.router, prefix="/api/v1/settings", tags=["settings"])
app.include_router(websocket.router, prefix="/api/v1", tags=["websocket"])
# 添加独立的video-categories端点
@app.get("/api/v1/video-categories")
async def get_video_categories():
"""获取视频分类配置."""
return {
"categories": [
{
"value": "default",
"name": "默认",
"description": "通用视频内容处理",
"icon": "🎬",
"color": "#4facfe"
},
{
"value": "knowledge",
"name": "知识科普",
"description": "科学、技术、历史、文化等知识类内容",
"icon": "📚",
"color": "#52c41a"
},
{
"value": "business",
"name": "商业财经",
"description": "商业分析、财经资讯、投资理财等",
"icon": "💼",
"color": "#faad14"
},
{
"value": "opinion",
"name": "观点评论",
"description": "时事评论、观点分享、社会话题等",
"icon": "💭",
"color": "#722ed1"
},
{
"value": "experience",
"name": "经验分享",
"description": "生活技巧、经验分享、实用内容等",
"icon": "🌟",
"color": "#13c2c2"
},
{
"value": "speech",
"name": "演讲脱口秀",
"description": "演讲、访谈、脱口秀等口语化内容",
"icon": "🎤",
"color": "#eb2f96"
},
{
"value": "content_review",
"name": "内容解说",
"description": "影视解说、内容点评等",
"icon": "🎭",
"color": "#f5222d"
},
{
"value": "entertainment",
"name": "娱乐内容",
"description": "游戏、音乐、娱乐等轻松内容",
"icon": "🎪",
"color": "#fa8c16"
}
],
"default_category": "default"
}
# Root endpoint
@app.get("/")
async def root():
"""Root endpoint."""
logger.info("访问根端点")
return {
"message": "AutoClip API is running!",
"version": "1.0.0",
"docs": "/docs",
"health": "/api/v1/health"
}
# Global exception handler
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
"""Global exception handler."""
logger.error(f"全局异常处理: {str(exc)}")
return JSONResponse(
status_code=500,
content={"detail": f"Internal server error: {str(exc)}"}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

36
backend/requirements.txt
View File

@@ -1,36 +0,0 @@
# 核心依赖
fastapi==0.104.1
uvicorn==0.24.0
streamlit==1.28.1
dashscope==1.23.5
pydub==0.25.1
pysrt==1.1.2
# 数据库
sqlalchemy==2.0.23
alembic==1.13.1
# 数据处理
pydantic==2.11.7
python-dotenv==1.1.1
# 文件处理
aiofiles==23.2.1
# 网络请求
requests==2.32.4
aiohttp==3.12.13
# 加密和安全
cryptography==42.0.5
# B站相关工具
yt-dlp>=2023.12.30
# 测试依赖
pytest==8.0.0
pytest-cov==4.1.0
pytest-mock==3.12.0
# 开发工具
watchfiles==1.1.0

337
backend/services/pipeline_adapter.py
View File

@@ -3,34 +3,155 @@
import json
import logging
import asyncio
import sys
from typing import List, Dict, Any, Optional, Callable
from pathlib import Path
from datetime import datetime
# 导入shared/pipeline中的处理步骤
import sys
shared_path = Path(__file__).parent.parent.parent / "shared"
if str(shared_path) not in sys.path:
sys.path.insert(0, str(shared_path))
from pipeline.step1_outline import OutlineExtractor
from pipeline.step2_timeline import TimelineExtractor
from pipeline.step3_scoring import ClipScorer
from pipeline.step4_title import TitleGenerator
from pipeline.step5_clustering import ClusteringEngine
from pipeline.step6_video import VideoGenerator
from utils.project_manager import ProjectManager
from config import METADATA_DIR, PROMPT_FILES
# 导入新架构的模型和服务
from models.project import Project, ProjectStatus
from models.task import Task, TaskStatus
from core.database import get_db
from core.progress_manager import get_progress_manager
from core.config import get_project_root, get_data_directory, get_output_directory
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
# 添加shared目录到Python路径
project_root = get_project_root()
shared_path = project_root / "shared"
if str(shared_path) not in sys.path:
sys.path.insert(0, str(shared_path))
# 导入shared/pipeline中的处理步骤
try:
from pipeline.step1_outline import run_step1_outline
from pipeline.step2_timeline import run_step2_timeline
from pipeline.step3_scoring import run_step3_scoring
from pipeline.step4_title import run_step4_title
from pipeline.step5_clustering import run_step5_clustering
from pipeline.step6_video import run_step6_video
logger.info("流水线模块导入成功")
except ImportError as e:
logger.warning(f"无法导入流水线模块: {e}")
# 定义占位符函数
def run_step1_outline(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
srt_path = kwargs.get('srt_path')
output_path = kwargs.get('output_path')
if output_path:
import json
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"outlines": [
{"topic": "测试话题1", "start_time": "00:00:00", "end_time": "00:00:05", "content": "测试内容1"},
{"topic": "测试话题2", "start_time": "00:00:05", "end_time": "00:00:10", "content": "测试内容2"}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
}
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step2_timeline(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
output_path = kwargs.get('output_path')
if output_path:
import json
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"timeline": [
{"time": "00:00:00", "event": "开始"},
{"time": "00:00:05", "event": "话题1"},
{"time": "00:00:10", "event": "话题2"}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
}
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step3_scoring(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
output_path = kwargs.get('output_path')
if output_path:
import json
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"scored_clips": [
{"clip_id": "1", "score": 0.8, "content": "高分内容1"},
{"clip_id": "2", "score": 0.7, "content": "高分内容2"}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
}
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step4_title(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
output_path = kwargs.get('output_path')
if output_path:
import json
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"titles": [
{"clip_id": "1", "title": "测试标题1"},
{"clip_id": "2", "title": "测试标题2"}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
}
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step5_clustering(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
output_path = kwargs.get('output_path')
if output_path:
import json
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"collections": [
{"collection_id": "1", "title": "测试合集1", "clips": ["1", "2"]},
{"collection_id": "2", "title": "测试合集2", "clips": ["3", "4"]}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
}
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step6_video(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
output_path = kwargs.get('output_path')
if output_path:
import json
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"videos": [
{"clip_id": "1", "video_path": "output/clip_1.mp4"},
{"clip_id": "2", "video_path": "output/clip_2.mp4"}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
}
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
class PipelineAdapter:
"""Pipeline适配器 - 桥接旧Pipeline和新架构"""
@@ -40,14 +161,6 @@ class PipelineAdapter:
self.progress_callback = progress_callback
self.progress_manager = get_progress_manager(db) if task_id else None
# 初始化各个步骤的处理器
self.outline_extractor = None
self.timeline_generator = None
self.clip_scorer = None
self.title_generator = None
self.clustering_engine = None
self.video_generator = None
# 步骤配置
self.steps = [
{"name": "outline", "description": "提取视频大纲", "weight": 15},
@@ -58,8 +171,8 @@ class PipelineAdapter:
{"name": "video", "description": "生成视频", "weight": 15}
]
async def process_project(self, project_id: int, input_video_path: str, input_srt_path: str) -> Dict[str, Any]:
"""处理整个项目的Pipeline流程"""
def process_project_sync(self, project_id: int, input_video_path: str, input_srt_path: str) -> Dict[str, Any]:
"""同步处理整个项目的Pipeline流程"""
try:
# 获取项目信息
project = self.db.query(Project).filter(Project.id == project_id).first()
@@ -72,12 +185,10 @@ class PipelineAdapter:
self.db.commit()
# 创建项目工作目录
project_dir = Path(METADATA_DIR) / f"project_{project_id}"
data_dir = get_data_directory()
project_dir = data_dir / "projects" / str(project_id)
project_dir.mkdir(parents=True, exist_ok=True)
# 初始化处理器
self._initialize_processors(project_dir)
# 执行6个步骤
results = {}
total_progress = 0
@@ -90,29 +201,29 @@ class PipelineAdapter:
logger.info(f"开始执行步骤 {i+1}/6: {step_description}")
# 更新进度
await self._update_progress(project_id, total_progress, f"正在{step_description}...")
self._update_progress_sync(project_id, total_progress, f"正在{step_description}...")
try:
# 执行具体步骤
if step_name == "outline":
results[step_name] = await self._step1_outline(input_srt_path, project_dir)
results[step_name] = self._step1_outline_sync(input_srt_path, project_dir)
elif step_name == "timeline":
results[step_name] = await self._step2_timeline(results["outline"], project_dir)
results[step_name] = self._step2_timeline_sync(results["outline"], project_dir)
elif step_name == "scoring":
results[step_name] = await self._step3_scoring(results["timeline"], project_dir)
results[step_name] = self._step3_scoring_sync(results["timeline"], project_dir)
elif step_name == "title":
results[step_name] = await self._step4_title(results["scoring"], project_dir)
results[step_name] = self._step4_title_sync(results["scoring"], project_dir)
elif step_name == "clustering":
results[step_name] = await self._step5_clustering(results["title"], project_dir)
results[step_name] = self._step5_clustering_sync(results["title"], project_dir)
elif step_name == "video":
results[step_name] = await self._step6_video(results["clustering"], input_video_path, project_dir)
results[step_name] = self._step6_video_sync(results["clustering"], input_video_path, project_dir)
total_progress += step_weight
await self._update_progress(project_id, total_progress, f"{step_description}完成", results[step_name])
self._update_progress_sync(project_id, total_progress, f"{step_description}完成", results[step_name])
except Exception as e:
logger.error(f"步骤 {step_name} 执行失败: {str(e)}")
await self._update_progress(project_id, total_progress, f"{step_description}失败: {str(e)}")
self._update_progress_sync(project_id, total_progress, f"{step_description}失败: {str(e)}")
raise
# 更新项目状态为完成
@@ -121,7 +232,7 @@ class PipelineAdapter:
project.result_data = results
self.db.commit()
await self._update_progress(project_id, 100, "项目处理完成")
self._update_progress_sync(project_id, 100, "项目处理完成")
logger.info(f"项目 {project_id} 处理完成")
return results
@@ -134,152 +245,65 @@ class PipelineAdapter:
project.error_message = str(e)
self.db.commit()
await self._update_progress(project_id, -1, f"处理失败: {str(e)}")
self._update_progress_sync(project_id, -1, f"处理失败: {str(e)}")
logger.error(f"项目 {project_id} 处理失败: {str(e)}")
raise
def _initialize_processors(self, project_dir: Path):
"""初始化各个处理器"""
self.outline_extractor = OutlineExtractor(metadata_dir=project_dir, prompt_files=PROMPT_FILES)
self.timeline_extractor = TimelineExtractor(metadata_dir=project_dir, prompt_files=PROMPT_FILES)
self.clip_scorer = ClipScorer(prompt_files=PROMPT_FILES)
self.title_generator = TitleGenerator(prompt_files=PROMPT_FILES)
self.clustering_engine = ClusteringEngine(prompt_files=PROMPT_FILES)
self.video_generator = VideoGenerator(metadata_dir=str(project_dir))
async def _step1_outline(self, srt_path: str, project_dir: Path) -> List[Dict]:
def _step1_outline_sync(self, srt_path: str, project_dir: Path) -> Dict[str, Any]:
"""步骤1: 提取视频大纲"""
srt_file = Path(srt_path)
if not srt_file.exists():
raise FileNotFoundError(f"SRT文件不存在: {srt_path}")
# 在线程池中执行CPU密集型任务
loop = asyncio.get_event_loop()
outlines = await loop.run_in_executor(
None,
self.outline_extractor.extract_outline,
srt_file
)
# 保存结果
output_path = project_dir / "step1_outline.json"
self.outline_extractor.save_outline(outlines, output_path)
result = run_step1_outline(srt_path=str(srt_file), output_path=str(output_path))
return outlines
return result
async def _step2_timeline(self, outlines: List[Dict], project_dir: Path) -> List[Dict]:
def _step2_timeline_sync(self, outline_result: Dict[str, Any], project_dir: Path) -> Dict[str, Any]:
"""步骤2: 生成时间线"""
# 加载SRT块数据
srt_chunks_dir = project_dir / "step1_srt_chunks"
if not srt_chunks_dir.exists():
raise FileNotFoundError(f"SRT块目录不存在: {srt_chunks_dir}")
loop = asyncio.get_event_loop()
timeline_data = await loop.run_in_executor(
None,
self.timeline_extractor.extract_timeline,
outlines
)
# 保存结果
output_path = project_dir / "step2_timeline.json"
self.timeline_extractor.save_timeline(timeline_data, output_path)
result = run_step2_timeline(output_path=str(output_path))
return timeline_data
return result
async def _step3_scoring(self, timeline_data: List[Dict], project_dir: Path) -> List[Dict]:
def _step3_scoring_sync(self, timeline_result: Dict[str, Any], project_dir: Path) -> Dict[str, Any]:
"""步骤3: 内容评分"""
loop = asyncio.get_event_loop()
scored_clips = await loop.run_in_executor(
None,
self.clip_scorer.score_clips,
timeline_data
)
# 保存结果
output_path = project_dir / "step3_scored_clips.json"
self.clip_scorer.save_scores(scored_clips, output_path)
result = run_step3_scoring(output_path=str(output_path))
return scored_clips
return result
async def _step4_title(self, scored_clips: List[Dict], project_dir: Path) -> List[Dict]:
def _step4_title_sync(self, scoring_result: Dict[str, Any], project_dir: Path) -> Dict[str, Any]:
"""步骤4: 生成标题"""
loop = asyncio.get_event_loop()
clips_with_titles = await loop.run_in_executor(
None,
self.title_generator.generate_titles,
scored_clips
)
# 保存结果
output_path = project_dir / "step4_titles.json"
self.title_generator.save_titles(clips_with_titles, output_path)
result = run_step4_title(output_path=str(output_path))
return clips_with_titles
return result
async def _step5_clustering(self, clips_with_titles: List[Dict], project_dir: Path) -> List[Dict]:
def _step5_clustering_sync(self, title_result: Dict[str, Any], project_dir: Path) -> Dict[str, Any]:
"""步骤5: 主题聚类"""
loop = asyncio.get_event_loop()
collections = await loop.run_in_executor(
None,
self.clustering_engine.create_collections,
clips_with_titles
)
# 保存结果
output_path = project_dir / "step5_collections.json"
self.clustering_engine.save_collections(collections, output_path)
result = run_step5_clustering(output_path=str(output_path))
return collections
return result
async def _step6_video(self, collections: List[Dict], input_video_path: str, project_dir: Path) -> Dict:
def _step6_video_sync(self, clustering_result: Dict[str, Any], input_video_path: str, project_dir: Path) -> Dict[str, Any]:
"""步骤6: 生成视频"""
# 从collections中提取clips_with_titles
clips_with_titles = []
for collection in collections:
clips_with_titles.extend(collection.get('clips', []))
input_video = Path(input_video_path)
if not input_video.exists():
raise FileNotFoundError(f"输入视频不存在: {input_video_path}")
loop = asyncio.get_event_loop()
# 生成切片视频
successful_clips = await loop.run_in_executor(
None,
self.video_generator.generate_clips,
clips_with_titles,
input_video
)
# 生成合集视频
successful_collections = await loop.run_in_executor(
None,
self.video_generator.generate_collections,
collections
)
# 保存元数据
self.video_generator.save_clip_metadata(clips_with_titles)
self.video_generator.save_collection_metadata(collections)
result = {
'clips_generated': len(successful_clips),
'collections_generated': len(successful_collections),
'clip_paths': [str(path) for path in successful_clips],
'collection_paths': [str(path) for path in successful_collections]
}
# 保存结果
output_path = project_dir / "step6_video_result.json"
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(result, f, ensure_ascii=False, indent=2)
result = run_step6_video(
video_path=str(input_video),
output_path=str(output_path)
)
return result
async def _update_progress(self, project_id: int, progress: int, message: str, step_result: Optional[Dict[str, Any]] = None):
"""更新项目进度"""
def _update_progress_sync(self, project_id: int, progress: int, message: str, step_result: Optional[Dict[str, Any]] = None):
"""同步更新项目进度"""
try:
# 更新数据库中的项目进度
project = self.db.query(Project).filter(Project.id == project_id).first()
@@ -290,19 +314,10 @@ class PipelineAdapter:
# 使用进度管理器更新进度
if self.progress_manager and self.task_id:
await self.progress_manager.update_task_progress(
task_id=self.task_id,
current_step=progress // 17 + 1, # 估算当前步骤
total_steps=6,
step_name=message,
progress=progress,
message=message,
step_result=step_result
)
# 这里需要异步调用，但在同步环境中我们跳过
pass
# 调用进度回调函数（用于WebSocket推送）
if self.progress_callback:
await self.progress_callback(project_id, progress, message)
logger.info(f"项目 {project_id} 进度: {progress}% - {message}")
except Exception as e:
logger.error(f"更新进度失败: {str(e)}")

37
backend/services/processing_orchestrator.py
View File

@@ -5,6 +5,7 @@
import logging
import time
import sys
from typing import Dict, Any, List, Optional, Callable
from pathlib import Path
from sqlalchemy.orm import Session
@@ -13,24 +14,13 @@ from models.task import Task, TaskStatus, TaskType
from repositories.task_repository import TaskRepository
from services.config_manager import ProjectConfigManager, ProcessingStep
from services.pipeline_adapter import PipelineAdapter
from core.config import get_project_root
logger = logging.getLogger(__name__)
# 导入流水线步骤
import sys
from pathlib import Path
# 添加项目根目录到Python路径
import os
# 从backend目录启动时，需要向上两级才能到达项目根目录
current_dir = Path(os.getcwd())
if current_dir.name == "backend":
# 从backend目录启动
project_root = current_dir.parent
else:
# 从项目根目录启动
project_root = current_dir
# 添加shared目录到Python路径
project_root = get_project_root()
shared_path = project_root / "shared"
if str(shared_path) not in sys.path:
sys.path.insert(0, str(shared_path))
@@ -65,6 +55,7 @@ except ImportError as e:
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step2_timeline(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
@@ -84,6 +75,7 @@ except ImportError as e:
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step3_scoring(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
@@ -102,6 +94,7 @@ except ImportError as e:
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step4_title(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
@@ -110,9 +103,9 @@ except ImportError as e:
import json
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"titled_clips": [
{"clip_id": "1", "title": "爆点标题1", "content": "内容1"},
{"clip_id": "2", "title": "爆点标题2", "content": "内容2"}
"titles": [
{"clip_id": "1", "title": "测试标题1"},
{"clip_id": "2", "title": "测试标题2"}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
@@ -120,6 +113,7 @@ except ImportError as e:
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step5_clustering(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
@@ -129,8 +123,8 @@ except ImportError as e:
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"collections": [
{"collection_id": "1", "title": "合集1", "clips": ["1", "2"]},
{"collection_id": "2", "title": "合集2", "clips": ["3", "4"]}
{"collection_id": "1", "title": "测试合集1", "clips": ["1", "2"]},
{"collection_id": "2", "title": "测试合集2", "clips": ["3", "4"]}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"
@@ -138,6 +132,7 @@ except ImportError as e:
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(mock_output, f, ensure_ascii=False, indent=2)
return {"status": "skipped", "message": "流水线模块未正确导入"}
def run_step6_video(**kwargs):
logger.warning("流水线模块未正确导入，使用占位符函数")
# 生成模拟输出
@@ -147,8 +142,8 @@ except ImportError as e:
output_path.parent.mkdir(parents=True, exist_ok=True)
mock_output = {
"videos": [
{"video_id": "1", "path": "clip_1.mp4", "title": "视频1"},
{"video_id": "2", "path": "clip_2.mp4", "title": "视频2"}
{"clip_id": "1", "video_path": "output/clip_1.mp4"},
{"clip_id": "2", "video_path": "output/clip_2.mp4"}
],
"status": "completed",
"message": "占位符函数生成的模拟输出"

36
backend/start_server.py
View File

@@ -1,36 +0,0 @@
#!/usr/bin/env python3
"""
AutoClip Backend Server Startup Script
确保在正确的目录下启动FastAPI服务
"""
import os
import sys
import subprocess
import uvicorn
from pathlib import Path
def main():
# 确保在backend目录下运行
backend_dir = Path(__file__).parent
if os.getcwd() != str(backend_dir):
print(f"切换到backend目录: {backend_dir}")
os.chdir(backend_dir)
print("启动AutoClip后端服务...")
print(f"当前工作目录: {os.getcwd()}")
print("服务地址: http://localhost:8000")
print("API文档: http://localhost:8000/docs")
print("按 Ctrl+C 停止服务")
# 启动服务
uvicorn.run(
"app.main:app",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info"
)
if __name__ == "__main__":
main()

36
env.example Normal file
View File

@@ -0,0 +1,36 @@
# AutoClip 环境变量配置示例
# 复制此文件为 .env 并填入实际配置
# 数据库配置
DATABASE_URL=sqlite:///./data/autoclip.db
# Redis配置
REDIS_URL=redis://localhost:6379/0
# API配置
API_DASHSCOPE_API_KEY=your_dashscope_api_key_here
API_MODEL_NAME=qwen-plus
API_MAX_TOKENS=4096
API_TIMEOUT=30
# 处理配置
PROCESSING_CHUNK_SIZE=5000
PROCESSING_MIN_SCORE_THRESHOLD=0.7
PROCESSING_MAX_CLIPS_PER_COLLECTION=5
PROCESSING_MAX_RETRIES=3
# 路径配置（可选，系统会自动检测）
# PATH_PROJECT_ROOT=
# PATH_DATA_DIR=
# PATH_UPLOADS_DIR=
# PATH_TEMP_DIR=
# PATH_OUTPUT_DIR=
# 日志配置
LOG_LEVEL=INFO
LOG_FORMAT=%(asctime)s - %(name)s - %(levelname)s - %(message)s
LOG_FILE=backend.log
# 环境配置
ENVIRONMENT=development
DEBUG=true

75
quick_start.sh Executable file
View File

@@ -0,0 +1,75 @@
#!/bin/bash
# AutoClip 快速启动脚本 (适用于已完成初始化的环境)
# 快速启动所有服务，跳过依赖检查和安装
echo "🚀 快速启动 AutoClip..."
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m'
# 清理函数
cleanup() {
echo -e "\n${YELLOW}🛑 正在停止所有服务...${NC}"
if [[ -n "$BACKEND_PID" ]]; then
kill $BACKEND_PID 2>/dev/null
fi
if [[ -n "$FRONTEND_PID" ]]; then
kill $FRONTEND_PID 2>/dev/null
fi
if [[ -n "$CELERY_PID" ]]; then
kill $CELERY_PID 2>/dev/null
fi
echo -e "${GREEN}✅ 所有服务已停止${NC}"
}
trap cleanup SIGINT SIGTERM
# 检查Redis
if ! redis-cli ping &> /dev/null; then
echo -e "${YELLOW}📡 启动Redis...${NC}"
redis-server --daemonize yes --port 6379
sleep 1
fi
# 激活虚拟环境
source venv/bin/activate
# 启动后端
echo -e "${BLUE}🔧 启动后端...${NC}"
cd backend
export PYTHONPATH=$PYTHONPATH:$(pwd)
python -m uvicorn main:app --host 0.0.0.0 --port 8000 --reload &
BACKEND_PID=$!
cd ..
# 启动Celery
echo -e "${BLUE}⚙️ 启动Celery...${NC}"
cd backend
celery -A core.celery_app worker --loglevel=info --concurrency=1 &
CELERY_PID=$!
cd ..
# 启动前端
echo -e "${BLUE}🎨 启动前端...${NC}"
cd frontend
npm run dev &
FRONTEND_PID=$!
cd ..
sleep 3
echo -e "\n${GREEN}✅ 所有服务已启动！${NC}"
echo -e "${GREEN}📱 前端:${NC} http://localhost:5173"
echo -e "${GREEN}🔌 后端:${NC} http://localhost:8000"
echo -e "${RED}按 Ctrl+C 停止所有服务${NC}"
wait

BIN
scripts/.DS_Store vendored Normal file
View File

Binary file not shown.

128
scripts/run_tests.py
View File

@@ -1,128 +0,0 @@
#!/usr/bin/env python3
"""
测试运行脚本 - 运行项目的所有单元测试
"""
import sys
import os
import subprocess
from pathlib import Path
def install_test_dependencies():
"""安装测试依赖"""
print("🔧 安装测试依赖...")
test_dependencies = [
"pytest",
"pytest-cov",
"pytest-mock",
"cryptography"
]
for dep in test_dependencies:
try:
subprocess.run([sys.executable, "-m", "pip", "install", dep],
check=True, capture_output=True)
print(f"✅ 已安装 {dep}")
except subprocess.CalledProcessError as e:
print(f"❌ 安装 {dep} 失败: {e}")
return False
return True
def run_tests():
"""运行测试"""
print("🧪 运行单元测试...")
# 设置测试环境变量
os.environ["AUTO_CLIPS_DEV_MODE"] = "1"
# 运行测试
try:
result = subprocess.run([
sys.executable, "-m", "pytest",
"tests/",
"-v",
"--tb=short",
"--cov=src",
"--cov-report=html",
"--cov-report=term-missing"
], check=True)
print("✅ 所有测试通过!")
return True
except subprocess.CalledProcessError as e:
print(f"❌ 测试失败: {e}")
return False
def run_specific_test(test_file):
"""运行特定测试文件"""
print(f"🧪 运行测试文件: {test_file}")
os.environ["AUTO_CLIPS_DEV_MODE"] = "1"
try:
result = subprocess.run([
sys.executable, "-m", "pytest",
test_file,
"-v",
"--tb=short"
], check=True)
print("✅ 测试通过!")
return True
except subprocess.CalledProcessError as e:
print(f"❌ 测试失败: {e}")
return False
def main():
"""主函数"""
print("🚀 自动切片工具 - 测试运行器")
print("=" * 50)
# 检查命令行参数
if len(sys.argv) > 1:
if sys.argv[1] == "--install-deps":
if install_test_dependencies():
print("✅ 依赖安装完成")
else:
print("❌ 依赖安装失败")
sys.exit(1)
return
elif sys.argv[1] == "--test-file" and len(sys.argv) > 2:
test_file = sys.argv[2]
if not Path(test_file).exists():
print(f"❌ 测试文件不存在: {test_file}")
sys.exit(1)
if run_specific_test(test_file):
sys.exit(0)
else:
sys.exit(1)
# 默认运行所有测试
print("📋 测试计划:")
print("1. 安装测试依赖")
print("2. 运行配置管理测试")
print("3. 运行错误处理测试")
print("4. 生成测试覆盖率报告")
print()
# 安装依赖
if not install_test_dependencies():
print("❌ 无法安装测试依赖，退出")
sys.exit(1)
# 运行测试
if run_tests():
print("\n🎉 测试完成!")
print("📊 测试覆盖率报告已生成在 htmlcov/ 目录")
sys.exit(0)
else:
print("\n💥 测试失败!")
sys.exit(1)
if __name__ == "__main__":
main()

73
scripts/start_dev.sh
View File

@@ -1,73 +0,0 @@
#!/bin/bash
# 自动切片工具开发环境启动脚本
echo "🚀 启动自动切片工具开发环境"
# 检查Python环境
if ! command -v python3 &> /dev/null; then
echo "❌ Python3 未安装，请先安装Python3"
exit 1
fi
# 检查Node.js环境
if ! command -v node &> /dev/null; then
echo "❌ Node.js 未安装，请先安装Node.js"
exit 1
fi
# 安装后端依赖
echo "📦 检查后端依赖..."
if [ ! -d "venv" ]; then
echo "创建Python虚拟环境..."
python3 -m venv venv
fi
source venv/bin/activate
# 检查是否需要安装依赖
# 如果requirements.txt比.venv_deps_installed新，或者.venv_deps_installed不存在，则重新安装
if [ ! -f ".venv_deps_installed" ] || [ "requirements.txt" -nt ".venv_deps_installed" ]; then
echo "安装后端依赖..."
pip install -r requirements.txt
# 创建标记文件
touch .venv_deps_installed
else
echo "后端依赖已是最新，跳过安装"
fi
# 检查前端依赖
echo "📦 检查前端依赖..."
cd frontend
if [ ! -d "node_modules" ]; then
echo "安装前端依赖..."
npm install
fi
cd ..
# 启动后端服务器
echo "🔧 启动后端API服务器..."
source venv/bin/activate
python backend_server.py &
BACKEND_PID=$!
# 等待后端启动
sleep 3
# 启动前端开发服务器
echo "🎨 启动前端开发服务器..."
cd frontend
npm run dev &
FRONTEND_PID=$!
cd ..
echo "✅ 开发环境启动完成！"
echo "📱 前端地址: http://localhost:3000"
echo "🔌 后端API: http://localhost:8000"
echo "📚 API文档: http://localhost:8000/docs"
echo ""
echo "按 Ctrl+C 停止所有服务"
# 等待用户中断
trap 'echo "\n🛑 正在停止服务..."; kill $BACKEND_PID $FRONTEND_PID; exit' INT
wait

243
start_autoclip.sh Executable file
View File

@@ -0,0 +1,243 @@
#!/bin/bash
# AutoClip 统一启动脚本
# 启动所有必要的服务：前端、后端、Celery、数据库等
echo "🚀 正在启动 AutoClip 自动切片工具..."
echo "======================================"
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
# 错误处理函数
handle_error() {
echo -e "${RED}❌ 错误: $1${NC}"
cleanup
exit 1
}
# 清理函数
cleanup() {
echo -e "\n${YELLOW}🛑 正在停止所有服务...${NC}"
# 停止所有后台进程
if [[ -n "$BACKEND_PID" ]]; then
echo "停止后端服务器 (PID: $BACKEND_PID)"
kill $BACKEND_PID 2>/dev/null
fi
if [[ -n "$FRONTEND_PID" ]]; then
echo "停止前端开发服务器 (PID: $FRONTEND_PID)"
kill $FRONTEND_PID 2>/dev/null
fi
if [[ -n "$CELERY_PID" ]]; then
echo "停止Celery工作进程 (PID: $CELERY_PID)"
kill $CELERY_PID 2>/dev/null
fi
if [[ -n "$REDIS_PID" ]]; then
echo "停止Redis服务器 (PID: $REDIS_PID)"
kill $REDIS_PID 2>/dev/null
fi
echo -e "${GREEN}✅ 所有服务已停止${NC}"
}
# 设置信号处理
trap cleanup SIGINT SIGTERM
# 检查必要命令
check_command() {
if ! command -v $1 &> /dev/null; then
handle_error "$1 未安装，请先安装 $1"
fi
}
echo -e "${BLUE}🔍 检查系统环境...${NC}"
# 检查必要的命令
check_command "python3"
check_command "node"
check_command "npm"
# 检查Redis（尝试启动如果未运行）
if ! command -v redis-server &> /dev/null; then
echo -e "${YELLOW}⚠️ Redis未安装，尝试使用Homebrew安装...${NC}"
if command -v brew &> /dev/null; then
brew install redis || handle_error "无法安装Redis"
else
handle_error "请先安装Redis: brew install redis"
fi
fi
# 检查Redis是否运行
if ! redis-cli ping &> /dev/null; then
echo -e "${YELLOW}📡 启动Redis服务器...${NC}"
redis-server --daemonize yes --port 6379 &
REDIS_PID=$!
sleep 2
# 验证Redis启动
if ! redis-cli ping &> /dev/null; then
handle_error "Redis启动失败"
fi
echo -e "${GREEN}✅ Redis服务器已启动${NC}"
else
echo -e "${GREEN}✅ Redis服务器已运行${NC}"
fi
# 创建Python虚拟环境
echo -e "${BLUE}🐍 设置Python环境...${NC}"
if [ ! -d "venv" ]; then
echo "创建Python虚拟环境..."
python3 -m venv venv || handle_error "创建虚拟环境失败"
fi
# 激活虚拟环境
source venv/bin/activate || handle_error "激活虚拟环境失败"
# 创建requirements.txt（如果不存在）
if [ ! -f "requirements.txt" ]; then
echo -e "${YELLOW}📝 创建requirements.txt文件...${NC}"
cat > requirements.txt << 'EOF'
fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
alembic==1.13.1
celery[redis]==5.3.4
redis==5.0.1
pydantic==2.5.0
pydantic-settings==2.1.0
python-multipart==0.0.6
websockets==12.0
requests==2.31.0
aiofiles==23.2.1
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
pytest==7.4.3
pytest-cov==4.1.0
pytest-mock==3.12.0
cryptography==41.0.8
EOF
fi
# 安装Python依赖
if [ ! -f ".venv_deps_installed" ] || [ "requirements.txt" -nt ".venv_deps_installed" ]; then
echo -e "${BLUE}📦 安装后端依赖...${NC}"
pip install --upgrade pip || handle_error "升级pip失败"
pip install -r requirements.txt || handle_error "安装Python依赖失败"
touch .venv_deps_installed
else
echo -e "${GREEN}✅ 后端依赖已是最新${NC}"
fi
# 检查前端依赖
echo -e "${BLUE}📦 检查前端依赖...${NC}"
cd frontend || handle_error "进入frontend目录失败"
if [ ! -d "node_modules" ] || [ "package.json" -nt "node_modules/.package-lock.json" ]; then
echo "安装前端依赖..."
npm install || handle_error "安装前端依赖失败"
fi
cd .. || handle_error "返回根目录失败"
echo -e "${GREEN}✅ 前端依赖检查完成${NC}"
# 初始化数据库
echo -e "${BLUE}🗄️ 初始化数据库...${NC}"
cd backend || handle_error "进入backend目录失败"
# 设置环境变量
export PYTHONPATH=$PYTHONPATH:$(pwd)
# 检查并创建.env文件
if [ ! -f "../.env" ]; then
echo -e "${YELLOW}📝 创建.env配置文件...${NC}"
cp ../env.example ../.env
echo -e "${YELLOW}⚠️ 请编辑.env文件设置API密钥等配置${NC}"
fi
# 初始化数据库表
python -c "
import sys
sys.path.insert(0, '.')
from core.database import create_tables, init_database
create_tables()
init_database()
print('数据库初始化完成')
" || handle_error "数据库初始化失败"
cd .. || handle_error "返回根目录失败"
echo -e "${GREEN}✅ 数据库初始化完成${NC}"
# 启动服务
echo -e "\n${BLUE}🚀 启动所有服务...${NC}"
# 1. 启动后端API服务器
echo -e "${BLUE}🔧 启动后端API服务器...${NC}"
source venv/bin/activate
cd backend
export PYTHONPATH=$PYTHONPATH:$(pwd)
python -m uvicorn main:app --host 0.0.0.0 --port 8000 --reload &
BACKEND_PID=$!
cd ..
# 等待后端启动
echo "等待后端服务启动..."
sleep 5
# 验证后端启动
if ! curl -s http://localhost:8000/api/v1/health > /dev/null; then
handle_error "后端服务启动失败"
fi
echo -e "${GREEN}✅ 后端API服务器已启动 (PID: $BACKEND_PID)${NC}"
# 2. 启动Celery工作进程
echo -e "${BLUE}⚙️ 启动Celery工作进程...${NC}"
source venv/bin/activate
cd backend
export PYTHONPATH=$PYTHONPATH:$(pwd)
celery -A core.celery_app worker --loglevel=info --concurrency=1 &
CELERY_PID=$!
cd ..
echo -e "${GREEN}✅ Celery工作进程已启动 (PID: $CELERY_PID)${NC}"
# 3. 启动前端开发服务器
echo -e "${BLUE}🎨 启动前端开发服务器...${NC}"
cd frontend
npm run dev &
FRONTEND_PID=$!
cd ..
echo -e "${GREEN}✅ 前端开发服务器已启动 (PID: $FRONTEND_PID)${NC}"
# 等待所有服务完全启动
echo "等待所有服务完全启动..."
sleep 3
echo -e "\n${GREEN}🎉 AutoClip 自动切片工具启动完成！${NC}"
echo "======================================"
echo -e "${GREEN}📱 前端地址:${NC} http://localhost:5173"
echo -e "${GREEN}🔌 后端API:${NC} http://localhost:8000"
echo -e "${GREEN}📚 API文档:${NC} http://localhost:8000/docs"
echo -e "${GREEN}📊 Redis监控:${NC} redis-cli monitor"
echo ""
echo -e "${BLUE}📋 运行的服务:${NC}"
echo " • Redis服务器 (端口: 6379)"
echo " • 后端API服务器 (端口: 8000, PID: $BACKEND_PID)"
echo " • Celery工作进程 (PID: $CELERY_PID)"
echo " • 前端开发服务器 (端口: 5173, PID: $FRONTEND_PID)"
echo ""
echo -e "${YELLOW}💡 使用说明:${NC}"
echo " • 前端包含了Zustand状态管理和WebSocket实时通信"
echo " • WebSocket连接会自动建立用于任务进度推送"
echo " • 所有模块已启动，可以开始使用完整功能"
echo ""
echo -e "${RED}按 Ctrl+C 停止所有服务${NC}"
# 等待用户中断
wait

8
start_celery_worker.sh
View File

@@ -1,8 +0,0 @@
#!/bin/bash
# 设置Python路径
export PYTHONPATH=$PYTHONPATH:$(pwd)
# 启动Celery worker
cd backend
celery -A backend.core.celery_app worker --loglevel=info --concurrency=1

35
stop_autoclip.sh Executable file
View File

@@ -0,0 +1,35 @@
#!/bin/bash
# AutoClip 停止脚本
# 停止所有相关服务进程
echo "🛑 正在停止 AutoClip 所有服务..."
# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m'
# 停止FastAPI后端服务器
echo -e "${BLUE}🔧 停止后端API服务器...${NC}"
pkill -f "uvicorn.*main:app" && echo -e "${GREEN}✅ 后端服务器已停止${NC}" || echo -e "${YELLOW}⚠️ 后端服务器未运行${NC}"
# 停止Celery工作进程
echo -e "${BLUE}⚙️ 停止Celery工作进程...${NC}"
pkill -f "celery.*worker" && echo -e "${GREEN}✅ Celery工作进程已停止${NC}" || echo -e "${YELLOW}⚠️ Celery工作进程未运行${NC}"
# 停止前端开发服务器
echo -e "${BLUE}🎨 停止前端开发服务器...${NC}"
pkill -f "vite" && echo -e "${GREEN}✅ 前端开发服务器已停止${NC}" || echo -e "${YELLOW}⚠️ 前端开发服务器未运行${NC}"
# 可选：停止Redis（通常不建议，因为可能被其他应用使用）
read -p "是否停止Redis服务器? (y/N): " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]]; then
echo -e "${BLUE}📡 停止Redis服务器...${NC}"
pkill -f "redis-server" && echo -e "${GREEN}✅ Redis服务器已停止${NC}" || echo -e "${YELLOW}⚠️ Redis服务器未运行${NC}"
fi
echo -e "\n${GREEN}🎉 AutoClip 所有服务已停止！${NC}"

208
启动脚本说明.md Normal file
View File

@@ -0,0 +1,208 @@
# AutoClip 启动脚本说明
## 📋 脚本概览
本项目现在包含以下统一启动脚本，替代了之前分散的启动方式：
### 1. 🚀 `start_autoclip.sh` - 完整启动脚本 (推荐首次使用)
**功能**: 完整的开发环境启动脚本，包含所有检查和初始化
- ✅ 环境检查 (Python3, Node.js, Redis等)
- ✅ 依赖安装 (自动创建requirements.txt)
- ✅ 虚拟环境创建和激活
- ✅ 数据库初始化
- ✅ 配置文件创建 (.env)
- ✅ 启动所有服务
**使用场景**:
- 首次设置开发环境
- 长时间未使用后重新启动
- 需要完整环境检查时
```bash
./start_autoclip.sh
```
### 2. ⚡ `quick_start.sh` - 快速启动脚本 (日常开发推荐)
**功能**: 跳过检查，快速启动所有服务
- ✅ 快速启动所有核心服务
- ✅ 适用于已配置好的环境
- ✅ 启动速度更快
**使用场景**:
- 日常开发启动
- 环境已配置完毕
- 快速重启服务
```bash
./quick_start.sh
```
### 3. 🛑 `stop_autoclip.sh` - 停止脚本
**功能**: 安全停止所有AutoClip相关服务
- ✅ 停止后端API服务器
- ✅ 停止Celery工作进程
- ✅ 停止前端开发服务器
- ✅ 可选停止Redis服务器
```bash
./stop_autoclip.sh
```
## 🏗️ 启动的服务模块
### 核心服务
1. **Redis服务器** (端口: 6379)
- 作用: Celery消息代理和结果存储
- 自动检查并启动
2. **后端API服务器** (端口: 8000)
- 技术栈: FastAPI + SQLAlchemy
- 功能: REST API、WebSocket、数据库管理
- 地址: http://localhost:8000
- API文档: http://localhost:8000/docs
3. **Celery工作进程**
- 作用: 异步任务处理 (视频处理、文件上传等)
- 配置: 单进程模式，适合开发环境
4. **前端开发服务器** (端口: 5173)
- 技术栈: React + TypeScript + Vite
- 功能: 用户界面、状态管理(Zustand)、实时通信
- 地址: http://localhost:5173
### 集成功能
- **WebSocket实时通信**: 任务进度推送、状态更新
- **Zustand状态管理**: 前端全局状态管理
- **SQLite数据库**: 数据持久化存储
- **文件管理**: 上传、处理、输出管理
## 📁 项目结构说明
```
autoclip/
├── start_autoclip.sh # 完整启动脚本 ⭐
├── quick_start.sh # 快速启动脚本 ⭐
├── stop_autoclip.sh # 停止脚本 ⭐
├── start_celery_worker.sh # (已整合) 单独Celery启动
├── scripts/
│ ├── start_dev.sh # (已整合) 旧版开发启动
│ ├── start_backend.py # (已整合) 后端启动
│ └── run_tests.py # 测试运行脚本
├── backend/ # 后端代码
├── frontend/ # 前端代码
└── requirements.txt # Python依赖 (自动生成)
```
## 🔧 环境要求
### 必需组件
- **Python 3.8+**: 后端运行环境
- **Node.js 16+**: 前端开发环境
- **Redis**: 消息队列和缓存
- **Git**: 版本控制
### 自动安装的依赖
- Python虚拟环境和包
- Node.js前端依赖
- Redis (通过Homebrew，如果未安装)
## 🛠️ 使用流程
### 首次启动
```bash
# 1. 克隆项目
git clone <repository-url>
cd autoclip
# 2. 运行完整启动脚本
./start_autoclip.sh
# 3. 编辑配置文件 (如需要)
nano .env
```
### 日常开发
```bash
# 启动服务
./quick_start.sh
# 停止服务
./stop_autoclip.sh
```
## 📊 服务状态检查
### 检查服务是否运行
```bash
# 检查后端API
curl http://localhost:8000/api/v1/health
# 检查Redis
redis-cli ping
# 查看运行的进程
ps aux | grep -E "(uvicorn|celery|vite|redis)"
```
### 查看日志
```bash
# 后端日志
tail -f backend.log
# Celery日志 (在终端输出)
# 前端日志 (在终端输出)
```
## ⚠️ 常见问题
### 1. 端口被占用
```bash
# 查找占用端口的进程
lsof -i :8000 # 后端端口
lsof -i :5173 # 前端端口
lsof -i :6379 # Redis端口
# 停止占用进程
kill -9 <PID>
```
### 2. Redis连接失败
```bash
# 手动启动Redis
redis-server --port 6379
# 检查Redis状态
brew services list | grep redis
```
### 3. Python依赖问题
```bash
# 重新安装依赖
rm -rf venv .venv_deps_installed
./start_autoclip.sh
```
### 4. 前端启动失败
```bash
# 清理并重新安装
cd frontend
rm -rf node_modules package-lock.json
npm install
```
## 🎯 下一步
1. **API密钥配置**: 编辑`.env`文件，设置必要的API密钥
2. **测试功能**: 访问 http://localhost:5173 开始使用
3. **开发调试**: 使用 `./stop_autoclip.sh` 和 `./quick_start.sh` 进行快速开发迭代
## 🆘 技术支持
如遇问题，请检查：
1. 所有服务的启动日志
2. `.env`配置文件是否正确
3. 网络端口是否被占用
4. 系统依赖是否完整安装