求助：实现T1插入行时自动更新T2的Python脚本开发

问题核心 #

现有脚本直接将T1查询结果全量追加到T2，会导致重复插入旧数据，且未考虑T2中客户已编辑的10列数据保留需求。我们需要实现仅同步T1新增行，同时不影响T2现有编辑内容的逻辑。

解决方案步骤 #

1. 确定唯一识别主键：从T1同步的C1-C5中选一个/多个列作为唯一标识（比如C1是唯一ID），用来区分新行和已存在的行。
2. 筛选T1中的新行：对比T2已有的主键，找出T1中存在但T2没有的行。
3. 补全T2的空列：给新行添加T2中那10个客户编辑列的默认空值，保证列数匹配T2的15列。
4. 插入新行到T2：仅将筛选出的新行追加到T2，避免重复。

完善后的代码 #

import pandas as pd
from sqlalchemy import text

def get_dispatches(db):
    query = """
        SELECT C1, C2, C3, C4, C5
        FROM T1
    """
    return pd.read_sql(text(query), db)

def get_plan(db):
    query = """
        SELECT * FROM T2;
    """
    return pd.read_sql(text(query), db)

# 连接数据库获取数据
con = get_sql_conn('T1schema')
df_dispatches = get_dispatches(con)
con.close()

plan_con = get_sql_conn('T2schema')
df_plan = get_plan(plan_con)

# 配置唯一主键（若为多列组合主键，改为列表形式如['C1', 'C2']）
primary_key = 'C1'

# 筛选T1中未同步到T2的新行
if primary_key in df_dispatches.columns and primary_key in df_plan.columns:
    existing_ids = df_plan[primary_key].tolist()
    new_rows = df_dispatches[~df_dispatches[primary_key].isin(existing_ids)]
else:
    raise ValueError(f"主键列{primary_key}在T1或T2中不存在，请检查列名")

if not new_rows.empty:
    # 补全T2所需的所有列（自动匹配T2的列名，无需硬编码）
    t2_all_columns = df_plan.columns.tolist()
    missing_cols = [col for col in t2_all_columns if col not in new_rows.columns]
    
    # 给新行添加空列，默认值设为None（可根据需求改为空字符串等）
    for col in missing_cols:
        new_rows[col] = None
    
    # 仅插入新行到T2
    inserted_count = new_rows.to_sql(
        'T2', plan_con, schema='T2schema', 
        if_exists='append', index=False
    )
    print(f"成功同步{inserted_count}条新行到T2")
else:
    print("T1无新行需要同步")

plan_con.close()

关键注意事项 #

• 主键正确性：必须确保主键能唯一标识每一行，否则会出现漏同步或重复插入问题。如果是组合主键，修改主键判断逻辑为：

  # 组合主键示例
  primary_keys = ['C1', 'C2']
  existing_keys = df_plan.set_index(primary_keys).index
  new_rows = df_dispatches[~df_dispatches.set_index(primary_keys).index.isin(existing_keys)]
  

• 列兼容性：代码会自动匹配T2的列结构，无需手动指定那10个编辑列的名称，只要T2包含C1-C5和编辑列即可。
• 数据安全：此逻辑仅新增数据，不会修改T2中已存在的行，客户编辑的内容会被完整保留。

内容的提问来源于stack exchange，提问作者Ash