qihuaiflow/backend/python/main.py

"""

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, channels, knowledge, market, token_usage, payment, profile, notifications, assets, evolution

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
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.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())

        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



            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(channels.router, prefix="/app/api/channels", tags=["频道"])

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(promotion_admin.router, prefix="/app/api/admin/promotion", tags=["promotion"])

app.include_router(
    achievement_admin.router,
    prefix="/app/api/admin/achievement",
    tags=["achievement"]
)




@app.get("/app/api/health")

async def health_check():

    return {"status": "ok", "service": "Qihu-AIFlow"}