LQAgentPlatform/server/app.py

import os
import sys
import time
import signal
import uvicorn
import datetime
import traceback
import threading
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE_DIR)

from views import lifespan
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from foundation.base.config import config_handler
from foundation.logger.loggering import server_logger
from foundation.base.celery_app import app as celery_app

# 导入所有路由
from views.test_views import test_router
from views.construction_review.file_upload import file_upload_router
from views.construction_review.review_results import review_results_router
from views.construction_review.launch_review import launch_review_router


class ServerUtils:
    """服务器工具函数类 - 集中管理工具方法"""

    @staticmethod
    def get_redis_connection():
        """获取Redis连接的统一工具函数

        Returns:
            redis.Redis: Redis连接对象
        """
        import redis
        from foundation.base.config import config_handler

        # 从配置文件获取Redis连接参数
        redis_host = config_handler.get('redis', 'REDIS_HOST', 'localhost')
        redis_port = config_handler.get('redis', 'REDIS_PORT', '6379')
        redis_password = config_handler.get('redis', 'REDIS_PASSWORD', '')
        redis_db = config_handler.get('redis', 'REDIS_DB', '0')

        # 构建Redis URL
        if redis_password:
            redis_url = f'redis://:{redis_password}@{redis_host}:{redis_port}/{redis_db}'
        else:
            redis_url = f'redis://{redis_host}:{redis_port}/{redis_db}'

        return redis.from_url(redis_url, decode_responses=True)


class RouteManager:
    """路由管理类 - 负责路由配置和中间件设置"""

    def __init__(self, app: FastAPI):
        """初始化路由管理器

        Args:
            app: FastAPI应用实例
        """
        self.app = app
        self._setup_cors()
        self._setup_routes()
        self._setup_exception_handlers()
        self._setup_health_checks()


    def _setup_cors(self):
        """配置CORS中间件"""
        self.app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],  # 允许所有的来源
            allow_credentials=True,
            allow_methods=["*"],  # 允许的HTTP方法
            allow_headers=["*"],  # 允许的请求头
        )

    def _setup_routes(self):
        """配置所有路由"""
        self.app.include_router(test_router)
        self.app.include_router(file_upload_router)
        self.app.include_router(review_results_router)
        self.app.include_router(launch_review_router)

    def _setup_exception_handlers(self):
        """配置全局异常处理"""
        @self.app.exception_handler(HTTPException)
        async def http_exception_handler(request, exc):
            return JSONResponse(
                status_code=exc.status_code,
                content=exc.detail
            )

    def _setup_health_checks(self):
        """配置健康检查接口"""
        @self.app.get("/health")
        async def health_check():
            timestamp = datetime.datetime.now().isoformat()
            return {"status": "healthy", "timestamp": timestamp}

        @self.app.get("/celery/status")
        async def get_celery_status():
            """获取Celery Worker状态"""
            # 延迟导入避免循环引用
            from server.app import celery_manager
            status = celery_manager.get_status()
            return {
                "celery_worker": status,
                "timestamp": datetime.datetime.now().isoformat()
            }



