PCNUSMSE/transcript_service

1"""音频转文字服务主应用程序2 3基于Gradio的音频转文字Web服务应用程序入口。4"""5 6import asyncio7import sys8import signal9import time10from pathlib import Path11from typing import Optional12 13# 添加项目根目录到Python路径14project_root = Path(__file__).parent15sys.path.insert(0, str(project_root))16 17# 加载环境变量18from dotenv import load_dotenv19load_dotenv(project_root / ".env")20 21from src.core.config import get_config, reload_config22from src.utils.logger import get_logger23from src.api.gradio_interface import get_gradio_interface24from src.core.task_manager import get_task_manager, TaskStatus25 26 27class TranscriptServiceApp:28    """音频转文字服务应用程序"""29    30    def __init__(self, environment: Optional[str] = None):31        """初始化应用程序32        33        Args:34            environment: 运行环境 (development/production)35        """36        # 加载配置37        if environment:38            self.config = reload_config(environment)39        else:40            self.config = get_config()41        42        # 初始化日志43        self.logger = get_logger("transcript_service.app")44        45        # 初始化界面46        self.gradio_interface = get_gradio_interface()47        48        # 添加健康检查端点49        self._setup_health_endpoint()50        51        # 运行状态52        self.is_running = False53        54        self.logger.info(f"应用程序初始化完成 - 环境: {self.config.environment}")55    56    def _setup_health_endpoint(self):57        """设置健康检查端点"""58        try:59            import gradio as gr60            61            def health_check():62                """健康检查函数"""63                import json64                import time65                66                health_data = {67                    "status": "healthy",68                    "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),69                    "environment": self.config.environment,70                    "version": self.config.app.version,71                    "uptime": time.time() - getattr(self, '_start_time', time.time()),72                    "services": {73                        "oss": self._check_oss_connection(),74                        "dashscope": self._check_dashscope_connection()75                    }76                }77                78                return json.dumps(health_data, indent=2, ensure_ascii=False)79            80            # 在Gradio应用中添加健康检查端点81            if hasattr(self.gradio_interface, 'app'):82                from fastapi.responses import JSONResponse83                84                @self.gradio_interface.app.get("/health")85                async def health_endpoint():86                    health_data = {87                        "status": "healthy",88                        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),89                        "environment": self.config.environment,90                        "version": self.config.app.version,91                        "uptime": time.time() - getattr(self, '_start_time', time.time()),92                        "services": {93                            "oss": self._check_oss_connection(),94                            "dashscope": self._check_dashscope_connection()95                        }96                    }97                    return JSONResponse(content=health_data)98            99        except Exception as e:100            self.logger.warning(f"设置健康检查端点失败: {e}")101    102    def _check_oss_connection(self) -> bool:103        """检查OSS连接"""104        try:105            if not (self.config.oss.access_key_id and self.config.oss.access_key_secret):106                return False107            108            import oss2109            auth = oss2.Auth(self.config.oss.access_key_id, self.config.oss.access_key_secret)110            service = oss2.Service(auth, "https://oss-cn-beijing.aliyuncs.com")111            112            # 简单的连接测试113            list(service.list_buckets(max_keys=1))114            return True115        except Exception:116            return False117    118    def _check_dashscope_connection(self) -> bool:119        """检查DashScope连接"""120        try:121            if not self.config.dashscope.api_key:122                return False123            124            # 简单的API key格式检查125            return self.config.dashscope.api_key.startswith("sk-")126        except Exception:127            return False128    129    def setup_signal_handlers(self):130        """设置信号处理器"""131        # 移除优雅关闭功能，允许应用直接终止132        pass133    134    def validate_environment(self) -> bool:135        """验证运行环境136        137        Returns:138            环境是否有效139        """140        try:141            # 检查必要的环境变量142            missing_vars = []143            if not self.config.oss.access_key_id:144                missing_vars.append("OSS_ACCESS_KEY_ID")145            if not self.config.oss.access_key_secret:146                missing_vars.append("OSS_ACCESS_KEY_SECRET")147            if not self.config.dashscope.api_key:148                missing_vars.append("DASHSCOPE_API_KEY")149            150            if missing_vars:151                self.logger.error(f"缺少必要的环境变量: {missing_vars}")152                return False153            154            # 检查目录权限155            logs_dir = self.config.get_logs_dir()156            temp_dir = self.config.get_temp_dir()157            158            for directory in [logs_dir, temp_dir]:159                if not directory.exists():160                    directory.mkdir(parents=True, exist_ok=True)161                162                # 测试写权限163                test_file = directory / ".write_test"164                try:165                    test_file.write_text("test")166                    test_file.unlink()167                except Exception as e:168                    self.logger.error(f"目录权限检查失败 {directory}: {str(e)}")169                    return False170            171            self.logger.info("环境验证通过")172            return True173            174        except Exception as e:175            self.logger.exception(f"环境验证失败: {str(e)}")176            return False177 178    def run(self, **launch_kwargs):179        """启动应用程序180        181        Args:182            **launch_kwargs: Gradio启动参数183        """184        try:185            # 设置信号处理器186            self.setup_signal_handlers()187            188            # 验证环境189            if not self.validate_environment():190                self.logger.error("环境验证失败，应用程序无法启动")191                sys.exit(1)192            193            # 启动应用194            self.is_running = True195            self._start_time = time.time()  # 记录启动时间196            self.logger.info("正在启动音频转文字服务...")197            198            # 启动Gradio界面199            self.gradio_interface.launch(**launch_kwargs)200 201        except OSError as e:202            if "address already in use" in str(e).lower():203                port = launch_kwargs.get('server_port', self.config.app.port)204                self.logger.warning(f"端口 {port} 已被占用。正在尝试使用一个可用的随机端口...")205                206                # 显式设置 server_port=None 来让 Gradio 自动查找可用端口207                launch_kwargs['server_port'] = None208                209                try:210                    # 再次尝试启动211                    self.gradio_interface.launch(**launch_kwargs)212                except Exception as final_e:213                    self.logger.exception(f"尝试使用随机端口后，应用程序启动仍然失败: {str(final_e)}")214                    sys.exit(1)215            else:216                self.logger.exception(f"启动时发生未处理的网络错误: {str(e)}")217                sys.exit(1)218        except KeyboardInterrupt:219            self.logger.info("接收到键盘中断信号")220            self.shutdown()221        except Exception as e:222            self.logger.exception(f"应用程序启动失败: {str(e)}")223            sys.exit(1)224    225    def shutdown(self):226        """关闭应用程序"""227        if not self.is_running:228            return229        230        self.logger.info("开始关闭应用程序...")231        self.is_running = False232        233        try:234            # 清理任务管理器235            task_manager = get_task_manager()236            237            # 取消所有待处理的任务238            pending_tasks = task_manager.get_tasks_by_status(TaskStatus.PENDING) 239            for task in pending_tasks:240                try:241                    loop = asyncio.get_running_loop()242                    asyncio.create_task(task_manager.cancel_task(task.id))243                except RuntimeError: # No running loop244                    asyncio.run(task_manager.cancel_task(task.id))245 246 247            # 等待正在处理的任务完成（最多等待30秒）248            active_tasks = (249                task_manager.get_tasks_by_status(TaskStatus.VALIDATING) +250                task_manager.get_tasks_by_status(TaskStatus.UPLOADING) +251                task_manager.get_tasks_by_status(TaskStatus.TRANSCRIBING)252            )253            254            if active_tasks:255                self.logger.info(f"等待 {len(active_tasks)} 个活跃任务完成...")256                # 这里可以添加更复杂的等待逻辑, 但为简单起见, 我们直接继续257            258            # 清理临时文件259            self.cleanup_temp_files()260            261            self.logger.info("应用程序已安全关闭")262            263        except Exception as e:264            self.logger.exception(f"关闭应用程序时发生错误: {str(e)}")265    266    def cleanup_temp_files(self):267        """清理临时文件"""268        try:269            temp_dir = self.config.get_temp_dir()270            if temp_dir.exists():271                for file_path in temp_dir.glob("*"):272                    if file_path.is_file():273                        file_path.unlink()274                self.logger.info("临时文件清理完成")275        except Exception as e:276            self.logger.warning(f"清理临时文件失败: {str(e)}")277    278    def get_app_info(self) -> dict:279        """获取应用程序信息280        281        Returns:282            应用程序信息字典283        """284        return {285            "name": self.config.app.name,286            "version": self.config.app.version,287            "environment": self.config.environment,288            "debug": self.config.app.debug,289            "host": self.config.app.host,290            "port": self.config.app.port,291            "is_running": self.is_running292        }293 294 295def create_app(environment: Optional[str] = None) -> TranscriptServiceApp:296    """创建应用程序实例297    298    Args:299        environment: 运行环境300        301    Returns:302        应用程序实例303    """304    return TranscriptServiceApp(environment)305 306 307def main():308    """主函数入口"""309    import argparse310    import os311    312    parser = argparse.ArgumentParser(description="音频转文字服务")313    parser.add_argument(314        "--env", 315        choices=["development", "production"],316        default=None,  # 改为None，从环境变量读取317        help="运行环境"318    )319    parser.add_argument(320        "--host",321        default=None,322        help="服务主机地址"323    )324    parser.add_argument(325        "--port",326        type=int,327        default=None,328        help="服务端口"329    )330    parser.add_argument(331        "--share",332        action="store_true",333        help="启用Gradio分享链接"334    )335    parser.add_argument(336        "--debug",337        action="store_true",338        help="启用调试模式"339    )340    341    args = parser.parse_args()342    343    # 从环境变量或命令行参数确定运行环境344    environment = args.env or os.getenv('ENVIRONMENT', 'production')345    346    # 创建应用347    app = create_app(environment)348    349    # 准备启动参数350    launch_kwargs = {351        'share': False,  # 生产环境禁用share352        'server_name': '0.0.0.0',  # Hugging Face Spaces 需要监听所有接口353        'server_port': 7860  # Hugging Face Spaces 默认端口354    }355    356    # 命令行参数可以覆盖默认值357    if args.host:358        launch_kwargs['server_name'] = args.host359    if args.port:360        launch_kwargs['server_port'] = args.port361    if args.share:362        launch_kwargs['share'] = True  # 如果用户明确要求share363    if args.debug:364        launch_kwargs['debug'] = True365    366    # 启动应用367    app.run(**launch_kwargs)368 369 370if __name__ == "__main__":371    main()372 373