77 lines
3.1 KiB
Python
77 lines
3.1 KiB
Python
|
# =============================================================================
|
|||
|
|
# 企微IT智能服务台 — 阶段5 自动化 北森 EHR 客户端（静态映射兜底）
|
|||
|
|
# =============================================================================
|
|||
|
|
# 说明：当联软（主映射源）无法解析员工→终端时，使用北森 EHR 提供的
|
|||
|
|
# 员工-部门-资产静态映射作为兜底。
|
|||
|
|
#
|
|||
|
|
# 认证：北森开放 API 通常使用 App Key + App Secret（Bearer 或签名）。
|
|||
|
|
# 此处用占位实现：AUTOMATION_EHR_BASE_URL / AUTOMATION_EHR_API_KEY。
|
|||
|
|
# 具体签名方式以真实环境文档为准，结构上可被单测 mock。
|
|||
|
|
#
|
|||
|
|
# 接口设计为占位骨架：方法签名完整、异常可捕获、出入参可被 ActionLog 记录，
|
|||
|
|
# 本地无真实密钥/环境时仅结构正确，可被单元测试 mock。
|
|||
|
|
# =============================================================================
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import logging
|
|||
|
|
from typing import Any, Dict, Optional
|
|||
|
|
|
|||
|
|
from app.config import settings
|
|||
|
|
from app.integrations.base import BaseClient, BaseClientError
|
|||
|
|
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class EhrClient(BaseClient):
|
|||
|
|
"""北森 EHR 客户端（静态映射兜底）。"""
|
|||
|
|
|
|||
|
|
system_name = "ehr"
|
|||
|
|
|
|||
|
|
def __init__(self, api_key: str, base_url: str, timeout=None, audit=None):
|
|||
|
|
super().__init__(base_url=base_url, timeout=timeout, audit=audit)
|
|||
|
|
self.api_key = api_key
|
|||
|
|
|
|||
|
|
def _headers(self) -> Dict[str, str]:
|
|||
|
|
return {
|
|||
|
|
"Authorization": f"Bearer {self.api_key}",
|
|||
|
|
"Content-Type": "application/json",
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async def get_employee_profile(self, employee_id: str) -> Dict[str, Any]:
|
|||
|
|
"""查询员工档案（部门、岗位、资产编号等），作为映射兜底。"""
|
|||
|
|
return await self.request(
|
|||
|
|
"GET",
|
|||
|
|
f"/api/v1/employees/{employee_id}",
|
|||
|
|
headers=self._headers(),
|
|||
|
|
event="ehr.get_employee_profile",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
async def get_terminal_by_employee(self, employee_id: str) -> Optional[Dict[str, Any]]:
|
|||
|
|
"""根据员工查兜底终端信息。
|
|||
|
|
|
|||
|
|
北森通常只给资产编号/部门，真正的终端 IP 仍需联软；
|
|||
|
|
此处返回 hint（如 last_known_hostname / asset_no），供 mapping_resolver 合并。
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
profile = await self.get_employee_profile(employee_id)
|
|||
|
|
return {
|
|||
|
|
"employee_id": employee_id,
|
|||
|
|
"department": profile.get("department", ""),
|
|||
|
|
"asset_no": profile.get("asset_no", ""),
|
|||
|
|
"terminal_hint": profile.get("last_known_hostname", ""),
|
|||
|
|
"source": "ehr",
|
|||
|
|
}
|
|||
|
|
except BaseClientError as e:
|
|||
|
|
logger.warning(f"EHR 映射兜底失败 employee={employee_id}: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def get_ehr_client(audit=None) -> Optional[EhrClient]:
|
|||
|
|
"""从 settings 构建 EHR 客户端；未配置返回 None。"""
|
|||
|
|
base_url = settings.automation_ehr_base_url
|
|||
|
|
api_key = settings.automation_ehr_api_key
|
|||
|
|
if not base_url or not api_key:
|
|||
|
|
return None
|
|||
|
|
return EhrClient(api_key=api_key, base_url=base_url, audit=audit)
|