class CeleryWorkerManager:
    """Celery Worker程序化管理器 - 独立的Celery管理模块"""

    def __init__(self, server_utils: ServerUtils = None):
        """初始化Celery Worker管理器

        Args:
            server_utils: 服务器工具类实例
        """
        self.worker = None
        self.is_running = False
        self.worker_thread = None
        self.shutdown_event = threading.Event()
        self.server_utils = server_utils or ServerUtils()

    def start_worker(self, **kwargs) -> bool:
        """启动Celery Worker

        Returns:
            bool: 启动是否成功
        """
        if self.is_running:
            server_logger.warning("Celery Worker已在运行")
            return True

        try:
            # 启动前清理残留任务
            self._cleanup_redis_tasks("启动前")

            # 创建Worker函数
            def run_celery_worker():
                try:
                    server_logger.info("Celery Worker开始运行...")
                    celery_app.worker_main(['worker'])
                except KeyboardInterrupt:
                    server_logger.info("收到停止信号，Celery Worker退出")
                except Exception as e:
                    server_logger.error(f"Celery Worker运行时出错: {e}")
                    server_logger.exception("详细错误信息:")
                finally:
                    self.is_running = False
                    server_logger.info("Celery Worker已停止")

            # 在单独线程中启动Worker
            self.worker_thread = threading.Thread(target=run_celery_worker, daemon=True)
            self.worker_thread.start()
            self.is_running = True

            # 等待启动
            time.sleep(2)

            success = self.is_running and self.worker_thread.is_alive()
            if success:
                server_logger.info("Celery Worker启动成功")
            else:
                server_logger.error("Celery Worker启动失败")
                self.is_running = False

            return success

        except ImportError as e:
            server_logger.error(f"导入Celery失败: {e}")
            server_logger.info("请先安装Celery: pip install celery redis")
            return False
        except Exception as e:
            server_logger.error(f"启动Celery Worker失败: {e}")
            server_logger.exception("详细错误信息:")
            return False

    def stop_worker(self, timeout: int = 5) -> bool:
        """优雅停止Celery Worker

        Args:
            timeout: 停止超时时间（秒）

        Returns:
            bool: 停止是否成功
        """
        if not self.is_running:
            server_logger.info("Celery Worker未运行")
            return True

        try:
            server_logger.info("停止Celery Worker...")
            self.shutdown_event.set()

            if self.worker_thread and self.worker_thread.is_alive():
                # 尝试优雅停止
                start_time = time.time()
                while self.is_running and (time.time() - start_time) < timeout:
                    time.sleep(0.1)

                if self.is_running:
                    server_logger.warning("Celery Worker优雅停止超时")
                else:
                    server_logger.info("Celery Worker已优雅停止")

            # 停止后清理Redis任务
            self._cleanup_redis_tasks("停止时")

            self.is_running = False
            self.shutdown_event.clear()
            return True

        except Exception as e:
            server_logger.error(f"停止Celery Worker失败: {e}")
            return False

    def stop_worker_immediately(self) -> bool:
        """立即停止Celery Worker，不等待

        Returns:
            bool: 停止是否成功
        """
        if not self.is_running:
            server_logger.info("Celery Worker未运行")
            return True

        try:
            server_logger.info("立即停止Celery Worker...")
            self.shutdown_event.set()

            # 发送中断信号
            if hasattr(os, 'kill'):
                try:
                    os.kill(os.getpid(), signal.SIGINT)
                    server_logger.info("已发送中断信号")
                except:
                    pass

            # 停止后清理Redis任务
            self._cleanup_redis_tasks("立即停止时")

            self.is_running = False
            self.shutdown_event.clear()
            server_logger.info("Celery Worker已立即停止")
            return True

        except Exception as e:
            server_logger.error(f"立即停止Celery Worker失败: {e}")
            self.is_running = False
            return False

    def get_status(self) -> dict:
        """获取Worker状态

        Returns:
            dict: 包含worker状态的字典
        """
        return {
            "is_running": self.is_running,
            "thread_alive": self.worker_thread.is_alive() if self.worker_thread else False,
        }

    def _cleanup_redis_tasks(self, phase: str):
        """清理Redis中的Celery任务

        Args:
            phase: 清理阶段（启动前/停止时/立即停止时）
        """
        try:
            r = self.server_utils.get_redis_connection()
            server_logger.info(f"{phase}清理Redis中的Celery任务...")

            # 清理任务相关键
            task_keys = r.keys('task:*')
            celery_meta_keys = r.keys('celery-task-meta-*')
            current_keys = r.keys('current:*')
            kombu_keys = r.keys('_kombu.binding.*')

            all_keys = task_keys + celery_meta_keys + current_keys + kombu_keys

            for key in all_keys:
                try:
                    r.delete(key)
                    server_logger.debug(f"{phase}清理: {key}")
                except Exception as e:
                    server_logger.warning(f"{phase}清理 {key} 失败: {e}")

            # 清理队列
            queues = ['celery', 'celery.pidbox', 'celeryev']
            for queue in queues:
                try:
                    r.delete(queue)
                    server_logger.debug(f"{phase}清理队列: {queue}")
                except Exception as e:
                    server_logger.warning(f"{phase}清理队列 {queue} 失败: {e}")

            if all_keys:
                server_logger.info(f"{phase}已清理 {len(all_keys)} 个Redis键")

        except Exception as e:
            server_logger.error(f"{phase}清理Redis任务失败: {e}")

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.stop_worker()

