""" Qihu-AIFlow FastAPI 主应用 完整后端API,监听18080端口,所有接口以 /app 为前缀 """ import sys import os import time import uuid import logging from contextlib import asynccontextmanager from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from starlette.middleware.base import BaseHTTPMiddleware from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.api import auth, chat, knowledge, market, token_usage, payment, profile, notifications, assets, evolution from app.casdoor_auth import router as casdoor_router from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin from app.api.im_bot_user import router as im_bot_user_router from app.api.im_bot_platform import router as im_bot_platform_router from app.api.portal import user_portal from app.api.admin.developer_tools_admin import router as developer_tools_router from app.api.admin.mcp_market_admin import router as mcp_market_router from app.api.admin import asset_admin, achievement_admin from app.config import setup_file_logging, SLOW_REQUEST_THRESHOLD from app.core.observability import set_trace_id from app.database import SessionLocal from app.models import OperationLog as OperationLogModel, TokenUsageLog as TokenUsageLogModel # 导入平台适配器触发注册(AstrBot借鉴优化方案 P1) import app.platform_adapters # noqa: F401 - 触发所有平台适配器装饰器注册 # 初始化文件日志(按天轮转) access_logger = setup_file_logging() logger = logging.getLogger("aiflow") class APITrackingMiddleware(BaseHTTPMiddleware): """API调用统计中间件 - 记录请求耗时,标记慢请求""" async def dispatch(self, request: Request, call_next): request_id = str(uuid.uuid4()) set_trace_id(request_id) start_time = time.time() path = request.url.path method = request.method # 跳过健康检查和静态资源 skip_paths = ["/app/api/health", "/favicon.ico"] should_skip = any(path.startswith(p) for p in skip_paths) try: response = await call_next(request) elapsed = time.time() - start_time status_code = response.status_code response.headers["X-Trace-Id"] = request_id if not should_skip: access_logger.info( f"[{request_id}] {method} {path} -> {status_code} ({elapsed:.3f}s)" ) # 异步记录API调用统计 try: db = SessionLocal() try: _record_api_call(db, method, path, status_code, elapsed) finally: db.close() except Exception as e: logger.warning(f"API统计记录失败: {e}") return response except Exception as e: elapsed = time.time() - start_time if not should_skip: access_logger.info( f"[{request_id}] {method} {path} -> ERROR ({elapsed:.3f}s): {str(e)}" ) raise async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.app(scope, receive, send) return await super().__call__(scope, receive, send) def _record_api_call(db, method: str, path: str, status_code: int, elapsed: float): """记录API调用到数据库统计表""" # 简化路径以减少基数(去掉动态ID) parts = path.split("/") simplified_parts = [] for part in parts: # 将UUID样式的路径段替换为{id} if len(part) == 36 and part.count("-") == 4: simplified_parts.append("{id}") # 将纯数字路径段替换为{id} elif part.isdigit(): simplified_parts.append("{id}") else: simplified_parts.append(part) simplified_path = "/".join(simplified_parts) # 统计记录通过数据库实现,简化版记录到日志 is_error = status_code >= 400 is_slow = elapsed >= SLOW_REQUEST_THRESHOLD if is_slow: logger.warning( f"慢请求: {method} {simplified_path} -> {status_code} ({elapsed:.3f}s, 阈值: {SLOW_REQUEST_THRESHOLD}s)" ) @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Qihu-AIFlow v2.0.0 启动中...") # 启动后台 Evolution Pipeline(知识库图谱已存在的用户配置自动恢复运行) try: from app.services.evolution_service import EvolutionWatcherManager from app.models import UserEvolutionConfig db = SessionLocal() try: configs = db.query(UserEvolutionConfig).filter( UserEvolutionConfig.watcher_status == "running" ).all() if configs: watcher = EvolutionWatcherManager() for cfg in configs: try: watcher.start_user_evolution(cfg.user_id, lambda: SessionLocal()) logger.info(f"[Evolution] 恢复用户 {cfg.user_id} 的进化Pipeline") except Exception as e: logger.warning(f"[Evolution] 恢复用户 {cfg.user_id} 失败: {e}") logger.info(f"[Evolution] 已恢复 {len(configs)} 个用户的进化Pipeline") else: logger.info("[Evolution] 无待恢复的进化配置") finally: db.close() except Exception as e: logger.warning(f"[Evolution] Pipeline自动启动异常(不影响正常服务): {e}") # 启动Redis事件总线 try: from app.event_bus import get_event_bus event_bus = get_event_bus() event_bus.start(num_consumers=2) logger.info("[EventBus] Redis事件总线已启动") except Exception as e: logger.warning(f"[EventBus] 启动失败(不影响正常服务): {e}") logger.info("Qihu-AIFlow v2.0.0 启动完成") yield # 停止Redis事件总线 try: from app.event_bus import get_event_bus get_event_bus().stop() logger.info("[EventBus] Redis事件总线已停止") except Exception: pass # 优雅关闭:停止所有 Evolution Pipeline try: from app.services.evolution_service import EvolutionWatcherManager EvolutionWatcherManager().shutdown_all() logger.info("[Evolution] 所有Pipeline已停止") except Exception: pass logger.info("Qihu-AIFlow 正在关闭...") app = FastAPI( title="Qihu-AIFlow - 智能自动化PaaS平台", version="1.0.0", lifespan=lifespan, ) # API调用追踪中间件 app.add_middleware(APITrackingMiddleware) # CORS配置 app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 前台 API app.include_router(auth.router, prefix="/app/api/auth", tags=["认证"]) app.include_router(chat.router, prefix="/app/api/chat", tags=["AI聊天"]) app.include_router(knowledge.router, prefix="/app/api/knowledge", tags=["知识库"]) app.include_router(market.router, prefix="/app/api/market", tags=["应用市场"]) app.include_router(token_usage.router, prefix="/app/api/token", tags=["Token驾驶舱"]) app.include_router(payment.router, prefix="/app/api/payment", tags=["付费中心"]) app.include_router(profile.router, prefix="/app/api/profile", tags=["个人中心"]) app.include_router(notifications.router, prefix="/app/api/notifications", tags=["消息通知"]) app.include_router(assets.router, prefix="/app/api/assets", tags=["资产管理"]) app.include_router(evolution.router, prefix="/app/api/evolution", tags=["AI自进化"]) # 后台 API app.include_router(dashboard.router, prefix="/app/api/admin/dashboard", tags=["后台-仪表盘"]) app.include_router(users.router, prefix="/app/api/admin/users", tags=["后台-用户管理"]) app.include_router(models.router, prefix="/app/api/admin/models", tags=["后台-模型管理"]) app.include_router(settings.router, prefix="/app/api/admin/settings", tags=["后台-网站设置"]) app.include_router(payment_admin.router, prefix="/app/api/admin/payment", tags=["后台-支付配置"]) app.include_router(services.router, prefix="/app/api/admin/services", tags=["后台-增值服务"]) app.include_router(membership_admin.router, prefix="/app/api/admin/membership", tags=["后台-会员管理"]) app.include_router(ads.router, prefix="/app/api/admin/ads", tags=["后台-广告管理"]) app.include_router(market_admin.router, prefix="/app/api/admin/market", tags=["后台-应用审核"]) app.include_router(logs.router, prefix="/app/api/admin/logs", tags=["后台-日志风控"]) app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=["后台-技能管理"]) app.include_router(agents_admin.router, prefix="/app/api/admin/agents", tags=["后台-Agent管理"]) app.include_router(knowledge_admin.router, prefix="/app/api/admin/knowledge", tags=["后台-知识库管理"]) app.include_router(workflow_admin.router, prefix="/app/api/admin/workflows", tags=["后台-工作流编排"]) app.include_router(asset_admin.router, prefix="/app/api/admin/assets", tags=["后台-资产管理"]) app.include_router(storage_admin.router, prefix="/app/api/admin/storage", tags=["后台-存储配置"]) app.include_router(automation_admin.router, prefix="/app/api/admin/automation", tags=["后台-自动化任务"]) app.include_router(mcp_admin.router, prefix="/app/api/admin/mcp", tags=["后台-MCP服务"]) app.include_router(developer_admin.router, prefix="/app/api/admin/developer", tags=["后台-开发者API"]) app.include_router(developer_tools_router, prefix="/app/api/admin/developer-tools", tags=["dev-tools"]) app.include_router(apps_admin.router, prefix="/app/api/admin/apps", tags=["admin-ai-apps"]) app.include_router(mcp_market_router, prefix="/app/api/admin/mcp-market", tags=["mcp-market"]) app.include_router(code_exec_admin.router, prefix="/app/api/admin/code-exec", tags=["code-exec"]) app.include_router(distro_admin.router, prefix="/app/api/admin/distributions", tags=["distributions"]) app.include_router(privacy_admin.router, prefix="/app/api/admin/privacy", tags=["privacy"]) app.include_router(im_bot_admin.router, prefix="/app/api/admin/im-bot", tags=["im-bot"]) app.include_router(im_bot_user_router, prefix="/app/api/im/wechat", tags=["im-wechat-user"]) app.include_router(im_bot_platform_router, prefix="/app/api/im/platform", tags=["im-platform-user"]) app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", tags=["promotion"]) app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"]) app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"]) app.include_router(analytics_admin.router, prefix="/app/api/admin/analytics", tags=["analytics"]) app.include_router(user_portal.router, prefix="/app/api/portal", tags=["portal"]) app.include_router(achievement_admin.router, prefix="/app/api/admin/achievement", tags=["achievement"]) app.include_router(casdoor_router, prefix="/app/api/auth", tags=["Casdoor SSO"]) @app.get("/app/api/health") async def health_check(): return {"status": "ok", "service": "Qihu-AIFlow"}