147 lines
4.9 KiB
Python
147 lines
4.9 KiB
Python
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import pytds
|
|
import requests
|
|
|
|
|
|
def env(name: str, default: str | None = None, required: bool = False) -> str:
|
|
value = os.getenv(name, default)
|
|
if isinstance(value, str):
|
|
value = value.strip()
|
|
if required and not value:
|
|
raise RuntimeError(f"missing env: {name}")
|
|
return value or ""
|
|
|
|
|
|
def get_connection():
|
|
return pytds.connect(
|
|
server=env("HR_SQLSERVER_HOST", required=True),
|
|
port=int(env("HR_SQLSERVER_PORT", "1433")),
|
|
database=env("HR_SQLSERVER_DATABASE", required=True),
|
|
user=env("HR_SQLSERVER_USER", required=True),
|
|
password=env("HR_SQLSERVER_PASSWORD", required=True),
|
|
as_dict=True,
|
|
login_timeout=15,
|
|
timeout=60,
|
|
)
|
|
|
|
|
|
def pick(row: dict[str, Any], *keys: str):
|
|
for key in keys:
|
|
if key in row and row[key] not in (None, ""):
|
|
return row[key]
|
|
return None
|
|
|
|
|
|
def stringify(value: Any) -> str | None:
|
|
if value in (None, ""):
|
|
return None
|
|
return str(value).strip()
|
|
|
|
|
|
def normalize_department(row: dict[str, Any]) -> dict[str, Any]:
|
|
external_id = stringify(pick(row, "Id", "id", "group_id"))
|
|
name = stringify(pick(row, "name", "Name"))
|
|
if not external_id or not name:
|
|
raise RuntimeError("department row missing external id or name")
|
|
|
|
return {
|
|
"externalId": external_id,
|
|
"name": name,
|
|
"parentExternalId": stringify(pick(row, "parent_id", "parentId", "ParentId")),
|
|
"orderIndex": int(pick(row, "sort", "sortOrder", "SortOrder") or 0),
|
|
"rawSnapshot": row,
|
|
}
|
|
|
|
|
|
def normalize_employee(row: dict[str, Any]) -> dict[str, Any]:
|
|
dept_list = pick(row, "dept_id_list", "deptIdList", "dept_ids")
|
|
department_external_ids = []
|
|
if isinstance(dept_list, str) and dept_list.strip():
|
|
department_external_ids = [item.strip() for item in dept_list.split(",") if item.strip()]
|
|
elif isinstance(dept_list, (list, tuple)):
|
|
department_external_ids = [str(item).strip() for item in dept_list if str(item).strip()]
|
|
|
|
primary_department = department_external_ids[0] if department_external_ids else None
|
|
name = stringify(pick(row, "name", "Name"))
|
|
if not name:
|
|
raise RuntimeError("employee row missing name")
|
|
|
|
return {
|
|
"dingtalkUserId": stringify(pick(row, "userid", "userId", "UserId")),
|
|
"dingtalkUnionId": stringify(pick(row, "unionid", "unionId", "UnionId")),
|
|
"jobNumber": stringify(pick(row, "job_number", "jobNumber", "JobNumber")),
|
|
"name": name,
|
|
"mobile": stringify(pick(row, "mobile", "Mobile")),
|
|
"email": stringify(pick(row, "email", "Email")),
|
|
"title": stringify(pick(row, "title", "Title")),
|
|
"primaryDepartmentExternalId": stringify(primary_department),
|
|
"departmentExternalIds": department_external_ids,
|
|
"rawSnapshot": row,
|
|
}
|
|
|
|
|
|
def normalize_dimission(row: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"dingtalkUserId": stringify(pick(row, "userid", "userId", "UserId")),
|
|
"dingtalkUnionId": stringify(pick(row, "unionid", "unionId", "UnionId")),
|
|
"mobile": stringify(pick(row, "mobile", "Mobile")),
|
|
"jobNumber": stringify(pick(row, "job_number", "jobNumber", "JobNumber")),
|
|
"rawSnapshot": row,
|
|
}
|
|
|
|
|
|
def fetch_rows(cursor, query: str):
|
|
cursor.execute(query)
|
|
return list(cursor.fetchall())
|
|
|
|
|
|
def build_payload() -> dict:
|
|
with get_connection() as conn:
|
|
with conn.cursor() as cursor:
|
|
departments = fetch_rows(cursor, "select * from dbo.hr_groups")
|
|
employees = fetch_rows(cursor, "select * from dbo.hr_employee")
|
|
dimissions = fetch_rows(cursor, "select * from dbo.hr_employee_dimission")
|
|
|
|
return {
|
|
"departments": [normalize_department(row) for row in departments if pick(row, "Id", "id", "group_id")],
|
|
"employees": [normalize_employee(row) for row in employees if pick(row, "name", "Name")],
|
|
"dimissions": [normalize_dimission(row) for row in dimissions],
|
|
}
|
|
|
|
|
|
def push_payload(payload: dict):
|
|
endpoint = env("HR_SYNC_ENDPOINT", required=True)
|
|
secret = env("HR_SYNC_SECRET", required=True)
|
|
response = requests.post(
|
|
endpoint,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"x-internal-sync-secret": secret,
|
|
},
|
|
data=json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8"),
|
|
timeout=120,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def main():
|
|
started_at = datetime.now().isoformat()
|
|
print(f"[{started_at}] start hr sync")
|
|
payload = build_payload()
|
|
result = push_payload(payload)
|
|
print(json.dumps(result, ensure_ascii=False, default=str))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"hr sync failed: {exc}", file=sys.stderr)
|
|
raise
|