class ApplicationFactory:
    """应用工厂类 - 负责创建和配置FastAPI应用"""

    def __init__(self):
        """初始化应用工厂"""
        self.server_utils = ServerUtils()
        self.celery_manager = CeleryWorkerManager(self.server_utils)

    def create_app(self) -> FastAPI:
        """创建FastAPI应用实例

        Returns:
            FastAPI: 配置完成的应用实例
        """
        app = FastAPI(
            title="Agent API - 施工方案审查系统",
            version="0.3",
            description="Agent API - 集成施工方案审查功能",
            lifespan=lifespan
        )

        # 使用路由管理器配置应用
        route_manager = RouteManager(app)

        return app

    def create_server_config(self) -> dict:
        """创建服务器配置

        Returns:
            dict: 服务器配置字典
        """
        # 确保端口号是整数类型
        port = config_handler.get('launch', 'LAUNCH_PORT', 8002)
        try:
            port = int(port)
        except (ValueError, TypeError):
            port = 8002

        return {
            'host': config_handler.get('launch', 'HOST', '0.0.0.0'),
            'port': port,
            'reload': False,
            'with_celery': True
        }


# 全局实例
app_factory = ApplicationFactory()
celery_manager = app_factory.celery_manager

# 创建 FastAPI 应用
def create_app() -> FastAPI:
    """创建主应用服务"""
    app = FastAPI(
        title="Agent API - 施工方案审查系统",
        version="0.3",
        description="Agent API - 集成施工方案审查功能",
        lifespan=lifespan
    )

    # 添加 CORS 中间件
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],  # 允许所有的来源
        allow_credentials=True,
        allow_methods=["*"],  # 允许的HTTP方法
        allow_headers=["*"],  # 允许的请求头
    )

    # 添加所有路由
    app.include_router(test_router)
    app.include_router(file_upload_router)
    app.include_router(review_results_router)
    app.include_router(launch_review_router)

    # 全局异常处理
    @app.exception_handler(HTTPException)
    async def http_exception_handler(request, exc):
        return JSONResponse(
            status_code=exc.status_code,
            content=exc.detail
        )

    # 健康检查
    @app.get("/health")
    async def health_check():
        timestamp = datetime.datetime.now().isoformat()
        return {"status": "healthy", "timestamp": timestamp}

    # Celery状态检查
    @app.get("/celery/status")
    async def get_celery_status():
        """获取Celery Worker状态"""
        global celery_manager
        status = celery_manager.get_status()
        return {
            "celery_worker": status,
            "timestamp": datetime.datetime.now().isoformat()
        }

    return app


def cleanup_redis_before_start():
    """启动前清理Redis中的残留Celery任务"""
    try:
        # 使用统一的Redis连接工具函数
        r = get_redis_connection()

        server_logger.info("清理Redis中的残留Celery任务...")

        # 清理所有Celery相关的键，包括更多模式
        keys_to_delete = []
        for key in r.keys():
            key_lower = key.lower()
            # 扩展匹配模式，包括你遇到的实际键格式
            if any(keyword in key_lower for keyword in [
                'celery', 'task:', 'celery-task', 'kombu', 'current:'
            ]):
                keys_to_delete.append(key)
            # 匹配特定模式
            elif key.startswith('celery-task-meta-') or key.startswith('current:'):
                keys_to_delete.append(key)
            # 临时键
            elif key == 't_key':
                keys_to_delete.append(key)

        # 清理消息队列
        try:
            # 清理所有Celery队列
            queues = ['celery', 'celery.pidbox', 'celeryev']
            for queue in queues:
                # 删除队列
                r.delete(queue)
                server_logger.debug(f"已清理队列: {queue}")

            # 清理Kombu绑定
            kombu_keys = r.keys('_kombu.binding.*')
            for key in kombu_keys:
                r.delete(key)
                server_logger.debug(f"已清理Kombu绑定: {key}")

        except Exception as e:
            server_logger.warning(f"清理队列失败: {e}")

        # 清理识别到的键
        if keys_to_delete:
            for key in keys_to_delete:
                try:
                    r.delete(key)
                    server_logger.debug(f"已清理: {key}")
                except Exception as e:
                    server_logger.warning(f"清理 {key} 失败: {e}")

            server_logger.info(f"成功清理 {len(keys_to_delete)} 个Redis键")
        else:
            server_logger.info("没有发现需要清理的残留任务")

        # 额外检查：确保关键队列被清空
        try:
            # 使用FLUSHDB只清空Celery相关的数据，而不是整个数据库
            # 这里我们检查是否还有残留，如果有则进行更彻底的清理
            remaining_keys = []
            for key in r.keys():
                if any(pattern in key.lower() for pattern in ['celery', 'kombu']):
                    remaining_keys.append(key)

            if remaining_keys:
                server_logger.warning(f"发现 {len(remaining_keys)} 个残留键，进行彻底清理")
                for key in remaining_keys:
                    try:
                        r.delete(key)
                        server_logger.debug(f"彻底清理: {key}")
                    except Exception as e:
                        server_logger.warning(f"彻底清理 {key} 失败: {e}")
        except Exception as e:
            server_logger.warning(f"彻底清理检查失败: {e}")

        return True

    except Exception as e:
        server_logger.error(f"清理Redis残留任务失败: {e}")
        return False

