Upload folder using huggingface_hub

README.md

@@ -1,12 +1,46 @@
 ---
 title: Transcript Service
-emoji: 💻
-colorFrom: indigo
-colorTo: purple
+emoji: 🎙️
+colorFrom: blue
+colorTo: green
 sdk: gradio
-sdk_version: 5.49.1
+sdk_version: 5.9.1
 app_file: app.py
 pinned: false
+license: apache-2.0
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🎙️ 音频转文字服务
+
+基于 Gradio 的智能音频转文字 Web 服务。
+
+## ✨ 功能特点
+
+- 🎤 支持多种音频格式（MP3, WAV, M4A 等）
+- 📝 自动语音识别转文字
+- ☁️ 阿里云 OSS 云存储
+- 🤖 阿里云 DashScope API 支持
+- 🌐 简洁易用的 Web 界面
+
+## 🚀 使用方法
+
+1. 上传音频文件
+2. 选择语言（自动检测或手动指定）
+3. 点击"转换"按钮
+4. 等待处理完成
+5. 查看或下载转换结果
+
+## 🛠️ 技术栈
+
+- **前端**: Gradio 5.9.1
+- **后端**: Python 3.10
+- **存储**: 阿里云 OSS
+- **AI 服务**: 阿里云 DashScope
+
+## 📝 许可证
+
+Apache License 2.0
+
+---
+
+**部署在 Hugging Face Spaces** 🤗

app.py

@@ -0,0 +1,365 @@
+"""音频转文字服务主应用程序
+
+基于Gradio的音频转文字Web服务应用程序入口。
+"""
+
+import asyncio
+import sys
+import signal
+import time
+from pathlib import Path
+from typing import Optional
+
+# 添加项目根目录到Python路径
+project_root = Path(__file__).parent
+sys.path.insert(0, str(project_root))
+
+# 加载环境变量
+from dotenv import load_dotenv
+load_dotenv(project_root / ".env")
+
+from src.core.config import get_config, reload_config
+from src.utils.logger import get_logger
+from src.api.gradio_interface import get_gradio_interface
+from src.core.task_manager import get_task_manager, TaskStatus
+
+
+class TranscriptServiceApp:
+    """音频转文字服务应用程序"""
+    
+    def __init__(self, environment: Optional[str] = None):
+        """初始化应用程序
+        
+        Args:
+            environment: 运行环境 (development/production)
+        """
+        # 加载配置
+        if environment:
+            self.config = reload_config(environment)
+        else:
+            self.config = get_config()
+        
+        # 初始化日志
+        self.logger = get_logger("transcript_service.app")
+        
+        # 初始化界面
+        self.gradio_interface = get_gradio_interface()
+        
+        # 添加健康检查端点
+        self._setup_health_endpoint()
+        
+        # 运行状态
+        self.is_running = False
+        
+        self.logger.info(f"应用程序初始化完成 - 环境: {self.config.environment}")
+    
+    def _setup_health_endpoint(self):
+        """设置健康检查端点"""
+        try:
+            import gradio as gr
+            
+            def health_check():
+                """健康检查函数"""
+                import json
+                import time
+                
+                health_data = {
+                    "status": "healthy",
+                    "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
+                    "environment": self.config.environment,
+                    "version": self.config.app.version,
+                    "uptime": time.time() - getattr(self, '_start_time', time.time()),
+                    "services": {
+                        "oss": self._check_oss_connection(),
+                        "dashscope": self._check_dashscope_connection()
+                    }
+                }
+                
+                return json.dumps(health_data, indent=2, ensure_ascii=False)
+            
+            # 在Gradio应用中添加健康检查端点
+            if hasattr(self.gradio_interface, 'app'):
+                from fastapi.responses import JSONResponse
+                
+                @self.gradio_interface.app.get("/health")
+                async def health_endpoint():
+                    health_data = {
+                        "status": "healthy",
+                        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
+                        "environment": self.config.environment,
+                        "version": self.config.app.version,
+                        "uptime": time.time() - getattr(self, '_start_time', time.time()),
+                        "services": {
+                            "oss": self._check_oss_connection(),
+                            "dashscope": self._check_dashscope_connection()
+                        }
+                    }
+                    return JSONResponse(content=health_data)
+            
+        except Exception as e:
+            self.logger.warning(f"设置健康检查端点失败: {e}")
+    
+    def _check_oss_connection(self) -> bool:
+        """检查OSS连接"""
+        try:
+            if not (self.config.oss.access_key_id and self.config.oss.access_key_secret):
+                return False
+            
+            import oss2
+            auth = oss2.Auth(self.config.oss.access_key_id, self.config.oss.access_key_secret)
+            service = oss2.Service(auth, "https://oss-cn-beijing.aliyuncs.com")
+            
+            # 简单的连接测试
+            list(service.list_buckets(max_keys=1))
+            return True
+        except Exception:
+            return False
+    
+    def _check_dashscope_connection(self) -> bool:
+        """检查DashScope连接"""
+        try:
+            if not self.config.dashscope.api_key:
+                return False
+            
+            # 简单的API key格式检查
+            return self.config.dashscope.api_key.startswith("sk-")
+        except Exception:
+            return False
+    
+    def setup_signal_handlers(self):
+        """设置信号处理器"""
+        # 移除优雅关闭功能，允许应用直接终止
+        pass
+    
+    def validate_environment(self) -> bool:
+        """验证运行环境
+        
+        Returns:
+            环境是否有效
+        """
+        try:
+            # 检查必要的环境变量
+            missing_vars = []
+            if not self.config.oss.access_key_id:
+                missing_vars.append("OSS_ACCESS_KEY_ID")
+            if not self.config.oss.access_key_secret:
+                missing_vars.append("OSS_ACCESS_KEY_SECRET")
+            if not self.config.dashscope.api_key:
+                missing_vars.append("DASHSCOPE_API_KEY")
+            
+            if missing_vars:
+                self.logger.error(f"缺少必要的环境变量: {missing_vars}")
+                return False
+            
+            # 检查目录权限
+            logs_dir = self.config.get_logs_dir()
+            temp_dir = self.config.get_temp_dir()
+            
+            for directory in [logs_dir, temp_dir]:
+                if not directory.exists():
+                    directory.mkdir(parents=True, exist_ok=True)
+                
+                # 测试写权限
+                test_file = directory / ".write_test"
+                try:
+                    test_file.write_text("test")
+                    test_file.unlink()
+                except Exception as e:
+                    self.logger.error(f"目录权限检查失败 {directory}: {str(e)}")
+                    return False
+            
+            self.logger.info("环境验证通过")
+            return True
+            
+        except Exception as e:
+            self.logger.exception(f"环境验证失败: {str(e)}")
+            return False
+
+    def run(self, **launch_kwargs):
+        """启动应用程序
+        
+        Args:
+            **launch_kwargs: Gradio启动参数
+        """
+        try:
+            # 设置信号处理器
+            self.setup_signal_handlers()
+            
+            # 验证环境
+            if not self.validate_environment():
+                self.logger.error("环境验证失败，应用程序无法启动")
+                sys.exit(1)
+            
+            # 启动应用
+            self.is_running = True
+            self._start_time = time.time()  # 记录启动时间
+            self.logger.info("正在启动音频转文字服务...")
+            
+            # 启动Gradio界面
+            self.gradio_interface.launch(**launch_kwargs)
+
+        except OSError as e:
+            if "address already in use" in str(e).lower():
+                port = launch_kwargs.get('server_port', self.config.app.port)
+                self.logger.warning(f"端口 {port} 已被占用。正在尝试使用一个可用的随机端口...")
+                
+                # 显式设置 server_port=None 来让 Gradio 自动查找可用端口
+                launch_kwargs['server_port'] = None
+                
+                try:
+                    # 再次尝试启动
+                    self.gradio_interface.launch(**launch_kwargs)
+                except Exception as final_e:
+                    self.logger.exception(f"尝试使用随机端口后，应用程序启动仍然失败: {str(final_e)}")
+                    sys.exit(1)
+            else:
+                self.logger.exception(f"启动时发生未处理的网络错误: {str(e)}")
+                sys.exit(1)
+        except KeyboardInterrupt:
+            self.logger.info("接收到键盘中断信号")
+            self.shutdown()
+        except Exception as e:
+            self.logger.exception(f"应用程序启动失败: {str(e)}")
+            sys.exit(1)
+    
+    def shutdown(self):
+        """关闭应用程序"""
+        if not self.is_running:
+            return
+        
+        self.logger.info("开始关闭应用程序...")
+        self.is_running = False
+        
+        try:
+            # 清理任务管理器
+            task_manager = get_task_manager()
+            
+            # 取消所有待处理的任务
+            pending_tasks = task_manager.get_tasks_by_status(TaskStatus.PENDING) 
+            for task in pending_tasks:
+                try:
+                    loop = asyncio.get_running_loop()
+                    asyncio.create_task(task_manager.cancel_task(task.id))
+                except RuntimeError: # No running loop
+                    asyncio.run(task_manager.cancel_task(task.id))
+
+
+            # 等待正在处理的任务完成（最多等待30秒）
+            active_tasks = (
+                task_manager.get_tasks_by_status(TaskStatus.VALIDATING) +
+                task_manager.get_tasks_by_status(TaskStatus.UPLOADING) +
+                task_manager.get_tasks_by_status(TaskStatus.TRANSCRIBING)
+            )
+            
+            if active_tasks:
+                self.logger.info(f"等待 {len(active_tasks)} 个活跃任务完成...")
+                # 这里可以添加更复杂的等待逻辑, 但为简单起见, 我们直接继续
+            
+            # 清理临时文件
+            self.cleanup_temp_files()
+            
+            self.logger.info("应用程序已安全关闭")
+            
+        except Exception as e:
+            self.logger.exception(f"关闭应用程序时发生错误: {str(e)}")
+    
+    def cleanup_temp_files(self):
+        """清理临时文件"""
+        try:
+            temp_dir = self.config.get_temp_dir()
+            if temp_dir.exists():
+                for file_path in temp_dir.glob("*"):
+                    if file_path.is_file():
+                        file_path.unlink()
+                self.logger.info("临时文件清理完成")
+        except Exception as e:
+            self.logger.warning(f"清理临时文件失败: {str(e)}")
+    
+    def get_app_info(self) -> dict:
+        """获取应用程序信息
+        
+        Returns:
+            应用程序信息字典
+        """
+        return {
+            "name": self.config.app.name,
+            "version": self.config.app.version,
+            "environment": self.config.environment,
+            "debug": self.config.app.debug,
+            "host": self.config.app.host,
+            "port": self.config.app.port,
+            "is_running": self.is_running
+        }
+
+
+def create_app(environment: Optional[str] = None) -> TranscriptServiceApp:
+    """创建应用程序实例
+    
+    Args:
+        environment: 运行环境
+        
+    Returns:
+        应用程序实例
+    """
+    return TranscriptServiceApp(environment)
+
+
+def main():
+    """主函数入口"""
+    import argparse
+    
+    parser = argparse.ArgumentParser(description="音频转文字服务")
+    parser.add_argument(
+        "--env", 
+        choices=["development", "production"],
+        default="development",
+        help="运行环境"
+    )
+    parser.add_argument(
+        "--host",
+        default=None,
+        help="服务主机地址"
+    )
+    parser.add_argument(
+        "--port",
+        type=int,
+        default=None,
+        help="服务端口"
+    )
+    parser.add_argument(
+        "--share",
+        action="store_true",
+        help="启用Gradio分享链接"
+    )
+    parser.add_argument(
+        "--debug",
+        action="store_true",
+        help="启用调试模式"
+    )
+    
+    args = parser.parse_args()
+    
+    # 创建应用
+    app = create_app(args.env)
+    
+    # 准备启动参数
+    launch_kwargs = {
+        'share': False  # 生产环境禁用share
+    }
+    
+    if args.host:
+        launch_kwargs['server_name'] = args.host
+    if args.port:
+        launch_kwargs['server_port'] = args.port
+    if args.share:
+        launch_kwargs['share'] = True  # 如果用户明确要求share
+    if args.debug:
+        launch_kwargs['debug'] = True
+    
+    # 启动应用
+    app.run(**launch_kwargs)
+
+
+if __name__ == "__main__":
+    main()
+

config/environments/development.yaml

@@ -0,0 +1,41 @@
+# 开发环境配置
+app:
+  name: "音频转文字服务"
+  version: "1.0.0"
+  debug: true
+  host: "127.0.0.1"
+  port: 7860
+  max_file_size: 2147483648  # 2GB
+  max_files_count: 100
+  concurrent_tasks: 5
+
+# OSS配置
+oss:
+  endpoint: "oss-cn-beijing.aliyuncs.com"
+  bucket_name: "audio-transcript-dev"
+  upload_timeout: 300
+  url_expire_hours: 24
+  temp_prefix: "temp/audio"
+  auto_cleanup_days: 7
+
+# 阿里云百炼API配置
+dashscope:
+  base_url: "https://dashscope.aliyuncs.com/api/v1"
+  model: "paraformer-v2"
+  timeout: 300
+  max_retries: 3
+  retry_delay: 5
+  language_hints: ["zh", "en"]
+
+# 任务配置
+task:
+  status_check_interval: 2
+  max_processing_time: 3600  # 1小时
+  queue_size: 1000
+
+# 日志配置
+logging:
+  level: "DEBUG"
+  format: "detailed"
+  file_max_size: "10MB"
+  backup_count: 5

config/environments/production.yaml

@@ -0,0 +1,41 @@
+# 生产环境配置
+app:
+  name: "音频转文字服务"
+  version: "1.0.0"
+  debug: false
+  host: "0.0.0.0"
+  port: 8080
+  max_file_size: 2147483648  # 2GB
+  max_files_count: 100
+  concurrent_tasks: 10
+
+# OSS配置
+oss:
+  endpoint: "oss-cn-beijing.aliyuncs.com"
+  bucket_name: "audio-transcript-prod"
+  upload_timeout: 300
+  url_expire_hours: 24
+  temp_prefix: "temp/audio"
+  auto_cleanup_days: 7
+
+# 阿里云百炼API配置
+dashscope:
+  base_url: "https://dashscope.aliyuncs.com/api/v1"
+  model: "paraformer-v2"
+  timeout: 300
+  max_retries: 5
+  retry_delay: 10
+  language_hints: ["zh", "en"]
+
+# 任务配置
+task:
+  status_check_interval: 5
+  max_processing_time: 3600  # 1小时
+  queue_size: 2000
+
+# 日志配置
+logging:
+  level: "INFO"
+  format: "structured"
+  file_max_size: "50MB"
+  backup_count: 10

config/logging.yaml

