194 lines
6.4 KiB
Python
194 lines
6.4 KiB
Python
|
# =============================================================================
|
|||
|
|
# 企微IT智能服务台 — 服务路由定义
|
|||
|
|
# =============================================================================
|
|||
|
|
# 说明：根据 SERVICE_NAME 环境变量定义各服务需要加载的路由
|
|||
|
|
# 支持 5 种服务：core, conversation, agent, ai, admin
|
|||
|
|
# 不设置 SERVICE_NAME 时加载全部路由（向后兼容）
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
from typing import Dict, List, Tuple
|
|||
|
|
from fastapi import APIRouter
|
|||
|
|
|
|||
|
|
# 路由模块导入
|
|||
|
|
from app.api import (
|
|||
|
|
wecom_callback,
|
|||
|
|
conversations,
|
|||
|
|
messages,
|
|||
|
|
agents,
|
|||
|
|
quick_replies,
|
|||
|
|
h5,
|
|||
|
|
agent_notes,
|
|||
|
|
system,
|
|||
|
|
wingman,
|
|||
|
|
todo_items,
|
|||
|
|
troubleshooting_templates,
|
|||
|
|
employees,
|
|||
|
|
upload,
|
|||
|
|
admin_api,
|
|||
|
|
portal,
|
|||
|
|
admin_roles,
|
|||
|
|
approval,
|
|||
|
|
wecom_jsapi,
|
|||
|
|
auth_qrcode,
|
|||
|
|
high_risk_routes,
|
|||
|
|
mfa,
|
|||
|
|
auth_wecom_sso,
|
|||
|
|
audit_logs,
|
|||
|
|
)
|
|||
|
|
# admin 子目录的路由需要单独导入
|
|||
|
|
from app.api.admin.security_comparison import router as security_comparison_router
|
|||
|
|
|
|||
|
|
# MFA 有两个 router，需要特殊处理
|
|||
|
|
_MFA_ROUTER = mfa.router
|
|||
|
|
_MFA_ADMIN_ROUTER = mfa.admin_router
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 路由定义：模块名 -> (router对象, tags, prefix)
|
|||
|
|
# prefix 为空时使用路由对象默认的 prefix
|
|||
|
|
_ROUTE_MODULES = {
|
|||
|
|
# 企微回调
|
|||
|
|
"wecom_callback": (wecom_callback.router, ["企微回调"], None),
|
|||
|
|
# 会话管理
|
|||
|
|
"conversations": (conversations.router, ["会话管理"], None),
|
|||
|
|
"messages": (messages.router, ["消息管理"], None),
|
|||
|
|
# 坐席管理
|
|||
|
|
"agents": (agents.router, ["坐席管理"], None),
|
|||
|
|
"quick_replies": (quick_replies.router, ["快速回复"], None),
|
|||
|
|
# H5 用户端
|
|||
|
|
"h5": (h5.router, ["H5用户端"], None),
|
|||
|
|
# 坐席备注
|
|||
|
|
"agent_notes": (agent_notes.router, ["坐席备注"], None),
|
|||
|
|
# 系统管理
|
|||
|
|
"system": (system.router, ["系统管理"], None),
|
|||
|
|
# AI Wingman
|
|||
|
|
"wingman": (wingman.router, ["AI Wingman"], None),
|
|||
|
|
# 待办事项
|
|||
|
|
"todo_items": (todo_items.router, ["待办事项"], None),
|
|||
|
|
# 排查模板
|
|||
|
|
"troubleshooting_templates": (troubleshooting_templates.router, ["排查模板"], None),
|
|||
|
|
# 员工管理
|
|||
|
|
"employees": (employees.router, ["员工管理"], None),
|
|||
|
|
# 文件上传
|
|||
|
|
"upload": (upload.router, ["文件上传"], None),
|
|||
|
|
# 管理后台
|
|||
|
|
"admin_api": (admin_api.router, ["管理后台"], None),
|
|||
|
|
# Portal 统一入口
|
|||
|
|
"portal": (portal.router, ["统一入口"], None),
|
|||
|
|
# 角色管理
|
|||
|
|
"admin_roles": (admin_roles.router, ["角色管理"], None),
|
|||
|
|
# 审批流程
|
|||
|
|
"approval": (approval.router, ["审批流程"], None),
|
|||
|
|
# 企微 JS-SDK
|
|||
|
|
"wecom_jsapi": (wecom_jsapi.router, ["企微JS-SDK"], None),
|
|||
|
|
# 扫码登录
|
|||
|
|
"auth_qrcode": (auth_qrcode.router, ["扫码登录"], None),
|
|||
|
|
# 高危操作
|
|||
|
|
"high_risk_routes": (high_risk_routes.router, ["高危操作"], None),
|
|||
|
|
# MFA 二次认证（用户端和管理端分开）
|
|||
|
|
"mfa": (_MFA_ROUTER, ["MFA二次认证"], None),
|
|||
|
|
"mfa_admin": (_MFA_ADMIN_ROUTER, ["MFA管理(管理员)"], None),
|
|||
|
|
# 企微 SSO
|
|||
|
|
"auth_wecom_sso": (auth_wecom_sso.router, ["企微SSO"], None),
|
|||
|
|
# 审计日志
|
|||
|
|
"audit_logs": (audit_logs.router, ["审计日志"], None),
|
|||
|
|
# 终端安全对比
|
|||
|
|
"security_comparison": (security_comparison_router, ["终端安全对比"], None),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 服务路由映射：服务名 -> 需要加载的路由模块列表
|
|||
|
|
SERVICE_ROUTE_MAP: Dict[str, List[str]] = {
|
|||
|
|
# Core 服务：核心服务（鉴权、员工、角色）
|
|||
|
|
"core": [
|
|||
|
|
"employees", # 员工管理
|
|||
|
|
"auth_qrcode", # 扫码登录
|
|||
|
|
"mfa", # MFA 二次认证
|
|||
|
|
"auth_wecom_sso", # 企微 SSO
|
|||
|
|
"portal", # 角色切换
|
|||
|
|
],
|
|||
|
|
# Conversation 服务：会话服务（会话、消息、H5、WebSocket）
|
|||
|
|
"conversation": [
|
|||
|
|
"wecom_callback", # 企微消息接收
|
|||
|
|
"conversations", # 会话管理
|
|||
|
|
"messages", # 消息管理
|
|||
|
|
"h5", # H5 用户端
|
|||
|
|
"todo_items", # 待办事项
|
|||
|
|
"troubleshooting_templates", # 排查模板
|
|||
|
|
],
|
|||
|
|
# Agent 服务：坐席服务
|
|||
|
|
"agent": [
|
|||
|
|
"agents", # 坐席管理
|
|||
|
|
"quick_replies", # 快速回复
|
|||
|
|
"agent_notes", # 坐席备注
|
|||
|
|
"approval", # 审批流程
|
|||
|
|
],
|
|||
|
|
# AI 服务：AI 服务（Dify 调用、Wingman）
|
|||
|
|
"ai": [
|
|||
|
|
"wingman", # AI Wingman
|
|||
|
|
],
|
|||
|
|
# Admin 服务：管理服务（仪表盘、配置、集成、审核）
|
|||
|
|
"admin": [
|
|||
|
|
"admin_api", # 管理后台 API（必须放第一个，因为 security_comparison 依赖它）
|
|||
|
|
"admin_roles", # 角色管理
|
|||
|
|
"audit_logs", # 审计日志
|
|||
|
|
"high_risk_routes", # 高危操作
|
|||
|
|
"system", # 系统管理
|
|||
|
|
"upload", # 文件上传
|
|||
|
|
"wecom_jsapi", # 企微 JS-SDK
|
|||
|
|
"security_comparison", # 终端安全对比
|
|||
|
|
"mfa_admin", # MFA 管理端
|
|||
|
|
],
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_routes_for_service(service_name: str) -> List[Tuple]:
|
|||
|
|
"""根据服务名获取需要加载的路由列表
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
service_name: 服务名 (core/conversation/agent/ai/admin) 或 None/空
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
路由元组列表：[(router, tags, prefix), ...]
|
|||
|
|
"""
|
|||
|
|
# 不设置 SERVICE_NAME 或设置为 "all" 时，加载全部路由（向后兼容）
|
|||
|
|
if not service_name or service_name.lower() == "all":
|
|||
|
|
return [
|
|||
|
|
(_ROUTE_MODULES[name][0], _ROUTE_MODULES[name][1], _ROUTE_MODULES[name][2])
|
|||
|
|
for name in _ROUTE_MODULES.keys()
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
# 根据服务名获取路由列表
|
|||
|
|
route_names = SERVICE_ROUTE_MAP.get(service_name.lower(), [])
|
|||
|
|
|
|||
|
|
# 查找并返回路由对象
|
|||
|
|
result = []
|
|||
|
|
for name in route_names:
|
|||
|
|
if name in _ROUTE_MODULES:
|
|||
|
|
result.append((
|
|||
|
|
_ROUTE_MODULES[name][0],
|
|||
|
|
_ROUTE_MODULES[name][1],
|
|||
|
|
_ROUTE_MODULES[name][2],
|
|||
|
|
))
|
|||
|
|
|
|||
|
|
return result
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_current_service_name() -> str:
|
|||
|
|
"""获取当前服务名称
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
SERVICE_NAME 环境变量，或空字符串
|
|||
|
|
"""
|
|||
|
|
return os.getenv("SERVICE_NAME", "")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def is_service_mode() -> bool:
|
|||
|
|
"""判断是否处于服务模式（非单体模式）
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
True 如果设置了有效的 SERVICE_NAME
|
|||
|
|
"""
|
|||
|
|
name = get_current_service_name()
|
|||
|
|
return bool(name and name.lower() != "all")
|