wecom_it_smart_desk/backend/app/api/byod.py

# =============================================================================
# IT智能服务台 — 自备电脑补贴（BYOD）资格查询 API
# =============================================================================
# 说明：提供自备电脑补贴资格的意图检测和资格检查功能
#   - 意图检测：关键词预过滤 → Dify → 关键词兜底（复用审批系统三级模式）
#   - 资格检查：通过企微通讯录API获取员工岗位 → 与资格清单匹配 → 返回结果
#   - 岗位匹配支持精确匹配和模糊匹配（如"高级前端开发岗"匹配"前端开发岗"）
# =============================================================================

import json
import logging
from typing import Optional

import httpx
from fastapi import APIRouter, Depends
from pydantic import BaseModel
import redis.asyncio as aioredis

from app.config import settings
from app.services.wecom_service import WecomService
from app.utils.response import success_response

logger = logging.getLogger(__name__)
router = APIRouter()


# Redis客户端（依赖注入）
async def get_redis() -> aioredis.Redis:
    """获取Redis客户端依赖"""
    from app.main import redis_client
    return redis_client


# =============================================================================
# BYOD 资格岗位清单（静态配置 — 13个岗位，4大类）
# 注：需求文档标注14个岗位，但实际清单列出13个岗位，以清单为准
# =============================================================================
# 数据来源：data/byod_eligible_positions.json
# 结构：序列 → 类别 → 岗位列表

BYOD_ELIGIBLE_POSITIONS: dict[str, dict[str, list[str]]] = {
    "技术序列": {
        "开发类": [
            "算法岗",
            "前端开发岗",
            "后端开发岗",
            "客户端开发岗",
            "运维开发岗",
            "移动端开发岗",
        ],
        "数据类": [
            "数据分析岗",
            "大数据开发岗",
        ],
        "测试类": [
            "测试开发岗",
            "业务测试岗",
        ],
    },
    "产品序列": {
        "产品策划与设计类": [
            "创意设计岗",
            "用户体验设计岗",
            "产品经理",
        ],
    },
}


# =============================================================================
# 岗位模糊匹配关键词
# =============================================================================
# 从每个资格岗位提取核心关键词，用于模糊匹配。
# 企微通讯录返回的 position 可能包含额外描述（如"高级前端开发岗"），
# 只要 position 包含核心关键词即视为匹配。
# 关键词按长度降序排列，优先匹配更精确的关键词，避免误匹配。
# 例："运维开发岗" 的关键词为 ["运维开发", "运维"]，
#     先匹配"运维开发"（精确），不中再匹配"运维"（宽泛）。

BYOD_POSITION_MATCH_KEYWORDS: dict[str, list[str]] = {
    "算法岗": ["算法"],
    "前端开发岗": ["前端开发", "前端"],
    "后端开发岗": ["后端开发", "后端"],
    "客户端开发岗": ["客户端开发", "客户端"],
    "运维开发岗": ["运维开发", "运维"],
    "移动端开发岗": ["移动端开发", "移动端", "移动开发"],
    "数据分析岗": ["数据分析"],
    "大数据开发岗": ["大数据开发", "大数据"],
    "测试开发岗": ["测试开发"],
    "业务测试岗": ["业务测试"],
    "创意设计岗": ["创意设计"],
    "用户体验设计岗": ["用户体验设计", "用户体验", "UX设计", "UE设计", "交互设计"],
    "产品经理": ["产品经理"],
}


# =============================================================================
# BYOD 预过滤关键词
# =============================================================================
# 用于快速过滤非 BYOD 相关消息，避免每条消息都调 Dify。

BYOD_PREFILTER_KEYWORDS: list[str] = [
    "自备电脑",
    "电脑补贴",
    "BYOD",
    "byod",
    "自带电脑",
    "个人电脑补贴",
    "补贴资格",
    "电脑补贴资格",
    "自备电脑补贴",
    "自带设备",
]


# =============================================================================
# 申请链接 & 注意事项
# =============================================================================

BYOD_APPLICATION_URL: str = (
    "https://ehr.servyou.com.cn/HRAPP/FlowMobile/InitiateIns.aspx"
    "?flowid=7747&desc=自备电脑使用申请"
)

BYOD_NOTES: list[str] = [
    "补贴按月发放，需提供个人电脑的购买凭证",
    "领取补贴期间不得再领用公司电脑",
    "如已领用公司电脑，需先退还后方可申请补贴",
    "补贴金额和发放规则以公司最新政策为准",
    "申请提交后由部门审批，审批进度可在eHR系统查看",
]

# 登记注意事项（非补贴岗位 — 仅登记无补贴）
BYOD_REGISTER_NOTES: list[str] = [
    "自备电脑登记不享受电脑补贴",
    "登记后仍需遵守公司信息安全管理规定",
    "如已领用公司电脑，需先退还后方可登记自备电脑",
    "自备电脑需满足公司办公基本配置要求",
    "申请提交后由部门审批，审批进度可在eHR系统查看",
]


# =============================================================================
# Schema 定义
# =============================================================================

class ByodEligibilityRequest(BaseModel):
    """BYOD 意图检测请求

    Attributes:
        text: 用户消息文本
        employee_id: 员工ID（可选，传给 Dify 作为用户标识）
    """
    text: str
    employee_id: Optional[str] = None


class ByodCheckEligibilityRequest(BaseModel):
    """BYOD 资格检查请求

    Attributes:
        employee_id: 员工的企微 UserID
    """
    employee_id: str


class ByodEligibilityResponse(BaseModel):
    """BYOD 资格查询响应

    Attributes:
        is_byod_intent: 是否为自备电脑补贴意图
        eligible: 是否有资格申请补贴（语义同 has_subsidy，保留兼容）
        has_subsidy: 是否有补贴（True=有补贴, False=仅登记无补贴）
        position: 员工岗位（企微通讯录返回的 position 字段）
        matched_category: 匹配到的资格类别（如"技术序列 - 开发类"），未匹配时为空
        application_url: 申请/登记链接（所有岗位都返回，获取失败除外）
        notes: 注意事项列表（有补贴=补贴须知, 无补贴=登记须知）
        reason: 提示信息（无补贴时说明"可登记但无补贴"）
        source: 结果来源 — dify(Dify识别) / keyword_prefilter(关键词预过滤未命中) / fallback(降级兜底) / wecom(企微通讯录查询)
    """
    is_byod_intent: bool
    eligible: bool
    has_subsidy: bool = False
    position: str = ""
    matched_category: str = ""
    application_url: str = ""
    notes: list[str] = []
    reason: str = ""
    source: str


# =============================================================================
# 岗位匹配逻辑
# =============================================================================

def _match_position(position: str) -> tuple[bool, str, str]:
    """将员工岗位与资格清单进行匹配。

    匹配策略（按优先级）：
    1. 精确匹配 — 员工岗位与资格岗位完全一致
    2. 包含匹配 — 资格岗位是员工岗位的子串（如"前端开发岗"在"高级前端开发岗"中）
    3. 关键词匹配 — 员工岗位包含资格岗位的核心关键词（如"前端"在"高级前端开发岗"中）

    Args:
        position: 企微通讯录返回的员工岗位字符串

    Returns:
        tuple: (是否匹配, 匹配到的资格岗位名称, 匹配到的类别)
               类别格式为"序列 - 类别"（如"技术序列 - 开发类"），未匹配时为空字符串
    """
    if not position:
        return False, "", ""

    position_stripped = position.strip()

    for sequence, categories in BYOD_ELIGIBLE_POSITIONS.items():
        for category, positions in categories.items():
            for eligible_pos in positions:
                # 1. 精确匹配
                if position_stripped == eligible_pos:
                    return True, eligible_pos, f"{sequence} - {category}"

                # 2. 包含匹配（资格岗位是员工岗位的子串）
                if eligible_pos in position_stripped:
                    return True, eligible_pos, f"{sequence} - {category}"

                # 3. 核心关键词匹配
                keywords = BYOD_POSITION_MATCH_KEYWORDS.get(eligible_pos, [])
                for keyword in keywords:
                    if keyword in position_stripped:
                        return True, eligible_pos, f"{sequence} - {category}"

    return False, "", ""


# =============================================================================
# Dify 意图识别（复用审批意图识别 Dify 应用）
# =============================================================================
# 说明：BYOD 意图识别复用审批系统的 Dify 应用（同一个 approval_dify_base_url /
#       approval_dify_api_key），Dify 后台的 System Prompt 已更新为同时支持
#       审批意图和 BYOD 意图识别。Dify 返回的 JSON 中包含 is_byod_intent 字段。

