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