Simon
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
|