async def _call_dify_byod_intent(text: str, employee_id: str = "") -> dict:
    """调用 Dify 意图识别应用，检测 BYOD 意图。

    复用审批意图识别的 Dify 应用（approval_dify_base_url / approval_dify_api_key），
    Dify 的 System Prompt 已更新为同时返回 is_byod_intent 字段。

    Dify 原生 API 调用方式与 approval.py 中的 _call_dify_approval_intent 一致：
    POST {base_url}/v1/chat-messages，blocking 模式。

    Args:
        text: 用户消息文本
        employee_id: 员工 ID（可选，传给 Dify 的 user 字段）

    Returns:
        dict: {
            "is_byod_intent": bool,
            "confidence": float,
        }

    Raises:
        Exception: Dify 调用失败或响应解析失败
    """
    base_url = settings.approval_dify_base_url
    api_key = settings.approval_dify_api_key
    timeout = settings.approval_dify_timeout

    if not base_url or not api_key:
        raise ValueError(
            "Dify 意图识别应用未配置（APPROVAL_DIFY_BASE_URL / APPROVAL_DIFY_API_KEY）"
        )

    # 构建请求 URL：base_url + /v1/chat-messages（Dify 原生 API）
    url = f"{base_url.rstrip('/')}/v1/chat-messages"

    body = {
        "inputs": {},
        "query": text,
        "response_mode": "blocking",
        "user": employee_id or "byod_detection",
    }
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    async with httpx.AsyncClient(timeout=httpx.Timeout(timeout)) as client:
        response = await client.post(url, json=body, headers=headers)
        response.raise_for_status()
        data = response.json()

        # 解析 Dify 原生响应：answer 字段包含 AI 返回的文本（JSON 字符串）
        answer = data.get("answer", "")
        parsed = json.loads(answer)

        return {
            "is_byod_intent": bool(parsed.get("is_byod_intent", False)),
            "confidence": float(parsed.get("confidence", 0.0)),
        }


# =============================================================================
# 关键词预过滤 & 降级兜底
# =============================================================================

def _byod_keyword_prefilter(text: str) -> bool:
    """BYOD 关键词预过滤：检查文本是否包含自备电脑补贴相关关键词。

    只要命中 BYOD_PREFILTER_KEYWORDS 中的任意一个关键词即返回 True，
    未命中返回 False。用于避免每条消息都调 Dify。

    Args:
        text: 用户消息文本

    Returns:
        bool: 是否包含 BYOD 相关关键词
    """
    if not text:
        return False
    lower_text = text.lower()
    return any(kw.lower() in lower_text for kw in BYOD_PREFILTER_KEYWORDS)


def _byod_fallback_detect(text: str) -> tuple[bool, float]:
    """BYOD 关键词兜底：Dify 不可用时通过关键词匹配判断 BYOD 意图。

    遍历 BYOD_PREFILTER_KEYWORDS，命中任意一个即认为有 BYOD 意图。
    置信度取 0.6（预过滤已通过说明有 BYOD 相关关键词）。

    Args:
        text: 用户消息文本

    Returns:
        tuple: (is_byod_intent, confidence)
    """
    lower_text = (text or "").lower()
    for kw in BYOD_PREFILTER_KEYWORDS:
        if kw.lower() in lower_text:
            return True, 0.6
    return False, 0.0


# =============================================================================
# API 端点
# =============================================================================

@router.post("/byod/detect-intent")
async def detect_byod_intent(request: ByodEligibilityRequest):
    """BYOD 意图检测端点。

    流程（复用审批系统三级模式）：
    1. 关键词预过滤 — 未命中 BYOD 关键词直接返回 false（避免每条消息都调 Dify）
    2. 命中关键词 → 调用 Dify 意图识别应用（复用审批 Dify 应用）
    3. Dify 调用失败 → 降级为关键词匹配（兜底）

    Args:
        request: 包含 text（用户消息）和可选的 employee_id

    Returns:
        ByodEligibilityResponse: 检测结果（仅 is_byod_intent 和 source 字段有效）
    """
    text = request.text or ""

    # 1. 关键词预过滤
    if not _byod_keyword_prefilter(text):
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=False,
            eligible=False,
            source="keyword_prefilter",
        ))

    # 2. 调用 Dify 意图识别
    try:
        result = await _call_dify_byod_intent(text, request.employee_id or "")
        threshold = settings.approval_confidence_threshold
        is_byod = result["is_byod_intent"] and result["confidence"] >= threshold
        logger.info(
            f"BYOD意图检测(Dify): is_byod={is_byod}, "
            f"confidence={result['confidence']}"
        )
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=is_byod,
            eligible=False,
            source="dify",
        ))
    except Exception as e:
        logger.warning(f"Dify BYOD 意图识别失败，降级为关键词匹配: {e}")
        # 3. 降级为关键词匹配
        is_byod, confidence = _byod_fallback_detect(text)
        logger.info(
            f"BYOD意图检测(兜底): is_byod={is_byod}, confidence={confidence}"
        )
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=is_byod,
            eligible=False,
            source="fallback",
        ))


