fix:修复财务数据导入问题。
This commit is contained in:
@@ -175,12 +175,24 @@ def batch_insert(table_name: str, df: pd.DataFrame, conn, conflict_columns: List
|
||||
# 注意:去重必须在此处 (NaN->None 替换之前) 执行,
|
||||
# 此时日期列仍为 datetime64 类型,sort_values(na_position="last")
|
||||
# 能正确处理 NaT;若在替换之后排序,object 类型混合日期/None 排序不可靠。
|
||||
#
|
||||
# 关键修复:PostgreSQL 对 VARCHAR/数值列会做隐式类型转换,因此数据库眼里
|
||||
# report_type = '1' (str) 与 report_type = 1 (int) 是"同一个约束值";
|
||||
# 但 pandas 的 drop_duplicates 认为 '1' != 1,导致去重不彻底、仍报错。
|
||||
# 因此去重前先将非日期冲突列统一转成 str 再判断重复 (不修改原始 DataFrame)。
|
||||
dedup_cols = [c for c in conflict_columns if c in columns]
|
||||
before_dedup = len(df)
|
||||
date_cols = [c for c in ["f_ann_date", "ann_date"] if c in columns]
|
||||
if date_cols:
|
||||
df = df.sort_values(date_cols, na_position="last")
|
||||
df = df.drop_duplicates(subset=dedup_cols, keep="last")
|
||||
if dedup_cols:
|
||||
# 规范化去重键:非 datetime64 列统一 astype(str),消除类型不一致的隐患
|
||||
key_df = df[dedup_cols].copy()
|
||||
for c in key_df.columns:
|
||||
if not pd.api.types.is_datetime64_any_dtype(key_df[c]):
|
||||
key_df[c] = key_df[c].astype(str)
|
||||
dup_mask = key_df.duplicated(keep="last")
|
||||
df = df[~dup_mask]
|
||||
after_dedup = len(df)
|
||||
if after_dedup < before_dedup:
|
||||
logger.warning(
|
||||
@@ -230,6 +242,25 @@ def batch_insert(table_name: str, df: pd.DataFrame, conn, conflict_columns: List
|
||||
return len(rows)
|
||||
except Exception as e:
|
||||
conn.rollback()
|
||||
# 兜底:若批内仍存在重复冲突键 (如极端类型差异未被识别),
|
||||
# 回退为逐行 INSERT。单行命令永远不会触发
|
||||
# "ON CONFLICT DO UPDATE command cannot affect row a second time"。
|
||||
if "cannot affect row a second time" in str(e):
|
||||
logger.warning(
|
||||
f" {table_name}: 批内仍存在重复冲突键,回退为逐行导入 ({len(rows)} 行)"
|
||||
)
|
||||
try:
|
||||
inserted = 0
|
||||
for r in rows:
|
||||
execute_values(cursor, upsert_sql.as_string(cursor), [r], page_size=1)
|
||||
inserted += 1
|
||||
conn.commit()
|
||||
logger.info(f" {table_name}: 逐行成功导入 {inserted} 条记录")
|
||||
return inserted
|
||||
except Exception as e2:
|
||||
conn.rollback()
|
||||
logger.error(f" {table_name}: 逐行导入失败 - {e2}")
|
||||
raise
|
||||
logger.error(f" {table_name}: 批量导入失败 - {e}")
|
||||
raise
|
||||
finally:
|
||||
|
||||
Reference in New Issue
Block a user