def start_celery_worker_background():
    """在后台启动Celery Worker（异步方式）"""
    # 启动前清理残留任务
    cleanup_redis_before_start()

    # 添加调用栈调试
    server_logger.info("=== Celery Worker启动调用栈 ===")
    for line in traceback.format_stack():
        server_logger.debug(f"  {line.strip()}")
    server_logger.info("=== 调用栈结束 ===")

    return celery_manager.start_worker()

def stop_celery_worker():
    """停止Celery Worker"""
    global celery_manager

    # 立即取消所有任务注册（使用DB0，与启动时保持一致）
    try:
        # 使用统一的Redis连接工具函数
        r = get_redis_connection()

        server_logger.info("停止时清理Redis中的Celery任务...")

        # 清理任务相关键
        task_keys = r.keys('task:*')  # 重复任务检查器的数据
        celery_meta_keys = r.keys('celery-task-meta-*')
        current_keys = r.keys('current:*')
        kombu_keys = r.keys('_kombu.binding.*')

        all_keys = task_keys + celery_meta_keys + current_keys + kombu_keys

        for key in all_keys:
            try:
                r.delete(key)
                server_logger.debug(f"停止时清理: {key}")
            except Exception as e:
                server_logger.warning(f"停止时清理 {key} 失败: {e}")

        # 清理队列
        queues = ['celery', 'celery.pidbox', 'celeryev']
        for queue in queues:
            try:
                r.delete(queue)
                server_logger.debug(f"停止时清理队列: {queue}")
            except Exception as e:
                server_logger.warning(f"停止时清理队列 {queue} 失败: {e}")

        server_logger.info(f"停止时已清理 {len(all_keys)} 个Redis键")

    except Exception as e:
        server_logger.error(f"停止时清理Redis任务失败: {e}")

    # 立即停止Worker，不等待
    return celery_manager.stop_worker_immediately()

def run_server(host: str = None, port: int = None, reload: bool = False,
                with_celery: bool = True):
    """运行服务器"""

    # 从配置文件获取默认值
    if host is None:
        host = config_handler.get('launch', 'HOST', '0.0.0.0')
    if port is None:
        port = config_handler.get('launch', 'LAUNCH_PORT')

    if with_celery:
        # 启动Celery Worker
        start_celery_worker_background()

        # 注册退出时的清理函数
        import atexit
        atexit.register(stop_celery_worker)

        # 设置信号处理
        def signal_handler(signum, frame):
            server_logger.info(f"收到信号 {signum}，正在停止服务...")
            stop_celery_worker()
            sys.exit(0)

        # Windows和Unix系统的信号处理
        try:
            signal.signal(signal.SIGINT, signal_handler)  # Ctrl+C
            signal.signal(signal.SIGTERM, signal_handler)  # 终止信号
        except AttributeError:
            # Windows可能不支持某些信号
            pass

        # Windows特有的控制台事件处理
        if sys.platform == 'win32':
            try:
                import win32api
                def win32_handler(dwCtrlType):
                    # 正确的控制台事件常量
                    CTRL_C_EVENT = 0
                    CTRL_BREAK_EVENT = 1
                    CTRL_CLOSE_EVENT = 2
                    CTRL_SHUTDOWN_EVENT = 6

                    if dwCtrlType in (CTRL_C_EVENT, CTRL_BREAK_EVENT, CTRL_CLOSE_EVENT, CTRL_SHUTDOWN_EVENT):
                        server_logger.info(f"收到Windows控制台事件 {dwCtrlType}，正在停止服务...")
                        stop_celery_worker()
                        sys.exit(0)
                    return False
                win32api.SetConsoleCtrlHandler(win32_handler, True)
            except (ImportError, AttributeError) as e:
                # 如果win32api不可用，跳过Windows控制台处理
                server_logger.debug(f"Windows控制台事件处理不可用: {e}")
                pass

    try:
        if reload:
            # 重载模式需要正确的模块路径
            app_import_path = "server.app:app"
            uvicorn.run(app_import_path, host=host, port=port, reload=reload)
        else:
            # 直接运行模式，直接使用app对象
            uvicorn.run(app, host=host, port=port)
    finally:
        if with_celery:
            stop_celery_worker()


app = create_app()

server_logger.info(msg="APP init successfully - 集成施工方案审查系统")

class ServerRunner:
    """服务器运行器 - 简化的主启动入口"""

    def __init__(self, app_factory: ApplicationFactory):
        """初始化服务器运行器

        Args:
            app_factory: 应用工厂实例
        """
        self.app_factory = app_factory
        self.celery_manager = app_factory.celery_manager

    def run_server(self, **kwargs):
        """运行服务器

        Args:
            **kwargs: 服务器配置参数
        """
        # 获取配置
        config = self.app_factory.create_server_config()
        config.update(kwargs)

        host = config.get('host', '0.0.0.0')
        port = config.get('port', 8002)
        # 确保端口号是整数类型
        try:
            port = int(port)
        except (ValueError, TypeError):
            port = 8002

        reload = config.get('reload', False)
        with_celery = config.get('with_celery', True)

        if with_celery:
            self._setup_celery_integration()

        # 创建应用实例
        app = self.app_factory.create_app()

        try:
            if reload:
                app_import_path = "server.app:app"
                uvicorn.run(app_import_path, host=host, port=port, reload=reload)
            else:
                uvicorn.run(app, host=host, port=port)
        finally:
            if with_celery:
                self.celery_manager.stop_worker()

    def _setup_celery_integration(self):
        """设置Celery集成"""
        # 启动Celery Worker
        self.celery_manager.start_worker()

        # 注册退出处理
        import atexit
        atexit.register(self.celery_manager.stop_worker_immediately)

        # 设置信号处理
        self._setup_signal_handlers()

    def _setup_signal_handlers(self):
        """设置信号处理器"""
        def signal_handler(signum, frame):
            server_logger.info(f"收到信号 {signum}，正在停止服务...")
            self.celery_manager.stop_worker_immediately()
            sys.exit(0)

        # 通用信号处理
        try:
            signal.signal(signal.SIGINT, signal_handler)  # Ctrl+C
            signal.signal(signal.SIGTERM, signal_handler)  # 终止信号
        except AttributeError:
            # Windows可能不支持某些信号
            pass

        # Windows特有处理
        if sys.platform == 'win32':
            self._setup_windows_signal_handler()

    def _setup_windows_signal_handler(self):
        """设置Windows信号处理器"""
        try:
            import win32api
            def win32_handler(dwCtrlType):
                CTRL_C_EVENT = 0
                CTRL_BREAK_EVENT = 1
                CTRL_CLOSE_EVENT = 2
                CTRL_SHUTDOWN_EVENT = 6

                if dwCtrlType in (CTRL_C_EVENT, CTRL_BREAK_EVENT, CTRL_CLOSE_EVENT, CTRL_SHUTDOWN_EVENT):
                    server_logger.info(f"收到Windows控制台事件 {dwCtrlType}，正在停止服务...")
                    self.celery_manager.stop_worker_immediately()
                    sys.exit(0)
                return False
            win32api.SetConsoleCtrlHandler(win32_handler, True)
        except (ImportError, AttributeError) as e:
            server_logger.debug(f"Windows控制台事件处理不可用: {e}")


# 创建应用实例和运行器
app = app_factory.create_app()
server_runner = ServerRunner(app_factory)

server_logger.info(msg="APP init successfully - 集成施工方案审查系统")


# 运行Uvicorn服务器
if __name__ == "__main__":
    # 使用新的服务器运行器启动
    config = app_factory.create_server_config()

    server_logger.info(f"Agent API服务启动中...运行在{config['host']}:{config['port']}")

    if config['with_celery']:
        server_logger.info("Celery Worker: 已集成启动")
    else:
        server_logger.warning("Celery Worker: 已禁用")

    server_runner.run_server(**config)