@router.post("/byod/check-eligibility")
async def check_byod_eligibility(
    request: ByodCheckEligibilityRequest,
    redis: aioredis.Redis = Depends(get_redis),
):
    """BYOD 资格检查端点。

    流程：
    1. 通过企微通讯录 API 获取员工岗位（position）
    2. 将岗位与资格清单进行匹配（支持精确匹配和模糊匹配）
    3. 返回判定结果：
       - 可申请 → 返回申请链接 + 注意事项
       - 不可申请 → 返回原因

    限制性条件（已领公司电脑等）当前版本作为注意事项提示，
    暂不接入资产系统自动查询。

    Args:
        request: 包含 employee_id（员工企微 UserID）
        redis: Redis 客户端（依赖注入，用于 WecomService 的 token 缓存）

    Returns:
        ByodEligibilityResponse: 资格检查结果
    """
    employee_id = request.employee_id

    if not employee_id:
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=True,
            eligible=False,
            reason="缺少员工ID，无法查询岗位信息",
            source="wecom",
        ))

    # 1. 通过企微通讯录 API 获取员工信息
    wecom_service = WecomService(redis_client=redis)
    try:
        user_info = await wecom_service.get_user_info(employee_id)
    except Exception as e:
        logger.error(f"获取员工信息失败: employee_id={employee_id}, error={e}")
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=True,
            eligible=False,
            reason=f"获取员工信息失败：{e}",
            source="wecom",
        ))
    finally:
        await wecom_service.close()

    # 2. 提取岗位信息
    position = user_info.get("position", "")
    employee_name = user_info.get("name", "")

    if not position:
        logger.warning(f"员工岗位为空: employee_id={employee_id}, name={employee_name}")
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=True,
            eligible=False,
            position="",
            reason="未能获取到您的岗位信息，请联系IT服务台人工核实",
            source="wecom",
        ))

    # 3. 岗位匹配
    matched, matched_position, matched_category = _match_position(position)

    if matched:
        logger.info(
            f"BYOD资格检查通过: employee_id={employee_id}, name={employee_name}, "
            f"position={position}, matched={matched_position}, category={matched_category}"
        )
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=True,
            eligible=True,
            has_subsidy=True,
            position=position,
            matched_category=matched_category,
            application_url=BYOD_APPLICATION_URL,
            notes=BYOD_NOTES,
            source="wecom",
        ))
    else:
        logger.info(
            f"BYOD资格检查未通过: employee_id={employee_id}, name={employee_name}, "
            f"position={position}, 无匹配的资格岗位"
        )
        return success_response(data=ByodEligibilityResponse(
            is_byod_intent=True,
            eligible=False,
            has_subsidy=False,
            position=position,
            application_url=BYOD_APPLICATION_URL,  # 非补贴也提供登记链接
            notes=BYOD_REGISTER_NOTES,  # 登记注意事项（非补贴）
            reason=f"您的岗位「{position}」不在自备电脑补贴资格清单中，但仍可进行自备电脑登记（无补贴）",
            source="wecom",
        ))


# =============================================================================
# 辅助端点：获取资格岗位清单（供前端展示或调试用）
# =============================================================================

@router.get("/byod/eligible-positions")
async def get_byod_eligible_positions():
    """获取自备电脑补贴资格岗位清单。

    返回所有有资格申请自备电脑补贴的岗位，按序列和类别分组。

    Returns:
        资格岗位清单字典
    """
    return success_response(data={
        "positions": BYOD_ELIGIBLE_POSITIONS,
        "total_count": sum(
            len(positions)
            for categories in BYOD_ELIGIBLE_POSITIONS.values()
            for positions in categories.values()
        ),
        "application_url": BYOD_APPLICATION_URL,
        "notes": BYOD_NOTES,
    })