@@ -0,0 +1,65 @@
+# 日志系统配置
+version: 1
+disable_existing_loggers: false
+
+formatters:
+  detailed:
+    format: '[%(asctime)s] [%(levelname)s] [%(name)s] [%(task_id)s] %(message)s'
+    datefmt: '%Y-%m-%d %H:%M:%S'
+  
+  structured:
+    format: '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "module": "%(name)s", "task_id": "%(task_id)s", "message": "%(message)s"}'
+    datefmt: '%Y-%m-%d %H:%M:%S'
+  
+  simple:
+    format: '[%(levelname)s] %(message)s'
+
+handlers:
+  console:
+    class: logging.StreamHandler
+    level: DEBUG
+    formatter: detailed
+    stream: ext://sys.stdout
+  
+  file_handler:
+    class: logging.handlers.RotatingFileHandler
+    level: INFO
+    formatter: structured
+    filename: logs/app.log
+    maxBytes: 10485760  # 10MB
+    backupCount: 5
+    encoding: utf8
+  
+  error_file_handler:
+    class: logging.handlers.RotatingFileHandler
+    level: ERROR
+    formatter: detailed
+    filename: logs/error.log
+    maxBytes: 10485760  # 10MB
+    backupCount: 5
+    encoding: utf8
+
+loggers:
+  transcript_service:
+    level: DEBUG
+    handlers: [console, file_handler, error_file_handler]
+    propagate: false
+  
+  transcript_service.oss:
+    level: INFO
+    handlers: [console, file_handler]
+    propagate: false
+  
+  transcript_service.api:
+    level: INFO
+    handlers: [console, file_handler]
+    propagate: false
+  
+  transcript_service.task:
+    level: DEBUG
+    handlers: [console, file_handler]
+    propagate: false
+
+root:
+  level: WARNING
+  handlers: [console, file_handler]

pyproject.toml

@@ -0,0 +1,148 @@
+[project]
+name = "transcript-service"
+version = "1.0.0"
+description = "智能音频转文字Web服务"
+authors = [{name = "Your Name", email = "your.email@example.com"}]
+license = {text = "MIT"}
+readme = "README.md"
+requires-python = ">=3.9"
+
+dependencies = [
+    # 核心Web框架依赖
+    "gradio>=4.44.0",
+    "fastapi>=0.104.0", 
+    "uvicorn>=0.24.0",
+    
+    # 云服务集成依赖
+    "oss2>=2.18.0",
+    "dashscope>=1.14.0",
+    
+    # 数据处理核心依赖
+    "pydantic>=2.5.0",
+    "pydantic-settings>=2.1.0",
+    
+    # 文件处理工具依赖
+    "python-multipart>=0.0.6",
+    "python-magic>=0.4.27",
+    
+    # 配置管理依赖
+    "PyYAML>=6.0.1",
+    "python-dotenv>=1.0.0",
+    
+    # 日志和监控依赖
+    "structlog>=23.2.0",
+    "rich>=13.7.0",
+    
+    # HTTP客户端依赖
+    "httpx>=0.25.2",
+    "aiohttp>=3.9.0",
+    
+    # 命令行工具依赖
+    "click>=8.1.7",
+    "typer>=0.9.0",
+    
+    # 性能优化可选依赖
+    "orjson>=3.9.0",
+    "ujson>=5.8.0"
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=7.4.0",
+    "pytest-asyncio>=0.21.0",
+    "pytest-cov>=4.1.0",
+    "black>=23.11.0",
+    "flake8>=7.0.0",
+    "isort>=5.12.0",
+    "mypy>=1.7.0",
+    "pre-commit>=3.5.0"
+]
+
+[project.scripts]
+transcript-service = "app:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src"]
+
+[dependency-groups]
+dev = [
+    "pytest>=7.4.0",
+    "pytest-asyncio>=0.21.0", 
+    "pytest-cov>=4.1.0",
+    "black>=23.11.0",
+    "flake8>=7.0.0",
+    "isort>=5.12.0",
+    "mypy>=1.7.0",
+    "pre-commit>=3.5.0"
+]
+
+[tool.black]
+line-length = 88
+target-version = ['py39']
+include = '\.pyi?$'
+extend-exclude = '''
+/(
+  # directories
+  \.eggs
+  | \.git
+  | \.hg
+  | \.mypy_cache
+  | \.tox
+  | \.venv
+  | _build
+  | buck-out
+  | build
+  | dist
+)/
+'''
+
+[tool.isort]
+profile = "black"
+multi_line_output = 3
+line_length = 88
+known_first_party = ["src"]
+
+[tool.flake8]
+max-line-length = 88
+extend-ignore = ["E203", "W503"]
+exclude = [
+    ".git",
+    "__pycache__",
+    ".venv",
+    "build",
+    "dist",
+    "*.egg-info"
+]
+
+[tool.mypy]
+python_version = "3.9"
+warn_return_any = true
+warn_unused_configs = true
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+check_untyped_defs = true
+disallow_untyped_decorators = true
+no_implicit_optional = true
+warn_redundant_casts = true
+warn_unused_ignores = true
+warn_no_return = true
+warn_unreachable = true
+strict_equality = true
+
+[[tool.mypy.overrides]]
+module = ["gradio.*", "oss2.*", "dashscope.*"]
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+minversion = "7.0"
+addopts = "-ra -q --strict-markers"
+testpaths = ["tests"]
+markers = [
+    "slow: marks tests as slow (deselect with '-m \"not slow\"')",
+    "integration: marks tests as integration tests",
+    "unit: marks tests as unit tests"
+]

requirements.txt

@@ -0,0 +1,43 @@
+# 核心依赖
+gradio>=4.44.0
+fastapi>=0.104.0
+uvicorn>=0.24.0
+
+# 云服务依赖
+oss2>=2.18.0
+dashscope>=1.14.0
+
+# 数据处理依赖
+pydantic>=2.5.0
+pydantic-settings>=2.1.0
+
+# 文件处理依赖
+python-multipart>=0.0.6
+python-magic>=0.4.27
+
+# 配置管理
+PyYAML>=6.0.1
+python-dotenv>=1.0.0
+
+# 日志和监控
+structlog>=23.2.0
+rich>=13.7.0
+
+# HTTP客户端
+httpx>=0.25.2
+aiohttp>=3.9.0
+
+# 工具依赖
+click>=8.1.7
+typer>=0.9.0
+
+# 开发依赖
+pytest>=7.4.0
+pytest-asyncio>=0.21.0
+black>=23.11.0
+flake8>=6.1.0
+isort>=5.12.0
+
+# 可选依赖（用于性能优化）
+orjson>=3.9.0
+ujson>=5.8.0

src/__init__.py

@@ -0,0 +1,44 @@
+"""源代码主模块
+
+应用程序源代码的根模块，集成所有功能组件。
+"""
+
+# 导入核心模块
+from .core import (
+    Config, get_config, reload_config,
+    TaskManager, TaskStatus, TaskPriority, Task, get_task_manager, task_manager
+)
+
+# 导入服务模块
+from .services import (
+    FileValidator, get_file_validator, file_validator,
+    OSSService, get_oss_service, oss_service,
+    ParaformerService, get_paraformer_service, paraformer_service
+)
+
+# 导入工具模块
+from .utils import (
+    Logger, TaskLogger, get_logger, get_task_logger, logger
+)
+
+# 导入API模块
+from .api import (
+    GradioInterface, get_gradio_interface, create_demo_interface, gradio_interface
+)
+
+__all__ = [
+    # 核心模块
+    "Config", "get_config", "reload_config",
+    "TaskManager", "TaskStatus", "TaskPriority", "Task", "get_task_manager", "task_manager",
+    
+    # 服务模块
+    "FileValidator", "get_file_validator", "file_validator",
+    "OSSService", "get_oss_service", "oss_service", 
+    "ParaformerService", "get_paraformer_service", "paraformer_service",
+    
+    # 工具模块
+    "Logger", "TaskLogger", "get_logger", "get_task_logger", "logger",
+    
+    # API模块
+    "GradioInterface", "get_gradio_interface", "create_demo_interface", "gradio_interface"
+]

src/api/__init__.py

@@ -0,0 +1,13 @@
+"""API模块
+
+包含应用程序的API接口和用户界面。
+"""
+
+from .gradio_interface import GradioInterface, get_gradio_interface, create_demo_interface, gradio_interface
+
+__all__ = [
+    "GradioInterface",
+    "get_gradio_interface",
+    "create_demo_interface",
+    "gradio_interface"
+]

src/api/gradio_interface.py

