This commit is contained in:
Marsway 2026-03-30 17:14:20 +08:00
parent ea01e8455a
commit 96b3439c5b
2 changed files with 85 additions and 9 deletions

View File

@ -295,15 +295,56 @@ class SyncEhrLeavesToOaApi:
start_date: date, start_date: date,
end_date: date, end_date: date,
) -> dict[tuple[str, str], int]: ) -> dict[tuple[str, str], int]:
rows = self.get_oa_rows_by_job_and_date(
table_name=table_name,
schema=schema,
job_no_column=job_no_column,
date_column=date_column,
leave_days_column=None,
name_column=None,
start_date=start_date,
end_date=end_date,
)
out: dict[tuple[str, str], int] = {}
for k, v in rows.items():
rid = _to_int_safe((v or {}).get("id"))
if rid > 0:
out[k] = rid
return out
def get_oa_rows_by_job_and_date(
self,
*,
table_name: str,
schema: str,
job_no_column: str,
date_column: str,
leave_days_column: str | None,
name_column: str | None,
start_date: date,
end_date: date,
) -> dict[tuple[str, str], dict[str, str]]:
t = str(table_name or "").strip() t = str(table_name or "").strip()
jc = str(job_no_column or "").strip() jc = str(job_no_column or "").strip()
dc = str(date_column or "").strip() dc = str(date_column or "").strip()
lc = str(leave_days_column or "").strip()
nc = str(name_column or "").strip()
s = str(schema or "").strip() or "dbo" s = str(schema or "").strip() or "dbo"
if not t or not jc or not dc: if not t or not jc or not dc:
raise ValueError("table_name/job_no_column/date_column are required") raise ValueError("table_name/job_no_column/date_column are required")
select_parts = [
"[id] AS row_id",
f"[{jc}] AS job_no",
f"[{dc}] AS leave_date",
]
if lc:
select_parts.append(f"[{lc}] AS leave_days")
if nc:
select_parts.append(f"[{nc}] AS staff_name")
sql = text( sql = text(
f"SELECT [id] AS row_id, [{jc}] AS job_no, [{dc}] AS leave_date " f"SELECT {', '.join(select_parts)} "
f"FROM [{s}].[{t}] WITH (NOLOCK) " f"FROM [{s}].[{t}] WITH (NOLOCK) "
f"WHERE TRY_CONVERT(date, [{dc}]) >= :start_date " f"WHERE TRY_CONVERT(date, [{dc}]) >= :start_date "
f"AND TRY_CONVERT(date, [{dc}]) <= :end_date " f"AND TRY_CONVERT(date, [{dc}]) <= :end_date "
@ -313,7 +354,7 @@ class SyncEhrLeavesToOaApi:
"start_date": start_date.isoformat(), "start_date": start_date.isoformat(),
"end_date": end_date.isoformat(), "end_date": end_date.isoformat(),
} }
out: dict[tuple[str, str], int] = {} out: dict[tuple[str, str], dict[str, str]] = {}
duplicate_keys = 0 duplicate_keys = 0
with self._sql_engine.connect() as conn: with self._sql_engine.connect() as conn:
rows = conn.execute(sql, params).fetchall() rows = conn.execute(sql, params).fetchall()
@ -328,9 +369,21 @@ class SyncEhrLeavesToOaApi:
if not job_no or not leave_date: if not job_no or not leave_date:
continue continue
key = (job_no, leave_date) key = (job_no, leave_date)
if key in out and out[key] != row_id: leave_days = ""
if lc:
leave_days = str(getattr(r, "leave_days", "") or "").strip()
staff_name = ""
if nc:
staff_name = str(getattr(r, "staff_name", "") or "").strip()
if key in out and _to_int_safe((out[key] or {}).get("id")) != row_id:
duplicate_keys += 1 duplicate_keys += 1
out[key] = row_id out[key] = {
"id": str(row_id),
"job_no": job_no,
"leave_date": leave_date,
"leave_days": leave_days,
"name": staff_name,
}
logger.info( logger.info(
"OA 现有记录索引完成table=%s month_range=%s~%s indexed=%s duplicate_keys=%s", "OA 现有记录索引完成table=%s month_range=%s~%s indexed=%s duplicate_keys=%s",
t, t,

View File

@ -80,6 +80,10 @@ def _to_int_safe(v: Any) -> int:
return 0 return 0
def _normalize_decimal_1(v: Any) -> str:
return _decimal_to_str(_to_decimal(v))
class SyncEhrLeavesToOaMonthJob(BaseJob): class SyncEhrLeavesToOaMonthJob(BaseJob):
""" """
EHR 请假 -> OA 月度同步按工号+日期汇总 EHR 请假 -> OA 月度同步按工号+日期汇总
@ -266,11 +270,13 @@ class SyncEhrLeavesToOaMonthJob(BaseJob):
if staff_name and key not in name_map: if staff_name and key not in name_map:
name_map[key] = staff_name name_map[key] = staff_name
existing_id_map = ehr.get_oa_row_id_map_by_job_and_date( existing_row_map = ehr.get_oa_rows_by_job_and_date(
table_name=_OA_SQLSERVER_TABLE, table_name=_OA_SQLSERVER_TABLE,
schema=_OA_SQLSERVER_SCHEMA, schema=_OA_SQLSERVER_SCHEMA,
job_no_column=field_job_no, job_no_column=field_job_no,
date_column=field_leave_date, date_column=field_leave_date,
leave_days_column=field_leave_days,
name_column=field_name,
start_date=month_start, start_date=month_start,
end_date=month_end, end_date=month_end,
) )
@ -278,21 +284,32 @@ class SyncEhrLeavesToOaMonthJob(BaseJob):
data_list: list[dict[str, Any]] = [] data_list: list[dict[str, Any]] = []
to_update = 0 to_update = 0
to_insert = 0 to_insert = 0
skipped_unchanged = 0
for (job_no, leave_date), leave_days in sorted(agg.items(), key=lambda x: (x[0][0], x[0][1])): for (job_no, leave_date), leave_days in sorted(agg.items(), key=lambda x: (x[0][0], x[0][1])):
leave_date_value = f"{leave_date} 00:00:00" leave_date_value = f"{leave_date} 00:00:00"
leave_days_value = _decimal_to_str(leave_days)
name_val = name_map.get((job_no, leave_date), "")
existing_row = existing_row_map.get((job_no, leave_date))
if existing_row is not None:
old_days = _normalize_decimal_1(existing_row.get("leave_days"))
old_name = str(existing_row.get("name") or "").strip()
if old_days == leave_days_value and (not field_name or old_name == name_val):
skipped_unchanged += 1
continue
fields_payload = [ fields_payload = [
{"name": field_job_no, "value": job_no, "showValue": job_no}, {"name": field_job_no, "value": job_no, "showValue": job_no},
{"name": field_leave_date, "value": leave_date_value, "showValue": leave_date}, {"name": field_leave_date, "value": leave_date_value, "showValue": leave_date},
{"name": field_leave_days, "value": _decimal_to_str(leave_days), "showValue": _decimal_to_str(leave_days)}, {"name": field_leave_days, "value": leave_days_value, "showValue": leave_days_value},
] ]
if field_name: if field_name:
name_val = name_map.get((job_no, leave_date), "")
fields_payload.append({"name": field_name, "value": name_val, "showValue": name_val}) fields_payload.append({"name": field_name, "value": name_val, "showValue": name_val})
record: dict[str, Any] = { record: dict[str, Any] = {
"fields": fields_payload, "fields": fields_payload,
} }
existing_id = existing_id_map.get((job_no, leave_date)) existing_id = _to_int_safe((existing_row or {}).get("id"))
if existing_id is not None: if existing_id > 0:
record["id"] = existing_id record["id"] = existing_id
to_update += 1 to_update += 1
else: else:
@ -351,6 +368,11 @@ class SyncEhrLeavesToOaMonthJob(BaseJob):
data_local = rj.get("data") or {} data_local = rj.get("data") or {}
chunk_success = int(data_local.get("successCount", 0) or 0) chunk_success = int(data_local.get("successCount", 0) or 0)
chunk_failed = int(data_local.get("failCount", 0) or 0) chunk_failed = int(data_local.get("failCount", 0) or 0)
if chunk_success == 0 and chunk_failed == 0:
fd2 = data_local.get("failedData") or {}
fd_len = len(fd2) if isinstance(fd2, dict) else 0
chunk_failed = fd_len
chunk_success = max(0, len(chunk) - chunk_failed)
success_count += chunk_success success_count += chunk_success
failed_count += chunk_failed failed_count += chunk_failed
fd = data_local.get("failedData") or {} fd = data_local.get("failedData") or {}
@ -377,6 +399,7 @@ class SyncEhrLeavesToOaMonthJob(BaseJob):
"aggregated_records": len(agg), "aggregated_records": len(agg),
"to_update": to_update, "to_update": to_update,
"to_insert": to_insert, "to_insert": to_insert,
"skipped_unchanged": skipped_unchanged,
"success_count": success_count, "success_count": success_count,
"failed_count": failed_count, "failed_count": failed_count,
"skipped_no_job_no": skipped_no_job_no, "skipped_no_job_no": skipped_no_job_no,