259 lines
8.4 KiB
Python
259 lines
8.4 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
from datetime import datetime, timedelta
|
||
from typing import Any
|
||
|
||
from app.integrations.ehr import EhrClient
|
||
|
||
|
||
logger = logging.getLogger("connecthub.extensions.sync_ehr_to_oa")
|
||
|
||
|
||
class SyncEhrToOaApi:
|
||
"""
|
||
北森 EHR -> OA 同步 API 封装。
|
||
|
||
已封装 API：
|
||
- 员工与单条任职时间窗滚动查询
|
||
- 组织单元时间窗滚动查询
|
||
- 职务时间窗滚动查询
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
secret_params: dict[str, str],
|
||
base_url: str = "https://openapi.italent.cn",
|
||
timeout_s: float = 10.0,
|
||
retries: int = 2,
|
||
retry_backoff_s: float = 0.5,
|
||
) -> None:
|
||
self._client = EhrClient(
|
||
base_url=base_url,
|
||
secret_params=secret_params,
|
||
timeout_s=timeout_s,
|
||
retries=retries,
|
||
retry_backoff_s=retry_backoff_s,
|
||
)
|
||
|
||
def close(self) -> None:
|
||
self._client.close()
|
||
|
||
@staticmethod
|
||
def _to_datetime(value: datetime | str | None) -> datetime:
|
||
if value is None:
|
||
return datetime.now()
|
||
if isinstance(value, datetime):
|
||
return value
|
||
s = str(value).strip()
|
||
if not s:
|
||
raise ValueError("datetime string cannot be empty")
|
||
if "T" in s:
|
||
return datetime.fromisoformat(s)
|
||
if " " in s:
|
||
return datetime.fromisoformat(s.replace(" ", "T"))
|
||
return datetime.strptime(s, "%Y-%m-%d")
|
||
|
||
@staticmethod
|
||
def _to_api_datetime(value: datetime | str | None) -> str:
|
||
return SyncEhrToOaApi._to_datetime(value).strftime("%Y-%m-%dT%H:%M:%S")
|
||
|
||
@staticmethod
|
||
def _iter_windows(start_dt: datetime, stop_dt: datetime, max_days: int = 90) -> list[tuple[datetime, datetime]]:
|
||
if stop_dt < start_dt:
|
||
raise ValueError("stop_time must be greater than or equal to start_time")
|
||
windows: list[tuple[datetime, datetime]] = []
|
||
cur = start_dt
|
||
max_delta = timedelta(days=max_days)
|
||
while cur < stop_dt:
|
||
nxt = min(cur + max_delta, stop_dt)
|
||
windows.append((cur, nxt))
|
||
cur = nxt
|
||
if not windows:
|
||
windows.append((start_dt, stop_dt))
|
||
return windows
|
||
|
||
def _get_all_by_time_window(
|
||
self,
|
||
*,
|
||
api_path: str,
|
||
api_name: str,
|
||
stop_time: datetime | str | None,
|
||
capacity: int,
|
||
time_window_query_type: int,
|
||
with_disabled: bool,
|
||
is_with_deleted: bool,
|
||
max_pages: int,
|
||
) -> dict[str, Any]:
|
||
if capacity <= 0 or capacity > 300:
|
||
raise ValueError("capacity must be in range [1, 300]")
|
||
if max_pages <= 0:
|
||
raise ValueError("max_pages must be > 0")
|
||
|
||
start_dt = datetime(2015, 1, 1, 0, 0, 0)
|
||
stop_dt = self._to_datetime(stop_time)
|
||
windows = self._iter_windows(start_dt=start_dt, stop_dt=stop_dt, max_days=90)
|
||
|
||
all_data: list[dict[str, Any]] = []
|
||
total_pages = 0
|
||
last_scroll_id = ""
|
||
|
||
for idx, (w_start, w_stop) in enumerate(windows, start=1):
|
||
start_time = self._to_api_datetime(w_start)
|
||
stop_time_s = self._to_api_datetime(w_stop)
|
||
scroll_id = ""
|
||
page = 0
|
||
window_total = 0
|
||
|
||
while True:
|
||
page += 1
|
||
total_pages += 1
|
||
if page > max_pages:
|
||
raise RuntimeError(f"scroll pages exceed max_pages={max_pages} in window index={idx}")
|
||
|
||
body: dict[str, Any] = {
|
||
"startTime": start_time,
|
||
"stopTime": stop_time_s,
|
||
"timeWindowQueryType": time_window_query_type,
|
||
"scrollId": scroll_id,
|
||
"capacity": capacity,
|
||
"withDisabled": with_disabled,
|
||
"isWithDeleted": is_with_deleted,
|
||
}
|
||
|
||
resp = self._client.request(
|
||
"POST",
|
||
api_path,
|
||
json=body,
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
payload = resp.json() if resp.content else {}
|
||
|
||
code = str(payload.get("code", "") or "")
|
||
if code != "200":
|
||
message = payload.get("message")
|
||
raise RuntimeError(f"EHR {api_name} failed code={code!r} message={message!r}")
|
||
|
||
batch = payload.get("data") or []
|
||
if not isinstance(batch, list):
|
||
raise RuntimeError(f"EHR {api_name} invalid response: data is not a list")
|
||
all_data.extend([x for x in batch if isinstance(x, dict)])
|
||
|
||
total_val = payload.get("total")
|
||
if total_val is not None:
|
||
try:
|
||
window_total = int(total_val)
|
||
except (TypeError, ValueError):
|
||
pass
|
||
|
||
is_last_data = bool(payload.get("isLastData", False))
|
||
scroll_id = str(payload.get("scrollId", "") or "")
|
||
last_scroll_id = scroll_id
|
||
|
||
logger.info(
|
||
"EHR %s window=%s/%s page=%s batch=%s window_total=%s isLastData=%s",
|
||
api_name,
|
||
idx,
|
||
len(windows),
|
||
page,
|
||
len(batch),
|
||
window_total,
|
||
is_last_data,
|
||
)
|
||
|
||
if is_last_data:
|
||
break
|
||
|
||
return {
|
||
"startTime": self._to_api_datetime(start_dt),
|
||
"stopTime": self._to_api_datetime(stop_dt),
|
||
"total": len(all_data),
|
||
"pages": total_pages,
|
||
"count": len(all_data),
|
||
"data": all_data,
|
||
"lastScrollId": last_scroll_id,
|
||
"windowCount": len(windows),
|
||
}
|
||
|
||
def get_all_employees_with_record_by_time_window(
|
||
self,
|
||
*,
|
||
stop_time: datetime | str | None = None,
|
||
capacity: int = 300,
|
||
time_window_query_type: int = 1,
|
||
with_disabled: bool = True,
|
||
is_with_deleted: bool = True,
|
||
max_pages: int = 100000,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
滚动查询“员工 + 单条任职”全量结果。
|
||
|
||
固定起始时间：
|
||
- 2015-01-01T00:00:00
|
||
"""
|
||
return self._get_all_by_time_window(
|
||
api_path="/TenantBaseExternal/api/v5/Employee/GetByTimeWindow",
|
||
api_name="Employee.GetByTimeWindow",
|
||
stop_time=stop_time,
|
||
capacity=capacity,
|
||
time_window_query_type=time_window_query_type,
|
||
with_disabled=with_disabled,
|
||
is_with_deleted=is_with_deleted,
|
||
max_pages=max_pages,
|
||
)
|
||
|
||
def get_all_organizations_by_time_window(
|
||
self,
|
||
*,
|
||
stop_time: datetime | str | None = None,
|
||
capacity: int = 300,
|
||
time_window_query_type: int = 1,
|
||
with_disabled: bool = True,
|
||
is_with_deleted: bool = True,
|
||
max_pages: int = 100000,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
滚动查询“组织单元”全量结果。
|
||
|
||
固定起始时间：
|
||
- 2015-01-01T00:00:00
|
||
"""
|
||
return self._get_all_by_time_window(
|
||
api_path="/TenantBaseExternal/api/v5/Organization/GetByTimeWindow",
|
||
api_name="Organization.GetByTimeWindow",
|
||
stop_time=stop_time,
|
||
capacity=capacity,
|
||
time_window_query_type=time_window_query_type,
|
||
with_disabled=with_disabled,
|
||
is_with_deleted=is_with_deleted,
|
||
max_pages=max_pages,
|
||
)
|
||
|
||
def get_all_job_posts_by_time_window(
|
||
self,
|
||
*,
|
||
stop_time: datetime | str | None = None,
|
||
capacity: int = 300,
|
||
time_window_query_type: int = 1,
|
||
with_disabled: bool = True,
|
||
is_with_deleted: bool = True,
|
||
max_pages: int = 100000,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
滚动查询“职务”全量结果。
|
||
|
||
固定起始时间：
|
||
- 2015-01-01T00:00:00
|
||
"""
|
||
return self._get_all_by_time_window(
|
||
api_path="/TenantBaseExternal/api/v5/JobPost/GetByTimeWindow",
|
||
api_name="JobPost.GetByTimeWindow",
|
||
stop_time=stop_time,
|
||
capacity=capacity,
|
||
time_window_query_type=time_window_query_type,
|
||
with_disabled=with_disabled,
|
||
is_with_deleted=is_with_deleted,
|
||
max_pages=max_pages,
|
||
)
|