@@ -0,0 +1,574 @@
+"""Gradio用户界面模块
+
+提供基于Gradio的Web界面，支持文件上传、进度显示和结果展示。
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple, Any
+import gradio as gr
+import pandas as pd
+
+from ..core.config import get_config
+from ..core.task_manager import get_task_manager, TaskStatus, TaskPriority
+from ..utils.logger import get_task_logger
+from ..services.file_validator import get_file_validator
+
+
+class GradioInterface:
+    """Gradio界面管理器"""
+    
+    def __init__(self):
+        """初始化Gradio界面"""
+        self.config = get_config()
+        self.task_manager = get_task_manager()
+        self.file_validator = get_file_validator()
+        self.logger = get_task_logger(logger_name="transcript_service.gradio")
+        
+        # 当前任务ID
+        self.current_task_id = None
+        
+        # 创建界面
+        self.interface = self._create_interface()
+        
+        # 注册任务状态回调
+        self.task_manager.add_status_callback(self._on_task_status_change)
+    
+    def _create_interface(self) -> gr.Blocks:
+        """创建Gradio界面"""
+        # 获取支持的格式信息
+        supported_formats = self.file_validator.get_supported_formats()
+        
+        with gr.Blocks(
+            title="音频转文字服务",
+            theme=gr.themes.Soft(),
+            css="""
+            .main-container { max-width: 1000px; margin: 0 auto; }
+            .upload-area { border: 2px dashed #ccc; border-radius: 10px; padding: 20px; text-align: center; }
+            .result-area { margin-top: 20px; }
+            .status-simple { font-size: 16px; font-weight: bold; }
+            """
+        ) as interface:
+            # 简洁标题
+            gr.Markdown("# 🎵 音频转文字服务")
+            
+            with gr.Row():
+                with gr.Column(scale=3):
+                    # 文件上传区
+                    file_upload = gr.File(
+                        label="📁 选择音频文件（支持多文件）",
+                        file_count="multiple",
+                        file_types=list(supported_formats['extensions']),
+                        height=120
+                    )
+                    
+                    # 简化的配置区
+                    with gr.Row():
+                        # 任务优先级
+                        priority_select = gr.Radio(
+                            label="优先级",
+                            choices=[("普通", "NORMAL"), ("高优先级", "HIGH")],
+                            value="NORMAL"
+                        )
+                    
+                    # 参数设置区（默认隐藏）
+                    with gr.Accordion("⚙️ 转录参数设置", open=False) as params_section:
+                        # 语言选择
+                        language_select = gr.CheckboxGroup(
+                            label="识别语言",
+                            choices=[
+                                ("中文", "zh"), ("英文", "en"), ("日语", "ja"), 
+                                ("粤语", "yue"), ("韩语", "ko"), ("德语", "de"),
+                                ("法语", "fr"), ("俄语", "ru")
+                            ],
+                            value=["zh", "en"]
+                        )
+                        
+                        with gr.Row():
+                            # 基础选项
+                            disfluency_removal = gr.Checkbox(
+                                label="过滤语气词",
+                                value=True
+                            )
+                            timestamp_alignment = gr.Checkbox(
+                                label="时间戳校准",
+                                value=True
+                            )
+                            diarization_enabled = gr.Checkbox(
+                                label="说话人分离",
+                                value=True
+                            )
+                        
+                        with gr.Row():
+                            speaker_count = gr.Number(
+                                label="说话人数量（可选）",
+                                value=None,
+                                minimum=None,
+                                maximum=100,
+                                step=1,
+                                info="留空则自动判断，如需指定请输入2-100之间的数值"
+                            )
+                            channel_select = gr.Textbox(
+                                label="音轨索引",
+                                value="0",
+                                info="多音轨文件的音轨索引，用逗号分隔"
+                            )
+                        
+                        # 高级选项（更深层折叠）
+                        with gr.Accordion("高级选项", open=False):
+                            vocabulary_id = gr.Textbox(
+                                label="热词ID v2",
+                                value="",
+                                info="v2模型的热词ID"
+                            )
+                            phrase_id = gr.Textbox(
+                                label="热词ID v1",
+                                value="",
+                                info="v1模型的热词ID"
+                            )
+                            special_word_filter = gr.Textbox(
+                                label="敏感词过滤配置",
+                                value="",
+                                lines=2,
+                                placeholder='JSON格式配置',
+                                info="敏感词过滤的JSON配置"
+                            )
+                    
+                    # 控制按钮
+                    with gr.Row():
+                        start_btn = gr.Button("🚀 开始转录", variant="primary", size="lg")
+                        cancel_btn = gr.Button("❌ 取消", variant="secondary")
+                        clear_btn = gr.Button("🗑️ 清空", variant="secondary")
+                
+                with gr.Column(scale=2):
+                    # 简化的状态显示
+                    status_text = gr.Textbox(
+                        label="📊 当前状态",
+                        value="等待上传文件...",
+                        interactive=False,
+                        elem_classes=["status-simple"]
+                    )
+                    
+                    # 转录结果
+                    result_text = gr.Textbox(
+                        label="📝 转录结果",
+                        placeholder="转录结果将在这里显示...",
+                        lines=12,
+                        max_lines=20,
+                        show_copy_button=True,
+                        elem_classes=["result-area"]
+                    )
+                    
+                    # 文件统计表格
+                    stats_df = gr.Dataframe(
+                        headers=["文件名", "时长", "文本长度", "置信度"],
+                        datatype=["str", "str", "number", "number"],
+                        label="📈 处理统计",
+                        visible=False
+                    )
+            
+            # 折叠的详细信息区域
+            with gr.Accordion("📋 详细信息", open=False) as detail_section:
+                with gr.Tabs():
+                    with gr.Tab("系统信息"):
+                        system_info = gr.JSON(
+                            label="服务状态",
+                            value=self._get_system_info()
+                        )
+                        format_info = gr.JSON(
+                            label="支持格式",
+                            value=supported_formats
+                        )
+                    
+                    with gr.Tab("任务信息"):
+                        task_info = gr.JSON(
+                            label="当前任务",
+                            value={}
+                        )
+                    
+                    with gr.Tab("完整结果"):
+                        result_json = gr.JSON(
+                            label="JSON结果",
+                            value={}
+                        )
+                    
+                    with gr.Tab("处理日志"):
+                        log_text = gr.Textbox(
+                            label="详细日志",
+                            lines=8,
+                            max_lines=12,
+                            interactive=False,
+                            show_copy_button=True
+                        )
+                        log_download = gr.File(
+                            label="下载日志文件",
+                            visible=False
+                        )
+            
+
+            
+            # 添加手动刷新按钮
+            with gr.Row():
+                refresh_btn = gr.Button("🔄 刷新状态", variant="secondary", size="sm")
+                refresh_btn.click(
+                    fn=self._update_interface,
+                    outputs=[status_text, task_info, result_text, result_json, stats_df, system_info, log_text]
+                )
+            
+            # 事件处理
+            start_btn.click(
+                fn=self._process_files,
+                inputs=[
+                    file_upload, priority_select, language_select, 
+                    disfluency_removal, timestamp_alignment, diarization_enabled,
+                    speaker_count, channel_select, vocabulary_id, 
+                    phrase_id, special_word_filter
+                ],
+                outputs=[status_text, task_info, log_text]
+            )
+            
+            cancel_btn.click(
+                fn=self._cancel_current_task,
+                outputs=[status_text, task_info]
+            )
+            
+            clear_btn.click(
+                fn=self._clear_interface,
+                outputs=[file_upload, result_text, result_json, stats_df, log_text, status_text, task_info]
+            )
+            
+            # 定时更新
+            interface.load(
+                fn=self._update_interface,
+                outputs=[status_text, task_info, result_text, result_json, stats_df, system_info, log_text]
+            )
+        
+        return interface
+    
+    def _get_custom_css(self) -> str:
+        """获取自定义CSS样式"""
+        return """
+        .gradio-container {
+            max-width: 1200px !important;
+        }
+        .gr-button-primary {
+            background: linear-gradient(45deg, #FF6B6B, #4ECDC4) !important;
+            border: none !important;
+        }
+        .gr-button-primary:hover {
+            transform: translateY(-2px) !important;
+            box-shadow: 0 4px 12px rgba(0,0,0,0.15) !important;
+        }
+        .progress-bar {
+            background: linear-gradient(90deg, #FF6B6B, #4ECDC4) !important;
+        }
+        """
+    
+    def _get_system_info(self) -> Dict:
+        """获取系统信息"""
+        stats = self.task_manager.get_statistics()
+        return {
+            "服务状态": "运行中",
+            "当前任务数": stats['total_tasks'],
+            "待处理": stats['pending'],
+            "处理中": stats['validating'] + stats['uploading'] + stats['transcribing'],
+            "已完成": stats['completed'],
+            "失败": stats['failed'],
+            "队列大小": stats['queue_size']
+        }
+    
+    def _get_timestamp(self) -> str:
+        """获取当前时间戳"""
+        from datetime import datetime
+        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    
+    async def _process_files(
+        self, 
+        files: List, 
+        priority: str,
+        languages: List[str], 
+        disfluency_removal: bool,
+        timestamp_alignment: bool,
+        diarization_enabled: bool,
+        speaker_count: Optional[int] | None,
+        channel_id: str,
+        vocabulary_id: str,
+        phrase_id: str,
+        special_word_filter: str
+    ) -> Tuple[str, Dict, str]:
+        """处理上传的文件
+        
+        Args:
+            files: 上传的文件列表
+            languages: 选择的语言
+            priority: 任务优先级
+            channel_id: 音轨索引
+            disfluency_removal: 是否过滤语气词
+            timestamp_alignment: 是否启用时间戳校准
+            diarization_enabled: 是否启用说话人分离
+            speaker_count: 说话人数量参考值
+            vocabulary_id: 热词ID v2
+            phrase_id: 热词ID v1
+            special_word_filter: 敏感词过滤配置
+            
+        Returns:
+            (状态信息, 任务信息, 日志信息)
+        """
+        try:
+            if not files:
+                return "请先上传音频文件", {}, "错误: 未选择任何文件"
+            
+            # 记录详细日志
+            log_messages = []
+            log_messages.append(f"[{self._get_timestamp()}] 开始处理文件上传请求")
+            log_messages.append(f"[{self._get_timestamp()}] 接收到 {len(files)} 个文件")
+            
+            # 转换文件路径
+            file_paths = [Path(f.name) for f in files]
+            log_messages.append(f"[{self._get_timestamp()}] 转换文件路径完成")
+            
+            # 显示文件信息
+            for i, file_path in enumerate(file_paths):
+                try:
+                    file_size = file_path.stat().st_size
+                    log_messages.append(f"[{self._get_timestamp()}] 文件 {i+1}: {file_path.name} (大小: {file_size} 字节)")
+                except Exception as e:
+                    log_messages.append(f"[{self._get_timestamp()}] 文件 {i+1}: {file_path.name} (无法获取文件信息: {str(e)})")
+            
+            # 解析音轨参数
+            try:
+                channel_list = [int(x.strip()) for x in channel_id.split(',') if x.strip()]
+            except ValueError:
+                channel_list = [0]  # 默认为第一条音轨
+            
+            # 验证说话人数量参数
+            validated_speaker_count = None
+            if speaker_count is not None:
+                if isinstance(speaker_count, (int, float)) and speaker_count >= 2 and speaker_count <= 100:
+                    validated_speaker_count = int(speaker_count)
+                else:
+                    log_messages.append(f"[{self._get_timestamp()}] 警告: 说话人数量无效（{speaker_count}），将使用自动判断")
+            
+            # 解析敏感词过滤参数
+            special_filter = None
+            if special_word_filter.strip():
+                try:
+                    special_filter = json.loads(special_word_filter)
+                except json.JSONDecodeError as e:
+                    log_messages.append(f"[{self._get_timestamp()}] 警告: 敏感词过滤配置格式错误，将使���默认设置")
+            
+            # 创建任务
+            task_priority = TaskPriority.HIGH if priority == "HIGH" else TaskPriority.NORMAL
+            
+            # 准备元数据，包含所有Paraformer参数
+            metadata = {
+                "languages": languages,
+                "file_count": len(file_paths),
+                "paraformer_params": {
+                    "language_hints": languages,
+                    "channel_id": channel_list,
+                    "disfluency_removal_enabled": disfluency_removal,
+                    "timestamp_alignment_enabled": timestamp_alignment,
+                    "diarization_enabled": diarization_enabled,
+                    "speaker_count": validated_speaker_count,
+                    "vocabulary_id": vocabulary_id.strip() if vocabulary_id.strip() else None,
+                    "phrase_id": phrase_id.strip() if phrase_id.strip() else None,
+                    "special_word_filter": json.dumps(special_filter) if special_filter else None
+                }
+            }
+            
+            log_messages.append(f"[{self._get_timestamp()}] 创建任务，优先级: {task_priority.value}")
+            log_messages.append(f"[{self._get_timestamp()}] 选择语言: {', '.join(languages) if languages else '自动识别'}")
+            
+            self.current_task_id = await self.task_manager.create_task(
+                file_paths=file_paths,
+                priority=task_priority,
+                metadata=metadata
+            )
+            
+            task = self.task_manager.get_task(self.current_task_id)
+            
+            log_messages.append(f"[{self._get_timestamp()}] 任务创建成功，任务ID: {self.current_task_id}")
+            
+            return (
+                f"任务已创建: {self.current_task_id}",
+                task.to_dict() if task else {},
+                "\n".join(log_messages) + f"\n开始处理 {len(file_paths)} 个文件...\n"
+            )
+            
+        except Exception as e:
+            error_msg = f"创建任务失败: {str(e)}"
+            self.logger.exception(error_msg)
+            return error_msg, {}, f"错误: {error_msg}\n"
+    
+    def _cancel_current_task(self) -> Tuple[str, Dict]:
+        """取消当前任务"""
+        if not self.current_task_id:
+            return "没有正在执行的任务", {}
+        
+        success = asyncio.create_task(
+            self.task_manager.cancel_task(self.current_task_id)
+        )
+        
+        if success:
+            return f"任务 {self.current_task_id} 已取消", {}
+        else:
+            return "取消任务失败", {}
+    
+    def _clear_interface(self) -> Tuple[None, str, Dict, List, str, str, Dict]:
+        """清空界面"""
+        self.current_task_id = None
+        return (
+            None,  # file_upload
+            "",    # result_text
+            {},    # result_json
+            [],    # stats_df
+            "",    # log_text
+            "界面已清空，等待上传文件...",  # status_text
+            {}     # task_info
+        )
+    
+    def _update_interface(self) -> Tuple[str, Dict, str, Dict, List, Dict, str]:
+        """更新界面状态"""
+        # 更新当前任务状态
+        status_text = "等待上传文件..."
+        task_info = {}
+        result_text = ""
+        result_json = {}
+        stats_data = []
+        log_text = ""
+        
+        if self.current_task_id:
+            task = self.task_manager.get_task(self.current_task_id)
+            if task:
+                task_info = task.to_dict()
+                status_text = f"[{task.status.value}] {task.progress.message}"
+                
+                # 收集详细日志
+                log_text = self._collect_task_logs(task)
+                
+                # 如果任务完成，显示结果
+                if task.status == TaskStatus.COMPLETED:
+                    self.logger.debug(f"任务已完成，检查转录结果: {task.result.transcription_results}")
+                    if task.result.transcription_results:
+                        result_json = task.result.transcription_results
+                        
+                        # 提取转录文本
+                        transcriptions = result_json.get('transcriptions', [])
+                        self.logger.debug(f"转录结果: {transcriptions}")
+                        result_text = "\n\n".join([
+                            f"文件: {t.get('file_url', '').split('/')[-1]}\n{t.get('text', '')}"
+                            for t in transcriptions if t.get('text')
+                        ])
+                        
+                        # 生成统计表格
+                        stats_data = []
+                        for t in transcriptions:
+                            if 'error' not in t:
+                                stats_data.append([
+                                    t.get('file_url', '').split('/')[-1],
+                                    f"{t.get('duration', 0):.1f}s",
+                                    len(t.get('text', '')),
+                                    t.get('language', 'unknown'),
+                                    round(t.get('confidence', 0), 3)
+                                ])
+                    else:
+                        self.logger.debug("任务已完成但没有转录结果")
+                elif task.status == TaskStatus.FAILED:
+                    # 如果任务失败，显示错误信息
+                    if task.result and task.result.error_message:
+                        log_text += f"\n[{self._get_timestamp()}] 任务失败: {task.result.error_message}"
+        
+        # 更新系统信息
+        system_info = self._get_system_info()
+        
+        return status_text, task_info, result_text, result_json, stats_data, system_info, log_text
+    
+    def _collect_task_logs(self, task) -> str:
+        """收集任务的详细日志
+        
+        Args:
+            task: 任务对象
+            
+        Returns:
+            格式化的日志字符串
+        """
+        if not task:
+            return "无任务信息"
+        
+        log_lines = []
+        log_lines.append(f"[{self._get_timestamp()}] 任务ID: {task.id}")
+        log_lines.append(f"[{self._get_timestamp()}] 任务状态: {task.status.value}")
+        log_lines.append(f"[{self._get_timestamp()}] 任务创建时间: {task.created_at}")
+        
+        # 添加进度信息
+        if task.progress:
+            log_lines.append(f"[{self._get_timestamp()}] 进度信息: {task.progress.message}")
+            # TaskProgress对象没有details属性，只使用message
+        
+        # 添加文件信息
+        if hasattr(task, 'file_paths') and task.file_paths:
+            log_lines.append(f"[{self._get_timestamp()}] 文件列表:")
+            for i, file_path in enumerate(task.file_paths):
+                try:
+                    file_size = file_path.stat().st_size
+                    log_lines.append(f"  {i+1}. {file_path.name} ({file_size} bytes)")
+                except Exception as e:
+                    log_lines.append(f"  {i+1}. {file_path.name} (无法获取文件信息: {str(e)})")
+        
+        # 添加结果信息（如果任务已完成）
+        if task.status == TaskStatus.COMPLETED and task.result:
+            log_lines.append(f"[{self._get_timestamp()}] 任务完成时间: {task.completed_at}")
+            if hasattr(task.result, 'transcription_results') and task.result.transcription_results:
+                transcriptions = task.result.transcription_results.get('transcriptions', [])
+                log_lines.append(f"[{self._get_timestamp()}] 转录结果: {len(transcriptions)} 个文件")
+        
+        # 添加错误信息（如果有的话）
+        # Task对象没有error属性，错误信息在result中
+        
+        return "\n".join(log_lines)
+    
+    def _on_task_status_change(self, task):
+        """任务状态变化回调"""
+        self.logger.debug(f"任务状态变化: {task.id} -> {task.status.value}")
+        # 当任务状态变化时，不直接更新界面，而是依赖定时更新机制
+        # Gradio的回调中不能直接更新界面组件
+    
+    def launch(self, **kwargs):
+        """启动Gradio界面"""
+        default_kwargs = {
+            'server_name': '0.0.0.0',  # 改为0.0.0.0以允许外部访问
+            'server_port': self.config.app.port,
+            'share': True,  # 开启分享链接
+            'debug': self.config.app.debug,
+            'show_error': True,
+            'quiet': not self.config.app.debug
+        }
+        default_kwargs.update(kwargs)
+        
+        self.logger.info(f"启动Gradio界面: http://{default_kwargs['server_name']}:{default_kwargs['server_port']}")
+        
+        return self.interface.launch(**default_kwargs)
+
+
+# 全局界面实例
+gradio_interface = GradioInterface()
+
+
+def get_gradio_interface() -> GradioInterface:
+    """获取Gradio界面实例
+    
+    Returns:
+        Gradio界面实例
+    """
+    return gradio_interface
+
+
+def create_demo_interface() -> gr.Blocks:
+    """创建演示界面
+    
+    Returns:
+        Gradio界面对象
+    """
+    return gradio_interface.interface

src/core/__init__.py

@@ -0,0 +1,19 @@
+"""核心模块
+
+包含应用程序的核心功能和基础组件。
+"""
+
+from .config import Config, get_config, reload_config
+from .task_manager import TaskManager, TaskStatus, TaskPriority, Task, get_task_manager, task_manager
+
+__all__ = [
+    "Config",
+    "get_config", 
+    "reload_config",
+    "TaskManager",
+    "TaskStatus",
+    "TaskPriority", 
+    "Task",
+    "get_task_manager",
+    "task_manager"
+]

src/core/config.py

@@ -0,0 +1,171 @@
+"""配置管理模块
+
+提供应用程序配置的加载和管理功能。
+支持多环境配置和环境变量覆盖。
+"""
+
+import os
+from pathlib import Path
+from typing import Any, Dict, Optional
+import yaml
+from pydantic import Field
+from pydantic_settings import BaseSettings
+from pydantic_settings import SettingsConfigDict
+
+
+class AppConfig(BaseSettings):
+    """应用程序配置"""
+    model_config = SettingsConfigDict(
+        env_prefix="APP_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore"  # 忽略额外字段
+    )
+    
+    name: str = "音频转文字服务"
+    version: str = "1.0.0"
+    debug: bool = False
+    host: str = "127.0.0.1"
+    port: int = 7860
+    max_file_size: int = 2147483648  # 2GB
+    max_files_count: int = 100
+    concurrent_tasks: int = 5
+
+
+class OSSConfig(BaseSettings):
+    """OSS配置"""
+    model_config = SettingsConfigDict(
+        env_prefix="OSS_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore"
+    )
+    
+    endpoint: str = Field(..., description="OSS服务端点")
+    access_key_id: str = Field(..., description="访问密钥ID")
+    access_key_secret: str = Field(..., description="访问密钥密码")
+    bucket_name: str = Field(..., description="存储桶名称")
+    upload_timeout: int = 300
+    url_expire_hours: int = 24
+    temp_prefix: str = "temp/audio"
+    auto_cleanup_days: int = 7
+
+
+class DashScopeConfig(BaseSettings):
+    """阿里云百炼API配置"""
+    model_config = SettingsConfigDict(
+        env_prefix="DASHSCOPE_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore"
+    )
+    
+    api_key: str = Field(..., description="API密钥")
+    base_url: str = "https://dashscope.aliyuncs.com/api/v1"
+    model: str = "paraformer-v2"
+    timeout: int = 300
+    max_retries: int = 3
+    retry_delay: int = 5
+    language_hints: list[str] = ["zh", "en"]
+
+
+class TaskConfig(BaseSettings):
+    """任务配置"""
+    model_config = SettingsConfigDict(
+        env_prefix="TASK_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore"
+    )
+    
+    status_check_interval: int = 2
+    max_processing_time: int = 3600  # 1小时
+    queue_size: int = 1000
+
+
+class LoggingConfig(BaseSettings):
+    """日志配置"""
+    model_config = SettingsConfigDict(
+        env_prefix="LOGGING_",
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore"
+    )
+    
+    level: str = "INFO"
+    format: str = "structured"
+    file_max_size: str = "10MB"
+    backup_count: int = 5
+
+
+class Config:
+    """配置管理器"""
+    
+    def __init__(self, environment: Optional[str] = None):
+        """初始化配置管理器
+        
+        Args:
+            environment: 环境名称（development/production）
+        """
+        self.environment = environment or os.getenv("ENVIRONMENT", "development")
+        self._config_data = self._load_config()
+        
+        # 初始化各个配置模块
+        self.app = AppConfig(**self._config_data.get("app", {}))
+        
+        # OSS配置 - 直接创建实例以支持环境变量覆盖
+        self.oss = OSSConfig()
+        
+        # DashScope配置 - 直接创建实例以支持环境变量覆盖
+        self.dashscope = DashScopeConfig()
+        
+        self.task = TaskConfig(**self._config_data.get("task", {}))
+        self.logging = LoggingConfig(**self._config_data.get("logging", {}))
+    
+    def _load_config(self) -> Dict[str, Any]:
+        """加载配置文件"""
+        config_dir = Path(__file__).parent.parent.parent / "config" / "environments"
+        config_file = config_dir / f"{self.environment}.yaml"
+        
+        if not config_file.exists():
+            raise FileNotFoundError(f"配置文件不存在: {config_file}")
+        
+        with open(config_file, 'r', encoding='utf-8') as file:
+            return yaml.safe_load(file)
+    
+    def get_project_root(self) -> Path:
+        """获取项目根目录"""
+        return Path(__file__).parent.parent.parent
+    
+    def get_logs_dir(self) -> Path:
+        """获取日志目录"""
+        logs_dir = self.get_project_root() / "logs"
+        logs_dir.mkdir(exist_ok=True)
+        return logs_dir
+    
+    def get_temp_dir(self) -> Path:
+        """获取临时文件目录"""
+        temp_dir = self.get_project_root() / "temp"
+        temp_dir.mkdir(exist_ok=True)
+        return temp_dir
+
+
+# 全局配置实例
+config = Config()
+
+
+def get_config() -> Config:
+    """获取配置实例"""
+    return config
+
+
+def reload_config(environment: Optional[str] = None) -> Config:
+    """重新加载配置"""
+    global config
+    config = Config(environment)
+    return config

src/core/task_manager.py

@@ -0,0 +1,462 @@
+"""任务管理模块
+
+提供任务状态跟踪、进度管理和任务队列功能。
+"""
+
+import asyncio
+import time
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from enum import Enum
+from pathlib import Path
+from typing import Dict, List, Optional, Callable, Any
+from concurrent.futures import ThreadPoolExecutor
+
+from ..core.config import get_config
+from ..utils.logger import get_task_logger
+from ..services.file_validator import get_file_validator
+from ..services.oss_service import get_oss_service
+from ..services.paraformer_service import get_paraformer_service
+
+
+class TaskStatus(Enum):
+    """任务状态"""
+    PENDING = "pending"
+    VALIDATING = "validating"
+    UPLOADING = "uploading"
+    TRANSCRIBING = "transcribing"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+class TaskPriority(Enum):
+    """任务优先级"""
+    LOW = 1
+    NORMAL = 2
+    HIGH = 3
+    URGENT = 4
+
+
+@dataclass
+class TaskProgress:
+    """任务进度信息"""
+    stage: str = ""
+    current: int = 0
+    total: int = 100
+    message: str = ""
+    percentage: float = 0.0
+    
+    def update(self, current: int = None, total: int = None, message: str = None):
+        """更新进度信息"""
+        if current is not None:
+            self.current = current
+        if total is not None:
+            self.total = total
+        if message is not None:
+            self.message = message
+        
+        if self.total > 0:
+            self.percentage = min(100.0, (self.current / self.total) * 100)
+
+
+@dataclass
+class TaskResult:
+    """任务结果"""
+    success: bool = False
+    data: Optional[Dict] = None
+    error_message: Optional[str] = None
+    processed_files: List[str] = field(default_factory=list)
+    failed_files: List[str] = field(default_factory=list)
+    transcription_results: Optional[Dict] = None
+    duration: float = 0.0
+    
+    def to_dict(self) -> Dict:
+        """转换为字典格式"""
+        return {
+            'success': self.success,
+            'data': self.data,
+            'error_message': self.error_message,
+            'processed_files': self.processed_files,
+            'failed_files': self.failed_files,
+            'transcription_results': self.transcription_results,
+            'duration': self.duration
+        }
+
+
+@dataclass
+class Task:
+    """任务信息"""
+    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
+    status: TaskStatus = TaskStatus.PENDING
+    priority: TaskPriority = TaskPriority.NORMAL
+    file_paths: List[Path] = field(default_factory=list)
+    progress: TaskProgress = field(default_factory=TaskProgress)
+    result: TaskResult = field(default_factory=TaskResult)
+    created_at: datetime = field(default_factory=datetime.now)
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+    callback: Optional[Callable] = None
+    metadata: Dict[str, Any] = field(default_factory=dict)
+    
+    def to_dict(self) -> Dict:
+        """转换为字典格式"""
+        return {
+            'id': self.id,
+            'status': self.status.value,
+            'priority': self.priority.value,
+            'file_count': len(self.file_paths),
+            'file_names': [fp.name for fp in self.file_paths],
+            'progress': {
+                'stage': self.progress.stage,
+                'current': self.progress.current,
+                'total': self.progress.total,
+                'percentage': self.progress.percentage,
+                'message': self.progress.message
+            },
+            'result': self.result.to_dict(),
+            'created_at': self.created_at.isoformat() if self.created_at else None,
+            'started_at': self.started_at.isoformat() if self.started_at else None,
+            'completed_at': self.completed_at.isoformat() if self.completed_at else None,
+            'metadata': self.metadata
+        }
+
+
+class TaskManager:
+    """任务管理器"""
+    
+    def __init__(self):
+        """初始化任务管理器"""
+        self.config = get_config()
+        self.logger = get_task_logger(logger_name="transcript_service.task")
+        
+        # 任务存储
+        self.tasks: Dict[str, Task] = {}
+        self.task_queue: asyncio.Queue = asyncio.Queue(maxsize=self.config.task.queue_size)
+        
+        # 服务实例
+        self.file_validator = get_file_validator()
+        self.oss_service = get_oss_service()
+        self.paraformer_service = get_paraformer_service()
+        
+        # 工作线程池
+        self.executor = ThreadPoolExecutor(max_workers=self.config.app.concurrent_tasks)
+        
+        # 状态回调
+        self.status_callbacks: List[Callable] = []
+        
+        # 任务处理器状态
+        self._processor_started = False
+        
+        # 启动任务处理器
+        self._start_task_processor()
+    
+    def add_status_callback(self, callback: Callable):
+        """添加状态变化回调函数
+        
+        Args:
+            callback: 回调函数
+        """
+        self.status_callbacks.append(callback)
+    
+    def _notify_status_change(self, task: Task):
+        """通知状态变化"""
+        for callback in self.status_callbacks:
+            try:
+                callback(task)
+            except Exception as e:
+                self.logger.error(f"回调函数执行失败: {str(e)}")
+    
+    async def create_task(self, file_paths: List[Path], priority: TaskPriority = TaskPriority.NORMAL, metadata = None) -> str:
+        """创建新任务
+        
+        Args:
+            file_paths: 文件路径列表
+            priority: 任务优先级
+            metadata: 任务元数据
+            
+        Returns:
+            任务ID
+        """
+        # 确保任务处理器已启动
+        if not self._processor_started:
+            self._ensure_processor_started()
+            
+        task = Task(
+            file_paths=file_paths,
+            priority=priority,
+            metadata=metadata or {}
+        )
+        
+        self.tasks[task.id] = task
+        
+        # 添加到队列
+        await self.task_queue.put(task.id)
+        
+        self.logger.info(f"创建任务: {task.id}, 文件数量: {len(file_paths)}")
+        return task.id
+    
+    def get_task(self, task_id: str) -> Optional[Task]:
+        """获取任务信息
+        
+        Args:
+            task_id: 任务ID
+            
+        Returns:
+            任务对象
+        """
+        return self.tasks.get(task_id)
+    
+    def get_all_tasks(self) -> List[Task]:
+        """获取所有任务"""
+        return list(self.tasks.values())
+    
+    def get_tasks_by_status(self, status: TaskStatus) -> List[Task]:
+        """根据状态获取任务"""
+        return [task for task in self.tasks.values() if task.status == status]
+    
+    async def cancel_task(self, task_id: str) -> bool:
+        """取消任务
+        
+        Args:
+            task_id: 任务ID
+            
+        Returns:
+            是否成功取消
+        """
+        task = self.get_task(task_id)
+        if not task:
+            return False
+        
+        if task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED]:
+            return False
+        
+        task.status = TaskStatus.CANCELLED
+        task.completed_at = datetime.now()
+        task.progress.message = "任务已取消"
+        
+        self._notify_status_change(task)
+        self.logger.info(f"任务已取消: {task_id}")
+        return True
+    
+    def _start_task_processor(self):
+        """启动任务处理器"""
+        try:
+            # 只有在有运行的事件循环时才启动任务处理器
+            loop = asyncio.get_running_loop()
+            asyncio.create_task(self._process_tasks())
+        except RuntimeError:
+            # 没有运行的事件循环，延迟启动
+            self.logger.debug("没有运行的事件循环，任务处理器将在需要时启动")
+            self._processor_started = False
+        else:
+            self._processor_started = True
+    
+    def _ensure_processor_started(self):
+        """确保任务处理器已启动"""
+        if not self._processor_started:
+            try:
+                loop = asyncio.get_running_loop()
+                asyncio.create_task(self._process_tasks())
+                self._processor_started = True
+            except RuntimeError:
+                self.logger.warning("无法启动任务处理器：没有运行的事件循环")
+    
+    async def _process_tasks(self):
+        """处理任务队列"""
+        while True:
+            try:
+                # 从队列获取任务
+                task_id = await self.task_queue.get()
+                task = self.get_task(task_id)
+                
+                if not task or task.status == TaskStatus.CANCELLED:
+                    self.task_queue.task_done()
+                    continue
+                
+                # 处理任务
+                await self._execute_task(task)
+                self.task_queue.task_done()
+                
+            except Exception as e:
+                self.logger.exception(f"处理任务队列时发生错误: {str(e)}")
+                await asyncio.sleep(1)
+    
+    async def _execute_task(self, task: Task):
+        """执行任务
+        
+        Args:
+            task: 任务对象
+        """
+        try:
+            # 设置任务日志上下文
+            self.logger.set_task_id(task.id)
+            
+            task.status = TaskStatus.VALIDATING
+            task.started_at = datetime.now()
+            task.progress.stage = "文件验证"
+            task.progress.update(0, 100, "开始验证文件")
+            self._notify_status_change(task)
+            
+            # 1. 文件验证
+            valid_files, invalid_files = await self._validate_files(task)
+            if not valid_files:
+                task.status = TaskStatus.FAILED
+                task.result.error_message = "没有有效的文件"
+                task.result.failed_files = [str(f[0]) for f in invalid_files]
+                task.completed_at = datetime.now()
+                self._notify_status_change(task)
+                return
+            
+            # 2. 文件上传
+            task.status = TaskStatus.UPLOADING
+            task.progress.stage = "文件上传"
+            task.progress.update(0, len(valid_files), "开始上传文件到OSS")
+            self._notify_status_change(task)
+            
+            upload_results = await self._upload_files(task, valid_files)
+            successful_uploads = [r for r in upload_results if r[1]]
+            
+            if not successful_uploads:
+                task.status = TaskStatus.FAILED
+                task.result.error_message = "文件上传失败"
+                task.completed_at = datetime.now()
+                self._notify_status_change(task)
+                return
+            
+            # 3. 转录处理
+            task.status = TaskStatus.TRANSCRIBING
+            task.progress.stage = "语音转录"
+            task.progress.update(0, 100, "开始语音转录")
+            self._notify_status_change(task)
+            
+            file_urls = [r[2] for r in successful_uploads]
+            success, transcription_result, error = await self._transcribe_audio(task, file_urls)
+            
+            # 4. 完成任务
+            task.completed_at = datetime.now()
+            task.result.duration = (task.completed_at - task.started_at).total_seconds()
+            
+            if success:
+                task.status = TaskStatus.COMPLETED
+                task.result.success = True
+                task.result.transcription_results = transcription_result
+                task.result.processed_files = [r[0] for r in successful_uploads]
+                task.progress.update(100, 100, "转录完成")
+            else:
+                task.status = TaskStatus.FAILED
+                task.result.error_message = error
+            
+            self._notify_status_change(task)
+            
+        except Exception as e:
+            task.status = TaskStatus.FAILED
+            task.result.error_message = f"任务执行失败: {str(e)}"
+            task.completed_at = datetime.now()
+            self.logger.exception(f"执行任务时发生错误: {task.id}")
+            self._notify_status_change(task)
+        finally:
+            self.logger.clear_task_id()
+    
+    async def _validate_files(self, task: Task) -> tuple:
+        """验证文件"""
+        self.logger.info(f"开始验证 {len(task.file_paths)} 个文件")
+        
+        valid_files, invalid_files = self.file_validator.validate_multiple_files(task.file_paths)
+        
+        task.progress.update(100, 100, f"验证完成: {len(valid_files)} 个有效文件")
+        self.logger.info(f"文件验证完成: {len(valid_files)} 个有效文件, {len(invalid_files)} 个无效文件")
+        
+        return valid_files, invalid_files
+    
+    async def _upload_files(self, task: Task, file_paths: List[Path]) -> List[tuple]:
+        """上传文件"""
+        self.logger.info(f"开始上传 {len(file_paths)} 个文件")
+        
+        results = []
+        for i, file_path in enumerate(file_paths):
+            if task.status == TaskStatus.CANCELLED:
+                break
+            
+            success, url_or_error, object_key = await self.oss_service.upload_file(file_path, task.id)
+            results.append((file_path.name, success, url_or_error, object_key))
+            
+            # 更新进度
+            task.progress.update(i + 1, len(file_paths), f"已上传 {i + 1}/{len(file_paths)} 个文件")
+            self._notify_status_change(task)
+        
+        self.logger.info(f"文件上传完成: {len([r for r in results if r[1]])} 个成功")
+        return results
+    
+    async def _transcribe_audio(self, task: Task, file_urls: List[str]) -> tuple:
+        """转录音频"""
+        self.logger.info(f"开始转录 {len(file_urls)} 个音频文件")
+        
+        # 提取Paraformer参数
+        paraformer_params = None
+        if 'paraformer_params' in task.metadata:
+            paraformer_params = task.metadata['paraformer_params']
+            self.logger.info(f"使用自定义Paraformer参数: {paraformer_params}")
+        
+        success, results, error = await self.paraformer_service.batch_process_with_retry(
+            file_urls, task.id, paraformer_params
+        )
+        
+        if success:
+            task.progress.update(100, 100, "转录完成")
+            self.logger.info(f"转录完成: {len(file_urls)} 个文件")
+        else:
+            self.logger.error(f"转录失败: {error}")
+        
+        return success, results, error
+    
+    def cleanup_completed_tasks(self, hours: int = 24):
+        """清理已完成的任务
+        
+        Args:
+            hours: 保留时间（小时）
+        """
+        cutoff_time = datetime.now() - timedelta(hours=hours)
+        to_remove = []
+        
+        for task_id, task in self.tasks.items():
+            if (task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED] and
+                task.completed_at and task.completed_at < cutoff_time):
+                to_remove.append(task_id)
+        
+        for task_id in to_remove:
+            del self.tasks[task_id]
+        
+        self.logger.info(f"清理了 {len(to_remove)} 个过期任务")
+    
+    def get_statistics(self) -> Dict:
+        """获取任务统计信息"""
+        stats = {
+            'total_tasks': len(self.tasks),
+            'pending': len(self.get_tasks_by_status(TaskStatus.PENDING)),
+            'validating': len(self.get_tasks_by_status(TaskStatus.VALIDATING)),
+            'uploading': len(self.get_tasks_by_status(TaskStatus.UPLOADING)),
+            'transcribing': len(self.get_tasks_by_status(TaskStatus.TRANSCRIBING)),
+            'completed': len(self.get_tasks_by_status(TaskStatus.COMPLETED)),
+            'failed': len(self.get_tasks_by_status(TaskStatus.FAILED)),
+            'cancelled': len(self.get_tasks_by_status(TaskStatus.CANCELLED)),
+            'queue_size': self.task_queue.qsize()
+        }
+        return stats
+
+
+# 全局任务管理器实例
+task_manager = None
+
+
+def get_task_manager() -> TaskManager:
+    """获取任务管理器实例
+    
+    Returns:
+        任务管理器实例
+    """
+    global task_manager
+    if task_manager is None:
+        task_manager = TaskManager()
+    return task_manager

src/services/__init__.py

@@ -0,0 +1,20 @@
+"""服务模块
+
+包含应用程序的核心业务逻辑服务。
+"""
+
+from .file_validator import FileValidator, get_file_validator, file_validator
+from .oss_service import OSSService, get_oss_service, oss_service
+from .paraformer_service import ParaformerService, get_paraformer_service, paraformer_service
+
+__all__ = [
+    "FileValidator",
+    "get_file_validator",
+    "file_validator",
+    "OSSService", 
+    "get_oss_service",
+    "oss_service",
+    "ParaformerService",
+    "get_paraformer_service", 
+    "paraformer_service"
+]

src/services/file_validator.py

@@ -0,0 +1,277 @@
+"""文件验证模块
+
+提供音频文件格式验证、大小检查等功能。
+"""
+
+import magic
+from pathlib import Path
+from typing import List, Optional, Tuple
+import mimetypes
+
+from ..core.config import get_config
+from ..utils.logger import get_task_logger
+
+
+class FileValidator:
+    """文件验证器"""
+    
+    # 支持的音频文件格式
+    SUPPORTED_EXTENSIONS = {
+        '.aac', '.amr', '.avi', '.flac', '.flv', '.m4a', '.mkv', 
+        '.mov', '.mp3', '.mp4', '.mpeg', '.ogg', '.opus', '.wav', 
+        '.webm', '.wma', '.wmv'
+    }
+    
+    # 支持的MIME类型
+    SUPPORTED_MIME_TYPES = {
+        'audio/aac', 'audio/amr', 'audio/flac', 'audio/mp3', 'audio/mpeg',
+        'audio/mp4', 'audio/ogg', 'audio/opus', 'audio/wav', 'audio/webm',
+        'audio/x-wav', 'audio/x-flac', 'audio/x-m4a',
+        'video/mp4', 'video/avi', 'video/x-flv', 'video/quicktime',
+        'video/x-msvideo', 'video/webm', 'video/x-ms-wmv'
+    }
+    
+    def __init__(self):
+        """初始化文件验证器"""
+        self.config = get_config()
+        self.logger = get_task_logger(logger_name="transcript_service.validator")
+        
+        # 初始化libmagic
+        try:
+            self.magic = magic.Magic(mime=True)
+        except Exception as e:
+            self.logger.warning(f"无法初始化libmagic: {str(e)}, 将使用基础验证")
+            self.magic = None
+    
+    def validate_file(self, file_path: Path) -> Tuple[bool, Optional[str]]:
+        """验证单个文件
+        
+        Args:
+            file_path: 文件路径
+            
+        Returns:
+            (是否有效, 错误信息)
+        """
+        try:
+            # 检查文件是否存在
+            if not file_path.exists():
+                return False, f"文件不存在: {file_path}"
+            
+            # 检查是否是文件
+            if not file_path.is_file():
+                return False, f"不是有效的文件: {file_path}"
+            
+            # 检查文件大小
+            file_size = file_path.stat().st_size
+            if file_size == 0:
+                return False, f"文件为空: {file_path.name}"
+            
+            if file_size > self.config.app.max_file_size:
+                size_mb = file_size / (1024 * 1024)
+                max_size_mb = self.config.app.max_file_size / (1024 * 1024)
+                return False, f"文件大小 {size_mb:.1f}MB 超过限制 {max_size_mb:.1f}MB: {file_path.name}"
+            
+            # 检查文件扩展名
+            file_ext = file_path.suffix.lower()
+            if file_ext not in self.SUPPORTED_EXTENSIONS:
+                return False, f"不支持的文件格式 {file_ext}: {file_path.name}"
+            
+            # 检查MIME类型
+            if not self._check_mime_type(file_path):
+                return False, f"文件内容与扩展名不匹配: {file_path.name}"
+            
+            # 检查文件完整性
+            if not self._check_file_integrity(file_path):
+                return False, f"文件可能损坏或不完整: {file_path.name}"
+            
+            self.logger.info(f"文件验证通过: {file_path.name}")
+            return True, None
+            
+        except Exception as e:
+            error_msg = f"验证文件时发生错误: {file_path.name}, 错误: {str(e)}"
+            self.logger.exception(error_msg)
+            return False, error_msg
+    
+    def validate_multiple_files(self, file_paths: List[Path]) -> Tuple[List[Path], List[Tuple[Path, str]]]:
+        """验证多个文件
+        
+        Args:
+            file_paths: 文件路径列表
+            
+        Returns:
+            (有效文件列表, 无效文件列表[(文件路径, 错误信息)])
+        """
+        # 检查文件数量
+        if len(file_paths) > self.config.app.max_files_count:
+            self.logger.warning(f"文件数量 {len(file_paths)} 超过限制 {self.config.app.max_files_count}")
+        
+        valid_files = []
+        invalid_files = []
+        
+        for file_path in file_paths[:self.config.app.max_files_count]:
+            is_valid, error_msg = self.validate_file(file_path)
+            if is_valid:
+                valid_files.append(file_path)
+            else:
+                invalid_files.append((file_path, error_msg))
+        
+        # 如果超过限制，记录被跳过的文件
+        if len(file_paths) > self.config.app.max_files_count:
+            skipped_count = len(file_paths) - self.config.app.max_files_count
+            self.logger.warning(f"跳过了 {skipped_count} 个文件（超过批处理限制）")
+        
+        self.logger.info(f"文件验证完成: {len(valid_files)} 个有效文件, {len(invalid_files)} 个无效文件")
+        return valid_files, invalid_files
+    
+    def _check_mime_type(self, file_path: Path) -> bool:
+        """检查文件MIME类型
+        
+        Args:
+            file_path: 文件路径
+            
+        Returns:
+            MIME类型是否匹配
+        """
+        try:
+            # 使用libmagic检查
+            if self.magic:
+                mime_type = self.magic.from_file(str(file_path))
+                if mime_type in self.SUPPORTED_MIME_TYPES:
+                    return True
+            
+            # 使用mimetypes作为备选方案
+            mime_type, _ = mimetypes.guess_type(str(file_path))
+            if mime_type and mime_type in self.SUPPORTED_MIME_TYPES:
+                return True
+            
+            # 对于某些格式，检查文件头
+            return self._check_file_header(file_path)
+            
+        except Exception as e:
+            self.logger.warning(f"检查MIME类型时发生错误: {file_path.name}, 错误: {str(e)}")
+            # 如果MIME检查失败，只要扩展名正确就通过
+            return True
+    
+    def _check_file_header(self, file_path: Path) -> bool:
+        """检查文件头部特征
+        
+        Args:
+            file_path: 文件路径
+            
+        Returns:
+            文件头是否匹配
+        """
+        try:
+            with open(file_path, 'rb') as f:
+                header = f.read(16)
+            
+            if not header:
+                return False
+            
+            # 检查常见音频格式的文件头
+            if header.startswith(b'ID3') or header[4:8] == b'ftyp':  # MP3, MP4
+                return True
+            elif header.startswith(b'RIFF') and b'WAVE' in header:  # WAV
+                return True
+            elif header.startswith(b'fLaC'):  # FLAC
+                return True
+            elif header.startswith(b'OggS'):  # OGG
+                return True
+            elif header.startswith(b'\xff\xfb') or header.startswith(b'\xff\xfa'):  # MP3
+                return True
+            
+            # 如果无法识别文件头，但扩展名正确，就通过验证
+            return True
+            
+        except Exception as e:
+            self.logger.warning(f"检查文件头时发生错误: {file_path.name}, 错误: {str(e)}")
+            return True
+    
+    def _check_file_integrity(self, file_path: Path) -> bool:
+        """检查文件完整性
+        
+        Args:
+            file_path: 文件路径
+            
+        Returns:
+            文件是否完整
+        """
+        try:
+            # 基础完整性检查：确保文件可以完全读取
+            with open(file_path, 'rb') as f:
+                # 读取文件开头和结尾
+                f.read(1024)  # 读取前1KB
+                f.seek(-min(1024, file_path.stat().st_size), 2)  # 读取后1KB
+                f.read()
+            
+            return True
+            
+        except Exception as e:
+            self.logger.warning(f"检查文件完整性时发生错误: {file_path.name}, 错误: {str(e)}")
+            return False
+    
+    def get_file_info(self, file_path: Path) -> dict:
+        """获取文件信息
+        
+        Args:
+            file_path: 文件路径
+            
+        Returns:
+            文件信息字典
+        """
+        try:
+            stat = file_path.stat()
+            
+            # 获取MIME类型
+            mime_type = None
+            if self.magic:
+                try:
+                    mime_type = self.magic.from_file(str(file_path))
+                except:
+                    pass
+            
+            if not mime_type:
+                mime_type, _ = mimetypes.guess_type(str(file_path))
+            
+            return {
+                'name': file_path.name,
+                'size': stat.st_size,
+                'size_mb': round(stat.st_size / (1024 * 1024), 2),
+                'extension': file_path.suffix.lower(),
+                'mime_type': mime_type,
+                'modified_time': stat.st_mtime,
+                'is_supported': file_path.suffix.lower() in self.SUPPORTED_EXTENSIONS
+            }
+            
+        except Exception as e:
+            self.logger.error(f"获取文件信息失败: {file_path.name}, 错误: {str(e)}")
+            return {
+                'name': file_path.name,
+                'error': str(e)
+            }
+    
+    def get_supported_formats(self) -> dict:
+        """获取支持的文件格式信息
+        
+        Returns:
+            支持的格式信息
+        """
+        return {
+            'extensions': sorted(list(self.SUPPORTED_EXTENSIONS)),
+            'mime_types': sorted(list(self.SUPPORTED_MIME_TYPES)),
+            'max_file_size_mb': self.config.app.max_file_size / (1024 * 1024),
+            'max_files_count': self.config.app.max_files_count
+        }
+
+
+# 全局文件验证器实例
+file_validator = FileValidator()
+
+
+def get_file_validator() -> FileValidator:
+    """获取文件验证器实例
+    
+    Returns:
+        文件验证器实例
+    """
+    return file_validator

src/services/oss_service.py

@@ -0,0 +1,293 @@
+"""OSS云存储服务模块
+
+提供阿里云OSS文件上传、下载和管理功能。
+"""
+
+import os
+import uuid
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import List, Optional, Tuple
+import asyncio
+import aiohttp
+import oss2
+from oss2.exceptions import OssError
+
+from ..core.config import get_config
+from ..utils.logger import get_task_logger
+
+
+class OSSService:
+    """OSS云存储服务"""
+    
+    def __init__(self):
+        """初始化OSS服务"""
+        self.config = get_config()
+        self.oss_config = self.config.oss
+        
+        # 初始化OSS客户端
+        auth = oss2.Auth(
+            self.oss_config.access_key_id,
+            self.oss_config.access_key_secret
+        )
+        self.bucket = oss2.Bucket(
+            auth,
+            self.oss_config.endpoint,
+            self.oss_config.bucket_name
+        )
+        
+        self.logger = get_task_logger(logger_name="transcript_service.oss")
+    
+    def _generate_object_key(self, filename: str, task_id: str) -> str:
+        """生成OSS对象键名
+        
+        Args:
+            filename: 原始文件名
+            task_id: 任务ID
+            
+        Returns:
+            OSS对象键名
+        """
+        now = datetime.now()
+        date_path = now.strftime("%Y/%m/%d")
+        timestamp = now.strftime("%Y%m%d_%H%M%S")
+        
+        # 获取文件扩展名
+        file_ext = Path(filename).suffix
+        safe_filename = f"{timestamp}_{task_id}_{uuid.uuid4().hex[:8]}{file_ext}"
+        
+        return f"{self.oss_config.temp_prefix}/{date_path}/{safe_filename}"
+    
+    async def upload_file(self, file_path: Path, task_id: str) -> Tuple[bool, str, Optional[str]]:
+        """上传文件到OSS
+        
+        Args:
+            file_path: 本地文件路径
+            task_id: 任务ID
+            
+        Returns:
+            (是否成功, 公网URL或错误信息, 对象键名)
+        """
+        try:
+            self.logger.info(f"开始上传文件到OSS: {file_path.name}")
+            
+            # 生成对象键名
+            object_key = self._generate_object_key(file_path.name, task_id)
+            
+            # 上传文件并设置公共读取权限
+            try:
+                # 首先上传文件
+                self.bucket.put_object_from_file(object_key, str(file_path))
+                
+                # 设置对象ACL为公共读取
+                self.bucket.put_object_acl(object_key, oss2.OBJECT_ACL_PUBLIC_READ)
+                
+                # 生成公网访问URL
+                url = self._generate_public_url(object_key)
+                self.logger.info(f"文件上传成功: {object_key}, URL: {url}")
+                return True, url, object_key
+                
+            except oss2.exceptions.OssError as oss_err:
+                # 如果设置ACL失败，尝试使用签名URL
+                if 'public-read' in str(oss_err).lower():
+                    self.logger.warning(f"ACL设置失败，使用签名URL: {oss_err}")
+                    url = self._generate_signed_url(object_key)
+                    self.logger.info(f"文件上传成功: {object_key}, URL: {url}")
+                    return True, url, object_key
+                else:
+                    raise
+                
+        except OssError as e:
+            error_msg = f"OSS错误: {str(e)}"
+            self.logger.error(error_msg)
+            return False, error_msg, None
+        except Exception as e:
+            error_msg = f"上传文件时发生未知错误: {str(e)}"
+            self.logger.exception(error_msg)
+            return False, error_msg, None
+    
+    async def upload_multiple_files(self, file_paths: List[Path], task_id: str) -> List[Tuple[str, bool, str, Optional[str]]]:
+        """批量上传文件到OSS
+        
+        Args:
+            file_paths: 本地文件路径列表
+            task_id: 任务ID
+            
+        Returns:
+            [(文件名, 是否成功, URL或错误信息, 对象键名), ...]
+        """
+        results = []
+        
+        # 创建异步任务
+        tasks = []
+        for file_path in file_paths:
+            task = self._upload_single_file_async(file_path, task_id)
+            tasks.append((file_path.name, task))
+        
+        # 等待所有上传完成
+        for filename, task in tasks:
+            success, url_or_error, object_key = await task
+            results.append((filename, success, url_or_error, object_key))
+        
+        return results
+    
+    async def _upload_single_file_async(self, file_path: Path, task_id: str) -> Tuple[bool, str, Optional[str]]:
+        """异步上传单个文件"""
+        return await asyncio.get_event_loop().run_in_executor(
+            None, 
+            lambda: asyncio.run(self.upload_file(file_path, task_id))
+        )
+    
+    def _generate_public_url(self, object_key: str) -> str:
+        """生成公网访问URL
+        
+        Args:
+            object_key: OSS对象键名
+            
+        Returns:
+            公网访问URL
+        """
+        # 生成简单的公网访问URL（不带签名）
+        # 正确的格式: https://bucket-name.endpoint/object-key
+        # 注意: endpoint不能包含协议前缀
+        endpoint = self.oss_config.endpoint
+        if endpoint.startswith('http://'):
+            endpoint = endpoint[7:]
+        elif endpoint.startswith('https://'):
+            endpoint = endpoint[8:]
+        
+        # 构造公网URL - 注意这里的格式必须正确
+        url = f"https://{self.oss_config.bucket_name}.{endpoint}/{object_key}"
+        
+        # 记录生成的URL以便调试
+        self.logger.debug(f"生成公网URL: {url}")
+        
+        return url
+    
+    def _generate_signed_url(self, object_key: str) -> str:
+        """生成签名URL（备用方案）
+        
+        Args:
+            object_key: OSS对象键名
+            
+        Returns:
+            签名URL
+        """
+        # 生成有时效性的签名URL
+        expire_time = int((datetime.now() + timedelta(hours=self.oss_config.url_expire_hours)).timestamp())
+        url = self.bucket.sign_url('GET', object_key, expire_time)
+        return url
+    
+    def delete_file(self, object_key: str) -> bool:
+        """删除OSS文件
+        
+        Args:
+            object_key: OSS对象键名
+            
+        Returns:
+            是否删除成功
+        """
+        try:
+            self.bucket.delete_object(object_key)
+            self.logger.info(f"文件删除成功: {object_key}")
+            return True
+        except OssError as e:
+            self.logger.error(f"删除文件失败: {object_key}, 错误: {str(e)}")
+            return False
+        except Exception as e:
+            self.logger.exception(f"删除文件时发生未知错误: {object_key}, 错误: {str(e)}")
+            return False
+    
+    def cleanup_old_files(self, days: Optional[int] = None) -> int:
+        """清理过期的临时文件
+        
+        Args:
+            days: 保留天数，默认使用配置中的值
+            
+        Returns:
+            删除的文件数量
+        """
+        cleanup_days = days or self.oss_config.auto_cleanup_days
+        cutoff_date = datetime.now() - timedelta(days=cleanup_days)
+        
+        deleted_count = 0
+        prefix = self.oss_config.temp_prefix
+        
+        try:
+            # 列出所有临时文件
+            for obj in oss2.ObjectIterator(self.bucket, prefix=prefix):
+                # 检查文件最后修改时间
+                if obj.last_modified.replace(tzinfo=None) < cutoff_date:
+                    if self.delete_file(obj.key):
+                        deleted_count += 1
+            
+            self.logger.info(f"清理完成，删除了 {deleted_count} 个过期文件")
+            return deleted_count
+            
+        except Exception as e:
+            self.logger.exception(f"清理过期文件时发生错误: {str(e)}")
+            return deleted_count
+    
+    def get_file_info(self, object_key: str) -> Optional[dict]:
+        """获取文件信息
+        
+        Args:
+            object_key: OSS对象键名
+            
+        Returns:
+            文件信息字典
+        """
+        try:
+            info = self.bucket.head_object(object_key)
+            return {
+                'size': info.content_length,
+                'last_modified': info.last_modified,
+                'etag': info.etag,
+                'content_type': info.content_type
+            }
+        except OssError as e:
+            self.logger.error(f"获取文件信息失败: {object_key}, 错误: {str(e)}")
+            return None
+    
+    def check_bucket_exists(self) -> bool:
+        """检查存储桶是否存在
+        
+        Returns:
+            存储桶是否存在
+        """
+        try:
+            return self.bucket.bucket_exists()
+        except Exception as e:
+            self.logger.error(f"检查存储桶失败: {str(e)}")
+            return False
+    
+    def get_bucket_info(self) -> Optional[dict]:
+        """获取存储桶信息
+        
+        Returns:
+            存储桶信息
+        """
+        try:
+            info = self.bucket.get_bucket_info()
+            return {
+                'name': info.name,
+                'location': info.location,
+                'creation_date': info.creation_date,
+                'storage_class': info.storage_class
+            }
+        except Exception as e:
+            self.logger.error(f"获取存储桶信息失败: {str(e)}")
+            return None
+
+
+# 全局OSS服务实例
+oss_service = OSSService()
+
+
+def get_oss_service() -> OSSService:
+    """获取OSS服务实例
+    
+    Returns:
+        OSS服务实例
+    """
+    return oss_service

src/services/paraformer_service.py

@@ -0,0 +1,407 @@
+"""Paraformer转录服务模块
+
+提供阿里云百炼平台Paraformer-v2模型的语音转录功能。
+"""
+
+import asyncio
+import json
+import time
+from typing import Dict, List, Optional, Tuple
+from enum import Enum
+import httpx
+from dashscope import audio
+
+from ..core.config import get_config
+from ..utils.logger import get_task_logger
+
+
+class TaskStatus(Enum):
+    """任务状态枚举"""
+    PENDING = "PENDING"
+    RUNNING = "RUNNING"
+    SUCCEEDED = "SUCCEEDED"
+    FAILED = "FAILED"
+    CANCELLED = "CANCELLED"
+
+
+class ParaformerService:
+    """Paraformer转录服务"""
+    
+    def __init__(self):
+        """初始化Paraformer服务"""
+        self.config = get_config()
+        self.api_config = self.config.dashscope
+        self.logger = get_task_logger(logger_name="transcript_service.api")
+        
+        # 设置API密钥
+        audio.api_key = self.api_config.api_key
+    
+    async def submit_transcription_task(
+        self, 
+        file_urls: List[str], 
+        task_id: str, 
+        paraformer_params: Optional[Dict] = None
+    ) -> Tuple[bool, str, Optional[str]]:
+        """提交转录任务
+        
+        Args:
+            file_urls: 音频文件URL列表
+            task_id: 任务ID
+            paraformer_params: Paraformer额外参数
+            
+        Returns:
+            (是否成功, 消息, API任务ID)
+        """
+        try:
+            self.logger.info(f"提交转录任务: {len(file_urls)} 个文件")
+            
+            # 准备请求参数
+            transcription_params = {
+                'model': self.api_config.model,
+                'file_urls': file_urls
+            }
+            
+            # 添加额外参数（如果提供）
+            if paraformer_params:
+                # 语言提示
+                if 'language_hints' in paraformer_params and paraformer_params['language_hints']:
+                    transcription_params['language_hints'] = paraformer_params['language_hints']
+                else:
+                    transcription_params['language_hints'] = self.api_config.language_hints
+                
+                # 音轨选择
+                if 'channel_id' in paraformer_params and paraformer_params['channel_id']:
+                    transcription_params['channel_id'] = paraformer_params['channel_id']
+                
+                # 语气词过滤
+                if 'disfluency_removal_enabled' in paraformer_params:
+                    transcription_params['disfluency_removal_enabled'] = paraformer_params['disfluency_removal_enabled']
+                
+                # 时间戳校准
+                if 'timestamp_alignment_enabled' in paraformer_params:
+                    transcription_params['timestamp_alignment_enabled'] = paraformer_params['timestamp_alignment_enabled']
+                
+                # 说话人分离
+                if 'diarization_enabled' in paraformer_params:
+                    transcription_params['diarization_enabled'] = paraformer_params['diarization_enabled']
+                
+                # 说话人数量
+                if 'speaker_count' in paraformer_params and paraformer_params['speaker_count']:
+                    transcription_params['speaker_count'] = paraformer_params['speaker_count']
+                
+                # 热词ID v2
+                if 'vocabulary_id' in paraformer_params and paraformer_params['vocabulary_id']:
+                    transcription_params['vocabulary_id'] = paraformer_params['vocabulary_id']
+                
+                # 热词ID v1
+                if 'phrase_id' in paraformer_params and paraformer_params['phrase_id']:
+                    transcription_params['phrase_id'] = paraformer_params['phrase_id']
+                
+                # 敏感词过滤
+                if 'special_word_filter' in paraformer_params and paraformer_params['special_word_filter']:
+                    transcription_params['special_word_filter'] = paraformer_params['special_word_filter']
+            else:
+                # 使用默认配置
+                transcription_params['language_hints'] = self.api_config.language_hints
+            
+            # 记录最终参数用于调试
+            self.logger.info(f"转录参数: {transcription_params}")
+            
+            # 调用API
+            response = audio.asr.Transcription.async_call(**transcription_params)
+            
+            if response.status_code == 200:
+                api_task_id = response.output.task_id
+                self.logger.info(f"任务提交成功, API任务ID: {api_task_id}")
+                return True, f"任务提交成功", api_task_id
+            else:
+                error_msg = f"API调用失败, 状态码: {response.status_code}, 错误: {response.message}"
+                self.logger.error(error_msg)
+                return False, error_msg, None
+                
+        except Exception as e:
+            error_msg = f"提交转录任务时发生错误: {str(e)}"
+            self.logger.exception(error_msg)
+            return False, error_msg, None
+    
+    async def check_task_status(self, api_task_id: str) -> Tuple[TaskStatus, Optional[dict], Optional[str]]:
+        """检查任务状态
+        
+        Args:
+            api_task_id: API任务ID
+            
+        Returns:
+            (任务状态, 结果数据, 错误信息)
+        """
+        try:
+            response = audio.asr.Transcription.fetch(task=api_task_id)
+            
+            if response.status_code == 200:
+                task_status = TaskStatus(response.output.task_status)
+                
+                if task_status == TaskStatus.SUCCEEDED:
+                    # 解析转录结果
+                    results = await self._parse_transcription_results(response.output.results)
+                    return task_status, results, None
+                elif task_status == TaskStatus.FAILED:
+                    error_msg = getattr(response.output, 'message', '转录失败')
+                    return task_status, None, error_msg
+                else:
+                    # 任务进行中
+                    return task_status, None, None
+            else:
+                error_msg = f"检查任务状态失败: {response.message}"
+                self.logger.error(error_msg)
+                return TaskStatus.FAILED, None, error_msg
+                
+        except Exception as e:
+            error_msg = f"检查任务状态时发生错误: {str(e)}"
+            self.logger.exception(error_msg)
+            return TaskStatus.FAILED, None, error_msg
+    
+    async def process_audio_files(
+        self, 
+        file_urls: List[str], 
+        task_id: str, 
+        paraformer_params: Optional[Dict] = None
+    ) -> Tuple[bool, Optional[dict], Optional[str]]:
+        """处理音频文件转录（完整流程）
+        
+        Args:
+            file_urls: 音频文件URL列表
+            task_id: 任务ID
+            paraformer_params: Paraformer额外参数
+            
+        Returns:
+            (是否成功, 转录结果, 错误信息)
+        """
+        try:
+            # 保存原始URL映射，用于结果处理
+            self._original_urls = file_urls.copy()
+            self.logger.info(f"保存原始URL: {self._original_urls}")
+            
+            # 1. 提交任务
+            success, message, api_task_id = await self.submit_transcription_task(file_urls, task_id, paraformer_params)
+            if not success:
+                return False, None, message
+            
+            self.logger.info(f"开始监控任务状态: {api_task_id}")
+            
+            # 2. 监控任务状态
+            max_wait_time = self.api_config.timeout
+            check_interval = self.config.task.status_check_interval
+            start_time = time.time()
+            
+            while time.time() - start_time < max_wait_time:
+                status, results, error = await self.check_task_status(api_task_id)
+                
+                if status == TaskStatus.SUCCEEDED:
+                    self.logger.info(f"转录完成: {api_task_id}")
+                    return True, results, None
+                elif status == TaskStatus.FAILED:
+                    self.logger.error(f"转录失败: {api_task_id}, 错误: {error}")
+                    return False, None, error
+                elif status in [TaskStatus.PENDING, TaskStatus.RUNNING]:
+                    self.logger.debug(f"任务进行中: {api_task_id}, 状态: {status.value}")
+                    await asyncio.sleep(check_interval)
+                else:
+                    error_msg = f"未知任务状态: {status}"
+                    self.logger.error(error_msg)
+                    return False, None, error_msg
+            
+            # 超时
+            error_msg = f"任务超时: {api_task_id} (等待时间: {max_wait_time}秒)"
+            self.logger.error(error_msg)
+            return False, None, error_msg
+            
+        except Exception as e:
+            error_msg = f"处理音频文件时发生错误: {str(e)}"
+            self.logger.exception(error_msg)
+            return False, None, error_msg
+    
+    async def _parse_transcription_results(self, results: List) -> dict:
+        """解析转录结果
+        
+        Args:
+            results: API返回的结果列表
+            
+        Returns:
+            解析后的结果字典
+        """
+        parsed_results = {
+            'transcriptions': [],
+            'summary': {
+                'total_files': len(results),
+                'total_duration': 0,
+                'total_text_length': 0,
+                'languages_detected': set()
+            }
+        }
+        
+        for i, result in enumerate(results):
+            try:
+                # 使用原始URL而不是API返回的file_url
+                original_url = ''
+                if hasattr(self, '_original_urls') and i < len(self._original_urls):
+                    original_url = self._original_urls[i]
+                    self.logger.info(f"使用原始URL[{i}]: {original_url}")
+                else:
+                    original_url = result.get('file_url', '')
+                    self.logger.warning(f"未找到原始URL[{i}]，使用API返回的URL: {original_url}")
+                
+                # 从transcription_url下载实际的转录结果
+                transcription_text = ''
+                duration = 0
+                language = 'unknown'
+                confidence = 0
+                segments = []
+                
+                if result.get('subtask_status') == 'SUCCEEDED' and result.get('transcription_url'):
+                    try:
+                        # 下载转录结果
+                        async with httpx.AsyncClient() as client:
+                            response = await client.get(result['transcription_url'])
+                            if response.status_code == 200:
+                                transcription_data = response.json()
+                                # 根据实际返回的数据结构解析
+                                # 获取原始时长（毫秒）
+                                original_duration_ms = transcription_data.get('properties', {}).get('original_duration_in_milliseconds', 0)
+                                duration = original_duration_ms / 1000.0  # 转换为秒
+                                language = 'en'  # 根据测试设置默认为英语
+                                
+                                # 从transcripts中提取文本
+                                transcription_text = ''
+                                all_sentences = []
+                                transcripts = transcription_data.get('transcripts', [])
+                                if transcripts:
+                                    # 提取第一个transcript的文本
+                                    first_transcript = transcripts[0]
+                                    transcription_text = first_transcript.get('text', '')
+                                    
+                                    # 获取句子信息
+                                    all_sentences = first_transcript.get('sentences', [])
+                                
+                                # 计算置信度平均值（如果有句子信息）
+                                confidence = 0
+                                if all_sentences:
+                                    confidences = [sentence.get('confidence', 0) for sentence in all_sentences if 'confidence' in sentence]
+                                    if confidences:
+                                        confidence = sum(confidences) / len(confidences)
+                            else:
+                                self.logger.warning(f"下载转录结果失败，状态码: {response.status_code}")
+                                self.logger.warning(f"响应内容: {response.text}")
+                    except Exception as e:
+                        self.logger.warning(f"下载转录结果时发生错误: {str(e)}")
+                
+                transcription = {
+                    'file_url': original_url,
+                    'text': transcription_text,
+                    'duration': duration,
+                    'language': language,
+                    'confidence': confidence,
+                    'segments': segments
+                }
+                
+                # 如果需要调试，保存API返回的原始file_url
+                api_file_url = result.get('file_url', '')
+                if api_file_url and api_file_url != original_url:
+                    transcription['api_file_url'] = api_file_url
+                
+                parsed_results['transcriptions'].append(transcription)
+                
+                # 更新摘要信息
+                parsed_results['summary']['total_duration'] += transcription['duration']
+                parsed_results['summary']['total_text_length'] += len(transcription['text'])
+                parsed_results['summary']['languages_detected'].add(transcription['language'])
+                
+            except Exception as e:
+                self.logger.warning(f"解析单个转录结果时发生错误: {str(e)}")
+                # 添加错误的结果项
+                original_url = ''
+                if hasattr(self, '_original_urls') and i < len(self._original_urls):
+                    original_url = self._original_urls[i]
+                
+                parsed_results['transcriptions'].append({
+                    'file_url': original_url,
+                    'error': str(e),
+                    'raw_result': result
+                })
+        
+        # 转换语言集合为列表
+        parsed_results['summary']['languages_detected'] = list(parsed_results['summary']['languages_detected'])
+        
+        return parsed_results
+    
+    async def batch_process_with_retry(
+        self, 
+        file_urls: List[str], 
+        task_id: str, 
+        paraformer_params: Optional[Dict] = None
+    ) -> Tuple[bool, Optional[dict], Optional[str]]:
+        """批量处理音频文件（带重试机制）
+        
+        Args:
+            file_urls: 音频文件URL列表
+            task_id: 任务ID
+            paraformer_params: Paraformer额外参数
+            
+        Returns:
+            (是否成功, 转录结果, 错误信息)
+        """
+        max_retries = self.api_config.max_retries
+        retry_delay = self.api_config.retry_delay
+        
+        for attempt in range(max_retries + 1):
+            try:
+                success, results, error = await self.process_audio_files(file_urls, task_id, paraformer_params)
+                
+                if success:
+                    return True, results, None
+                
+                # 如果是最后一次重试，返回错误
+                if attempt == max_retries:
+                    return False, None, error
+                
+                # 等待后重试
+                self.logger.warning(f"第 {attempt + 1} 次尝试失败，{retry_delay} 秒后重试: {error}")
+                await asyncio.sleep(retry_delay * (attempt + 1))  # 递增延迟
+                
+            except Exception as e:
+                error_msg = f"重试过程中发生错误: {str(e)}"
+                self.logger.exception(error_msg)
+                
+                if attempt == max_retries:
+                    return False, None, error_msg
+                
+                await asyncio.sleep(retry_delay * (attempt + 1))
+        
+        return False, None, "重试次数已达上限"
+    
+    def get_service_info(self) -> dict:
+        """获取服务信息
+        
+        Returns:
+            服务配置信息
+        """
+        return {
+            'model': self.api_config.model,
+            'base_url': self.api_config.base_url,
+            'timeout': self.api_config.timeout,
+            'max_retries': self.api_config.max_retries,
+            'retry_delay': self.api_config.retry_delay,
+            'language_hints': self.api_config.language_hints,
+            'status_check_interval': self.config.task.status_check_interval
+        }
+
+
+# 全局Paraformer服务实例
+paraformer_service = ParaformerService()
+
+
+def get_paraformer_service() -> ParaformerService:
+    """获取Paraformer服务实例
+    
+    Returns:
+        Paraformer服务实例
+    """
+    return paraformer_service

src/utils/__init__.py

@@ -0,0 +1,34 @@
+"""工具模块
+
+包含应用程序的工具函数和辅助类。
+"""
+
+from .logger import Logger, TaskLogger, get_logger, get_task_logger, logger
+from .error_handler import (
+    ErrorCode, TranscriptServiceError, FileValidationError, NetworkError, 
+    APIError, OSSError, SystemError, RetryStrategy, ErrorHandler,
+    retry_async, retry_sync, safe_execute, safe_execute_async, get_error_handler, error_handler
+)
+
+__all__ = [
+    "Logger",
+    "TaskLogger", 
+    "get_logger",
+    "get_task_logger",
+    "logger",
+    "ErrorCode",
+    "TranscriptServiceError",
+    "FileValidationError",
+    "NetworkError",
+    "APIError", 
+    "OSSError",
+    "SystemError",
+    "RetryStrategy",
+    "ErrorHandler",
+    "retry_async",
+    "retry_sync",
+    "safe_execute",
+    "safe_execute_async",
+    "get_error_handler",
+    "error_handler"
+]

src/utils/error_handler.py

@@ -0,0 +1,380 @@
+"""错误处理和容错机制模块
+
+提供统一的错误处理、重试逻辑和异常恢复功能。
+"""
+
+import asyncio
+import functools
+import time
+from typing import Any, Callable, Dict, Optional, Type, Union
+from enum import Enum
+
+from ..core.config import get_config
+from ..utils.logger import get_task_logger
+
+
+class ErrorCode(Enum):
+    """错误代码"""
+    # 文件相关错误
+    FILE_NOT_FOUND = "FILE_001"
+    FILE_TOO_LARGE = "FILE_002"
+    FILE_FORMAT_UNSUPPORTED = "FILE_003"
+    FILE_CORRUPTED = "FILE_004"
+    
+    # 网络相关错误
+    NETWORK_TIMEOUT = "NET_001"
+    NETWORK_CONNECTION_ERROR = "NET_002"
+    NETWORK_DNS_ERROR = "NET_003"
+    
+    # API相关错误
+    API_KEY_INVALID = "API_001"
+    API_QUOTA_EXCEEDED = "API_002"
+    API_SERVICE_UNAVAILABLE = "API_003"
+    API_RATE_LIMITED = "API_004"
+    
+    # OSS相关错误
+    OSS_ACCESS_DENIED = "OSS_001"
+    OSS_BUCKET_NOT_FOUND = "OSS_002"
+    OSS_UPLOAD_FAILED = "OSS_003"
+    
+    # 系统相关错误
+    SYSTEM_OUT_OF_MEMORY = "SYS_001"
+    SYSTEM_DISK_FULL = "SYS_002"
+    SYSTEM_PERMISSION_DENIED = "SYS_003"
+    
+    # 通用错误
+    UNKNOWN_ERROR = "GEN_001"
+    TIMEOUT_ERROR = "GEN_002"
+    VALIDATION_ERROR = "GEN_003"
+
+
+class TranscriptServiceError(Exception):
+    """服务自定义异常基类"""
+    
+    def __init__(self, message: str, error_code: ErrorCode = ErrorCode.UNKNOWN_ERROR, details: Dict = None):
+        """初始化异常
+        
+        Args:
+            message: 错误消息
+            error_code: 错误代码
+            details: 额外详情
+        """
+        super().__init__(message)
+        self.message = message
+        self.error_code = error_code
+        self.details = details or {}
+        self.timestamp = time.time()
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典格式"""
+        return {
+            'error_code': self.error_code.value,
+            'message': self.message,
+            'details': self.details,
+            'timestamp': self.timestamp
+        }
+
+
+class FileValidationError(TranscriptServiceError):
+    """文件验证错误"""
+    pass
+
+
+class NetworkError(TranscriptServiceError):
+    """网络相关错误"""
+    pass
+
+
+class APIError(TranscriptServiceError):
+    """API调用错误"""
+    pass
+
+
+class OSSError(TranscriptServiceError):
+    """OSS操作错误"""
+    pass
+
+
+class SystemError(TranscriptServiceError):
+    """系统错误"""
+    pass
+
+
+class RetryStrategy:
+    """重试策略"""
+    
+    def __init__(
+        self,
+        max_attempts: int = 3,
+        base_delay: float = 1.0,
+        max_delay: float = 60.0,
+        exponential_base: float = 2.0,
+        jitter: bool = True
+    ):
+        """初始化重试策略
+        
+        Args:
+            max_attempts: 最大重试次数
+            base_delay: 基础延迟时间（秒）
+            max_delay: 最大延迟时间（秒）
+            exponential_base: 指数退避基数
+            jitter: 是否添加随机抖动
+        """
+        self.max_attempts = max_attempts
+        self.base_delay = base_delay
+        self.max_delay = max_delay
+        self.exponential_base = exponential_base
+        self.jitter = jitter
+    
+    def calculate_delay(self, attempt: int) -> float:
+        """计算延迟时间
+        
+        Args:
+            attempt: 当前尝试次数（从1开始）
+            
+        Returns:
+            延迟时间（秒）
+        """
+        delay = self.base_delay * (self.exponential_base ** (attempt - 1))
+        delay = min(delay, self.max_delay)
+        
+        if self.jitter:
+            import random
+            delay *= (0.5 + random.random() * 0.5)  # 添加±50%的随机抖动
+        
+        return delay
+
+
+class ErrorHandler:
+    """错误处理器"""
+    
+    def __init__(self):
+        """初始化错误处理器"""
+        self.config = get_config()
+        self.logger = get_task_logger(logger_name="transcript_service.error")
+        
+        # 错误分类映射
+        self.error_mapping = {
+            # 文件错误
+            FileNotFoundError: (FileValidationError, ErrorCode.FILE_NOT_FOUND),
+            PermissionError: (SystemError, ErrorCode.SYSTEM_PERMISSION_DENIED),
+            
+            # 网络错误
+            asyncio.TimeoutError: (NetworkError, ErrorCode.NETWORK_TIMEOUT),
+            ConnectionError: (NetworkError, ErrorCode.NETWORK_CONNECTION_ERROR),
+            
+            # 通用错误
+            ValueError: (TranscriptServiceError, ErrorCode.VALIDATION_ERROR),
+            RuntimeError: (TranscriptServiceError, ErrorCode.UNKNOWN_ERROR),
+        }
+        
+        # 可重试的错误类型
+        self.retryable_errors = {
+            ErrorCode.NETWORK_TIMEOUT,
+            ErrorCode.NETWORK_CONNECTION_ERROR,
+            ErrorCode.API_RATE_LIMITED,
+            ErrorCode.OSS_UPLOAD_FAILED,
+            ErrorCode.API_SERVICE_UNAVAILABLE
+        }
+    
+    def classify_error(self, error: Exception) -> TranscriptServiceError:
+        """分类和包装错误
+        
+        Args:
+            error: 原始异常
+            
+        Returns:
+            分类后的服务异常
+        """
+        if isinstance(error, TranscriptServiceError):
+            return error
+        
+        error_type = type(error)
+        if error_type in self.error_mapping:
+            exception_class, error_code = self.error_mapping[error_type]
+            return exception_class(str(error), error_code)
+        
+        # 根据错误消息内容进行分类
+        error_msg = str(error).lower()
+        
+        if "timeout" in error_msg:
+            return NetworkError(str(error), ErrorCode.NETWORK_TIMEOUT)
+        elif "permission denied" in error_msg:
+            return SystemError(str(error), ErrorCode.SYSTEM_PERMISSION_DENIED)
+        elif "api key" in error_msg:
+            return APIError(str(error), ErrorCode.API_KEY_INVALID)
+        elif "quota" in error_msg or "limit" in error_msg:
+            return APIError(str(error), ErrorCode.API_QUOTA_EXCEEDED)
+        else:
+            return TranscriptServiceError(str(error), ErrorCode.UNKNOWN_ERROR)
+    
+    def is_retryable(self, error: TranscriptServiceError) -> bool:
+        """判断错误是否可重试
+        
+        Args:
+            error: 服务异常
+            
+        Returns:
+            是否可重试
+        """
+        return error.error_code in self.retryable_errors
+    
+    def handle_error(self, error: Exception, context: str = "") -> TranscriptServiceError:
+        """处理错误
+        
+        Args:
+            error: 原始异常
+            context: 错误上下文
+            
+        Returns:
+            处理后的服务异常
+        """
+        classified_error = self.classify_error(error)
+        
+        # 记录错误日志
+        log_msg = f"错误处理 - {context}: {classified_error.message}"
+        if classified_error.error_code in [ErrorCode.UNKNOWN_ERROR, ErrorCode.SYSTEM_OUT_OF_MEMORY]:
+            self.logger.exception(log_msg)
+        else:
+            self.logger.error(log_msg)
+        
+        return classified_error
+
+
+# 全局错误处理器实例
+error_handler = ErrorHandler()
+
+
+def retry_async(
+    strategy: Optional[RetryStrategy] = None,
+    exceptions: tuple = (Exception,),
+    context: str = ""
+):
+    """异步函数重试装饰器
+    
+    Args:
+        strategy: 重试策略
+        exceptions: 需要重试的异常类型
+        context: 上下文信息
+    """
+    if strategy is None:
+        strategy = RetryStrategy()
+    
+    def decorator(func: Callable):
+        @functools.wraps(func)
+        async def wrapper(*args, **kwargs):
+            logger = get_task_logger(logger_name="transcript_service.retry")
+            
+            for attempt in range(1, strategy.max_attempts + 1):
+                try:
+                    return await func(*args, **kwargs)
+                except exceptions as e:
+                    classified_error = error_handler.classify_error(e)
+                    
+                    # 检查是否可重试
+                    if attempt == strategy.max_attempts or not error_handler.is_retryable(classified_error):
+                        logger.error(f"{context} 最终失败 (尝试 {attempt}/{strategy.max_attempts}): {str(e)}")
+                        raise classified_error
+                    
+                    # 计算延迟时间
+                    delay = strategy.calculate_delay(attempt)
+                    logger.warning(f"{context} 第 {attempt} 次尝试失败，{delay:.1f}秒后重试: {str(e)}")
+                    
+                    await asyncio.sleep(delay)
+            
+            # 理论上不会执行到这里
+            raise TranscriptServiceError("重试逻辑异常", ErrorCode.UNKNOWN_ERROR)
+        
+        return wrapper
+    return decorator
+
+
+def retry_sync(
+    strategy: Optional[RetryStrategy] = None,
+    exceptions: tuple = (Exception,),
+    context: str = ""
+):
+    """同步函数重试装饰器
+    
+    Args:
+        strategy: 重试策略
+        exceptions: 需要重试的异常类型
+        context: 上下文信息
+    """
+    if strategy is None:
+        strategy = RetryStrategy()
+    
+    def decorator(func: Callable):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            logger = get_task_logger(logger_name="transcript_service.retry")
+            
+            for attempt in range(1, strategy.max_attempts + 1):
+                try:
+                    return func(*args, **kwargs)
+                except exceptions as e:
+                    classified_error = error_handler.classify_error(e)
+                    
+                    # 检查是否可重试
+                    if attempt == strategy.max_attempts or not error_handler.is_retryable(classified_error):
+                        logger.error(f"{context} 最终失败 (尝试 {attempt}/{strategy.max_attempts}): {str(e)}")
+                        raise classified_error
+                    
+                    # 计算延迟时间
+                    delay = strategy.calculate_delay(attempt)
+                    logger.warning(f"{context} 第 {attempt} 次尝试失败，{delay:.1f}秒后重试: {str(e)}")
+                    
+                    time.sleep(delay)
+            
+            # 理论上不会执行到这里
+            raise TranscriptServiceError("重试逻辑异常", ErrorCode.UNKNOWN_ERROR)
+        
+        return wrapper
+    return decorator
+
+
+def safe_execute(func: Callable, *args, **kwargs) -> tuple[bool, Any, Optional[TranscriptServiceError]]:
+    """安全执行函数
+    
+    Args:
+        func: 要执行的函数
+        *args: 位置参数
+        **kwargs: 关键字参数
+        
+    Returns:
+        (是否成功, 结果或None, 错误或None)
+    """
+    try:
+        result = func(*args, **kwargs)
+        return True, result, None
+    except Exception as e:
+        error = error_handler.handle_error(e, f"执行 {func.__name__}")
+        return False, None, error
+
+
+async def safe_execute_async(func: Callable, *args, **kwargs) -> tuple[bool, Any, Optional[TranscriptServiceError]]:
+    """安全执行异步函数
+    
+    Args:
+        func: 要执行的异步函数
+        *args: 位置参数
+        **kwargs: 关键字参数
+        
+    Returns:
+        (是否成功, 结果或None, 错误或None)
+    """
+    try:
+        result = await func(*args, **kwargs)
+        return True, result, None
+    except Exception as e:
+        error = error_handler.handle_error(e, f"执行 {func.__name__}")
+        return False, None, error
+
+
+def get_error_handler() -> ErrorHandler:
+    """获取错误处理器实例
+    
+    Returns:
+        错误处理器实例
+    """
+    return error_handler

src/utils/logger.py

@@ -0,0 +1,260 @@
+"""日志管理模块
+
+提供结构化日志记录功能，支持任务跟踪和状态记录。
+"""
+
+import logging
+import logging.config
+import logging.handlers
+import uuid
+from pathlib import Path
+from typing import Any, Dict, Optional
+import yaml
+
+try:
+    from rich.console import Console
+    from rich.logging import RichHandler
+    RICH_AVAILABLE = True
+except ImportError:
+    RICH_AVAILABLE = False
+
+from ..core.config import get_config
+
+
+class TaskContextFilter(logging.Filter):
+    """任务上下文过滤器
+    
+    为日志记录添加任务ID上下文信息。
+    """
+    
+    def __init__(self):
+        super().__init__()
+        self.task_id = 'system'
+    
+    def filter(self, record):
+        """添加任务ID到日志记录"""
+        # 确保所有记录都有task_id字段
+        if not hasattr(record, 'task_id'):
+            record.task_id = getattr(self, 'task_id', 'system')
+        elif getattr(record, 'task_id', None) is None:
+            record.task_id = getattr(self, 'task_id', 'system')
+        return True
+
+
+class Logger:
+    """日志管理器"""
+    
+    def __init__(self, name: str = "transcript_service"):
+        """初始化日志管理器
+        
+        Args:
+            name: 日志器名称
+        """
+        self.name = name
+        self.config = get_config()
+        self._setup_logging()
+        self.logger = logging.getLogger(name)
+        self.task_filter = TaskContextFilter()
+        
+        # 为所有处理器添加任务过滤器
+        for handler in self.logger.handlers:
+            handler.addFilter(self.task_filter)
+            
+        # 同时为根日志器的处理器添加过滤器
+        root_logger = logging.getLogger()
+        for handler in root_logger.handlers:
+            if not any(isinstance(f, TaskContextFilter) for f in handler.filters):
+                handler.addFilter(self.task_filter)
+    
+    def _setup_logging(self):
+        """设置日志配置"""
+        # 确保日志目录存在
+        logs_dir = self.config.get_logs_dir()
+        
+        # 加载日志配置文件
+        config_file = self.config.get_project_root() / "config" / "logging.yaml"
+        
+        if config_file.exists():
+            with open(config_file, 'r', encoding='utf-8') as file:
+                logging_config = yaml.safe_load(file)
+            
+            # 更新文件路径为绝对路径
+            for handler_name, handler_config in logging_config.get('handlers', {}).items():
+                if 'filename' in handler_config:
+                    handler_config['filename'] = str(logs_dir / Path(handler_config['filename']).name)
+            
+            logging.config.dictConfig(logging_config)
+        else:
+            # 使用默认配置
+            self._setup_default_logging()
+    
+    def _setup_default_logging(self):
+        """设置默认日志配置"""
+        # 控制台处理器
+        if RICH_AVAILABLE:
+            console = Console()
+            console_handler = RichHandler(
+                console=console,
+                show_time=True,
+                show_path=True,
+                markup=True
+            )
+        else:
+            console_handler = logging.StreamHandler()
+            console_formatter = logging.Formatter(
+                '[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s',
+                datefmt='%Y-%m-%d %H:%M:%S'
+            )
+            console_handler.setFormatter(console_formatter)
+        
+        console_handler.setLevel(logging.DEBUG if self.config.app.debug else logging.INFO)
+        
+        # 文件处理器
+        log_file = self.config.get_logs_dir() / "app.log"
+        file_handler = logging.handlers.RotatingFileHandler(
+            log_file,
+            maxBytes=10*1024*1024,  # 10MB
+            backupCount=5,
+            encoding='utf-8'
+        )
+        file_handler.setLevel(logging.INFO)
+        
+        # 格式化器（简化版本）
+        formatter = logging.Formatter(
+            '[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s',
+            datefmt='%Y-%m-%d %H:%M:%S'
+        )
+        file_handler.setFormatter(formatter)
+        
+        # 配置根日志器
+        root_logger = logging.getLogger()
+        root_logger.setLevel(logging.DEBUG if self.config.app.debug else logging.INFO)
+        root_logger.addHandler(console_handler)
+        root_logger.addHandler(file_handler)
+    
+    def set_task_id(self, task_id: str):
+        """设置当前任务ID
+        
+        Args:
+            task_id: 任务ID
+        """
+        self.task_filter.task_id = task_id
+    
+    def clear_task_id(self):
+        """清除当前任务ID"""
+        self.task_filter.task_id = 'system'
+    
+    def debug(self, message: str, **kwargs):
+        """记录调试信息"""
+        self.logger.debug(message, extra=kwargs)
+    
+    def info(self, message: str, **kwargs):
+        """记录一般信息"""
+        self.logger.info(message, extra=kwargs)
+    
+    def warning(self, message: str, **kwargs):
+        """记录警告信息"""
+        self.logger.warning(message, extra=kwargs)
+    
+    def error(self, message: str, **kwargs):
+        """记录错误信息"""
+        self.logger.error(message, extra=kwargs)
+    
+    def critical(self, message: str, **kwargs):
+        """记录严重错误"""
+        self.logger.critical(message, extra=kwargs)
+    
+    def exception(self, message: str, **kwargs):
+        """记录异常信息（包含堆栈跟踪）"""
+        self.logger.exception(message, extra=kwargs)
+
+
+class TaskLogger:
+    """任务日志记录器
+    
+    为特定任务提供上下文日志记录。
+    """
+    
+    def __init__(self, task_id: Optional[str] = None, logger_name: str = "transcript_service"):
+        """初始化任务日志记录器
+        
+        Args:
+            task_id: 任务ID，如果为None则自动生成
+            logger_name: 基础日志器名称
+        """
+        self.task_id = task_id or str(uuid.uuid4())[:8]
+        self.logger = Logger(logger_name)
+        self.logger.set_task_id(self.task_id)
+    
+    def __enter__(self):
+        """进入上下文管理器"""
+        return self
+    
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """退出上下文管理器"""
+        self.logger.clear_task_id()
+    
+    def debug(self, message: str, **kwargs):
+        """记录调试信息"""
+        self.logger.debug(message, **kwargs)
+    
+    def info(self, message: str, **kwargs):
+        """记录一般信息"""
+        self.logger.info(message, **kwargs)
+    
+    def warning(self, message: str, **kwargs):
+        """记录警告信息"""
+        self.logger.warning(message, **kwargs)
+    
+    def error(self, message: str, **kwargs):
+        """记录错误信息"""
+        self.logger.error(message, **kwargs)
+    
+    def critical(self, message: str, **kwargs):
+        """记录严重错误"""
+        self.logger.critical(message, **kwargs)
+    
+    def exception(self, message: str, **kwargs):
+        """记录异常信息"""
+        self.logger.exception(message, **kwargs)
+    
+    def set_task_id(self, task_id: str):
+        """设置当前任务ID
+        
+        Args:
+            task_id: 任务ID
+        """
+        self.logger.set_task_id(task_id)
+    
+    def clear_task_id(self):
+        """清除当前任务ID"""
+        self.logger.clear_task_id()
+
+
+# 全局日志实例
+logger = Logger()
+
+
+def get_logger(name: str = "transcript_service") -> Logger:
+    """获取日志实例
+    
+    Args:
+        name: 日志器名称
+        
+    Returns:
+        日志实例
+    """
+    return Logger(name)
+
+
+def get_task_logger(task_id: Optional[str] = None, logger_name: str = "transcript_service") -> TaskLogger:
+    """获取任务日志实例
+    
+    Args:
+        task_id: 任务ID
+        logger_name: 日志器名称
+        
+    Returns:
+        任务日志实例
+    """
+    return TaskLogger(task_id, logger_name)