app.py

"""音频转文字服务主应用程序

基于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
    import os
    
    parser = argparse.ArgumentParser(description="音频转文字服务")
    parser.add_argument(
        "--env", 
        choices=["development", "production"],
        default=None,  # 改为None，从环境变量读取
        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()
    
    # 从环境变量或命令行参数确定运行环境
    environment = args.env or os.getenv('ENVIRONMENT', 'production')
    
    # 创建应用
    app = create_app(environment)
    
    # 准备启动参数
    launch_kwargs = {
        'share': False,  # 生产环境禁用share
        'server_name': '0.0.0.0',  # Hugging Face Spaces 需要监听所有接口
        'server_port': 7860  # Hugging Face Spaces 默认端口
    }
    
    # 命令行参数可以覆盖默认值
    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()