325 lines
12 KiB
Python
325 lines
12 KiB
Python
import configparser
|
||
import pandas as pd
|
||
from sqlalchemy import create_engine, text
|
||
from urllib.parse import quote_plus
|
||
import logging
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 读取配置文件
|
||
config = configparser.ConfigParser()
|
||
config.read('config.ini')
|
||
|
||
# 获取数据库连接配置
|
||
source_config = {
|
||
'host': config.get('source_db', 'host'),
|
||
'user': config.get('source_db', 'user'),
|
||
'password': config.get('source_db', 'password'),
|
||
'database': config.get('source_db', 'database'),
|
||
'port': config.getint('source_db', 'port')
|
||
}
|
||
|
||
target_config = {
|
||
'host': config.get('target_db', 'host'),
|
||
'user': config.get('target_db', 'user'),
|
||
'password': config.get('target_db', 'password'),
|
||
'database': config.get('target_db', 'database'),
|
||
'port': config.getint('target_db', 'port')
|
||
}
|
||
|
||
# 创建数据库引擎
|
||
source_engine = create_engine(
|
||
f"mysql+pymysql://{source_config['user']}:{quote_plus(source_config['password'])}@{source_config['host']}:{source_config['port']}/{source_config['database']}"
|
||
)
|
||
target_engine = create_engine(
|
||
f"mysql+pymysql://{target_config['user']}:{quote_plus(target_config['password'])}@{target_config['host']}:{target_config['port']}/{target_config['database']}"
|
||
)
|
||
|
||
|
||
class DataComparisonUpdater:
|
||
"""数据对比和更新器"""
|
||
|
||
def __init__(self):
|
||
self.source_engine = source_engine
|
||
self.target_engine = target_engine
|
||
self.update_rules = {} # 存储更新规则
|
||
self.comparison_results = [] # 存储对比结果
|
||
|
||
def set_update_rules(self, rules):
|
||
"""
|
||
设置更新规则
|
||
rules: 字典格式，包含表名、条件、要更新的字段等
|
||
示例：
|
||
{
|
||
'table_name': 'ma_supplier_info',
|
||
'source_table': 'bm_supplier',
|
||
'join_conditions': {'supplier': 'NAME'},
|
||
'update_fields': ['address', 'phone'],
|
||
'where_conditions': {'status': 'active'}
|
||
}
|
||
"""
|
||
self.update_rules = rules
|
||
logger.info(f"已设置更新规则: {rules}")
|
||
|
||
def query_target_data(self, table_name, columns=None, conditions=None):
|
||
"""
|
||
从目标数据库查询数据
|
||
|
||
Args:
|
||
table_name: 表名
|
||
columns: 要查询的列，默认为所有列
|
||
conditions: 查询条件，字典格式
|
||
|
||
Returns:
|
||
DataFrame: 查询结果
|
||
"""
|
||
try:
|
||
# 构建查询SQL
|
||
if columns:
|
||
columns_str = ', '.join(columns)
|
||
else:
|
||
columns_str = '*'
|
||
|
||
sql = f"SELECT {columns_str} FROM {table_name}"
|
||
|
||
# 添加WHERE条件
|
||
if conditions:
|
||
where_clauses = []
|
||
for key, value in conditions.items():
|
||
if isinstance(value, str):
|
||
where_clauses.append(f"{key} = '{value}'")
|
||
else:
|
||
where_clauses.append(f"{key} = {value}")
|
||
|
||
if where_clauses:
|
||
sql += " WHERE " + " AND ".join(where_clauses)
|
||
|
||
logger.info(f"执行目标数据库查询: {sql}")
|
||
df = pd.read_sql(sql, self.target_engine)
|
||
logger.info(f"查询到 {len(df)} 条记录")
|
||
return df
|
||
|
||
except Exception as e:
|
||
logger.error(f"查询目标数据库时发生错误: {str(e)}")
|
||
return pd.DataFrame()
|
||
|
||
def query_source_data(self, table_name, columns=None, conditions=None):
|
||
"""
|
||
从源数据库查询数据
|
||
|
||
Args:
|
||
table_name: 表名
|
||
columns: 要查询的列，默认为所有列
|
||
conditions: 查询条件，字典格式
|
||
|
||
Returns:
|
||
DataFrame: 查询结果
|
||
"""
|
||
try:
|
||
# 构建查询SQL
|
||
if columns:
|
||
columns_str = ', '.join(columns)
|
||
else:
|
||
columns_str = '*'
|
||
|
||
sql = f"SELECT {columns_str} FROM {table_name}"
|
||
|
||
# 添加WHERE条件
|
||
if conditions:
|
||
where_clauses = []
|
||
for key, value in conditions.items():
|
||
if isinstance(value, str):
|
||
where_clauses.append(f"{key} = '{value}'")
|
||
else:
|
||
where_clauses.append(f"{key} = {value}")
|
||
|
||
if where_clauses:
|
||
sql += " WHERE " + " AND ".join(where_clauses)
|
||
|
||
logger.info(f"执行源数据库查询: {sql}")
|
||
df = pd.read_sql(sql, self.source_engine)
|
||
logger.info(f"查询到 {len(df)} 条记录")
|
||
return df
|
||
|
||
except Exception as e:
|
||
logger.error(f"查询源数据库时发生错误: {str(e)}")
|
||
return pd.DataFrame()
|
||
|
||
def compare_data(self, target_df, source_df, join_columns):
|
||
"""
|
||
对比目标数据库和源数据库的数据
|
||
|
||
Args:
|
||
target_df: 目标数据库数据
|
||
source_df: 源数据库数据
|
||
join_columns: 用于关联的列名，字典格式 {'target_col': 'source_col'}
|
||
|
||
Returns:
|
||
DataFrame: 对比结果
|
||
"""
|
||
try:
|
||
# 重命名源数据库列名以便关联
|
||
source_df_renamed = source_df.copy()
|
||
for target_col, source_col in join_columns.items():
|
||
if source_col in source_df_renamed.columns:
|
||
source_df_renamed = source_df_renamed.rename(columns={source_col: target_col})
|
||
|
||
# 执行关联
|
||
merged_df = pd.merge(target_df, source_df_renamed,
|
||
on=list(join_columns.keys()),
|
||
how='inner',
|
||
suffixes=('_target', '_source'))
|
||
|
||
logger.info(f"数据对比完成，找到 {len(merged_df)} 条匹配记录")
|
||
return merged_df
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据对比时发生错误: {str(e)}")
|
||
return pd.DataFrame()
|
||
|
||
def update_source_data(self, comparison_df, update_fields, table_name):
|
||
"""
|
||
根据对比结果更新源数据库
|
||
|
||
Args:
|
||
comparison_df: 对比结果数据
|
||
update_fields: 要更新的字段列表
|
||
table_name: 要更新的表名
|
||
|
||
Returns:
|
||
int: 更新的记录数
|
||
"""
|
||
try:
|
||
update_count = 0
|
||
|
||
with self.source_engine.connect() as conn:
|
||
for _, row in comparison_df.iterrows():
|
||
# 构建UPDATE语句
|
||
set_clauses = []
|
||
for field in update_fields:
|
||
target_field = f"{field}_target"
|
||
if target_field in row and pd.notna(row[target_field]):
|
||
if isinstance(row[target_field], str):
|
||
set_clauses.append(f"{field} = '{row[target_field]}'")
|
||
else:
|
||
set_clauses.append(f"{field} = {row[target_field]}")
|
||
|
||
if set_clauses:
|
||
# 构建WHERE条件（使用主键或唯一标识）
|
||
where_conditions = []
|
||
for key in comparison_df.columns:
|
||
if key.endswith('_source') and not key.endswith('_target'):
|
||
if pd.notna(row[key]):
|
||
if isinstance(row[key], str):
|
||
where_conditions.append(f"{key.replace('_source', '')} = '{row[key]}'")
|
||
else:
|
||
where_conditions.append(f"{key.replace('_source', '')} = {row[key]}")
|
||
|
||
if where_conditions:
|
||
sql = f"UPDATE {table_name} SET {', '.join(set_clauses)} WHERE {' AND '.join(where_conditions)}"
|
||
logger.debug(f"执行更新SQL: {sql}")
|
||
|
||
result = conn.execute(text(sql))
|
||
update_count += result.rowcount
|
||
|
||
conn.commit()
|
||
|
||
logger.info(f"成功更新 {update_count} 条记录")
|
||
return update_count
|
||
|
||
except Exception as e:
|
||
logger.error(f"更新源数据库时发生错误: {str(e)}")
|
||
return 0
|
||
|
||
def execute_comparison_update(self):
|
||
"""
|
||
执行完整的数据对比和更新流程
|
||
"""
|
||
if not self.update_rules:
|
||
logger.error("未设置更新规则，请先调用 set_update_rules()")
|
||
return False
|
||
|
||
try:
|
||
rules = self.update_rules
|
||
|
||
# 1. 查询目标数据库数据
|
||
target_df = self.query_target_data(
|
||
rules['table_name'],
|
||
rules.get('target_columns'),
|
||
rules.get('target_conditions')
|
||
)
|
||
|
||
if target_df.empty:
|
||
logger.warning("目标数据库查询结果为空")
|
||
return False
|
||
|
||
# 2. 查询源数据库数据
|
||
source_df = self.query_source_data(
|
||
rules['source_table'],
|
||
rules.get('source_columns'),
|
||
rules.get('source_conditions')
|
||
)
|
||
|
||
if source_df.empty:
|
||
logger.warning("源数据库查询结果为空")
|
||
return False
|
||
|
||
# 3. 对比数据
|
||
comparison_df = self.compare_data(
|
||
target_df,
|
||
source_df,
|
||
rules['join_conditions']
|
||
)
|
||
|
||
if comparison_df.empty:
|
||
logger.warning("没有找到匹配的数据")
|
||
return False
|
||
|
||
# 4. 更新源数据库
|
||
update_count = self.update_source_data(
|
||
comparison_df,
|
||
rules['update_fields'],
|
||
rules['source_table']
|
||
)
|
||
|
||
logger.info(f"数据对比和更新完成，共更新 {update_count} 条记录")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"执行数据对比更新时发生错误: {str(e)}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主函数 - 示例用法"""
|
||
updater = DataComparisonUpdater()
|
||
|
||
# 示例：设置更新规则
|
||
# 这里您需要根据具体需求设置规则
|
||
example_rules = {
|
||
'table_name': 'ma_supplier_info', # 目标表
|
||
'source_table': 'bm_supplier', # 源表
|
||
'join_conditions': {'supplier': 'NAME'}, # 关联条件
|
||
'update_fields': ['address', 'phone'], # 要更新的字段
|
||
'target_conditions': {'status': 'active'}, # 目标表查询条件
|
||
'source_conditions': None, # 源表查询条件
|
||
'target_columns': ['supplier', 'address', 'phone'], # 目标表查询列
|
||
'source_columns': ['NAME', 'ADDRESS', 'PHONE_NUMBER'] # 源表查询列
|
||
}
|
||
|
||
# 设置更新规则
|
||
updater.set_update_rules(example_rules)
|
||
|
||
# 执行对比和更新
|
||
success = updater.execute_comparison_update()
|
||
|
||
if success:
|
||
print("✅ 数据对比和更新完成")
|
||
else:
|
||
print("❌ 数据对比和更新失败")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |