# ============================================================================= # 企微IT智能服务台 — 坐席管理 API # ============================================================================= # 说明:坐席端的管理接口,包括: # 1. POST /api/agents/login — 坐席登录(用户名密码,返回JWT token) # 2. GET /api/agents/me — 获取当前坐席信息 # 3. PUT /api/agents/me/status — 更新坐席状态(online/busy/offline) # 4. GET /api/agents — 获取坐席列表(用于转接选择) # 坐席认证使用 JWT,token 存 Redis(TTL 8小时) # ============================================================================= import base64 import io import json import logging import secrets from datetime import datetime from typing import Optional from uuid import UUID import qrcode import redis.asyncio as aioredis import bcrypt # P1 修复: 直接使用 bcrypt 库替代 passlib from fastapi import APIRouter, Depends, Header, Query, Request from pydantic import BaseModel, Field from slowapi import Limiter from slowapi.util import get_remote_address from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.database import get_db from app.dependencies import get_current_user, require_role, dep_wecom_service from app.models.agent import Agent from app.schemas.agent import AgentLogin, AgentResponse, AgentStatusUpdate from app.services.wecom_service import WecomService from app.services.mfa_service import MFAService from app.utils.response import AppException, ERR_UNAUTHORIZED, success_response from app.utils.error_codes import ErrorCode # 速率限制器实例(与 main.py 共享同一配置) # 移除 env_file=None 参数:slowapi 0.1.9 不支持该参数 # python-dotenv 已在应用启动时处理 .env 文件 limiter = Limiter(key_func=get_remote_address) logger = logging.getLogger(__name__) # 创建路由器 router = APIRouter() # JWT 简化版:使用随机 token 存 Redis,TTL 8 小时 # 为什么不用标准 JWT:第一步简化实现,token 存 Redis 更容易实现登出和状态管理 TOKEN_TTL_SECONDS = 8 * 60 * 60 # 8小时 def _get_redis() -> aioredis.Redis: """获取 Redis 客户端。""" return settings.create_redis_client() # -------------------------------------------------------------------------- # 坐席认证依赖 # -------------------------------------------------------------------------- async def get_current_agent( authorization: Optional[str] = Header(None, alias="Authorization"), db: AsyncSession = Depends(get_db), ) -> Agent: """从请求头中提取坐席身份(认证依赖)。 支持两种 Token 格式: 1. 统一格式:user:token:{token} → JSON 包含 employee_id 和 roles 2. 旧格式:agent:token:{token} → 直接存储 user_id Args: authorization: 请求头中的 Authorization 字段(格式:Bearer token) db: 数据库会话 Returns: Agent: 当前坐席对象 Raises: AppException: 未授权(token 缺失、无效或过期) """ if not authorization: raise ERR_UNAUTHORIZED # 提取 token(支持 "Bearer xxx" 格式) token = authorization.replace("Bearer ", "") if authorization.startswith("Bearer ") else authorization if not token: raise ERR_UNAUTHORIZED # 从 Redis 查找坐席ID redis_client = _get_redis() try: # 1. 尝试统一格式(新) unified_data = await redis_client.get(f"user:token:{token}") if unified_data: try: user_info = json.loads(unified_data) agent_user_id = user_info.get("employee_id") if agent_user_id: # 从数据库查找坐席 stmt = select(Agent).where(Agent.user_id == agent_user_id) result = await db.execute(stmt) agent = result.scalars().first() if agent: return agent except json.JSONDecodeError: logger.warning(f"统一 Token 数据解析失败: {token[:10]}...") # 2. 尝试旧格式(兼容) agent_user_id = await redis_client.get(f"agent:token:{token}") if not agent_user_id: raise ERR_UNAUTHORIZED # 从数据库查找坐席 # agent_user_id 可能是 bytes(Redis 返回)或 str uid = agent_user_id.decode("utf-8") if isinstance(agent_user_id, bytes) else agent_user_id stmt = select(Agent).where(Agent.user_id == uid) result = await db.execute(stmt) agent = result.scalars().first() if not agent: raise ERR_UNAUTHORIZED return agent except AppException: # 业务异常直接抛出(如 ERR_UNAUTHORIZED) raise except Exception as e: # Redis 连接失败等底层异常 logger.error(f"Redis 读取失败: {e}") raise ERR_UNAUTHORIZED finally: try: await redis_client.close() except Exception: pass # -------------------------------------------------------------------------- # POST /api/agents/login — 坐席登录 # -------------------------------------------------------------------------- @router.post("/agents/login") @limiter.limit("10/minute") # 登录接口限流:每IP每分钟最多10次,防暴力破解 async def agent_login( request: Request, body: AgentLogin, db: AsyncSession = Depends(get_db), ): """坐席登录。 第一步使用简单的用户名密码登录。 登录成功后生成 token 存入 Redis(TTL 8小时)。 流程: 1. 查找坐席记录(按 user_id),不存在则自动创建 2. 生成随机 token 3. token 存 Redis(key: agent:token:{token}, value: user_id) 4. 更新坐席状态为 online 5. 返回坐席信息和 token Args: body: 登录请求体(包含 user_id 和 name) db: 数据库会话 Returns: Dict: 统一响应格式,包含坐席信息和 token """ try: # 0. 企微通讯录身份验证(防止任意 user_id 冒充坐席) # 调用企微API校验 user_id 是否存在于通讯录中 # 安全策略: # - 企微验证通过 → 正常登录,用企微真实姓名覆盖前端传入值 # - 企微验证失败(用户不存在) → 拒绝登录 # - 企微API不可达(网络故障) → 仅允许已注册坐席降级登录,新注册必须验证 wecom_verified = False # 默认空头像,企微验证成功时覆盖;确保在 wecom 不可达(降级)时仍可安全引用 avatar = "" try: redis_client_verify = _get_redis() try: wecom_service = WecomService(redis_client_verify) user_info = await wecom_service.get_user_info(body.user_id) # 验证通过:用户存在于企微通讯录 wecom_verified = True # 用企微返回的真实姓名覆盖前端传入的姓名(防止冒用他人身份) real_name = user_info.get("name", "") if real_name: body.name = real_name # 【P1-02】每次坐席登录也强制更新头像(与 H5 登录保持一致,统一走 avatar_service) avatar = user_info.get("avatar", "") if avatar: try: from app.services.avatar_service import sync_employee_avatar await sync_employee_avatar(db, redis_client_verify, body.user_id, avatar) except Exception as e: logger.warning(f"同步员工头像失败(不阻塞登录): user_id={body.user_id}, error={e}") logger.info(f"坐席企微身份验证通过: user_id={body.user_id}, name={real_name}") finally: try: await redis_client_verify.close() except Exception: pass try: await wecom_service.close() except Exception: pass except Exception as wecom_err: # 企微API不可达时:仅允许已注册坐席降级登录,新注册必须验证 # 原因:网络故障不应阻断已注册坐席工作,但不能让未验证用户注册新账号 logger.warning( f"企微通讯录验证失败: user_id={body.user_id}, " f"error={wecom_err}" ) # 检查是否为已注册坐席(数据库已有记录才允许降级登录) check_stmt = select(Agent).where(Agent.user_id == body.user_id) check_result = await db.execute(check_stmt) existing_agent = check_result.scalars().first() if not existing_agent: # 新坐席注册必须通过企微验证,防止任意 user_id 冒充 raise AppException( ErrorCode.AUTH_TOKEN_INVALID, "企微通讯录验证失败,新坐席注册需要企微身份验证。请稍后重试或联系管理员。" ) logger.warning( f"企微API不可达,已注册坐席降级放行: user_id={body.user_id}" ) # P0 修复: 降级放行时,如果 agent 已设置密码则必须验证本地密码 if existing_agent: if existing_agent.password_hash is None: # 已注册坐席但未设置密码,要求先设置密码 raise AppException( ErrorCode.AUTH_PASSWORD_REQUIRED, "首次登录请先设置密码。管理后台 → 坐席管理 → 设置本地密码" ) if not body.password: raise AppException(ErrorCode.AUTH_PASSWORD_WRONG, "请输入本地密码") if not bcrypt.checkpw(body.password.encode('utf-8'), existing_agent.password_hash.encode('utf-8')): raise AppException(ErrorCode.AUTH_PASSWORD_WRONG, "本地密码错误") # 1. 查找或创建坐席记录 stmt = select(Agent).where(Agent.user_id == body.user_id) result = await db.execute(stmt) agent = result.scalars().first() if not agent: # 首次登录,创建坐席记录 agent = Agent( user_id=body.user_id, name=body.name, status="online", current_load=0, max_load=5, ) db.add(agent) await db.flush() logger.info(f"新坐席注册: user_id={body.user_id}, name={body.name}") else: # 更新坐席名称(可能改名了) agent.name = body.name agent.status = "online" agent.updated_at = datetime.now() db.add(agent) await db.flush() logger.info(f"坐席登录: user_id={body.user_id}, name={body.name}") # 2. MFA 二次验证(已绑定 MFA 的坐席/管理员) # 决策3(三端认证重构 AUTH-04):移除「企微已登录+角色→免密直接进入」分支, # 所有登录方式(扫码/账密/企微验证)均需 OTP 验证,统一安全水位。 # 执行MFA验证 if agent.mfa_enabled: if not body.otp_code: # 需要 OTP 验证,返回 require_otp 标记(必须包含role字段,否则前端校验会失败) return success_response(data={ "require_otp": True, "message": "请输入OTP动态码", "user_id": agent.user_id, "name": agent.name, "role": agent.role, # 必须包含role字段,供前端校验权限 }) else: # 验证 OTP 码(决策3:复用 MFAService 统一校验逻辑) if not MFAService.verify_code(agent.mfa_secret, body.otp_code, valid_window=1): raise AppException(1006, "OTP验证码错误,请重新输入") # 3. 生成随机 token(使用统一格式) from app.services.token_service import TokenService from app.dependencies import get_redis # 使用共享 Redis 连接(从连接池获取,不要手动关闭) redis_client = await get_redis() token_service = TokenService(redis_client) # 查询用户角色 from app.services.role_mapping_service import RoleMappingService role_service = RoleMappingService(db) roles = await role_service.get_user_roles(body.user_id) # 创建统一格式的 Token token = await token_service.create_token( employee_id=body.user_id, name=body.name, roles=roles, avatar=avatar, login_source="agent", ) # 5. 返回坐席信息和 token agent_data = AgentResponse.model_validate(agent).model_dump() agent_data["token"] = token return success_response(data=agent_data) except AppException: # 业务异常直接抛出 raise except Exception as e: # 未预期的异常:记录日志,返回友好错误 logger.error(f"登录异常: {e}", exc_info=True) raise AppException(1005, f"登录失败: {str(e)}") # -------------------------------------------------------------------------- # GET /api/agents/me — 获取当前坐席信息 # -------------------------------------------------------------------------- @router.get("/agents/me") async def get_agent_me( agent: Agent = Depends(get_current_agent), ): """获取当前坐席信息。 需要在请求头中携带有效的 token。 Args: agent: 当前坐席(通过认证依赖注入) Returns: Dict: 统一响应格式,包含坐席信息 """ agent_data = AgentResponse.model_validate(agent).model_dump() return success_response(data=agent_data) # -------------------------------------------------------------------------- # PUT /api/agents/me/status — 更新坐席状态 # -------------------------------------------------------------------------- @router.put("/agents/me/status") async def update_agent_status( body: AgentStatusUpdate, agent: Agent = Depends(get_current_agent), db: AsyncSession = Depends(get_db), ): """更新坐席状态。 坐席可以切换为 online/busy/offline。 - online: 在线,可以接收新会话 - busy: 忙碌,不接收新会话但继续处理已有的 - offline: 离线,不接收任何会话 Args: body: 状态更新请求体 agent: 当前坐席 db: 数据库会话 Returns: Dict: 统一响应格式,包含更新后的坐席信息 """ agent.status = body.status agent.updated_at = datetime.now() db.add(agent) await db.flush() logger.info(f"坐席状态更新: agent={agent.user_id}, status={body.status}") agent_data = AgentResponse.model_validate(agent).model_dump() return success_response(data=agent_data) # -------------------------------------------------------------------------- # GET /api/agents — 获取坐席列表(需要 agent 或 admin 角色) # -------------------------------------------------------------------------- @router.get("/agents") @require_role("agent", "admin") async def list_agents( status: Optional[str] = Query(None, description="按状态过滤: online/busy/offline"), db: AsyncSession = Depends(get_db), ): """获取坐席列表。 用于转接选择时展示可用的坐席列表。 Args: status: 按状态过滤(可选) db: 数据库会话 Returns: Dict: 统一响应格式,包含坐席列表 """ stmt = select(Agent).order_by(Agent.name) if status: stmt = stmt.where(Agent.status == status) result = await db.execute(stmt) agents = list(result.scalars().all()) items = [AgentResponse.model_validate(a).model_dump() for a in agents] return success_response(data={"items": items}) # -------------------------------------------------------------------------- # 本地密码管理接口(P0-#5) # -------------------------------------------------------------------------- class AgentPasswordUpdate(BaseModel): """坐席修改密码请求 Schema""" old_password: Optional[str] = Field(None, description="旧密码(验证用)") new_password: str = Field(..., min_length=6, max_length=128, description="新密码") @router.post("/agents/password") async def update_agent_password( body: AgentPasswordUpdate, agent: Agent = Depends(get_current_agent), db: AsyncSession = Depends(get_db), ): """修改坐席本地密码。 可选功能,允许坐席设置本地密码作为备用认证方式。 企微验证失败时,可使用本地密码认证。 P0-#5 新增端点。 Args: body.new_password: 新密码(6-128位) Returns: Dict: 修改结果 """ try: # 如果已有旧密码,验证旧密码 if agent.password_hash: if not body.old_password: # 2026-06-15 修复: 改用专用 ErrorCode,避免与登录 1012 冲突 raise AppException(ErrorCode.AUTH_OLD_PASSWORD_REQUIRED, "请输入旧密码") if not bcrypt.checkpw(body.old_password.encode('utf-8'), agent.password_hash.encode('utf-8')): # 2026-06-15 修复: 改用专用 ErrorCode raise AppException(ErrorCode.AUTH_OLD_PASSWORD_WRONG, "旧密码错误") # 设置新密码 agent.password_hash = bcrypt.hashpw(body.new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') agent.updated_at = datetime.now() db.add(agent) await db.flush() logger.info(f"本地密码已更新: agent={agent.user_id}") return success_response(data={"message": "密码已更新"}) except AppException: raise except Exception as e: logger.error(f"密码更新异常: {e}", exc_info=True) raise AppException(1014, f"密码更新失败: {str(e)}") # ============================================================================ # 忘记密码 - 企微扫码重置 # ============================================================================ class AgentPasswordResetByWecom(BaseModel): """通过企微扫码重置密码请求 Schema""" code: str = Field(..., description="企微OAuth2授权码") new_password: str = Field(..., min_length=6, max_length=128, description="新密码") @router.post("/agents/password/reset-by-wecom") async def reset_password_by_wecom( body: AgentPasswordResetByWecom, db: AsyncSession = Depends(get_db), wecom_service: WecomService = Depends(dep_wecom_service), ): """通过企微扫码验证后重置密码。 适用于坐席忘记原密码的情况。通过企微OAuth2扫码验证身份后, 无需旧密码即可重置密码。 #91 新增端点。 Args: body.code: 企微OAuth2授权码 body.new_password: 新密码(6-128位) Returns: Dict: 重置结果 """ try: # 1. 用 code 换取员工身份 user_info = await wecom_service.get_oauth_user_info(body.code) employee_id = user_info.get("userid", "") if not employee_id: raise AppException(2007, "OAuth2授权失败:未获取到员工ID") # 2. 查询该员工是否是坐席 from sqlalchemy import select from app.models.agent import Agent stmt = select(Agent).where(Agent.user_id == employee_id) result = await db.execute(stmt) agent = result.scalar_one_or_none() if not agent: raise AppException(1015, "该员工不是坐席,无法重置密码") # 3. 重置密码 agent.password_hash = bcrypt.hashpw(body.new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') agent.updated_at = datetime.now() db.add(agent) await db.flush() logger.info(f"密码已通过企微扫码重置: agent={agent.user_id}") return success_response(data={"message": "密码已重置"}) except AppException: raise except Exception as e: logger.error(f"密码重置异常: {e}", exc_info=True) raise AppException(1016, f"密码重置失败: {str(e)}") # ============================================================================ # 企微 OAuth2 一键登录(坐席端) # ============================================================================ import urllib.parse import secrets as secrets_module def _build_agent_oauth_url(redirect_uri: str) -> str: """构建坐席端企微OAuth2授权URL。 文档: https://developer.work.weixin.qq.com/document/path/91022 """ params = { "appid": settings.wecom_corp_id, "redirect_uri": redirect_uri, "response_type": "code", "scope": "snsapi_base", # 静默授权 "state": "agent_login", # 标记为坐席登录 } # 如果有 agentid 也加上 if getattr(settings, "wecom_agent_id", None): params["agentid"] = str(settings.wecom_agent_id) query = urllib.parse.urlencode(params) # 企业微信 OAuth2 地址(注意是 open.work.weixin.qq.com) return f"https://open.work.weixin.qq.com/connect/oauth2/authorize?{query}#wechat_redirect" @router.get("/agents/oauth/authorize") async def get_oauth_authorize_url( redirect_uri: str = Query(None, description="OAuth回调地址(可选,默认坐席端地址)"), ): """获取企微OAuth2授权URL(JSON格式,供前端跳转)。 前端调用此接口获取授权URL,然后自行跳转到企微授权页。 授权成功后企微会携带 code 回调到此接口的 redirect_uri。 Args: redirect_uri: 授权成功后的回调地址(可选) 默认: https://itsupport.servyou.com.cn/itagent/ Returns: JSON: { code: 0, data: { authorize_url: "https://open.weixin.qq.com/..." } } """ # 确定回调地址 if redirect_uri: # 前端传入的回调地址 pass else: # 默认回调地址:坐席端首页 redirect_uri = "https://itsupport.servyou.com.cn/itagent/" # 编码回调地址 encoded_redirect = urllib.parse.quote(redirect_uri, safe='') # 构建授权URL authorize_url = _build_agent_oauth_url(redirect_uri) logger.info(f"生成坐席端OAuth授权URL: redirect_uri={redirect_uri}") return success_response(data={ "authorize_url": authorize_url, "redirect_uri": redirect_uri, }) # OAuth 回调请求模型 class OAuthCallbackRequest(BaseModel): code: str = Field(..., description="企微授权码") state: str = Field(default="agent_login", description="state参数") @router.post("/agents/oauth/callback") async def oauth_callback( body: OAuthCallbackRequest, db: AsyncSession = Depends(get_db), ): """企微OAuth2回调处理(坐席端)。 用授权码换取员工ID,验证坐席身份,生成登录token。 Args: body: { code: "xxx", state: "agent_login" } db: 数据库会话 Returns: JSON: { code: 0, data: { token, user_id, name, roles } } """ code = body.code state = body.state if not code: raise AppException(2007, "授权码不能为空") # 1. 用 code 换取员工身份 wecom_service = WecomService() try: oauth_info = await wecom_service.get_oauth_user_info(code) user_id = oauth_info.get("userid", "") if not user_id: raise AppException(2007, "OAuth授权失败:未获取到员工ID") except Exception as e: logger.error(f"企微OAuth换取userid失败: {e}") raise AppException(2007, f"OAuth授权失败: {str(e)}") # 2. 获取员工详细信息(包含姓名) employee_name = "" try: detail = await wecom_service.get_user_info(user_id) employee_name = detail.get("name", "") except Exception as e: logger.warning(f"获取员工详细信息失败: user_id={user_id}, error={e}") # 3. 验证是否为坐席 stmt = select(Agent).where(Agent.user_id == user_id) result = await db.execute(stmt) agent = result.scalars().first() if not agent: raise AppException(2008, f"您不是坐席,无法通过企业微信登录") # 4. 生成登录token token = secrets_module.token_urlsafe(32) redis_client = _get_redis() if redis_client: try: # 存储 token -> agent信息(JSON格式) token_data = { "user_id": agent.user_id, "name": agent.name, "roles": [agent.role], "login_source": "agent_oauth", } import json as json_module await redis_client.setex( f"user:token:{token}", TOKEN_TTL_SECONDS, json_module.dumps(token_data), ) # 记录登录日志 logger.info(f"企微OAuth登录成功: user_id={user_id}, name={employee_name}") except Exception as e: logger.error(f"Token存储Redis失败: {e}") raise AppException(1003, "登录失败,请重试") return success_response(data={ "token": token, "user_id": agent.user_id, "name": employee_name or agent.name, "role": agent.role, "require_otp": agent.mfa_secret is not None, })