diff --git a/quantitative_data/importer.py b/quantitative_data/importer.py
index ab83f76..4433d98 100644
--- a/quantitative_data/importer.py
+++ b/quantitative_data/importer.py
@@ -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: