99 lines
3.6 KiB
Python
99 lines
3.6 KiB
Python
|
# =============================================================================
|
|||
|
|
# 企微IT智能服务台 — 阶段5 自动化 依赖注入
|
|||
|
|
# =============================================================================
|
|||
|
|
# 说明：提供自动化接口的认证/鉴权依赖：
|
|||
|
|
# 1. get_current_employee_id — 从 Bearer Token 解析 H5 员工身份
|
|||
|
|
# （Redis key 与 app/api/ws.py 的 employee:token:{token} 保持一致）
|
|||
|
|
# 2. get_automation_agent — 复用坐席认证（转人工接管/审批用）
|
|||
|
|
# 3. get_scenario_config — 加载并校验场景配置是否启用
|
|||
|
|
# 高危配置写接口复用 app.dependencies.require_high_risk_otp（管理端 OTP）。
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
from typing import Optional
|
|||
|
|
|
|||
|
|
from fastapi import Depends
|
|||
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|||
|
|
from sqlalchemy import select
|
|||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
|
|
|||
|
|
from app.constants import AutomationErrorCode
|
|||
|
|
from app.database import get_db
|
|||
|
|
from app.dependencies import get_redis
|
|||
|
|
from app.models.automation import ScenarioConfig
|
|||
|
|
from app.services.cache_service import cache_service
|
|||
|
|
from app.utils.response import AppException, ERR_UNAUTHORIZED
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
# Bearer 认证方案（auto_error=False：缺失时由我们自行返回 1002）
|
|||
|
|
_security = HTTPBearer(auto_error=False)
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def get_current_employee_id(
|
|||
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(_security),
|
|||
|
|
) -> str:
|
|||
|
|
"""从 Bearer Token 解析 H5 员工身份。
|
|||
|
|
|
|||
|
|
与 app/api/ws.py 中 H5 WebSocket 的鉴权保持一致：
|
|||
|
|
Redis key = employee:token:{token} → employee_id。
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
str: 员工企微 UserID
|
|||
|
|
|
|||
|
|
Raises:
|
|||
|
|
AppException(1002): 令牌缺失或无效
|
|||
|
|
"""
|
|||
|
|
token = credentials.credentials if credentials else ""
|
|||
|
|
if not token:
|
|||
|
|
raise AppException(ERR_UNAUTHORIZED.code, "缺少认证令牌")
|
|||
|
|
|
|||
|
|
# 优先从共享 Redis（cache_service），降级到独立连接
|
|||
|
|
employee_id = None
|
|||
|
|
try:
|
|||
|
|
employee_id = await cache_service.get(f"employee:token:{token}")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.warning(f"employee token 校验 Redis 异常: {e}")
|
|||
|
|
if not employee_id:
|
|||
|
|
redis_client = await get_redis()
|
|||
|
|
if redis_client is not None:
|
|||
|
|
try:
|
|||
|
|
employee_id = await redis_client.get(f"employee:token:{token}")
|
|||
|
|
except Exception as e: # noqa: BLE001
|
|||
|
|
logger.warning(f"employee token 校验失败: {e}")
|
|||
|
|
|
|||
|
|
if not employee_id:
|
|||
|
|
raise AppException(ERR_UNAUTHORIZED.code, "员工令牌无效或已过期")
|
|||
|
|
return employee_id
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def get_automation_agent(agent=Depends(lambda: None)) -> str: # pragma: no cover
|
|||
|
|
"""占位：坐席认证由 app.api.agents.get_current_agent 直接提供。
|
|||
|
|
|
|||
|
|
本函数保留以便路由层统一引用；实际坐席端点应使用
|
|||
|
|
`from app.api.agents import get_current_agent`。
|
|||
|
|
"""
|
|||
|
|
return agent
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def get_scenario_config(
|
|||
|
|
scenario_key: str,
|
|||
|
|
db: AsyncSession = Depends(get_db),
|
|||
|
|
) -> ScenarioConfig:
|
|||
|
|
"""加载场景配置，并校验是否启用。
|
|||
|
|
|
|||
|
|
Raises:
|
|||
|
|
AppException(4001): 场景未配置或已禁用
|
|||
|
|
"""
|
|||
|
|
stmt = select(ScenarioConfig).where(ScenarioConfig.scenario_key == scenario_key)
|
|||
|
|
result = await db.execute(stmt)
|
|||
|
|
config = result.scalar_one_or_none()
|
|||
|
|
if config is None or not config.enabled:
|
|||
|
|
raise AppException(
|
|||
|
|
AutomationErrorCode.SCENARIO_NOT_FOUND,
|
|||
|
|
f"场景未启用或未配置: {scenario_key}",
|
|||
|
|
)
|
|||
|
|
return config
|