diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..94a25f7 --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/machines/结算记录.py b/machines/结算记录.py new file mode 100644 index 0000000..78da505 --- /dev/null +++ b/machines/结算记录.py @@ -0,0 +1,186 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus + +# 读取配置文件 +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']}?charset=utf8mb4" +) +target_engine = create_engine( + f"mysql+pymysql://{target_config['user']}:{quote_plus(target_config['password'])}@{target_config['host']}:{target_config['port']}/{target_config['database']}?charset=utf8mb4" +) + + +def safe_convert_to_int(series): + """安全转换为整数""" + return pd.to_numeric(series, errors='coerce').fillna(0).astype(int) + + +def safe_convert_to_float(series): + """安全转换为浮点数""" + return pd.to_numeric(series, errors='coerce').fillna(0.0) + + +def process_slt_agreement_apply(): + """处理结算申请主表""" + try: + sql = """ + SELECT + bma.ID as id, + bma.ID as agreement_id, + bma.`CODE` as code, + bma.SETTLEMENT_TIME as create_time, + IF(bma.IS_BALANCE = 0, 2, 1) as status, + bma.SETTLEMENT_TIME as audit_time + FROM + ba_ma_agreement bma + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() + """ + df = pd.read_sql(sql, source_engine) + + # 转换状态字段 + df['status'] = safe_convert_to_int(df['status']) + + # 确保时间字段格式正确 + time_cols = ['create_time', 'audit_time'] + for col in time_cols: + df[col] = pd.to_datetime(df[col]) + + df.to_sql('slt_agreement_apply', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条结算申请主表数据") + return True + except Exception as e: + print(f"处理结算申请主表时出错: {str(e)}") + return False + + +def process_slt_agreement_details(): + """处理结算申请明细表""" + try: + # 四个UNION查询合并 + sql = """ + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 1, bsd.lease_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 1 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 1 + + UNION + + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 2, bsd.buy_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 2 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 2 + + UNION + + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 2, bsd.buy_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 3 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 4 + + UNION + + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 2, bsd.buy_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 4 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 5 + """ + df = pd.read_sql(sql, source_engine) + + # 转换数值字段 + num_cols = ['type_id', 'NUM', 'slt_type', 'is_charge'] + for col in num_cols: + df[col] = safe_convert_to_int(df[col]) + + # 转换金额字段 + money_cols = ['price', 'MONEY'] + for col in money_cols: + df[col] = safe_convert_to_float(df[col]) + + # 确保时间字段格式正确 + time_cols = ['start_time', 'end_time'] + for col in time_cols: + # 先用正则提取 YYYY-MM-DD 部分 + df[col] = df[col].str.extract(r'(\d{4}-\d{2}-\d{2})') + # 再转换为 datetime + df[col] = pd.to_datetime(df[col], errors='coerce') + + df.to_sql('slt_agreement_details', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条结算申请明细数据") + return True + except Exception as e: + print(f"处理结算申请明细时出错: {str(e)}") + return False + + +if __name__ == "__main__": + # 执行转换流程 + success1 = process_slt_agreement_apply() + success2 = process_slt_agreement_details() + + if success1 and success2: + print("所有结算申请相关表转换完成！") + else: + print("!!! 部分转换失败，请检查错误日志 !!!") diff --git a/machines/维修.py b/machines/维修.py new file mode 100644 index 0000000..9726b38 --- /dev/null +++ b/machines/维修.py @@ -0,0 +1,351 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus + +# 读取配置文件 +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']}?charset=utf8mb4" +) +target_engine = create_engine( + f"mysql+pymysql://{target_config['user']}:{quote_plus(target_config['password'])}@{target_config['host']}:{target_config['port']}/{target_config['database']}?charset=utf8mb4" +) + + +def safe_convert_to_int(series): + """安全转换为整数""" + return pd.to_numeric(series, errors='coerce').fillna(0).astype(int) + + +def safe_convert_to_float(series): + """安全转换为浮点数""" + return pd.to_numeric(series, errors='coerce').fillna(0.0) + + +def process_tm_task(): + """处理维修任务主表""" + try: + sql = """ + SELECT + tt.ID as task_id, + 4 as task_type, + IF(tt.`STATUS` = 6, 1, 0) AS task_status, + bmr.APPLY_NUMBER AS code, + bmr.CREATOR as create_by, + bmr.CREATE_TIME + FROM + ba_ma_repair bmr + LEFT JOIN tm_task tt on bmr.ID = tt.ID + LEFT JOIN tm_task_status tts on tt.`STATUS` = tts.`CODE` + WHERE bmr.company_id = 1 + """ + df = pd.read_sql(sql, source_engine) + + # 转换状态字段 + df['task_status'] = safe_convert_to_int(df['task_status']) + + df.to_sql('tm_task', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修任务主表数据") + return True + except Exception as e: + print(f"处理维修任务主表时出错: {str(e)}") + return False + + +def process_tm_task_agreement(): + """处理维修任务关联表""" + try: + sql = """ + SELECT + bmr.ID as task_id, + bat.AGREEMENT_ID as agreement_id + FROM + ba_ma_repair bmr + LEFT JOIN ba_agreement_task bat on bmr.ID = bat.TASK_ID + WHERE bmr.company_id = 1 + """ + df = pd.read_sql(sql, source_engine) + + # 过滤空agreement_id + df = df.dropna(subset=['agreement_id']) + + df.to_sql('tm_task_agreement', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修任务关联数据") + return True + except Exception as e: + print(f"处理维修任务关联表时出错: {str(e)}") + return False + + +def process_repair_apply_details(): + """处理维修申请明细（合并三个来源）""" + try: + # 来源1：数量申请维修 + sql1 = """ + SELECT + bmr.ID as task_id, + ttmt.MA_TYPE_ID as type_id, + NULL as ma_id, + ttmt.MACHINES_NUM as repair_num, + ttmt.ACTUAL_NUM as repaired_num, + ttmt2.MACHINES_NUM as scrap_num, + ttmt.IS_SURE as status, + bmr.CREATOR as create_by, + bmr.CREATE_TIME, + bmr.BACK_ID as back_id, + IF(tt.`STATUS` = 6, 0, 1) AS is_ds + FROM + ba_ma_repair bmr + LEFT JOIN tm_task tt on bmr.ID = tt.ID + LEFT JOIN tm_task_ma_type ttmt on bmr.ID = ttmt.TASK_ID + LEFT JOIN ba_ma_scarp bms on bms.REPAIR_ID = bmr.ID + LEFT JOIN tm_task_ma_type ttmt2 on bms.ID = ttmt2.TASK_ID AND ttmt.MA_TYPE_ID = ttmt2.MA_TYPE_ID + WHERE ttmt.IS_COUNT = 1 + GROUP BY ttmt.TASK_ID, ttmt.MA_TYPE_ID + """ + + # 来源2：编码申请维修 + sql2 = """ + SELECT + bmr.ID AS task_id, + mm.type AS type_id, + mm.ID AS ma_id, + 1 AS repair_num, +IF + ( tt.`STATUS` = 6, 1, 0 ) AS repaired_num, + 0 AS scrap_num, +IF + ( tt.`STATUS` = 6, 1, 0 ) AS status, + bmr.CREATOR AS create_by, + bmr.CREATE_TIME, + bmr.BACK_ID AS back_id, +IF + ( tt.`STATUS` = 6, 0, 1 ) AS is_ds +FROM + ba_ma_repair_pass brp + LEFT JOIN ba_ma_repair bmr ON brp.repair_id = bmr.ID + LEFT JOIN tm_task tt ON bmr.ID = tt.ID + LEFT JOIN ma_machines mm ON brp.ma_id = mm.id + """ + + # 来源3：编码申请维修含报废 + sql3 = """ + SELECT + bmr.ID as task_id, + mm.type AS type_id, + tmsr.MA_ID as ma_id, + 1 as repair_num, + 0 as repaired_num, + 1 as scrap_num, + 1 as status, + bmr.CREATOR as create_by, + bmr.CREATE_TIME, + bmr.BACK_ID as back_id, + IF(tt.`STATUS` = 6, 0, 1) AS is_ds + FROM + ba_ma_scarp bms + LEFT JOIN ba_ma_repair bmr on bms.repair_id = bmr.ID + LEFT JOIN tm_task tt on bmr.ID = tt.ID + LEFT JOIN tm_ma_scarp_reason tmsr on bms.ID = tmsr.TASK_ID + LEFT JOIN ma_machines mm on tmsr.ma_id = mm.id + WHERE tmsr.IS_COUNT = 0 + """ + + # 合并三个来源的数据 + df1 = pd.read_sql(sql1, source_engine) + df2 = pd.read_sql(sql2, source_engine) + df3 = pd.read_sql(sql3, source_engine) + df = pd.concat([df1, df2, df3], ignore_index=True) + + # 转换数值字段 + num_cols = ['repair_num', 'repaired_num', 'scrap_num', 'status', 'is_ds'] + for col in num_cols: + df[col] = safe_convert_to_int(df[col]) + + df.to_sql('repair_apply_details', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修申请明细（合并3个来源）") + return True + except Exception as e: + print(f"处理维修申请明细时出错: {str(e)}") + return False + + +def process_repair_apply_record(): + """处理维修申请记录（合并四个来源）""" + try: + # 来源1：数量维修合格 + sql1 = """ + SELECT + bmr.ID as task_id, + ttmt.MA_TYPE_ID as type_id, + NULL as ma_id, + ttmt.ACTUAL_NUM as repair_num, + 0 as scrap_num, + ttmt.IS_SURE as status, + bmr.CREATOR as create_by, + IFNULL(bmr.REPAIR_TIME, bmr.CREATE_TIME) as create_time, + NULL as scrap_type, + NULL as scrap_reason + FROM + ba_ma_repair bmr + LEFT JOIN tm_task_ma_type ttmt on bmr.ID = ttmt.TASK_ID + LEFT JOIN ba_ma_scarp bms on bms.REPAIR_ID = bmr.ID + LEFT JOIN tm_task_ma_type ttmt2 on bms.ID = ttmt2.TASK_ID AND ttmt.MA_TYPE_ID = ttmt2.MA_TYPE_ID + WHERE ttmt.IS_COUNT = 1 + GROUP BY ttmt.TASK_ID, ttmt.MA_TYPE_ID + HAVING repair_num > 0 + """ + + # 来源2：数量维修报废 + sql2 = """ + SELECT + bmr.ID as task_id, + ttmt.MA_TYPE_ID as type_id, + NULL as ma_id, + 0 as repair_num, + ttmt2.MACHINES_NUM as scrap_num, + ttmt.IS_SURE as status, + bmr.CREATOR as create_by, + bmr.REPAIR_TIME as create_time, + tmsr.DAMAGE as scrap_type, + tmsr.SCARP_REASON as scrap_reason + FROM + ba_ma_repair bmr + LEFT JOIN tm_task_ma_type ttmt on bmr.ID = ttmt.TASK_ID + LEFT JOIN ba_ma_scarp bms on bms.REPAIR_ID = bmr.ID + LEFT JOIN tm_ma_scarp_reason tmsr on bms.ID = tmsr.TASK_ID + LEFT JOIN tm_task_ma_type ttmt2 on bms.ID = ttmt2.TASK_ID AND ttmt.MA_TYPE_ID = ttmt2.MA_TYPE_ID + WHERE ttmt.IS_COUNT = 1 + GROUP BY ttmt.TASK_ID, ttmt.MA_TYPE_ID + HAVING scrap_num > 0 + """ + + # 来源3：编码维修合格 + sql3 = """ + SELECT + bmr.ID as task_id, + mm.type as type_id, + mm.ID as ma_id, + 1 as repair_num, + 0 as scrap_num, + 1 as status, + bmr.CREATOR as create_by, + bmr.REPAIR_TIME as create_time, + NULL as scrap_type, + NULL as scrap_reason + FROM + ba_ma_repair_pass brp + LEFT JOIN ba_ma_repair bmr on brp.repair_id = bmr.ID + LEFT JOIN ma_machines mm on brp.ma_id = mm.id + """ + + # 来源4：编码维修报废 + sql4 = """ + SELECT + bmr.ID as task_id, + mm.type as type_id, + tmsr.MA_ID as ma_id, + 0 as repair_num, + 1 as scrap_num, + 1 as status, + bmr.CREATOR as create_by, + bmr.REPAIR_TIME as create_time, + tmsr.DAMAGE as scrap_type, + tmsr.SCARP_REASON as scrap_reason + FROM + ba_ma_scarp bms + LEFT JOIN ba_ma_repair bmr on bms.repair_id = bmr.ID + LEFT JOIN tm_ma_scarp_reason tmsr on bms.ID = tmsr.TASK_ID + LEFT JOIN ma_machines mm on tmsr.ma_id = mm.id + WHERE tmsr.IS_COUNT = 0 + """ + + # 合并四个来源的数据 + df1 = pd.read_sql(sql1, source_engine) + df2 = pd.read_sql(sql2, source_engine) + df3 = pd.read_sql(sql3, source_engine) + df4 = pd.read_sql(sql4, source_engine) + df = pd.concat([df1, df2, df3, df4], ignore_index=True) + + # 转换数值字段 + num_cols = ['repair_num', 'scrap_num', 'status'] + for col in num_cols: + df[col] = safe_convert_to_int(df[col]) + + df.to_sql('repair_apply_record', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修申请记录（合并4个来源）") + return True + except Exception as e: + print(f"处理维修申请记录时出错: {str(e)}") + return False + + +def process_repair_cost(): + """处理维修费用""" + try: + sql = """ + SELECT + rp.TASK_ID as task_id, + rp.MA_TYPE_ID as type_id, + rp.MA_ID as ma_id, + rp.REPAIR_NUM as repair_num, + rp.PRICE as costs, + rp.IS_BUCKLE as part_type, + 1 as status + FROM + repair_part rp + LEFT JOIN ba_ma_repair bmr on rp.TASK_ID = bmr.ID + """ + df = pd.read_sql(sql, source_engine) + + # 转换数值字段 + df['repair_num'] = safe_convert_to_int(df['repair_num']) + df['costs'] = safe_convert_to_float(df['costs']) + df['part_type'] = safe_convert_to_int(df['part_type']) + df['status'] = safe_convert_to_int(df['status']) + + df.to_sql('repair_cost', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修费用记录") + return True + except Exception as e: + print(f"处理维修费用时出错: {str(e)}") + return False + + +if __name__ == "__main__": + # 执行所有转换流程 + processes = [ + process_tm_task, + process_tm_task_agreement, + process_repair_apply_details, + process_repair_apply_record, + process_repair_cost + ] + + success = all([p() for p in processes]) + + if success: + print("所有维修相关表转换完成！") + else: + print("!!! 部分转换失败，请检查错误日志 !!!") \ No newline at end of file diff --git a/机具/repair_audit_details.py b/机具/repair_audit_details.py new file mode 100644 index 0000000..79d2adf --- /dev/null +++ b/机具/repair_audit_details.py @@ -0,0 +1,243 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus + +# @author: 阮世耀 +# 描述：处理ba_ma_input_check表数据到repair_audit_details表的迁移 + +# 读取配置文件 +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']}" +) + + +def process_repair_audit_details(): + """处理ba_ma_input_check表数据到repair_audit_details表的迁移""" + try: + # 源表查询SQL（包含UNION的复杂查询） + sql = """ + SELECT + bic.ID as task_id, + bic.REPAIR_ID as repair_id, + mm.TYPE AS type_id, + ttm.MA_ID as ma_id, + 1 as repair_num, + IF(ttm.MA_ID = tmsr.MA_ID, 1, 0) as scrap_num, + 1 - IF(ttm.MA_ID = tmsr.MA_ID, 1, 0) as repaired_num, + bic.IS_SURE as status, + bic.CREATE_TIME, + bic.CREATOR as create_by + FROM + ba_ma_input_check bic + LEFT JOIN ba_ma_scarp bms on bic.REPAIR_ID = bms.REPAIR_ID + LEFT JOIN tm_ma_scarp_reason tmsr on bms.ID = tmsr.TASK_ID and tmsr.IS_COUNT = 0 + LEFT JOIN tm_task_ma ttm on ttm.task_id = bic.id + LEFT JOIN ma_machines mm on ttm.ma_id = mm.id + WHERE mm.id is not null + + UNION + + SELECT + bic.ID as task_id, + bic.REPAIR_ID as repair_id, + ttmt.MA_TYPE_ID as type_id, + null as ma_id, + ttmt.MACHINES_NUM as repair_num, + ttmt2.MACHINES_NUM as scrap_num, + ttmt.MACHINES_NUM - ttmt2.MACHINES_NUM as repaired_num, + bic.IS_SURE as status, + bic.CREATE_TIME, + bic.CREATOR as create_by + FROM + ba_ma_input_check bic + LEFT JOIN tm_task_ma_type ttmt on bic.ID = ttmt.TASK_ID + LEFT JOIN ba_ma_scarp bms on bic.REPAIR_ID = bms.REPAIR_ID + LEFT JOIN tm_task_ma_type ttmt2 on bms.ID = ttmt2.TASK_ID and ttmt.MA_TYPE_ID = ttmt2.MA_TYPE_ID + WHERE ttmt.IS_COUNT = 1 + """ + + print("正在执行复杂查询，请稍候...") + + # 执行查询获取源数据 + df = pd.read_sql(sql, source_engine) + + print(f"查询完成，获取到 {len(df)} 条原始记录") + + # 数据转换和字段映射 + result = pd.DataFrame() + result['task_id'] = df['task_id'] + result['repair_id'] = df['repair_id'] + result['ma_id'] = df['ma_id'] # 可能为null + result['type_id'] = df['type_id'] + result['repair_num'] = df['repair_num'] + result['repaired_num'] = df['repaired_num'] + result['scrap_num'] = df['scrap_num'] + result['create_by'] = df['create_by'] + result['create_time'] = df['CREATE_TIME'] + result['status'] = df['status'] + + # 数据清洗 + # 移除关键字段为空的记录 + before_clean = len(result) + result = result.dropna(subset=['task_id', 'repair_id', 'type_id']) + after_clean = len(result) + + if before_clean != after_clean: + print(f"数据清洗：移除了 {before_clean - after_clean} 条关键字段为空的记录") + + # 处理数值字段的空值，设为0 + numeric_fields = ['repair_num', 'repaired_num', 'scrap_num'] + for field in numeric_fields: + result[field] = result[field].fillna(0) + + # 写入目标表 + result.to_sql('repair_audit_details', target_engine, + if_exists='append', index=False) + + print(f"成功转换并导入 {len(result)} 条记录到 repair_audit_details") + + # 生成统计报告 + generate_statistics_report(result) + + return True + + except Exception as e: + print(f"处理 repair_audit_details 时发生错误: {str(e)}") + import traceback + print("详细错误信息:") + traceback.print_exc() + return False + + +def generate_statistics_report(result): + """生成详细的统计报告""" + print("\n=== 数据统计报告 ===") + + # 基本统计 + print(f"总记录数: {len(result)}") + + # 按状态分组统计 + print(f"\n审核状态分布:") + status_counts = result['status'].value_counts() + for status, count in status_counts.items(): + status_desc = "已审核" if status == 1 else "待审核" + print(f" - status = {status} ({status_desc}): {count} 条") + + # 有ma_id和无ma_id的记录分布 + print(f"\n设备记录类型分布:") + ma_id_not_null = result['ma_id'].notna().sum() + ma_id_null = result['ma_id'].isna().sum() + print(f" - 具体设备记录 (ma_id不为空): {ma_id_not_null} 条") + print(f" - 设备类型记录 (ma_id为空): {ma_id_null} 条") + + # 数量统计 + print(f"\n维修数量统计:") + print(f" - 总维修数量: {result['repair_num'].sum()}") + print(f" - 总修复数量: {result['repaired_num'].sum()}") + print(f" - 总报废数量: {result['scrap_num'].sum()}") + + # 按repair_id分组统计 + repair_groups = result.groupby('repair_id').size() + print(f"\n维修单分布:") + print(f" - 涉及维修单数量: {len(repair_groups)}") + print(f" - 平均每个维修单记录数: {repair_groups.mean():.2f}") + + +def validate_data(): + """验证迁移数据的完整性和逻辑一致性""" + try: + print("\n=== 开始数据验证 ===") + + # 检查目标表记录数 + target_count_sql = "SELECT COUNT(*) as total_count FROM repair_audit_details" + target_count = pd.read_sql(target_count_sql, target_engine)['total_count'].iloc[0] + print(f"目标表总记录数: {target_count}") + + # 检查数据逻辑一致性 + validation_sql = """ + SELECT + COUNT(*) as total_records, + COUNT(CASE WHEN ma_id IS NOT NULL THEN 1 END) as with_ma_id, + COUNT(CASE WHEN ma_id IS NULL THEN 1 END) as without_ma_id, + SUM(repair_num) as total_repair, + SUM(repaired_num) as total_repaired, + SUM(scrap_num) as total_scrapped + FROM repair_audit_details + """ + + validation_result = pd.read_sql(validation_sql, target_engine) + print(f"\n逻辑验证结果:") + print(f" - 总记录数: {validation_result['total_records'].iloc[0]}") + print(f" - 有设备ID记录: {validation_result['with_ma_id'].iloc[0]}") + print(f" - 无设备ID记录: {validation_result['without_ma_id'].iloc[0]}") + print(f" - 总维修数: {validation_result['total_repair'].iloc[0]}") + print(f" - 总修复数: {validation_result['total_repaired'].iloc[0]}") + print(f" - 总报废数: {validation_result['total_scrapped'].iloc[0]}") + + # 检查数据完整性 + print(f"\n✅ 数据验证完成") + return True + + except Exception as e: + print(f"数据验证时发生错误: {str(e)}") + return False + + +def clear_existing_data(): + """清理目标表中的现有数据（可选）""" + try: + confirmation = input("是否要清理目标表 repair_audit_details 中的现有数据？(y/N): ") + if confirmation.lower() == 'y': + with target_engine.connect() as conn: + result = conn.execute("DELETE FROM repair_audit_details") + print(f"已清理 {result.rowcount} 条现有数据") + return True + else: + print("跳过数据清理") + return True + except Exception as e: + print(f"清理数据时发生错误: {str(e)}") + return False + + +if __name__ == "__main__": + print("=== repair_audit_details 审核明细数据迁移 ===") + print("注意：此脚本将执行复杂的UNION查询，可能需要较长时间") + print("开始执行数据迁移...") + + # 可选：清理现有数据 + if clear_existing_data(): + # 执行数据迁移 + if process_repair_audit_details(): + # 验证数据完整性 + validate_data() + print("\n=== 迁移完成 ===") + else: + print("\n=== 迁移失败 ===") + else: + print("\n=== 迁移中止 ===") \ No newline at end of file diff --git a/机具/tm_task.py b/机具/tm_task.py new file mode 100644 index 0000000..90186b9 --- /dev/null +++ b/机具/tm_task.py @@ -0,0 +1,155 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus + +# @author: 阮世耀 +# 描述：处理ba_ma_input_check表数据到tm_task表的迁移 + +# 读取配置文件 +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']}" +) + + +def process_tm_task(): + """处理tm_task表数据迁移""" + try: + # 源表查询SQL + sql = """ + SELECT + bic.ID as task_id, + 5 as task_type, + IF(bic.IS_SURE = 1, 11, 10) as task_status, + bmr.APPLY_NUMBER as code, + tt.CREATOR, + tt.CREATE_TIME + FROM + ba_ma_input_check bic + LEFT JOIN tm_task tt on bic.ID = tt.ID + LEFT JOIN ba_ma_repair bmr on bic.REPAIR_ID = bmr.ID + """ + + # 执行查询获取源数据 + df = pd.read_sql(sql, source_engine) + + # 数据转换和字段映射 + result = pd.DataFrame() + result['task_id'] = df['task_id'] + result['task_type'] = df['task_type'] + result['task_status'] = df['task_status'] + result['code'] = df['code'] + result['create_by'] = df['CREATOR'] + result['create_time'] = df['CREATE_TIME'] + + # 数据清洗：移除空值行 + result = result.dropna(subset=['task_id']) + + # 写入目标表 + result.to_sql('tm_task', target_engine, + if_exists='append', index=False) + + print(f"成功转换并导入 {len(result)} 条记录到 tm_task") + print(f"任务类型分布：") + print(f" - task_type = 5: {len(result)} 条") + print(f"任务状态分布：") + status_counts = result['task_status'].value_counts() + for status, count in status_counts.items(): + status_desc = "已确认" if status == 11 else "待确认" + print(f" - task_status = {status} ({status_desc}): {count} 条") + + return True + + except Exception as e: + print(f"处理 tm_task 时发生错误: {str(e)}") + return False + + +def validate_data(): + """验证迁移数据的完整性""" + try: + # 检查源数据总数 + source_count_sql = """ + SELECT COUNT(*) as total_count + FROM ba_ma_input_check bic + LEFT JOIN tm_task tt on bic.ID = tt.ID + LEFT JOIN ba_ma_repair bmr on bic.REPAIR_ID = bmr.ID + """ + source_count = pd.read_sql(source_count_sql, source_engine)['total_count'].iloc[0] + + # 检查目标数据总数 + target_count_sql = "SELECT COUNT(*) as total_count FROM tm_task WHERE task_type = 5" + target_count = pd.read_sql(target_count_sql, target_engine)['total_count'].iloc[0] + + print(f"\n数据验证结果：") + print(f"源数据记录数: {source_count}") + print(f"目标数据记录数: {target_count}") + + if source_count == target_count: + print("✅ 数据迁移完整性验证通过") + return True + else: + print("❌ 数据迁移完整性验证失败，记录数不匹配") + return False + + except Exception as e: + print(f"数据验证时发生错误: {str(e)}") + return False + + +def clear_existing_data(): + """清理目标表中的现有数据（可选）""" + try: + confirmation = input("是否要清理目标表中 task_type=5 的现有数据？(y/N): ") + if confirmation.lower() == 'y': + with target_engine.connect() as conn: + result = conn.execute("DELETE FROM tm_task WHERE task_type = 5") + print(f"已清理 {result.rowcount} 条现有数据") + return True + else: + print("跳过数据清理") + return True + except Exception as e: + print(f"清理数据时发生错误: {str(e)}") + return False + + +if __name__ == "__main__": + print("=== tm_task 审核表数据迁移 ===") + print("开始执行数据迁移...") + + # 可选：清理现有数据 + if clear_existing_data(): + # 执行数据迁移 + if process_tm_task(): + # 验证数据完整性 + validate_data() + print("\n=== 迁移完成 ===") + else: + print("\n=== 迁移失败 ===") + else: + print("\n=== 迁移中止 ===") \ No newline at end of file diff --git a/机具/新购机具.py b/机具/新购机具.py new file mode 100644 index 0000000..86b45e7 --- /dev/null +++ b/机具/新购机具.py @@ -0,0 +1,144 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus +# 读取配置文件 +config = configparser.ConfigParser() +config.read(r'D:\code\Bonus-Transfer-Machines\machines\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']}" +) + +def process_tm_task(): + """处理新购待验收数据""" + try: + # 执行转换SQL + sql = """ + SELECT + bms.ID as task_id, + 0 as task_type, + 2 as task_status, + bms.APPLY_NUMBER as code, + bms.BUYER as create_by, + tt.CREATE_TIME as create_time + FROM + ba_ma_shop bms + LEFT JOIN tm_task tt on bms.ID = tt.ID + LEFT JOIN tm_task_status tts on tt.`STATUS` = tts.`CODE` + LEFT JOIN tm_task_ma_type ttmt on bms.ID = ttmt.TASK_ID + WHERE tt.CREATE_TIME BETWEEN '2025-01-01' and NOW() and tt.`STATUS` in (18,19) + GROUP BY bms.ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('tm_task', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条新购机具任务表数据") + return True + + except Exception as e: + print(f"处理新购机具任务表数据时出错: {str(e)}") + return False + +def process_purchase_check_info(): + try: + # 执行转换SQL + sql = """ + SELECT + bms.ID as task_id, + bms.BUY_TIME as purchase_time, + bms.ACCEPT_TIME as arrival_time, + ttmt.MANUFACTURER_ID as supplier_id, + bms.BUYER as create_by, + tt.CREATE_TIME as create_time + FROM + ba_ma_shop bms + LEFT JOIN tm_task tt on bms.ID = tt.ID + LEFT JOIN tm_task_status tts on tt.`STATUS` = tts.`CODE` + LEFT JOIN tm_task_ma_type ttmt on bms.ID = ttmt.TASK_ID + WHERE tt.CREATE_TIME BETWEEN '2025-01-01' and NOW() + GROUP BY bms.ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('purchase_check_info', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条新购主表数据") + return True + + except Exception as e: + print(f"处理新购主表数据时出错: {str(e)}") + return False + +def process_purchase_check_details(): + try: + # 执行转换SQL + sql = """ + SELECT + bms.ID as task_id, + ttmt.MA_TYPE_ID as type_id, + ttmt.PRICE as purchase_price, + ttmt.MACHINES_NUM as purchase_num, + ttmt.MACHINES_NUM as check_num, + ttmt.ACTUAL_NUM as bind_num, + ttmt.INPUT_NUM as input_num, + ttmt.MANUFACTURER_ID as supplier_id, + IF(ttmt.MACHINES_NUM - IFNULL(ttmt.INPUT_NUM,0) = 0,19,4) as status, + bms.BUYER as create_by, + tt.CREATE_TIME as create_time + FROM + ba_ma_shop bms + LEFT JOIN tm_task tt on bms.ID = tt.ID + LEFT JOIN tm_task_ma_type ttmt on bms.ID = ttmt.TASK_ID + WHERE tt.CREATE_TIME BETWEEN '2025-01-01' and NOW() + GROUP BY bms.ID,ttmt.MA_TYPE_ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('purchase_check_details', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条租赁申请主表数据") + return True + + except Exception as e: + print(f"处理租赁申请主表数据时出错: {str(e)}") + return False + +if __name__ == "__main__": + # 执行3个转换流程 + success1 = process_tm_task() + success2 = process_purchase_check_info() + success3 = process_purchase_check_details() + + if success1 and success2 and success3: + print("所有数据转换完成！") + else: + print("数据转换过程中出现错误，请检查日志") \ No newline at end of file diff --git a/机具/新购配件.py b/机具/新购配件.py new file mode 100644 index 0000000..eb3baea --- /dev/null +++ b/机具/新购配件.py @@ -0,0 +1,143 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus +# 读取配置文件 +config = configparser.ConfigParser() +config.read(r'D:\code\Bonus-Transfer-Machines\machines\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']}" +) + +def process_tm_task(): + """处理新购任务数据""" + try: + # 执行转换SQL + sql = """ + SELECT + bps.ID as task_id, + 12 as task_type, + IF( SUM(ttmt.MACHINES_NUM) - IFNULL(SUM(ttmt.INPUT_NUM),0) =0,1,0) as task_status, + tt.`CODE` as code, + bps.BUYER as create_by, + tt.CREATE_TIME as create_time + FROM + ba_pa_shop bps + LEFT JOIN tm_task tt on bps.ID = tt.ID + LEFT JOIN tm_task_status tts on tt.`STATUS` = tts.`CODE` + LEFT JOIN tm_task_ma_type ttmt on bps.ID = ttmt.TASK_ID + WHERE tt.CREATE_TIME BETWEEN '2025-01-01' and NOW() + GROUP BY bps.ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('tm_task', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条新购配件任务表数据") + return True + + except Exception as e: + print(f"处理新购配件任务表数据时出错: {str(e)}") + return False + +def process_purchase_part_info(): + try: + # 执行转换SQL + sql = """ + SELECT + bps.ID as task_id, + bps.BUY_TIME as purchase_time, + bps.ACCEPT_TIME as arrival_time, + ttmt.MANUFACTURER_ID as supplier_id, + bps.BUYER as create_by, + tt.CREATE_TIME as create_time + FROM + ba_pa_shop bps + LEFT JOIN tm_task tt on bps.ID = tt.ID + LEFT JOIN tm_task_status tts on tt.`STATUS` = tts.`CODE` + LEFT JOIN tm_task_ma_type ttmt on bps.ID = ttmt.TASK_ID + WHERE tt.CREATE_TIME BETWEEN '2025-01-01' and NOW() + GROUP BY bps.ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('purchase_part_info', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条新购配件主表数据") + return True + + except Exception as e: + print(f"处理新购配件主表数据时出错: {str(e)}") + return False + +def process_purchase_part_details(): + try: + # 执行转换SQL + sql = """ + SELECT + bps.ID as task_id, + ttmt.MA_TYPE_ID as part_id, + ttmt.PRICE as purchase_price, + ttmt.MACHINES_NUM as purchase_num, + ttmt.MACHINES_NUM as check_num, + ttmt.INPUT_NUM as input_num, + ttmt.MANUFACTURER_ID as supplier_id, + IF(ttmt.MACHINES_NUM - IFNULL(ttmt.INPUT_NUM,0) = 0,1,0) as status, + bps.BUYER as create_by, + tt.CREATE_TIME as as create_time + FROM + ba_pa_shop bps + LEFT JOIN tm_task tt on bps.ID = tt.ID + LEFT JOIN tm_task_ma_type ttmt on bps.ID = ttmt.TASK_ID + WHERE tt.CREATE_TIME BETWEEN '2025-01-01' and NOW() + GROUP BY bps.ID,ttmt.MA_TYPE_ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('purchase_part_details', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条新购配件详情表数据") + return True + + except Exception as e: + print(f"处理新购配件详情表数据时出错: {str(e)}") + return False + +if __name__ == "__main__": + # 执行3个转换流程 + success1 = process_tm_task() + success2 = process_purchase_part_info() + success3 = process_purchase_part_details() + + if success1 and success2 and success3: + print("所有数据转换完成！") + else: + print("数据转换过程中出现错误，请检查日志") \ No newline at end of file diff --git a/机具/标准箱.py b/机具/标准箱.py new file mode 100644 index 0000000..2a7bd70 --- /dev/null +++ b/机具/标准箱.py @@ -0,0 +1,105 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus +# 读取配置文件 +config = configparser.ConfigParser() +config.read(r'D:\code\Bonus-Transfer-Machines\machines\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']}" +) + +def process_bm_qrcode_box(): + try: + # 执行转换SQL + sql = """ + SELECT + bqm.BOX_ID as box_id, + bqb.BOX_QRCODE as box_code, + CASE + WHEN bqb.box_status = 1 THEN 4 + WHEN bqb.box_status = 2 THEN 3 + WHEN bqb.box_status = 4 THEN 6 + ELSE 5 + END AS box_status, + 132 as input_user + FROM + bm_qrcode_ma bqm + LEFT JOIN bm_qrcode_box bqb on bqb.ID = bqm.BOX_ID + WHERE bqm.TIME BETWEEN '2025-01-01' and NOW() + GROUP BY bqm.BOX_ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('bm_qrcode_box', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条标准箱表数据") + return True + + except Exception as e: + print(f"处理标准箱表数据时出错: {str(e)}") + return False + +def process_bm_qrcode_box_bind(): + try: + # 执行转换SQL + sql = """ + SELECT + bqm.BOX_ID as box_id, + bqm.MA_ID as ma_id, + pu.`NAME` as create_by, + bqm.TIME as create_time + FROM + bm_qrcode_ma bqm + LEFT JOIN bm_qrcode_box bqb on bqb.ID = bqm.BOX_ID + LEFT JOIN pm_user pu on bqm.`USER` = pu.ID + WHERE bqm.TIME BETWEEN '2025-01-01' and NOW() + GROUP BY bqm.BOX_ID,bqm.MA_ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('bm_qrcode_box_bind', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条标准箱表绑定数据") + return True + + except Exception as e: + print(f"处理标准箱表绑定数据时出错: {str(e)}") + return False + + +if __name__ == "__main__": + # 执行2个转换流程 + success1 = process_bm_qrcode_box() + success2 = process_bm_qrcode_box_bind() + + if success1 and success2: + print("所有数据转换完成！") + else: + print("数据转换过程中出现错误，请检查日志") \ No newline at end of file diff --git a/机具/结算记录.py b/机具/结算记录.py new file mode 100644 index 0000000..78da505 --- /dev/null +++ b/机具/结算记录.py @@ -0,0 +1,186 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus + +# 读取配置文件 +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']}?charset=utf8mb4" +) +target_engine = create_engine( + f"mysql+pymysql://{target_config['user']}:{quote_plus(target_config['password'])}@{target_config['host']}:{target_config['port']}/{target_config['database']}?charset=utf8mb4" +) + + +def safe_convert_to_int(series): + """安全转换为整数""" + return pd.to_numeric(series, errors='coerce').fillna(0).astype(int) + + +def safe_convert_to_float(series): + """安全转换为浮点数""" + return pd.to_numeric(series, errors='coerce').fillna(0.0) + + +def process_slt_agreement_apply(): + """处理结算申请主表""" + try: + sql = """ + SELECT + bma.ID as id, + bma.ID as agreement_id, + bma.`CODE` as code, + bma.SETTLEMENT_TIME as create_time, + IF(bma.IS_BALANCE = 0, 2, 1) as status, + bma.SETTLEMENT_TIME as audit_time + FROM + ba_ma_agreement bma + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() + """ + df = pd.read_sql(sql, source_engine) + + # 转换状态字段 + df['status'] = safe_convert_to_int(df['status']) + + # 确保时间字段格式正确 + time_cols = ['create_time', 'audit_time'] + for col in time_cols: + df[col] = pd.to_datetime(df[col]) + + df.to_sql('slt_agreement_apply', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条结算申请主表数据") + return True + except Exception as e: + print(f"处理结算申请主表时出错: {str(e)}") + return False + + +def process_slt_agreement_details(): + """处理结算申请明细表""" + try: + # 四个UNION查询合并 + sql = """ + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 1, bsd.lease_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 1 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 1 + + UNION + + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 2, bsd.buy_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 2 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 2 + + UNION + + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 2, bsd.buy_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 3 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 4 + + UNION + + SELECT + bma.ID as apply_id, + bsd.MA_TYPE_ID as type_id, + bsd.NUM, + bsd.START_TIME as start_time, + bsd.END_TIME as end_time, + IF(bsd.TYPE = 2, bsd.buy_price, 0) as price, + bsd.MONEY, + bsd.charge as is_charge, + 4 as slt_type + FROM + ba_ma_agreement bma + LEFT JOIN ba_ma_settlement_detail bsd on bma.ID = bsd.AGREEMENT_ID + WHERE bma.SETTLEMENT_TIME BETWEEN '2025-01-01' AND NOW() AND bsd.TYPE = 5 + """ + df = pd.read_sql(sql, source_engine) + + # 转换数值字段 + num_cols = ['type_id', 'NUM', 'slt_type', 'is_charge'] + for col in num_cols: + df[col] = safe_convert_to_int(df[col]) + + # 转换金额字段 + money_cols = ['price', 'MONEY'] + for col in money_cols: + df[col] = safe_convert_to_float(df[col]) + + # 确保时间字段格式正确 + time_cols = ['start_time', 'end_time'] + for col in time_cols: + # 先用正则提取 YYYY-MM-DD 部分 + df[col] = df[col].str.extract(r'(\d{4}-\d{2}-\d{2})') + # 再转换为 datetime + df[col] = pd.to_datetime(df[col], errors='coerce') + + df.to_sql('slt_agreement_details', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条结算申请明细数据") + return True + except Exception as e: + print(f"处理结算申请明细时出错: {str(e)}") + return False + + +if __name__ == "__main__": + # 执行转换流程 + success1 = process_slt_agreement_apply() + success2 = process_slt_agreement_details() + + if success1 and success2: + print("所有结算申请相关表转换完成！") + else: + print("!!! 部分转换失败，请检查错误日志 !!!") diff --git a/机具/维修.py b/机具/维修.py new file mode 100644 index 0000000..9726b38 --- /dev/null +++ b/机具/维修.py @@ -0,0 +1,351 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus + +# 读取配置文件 +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']}?charset=utf8mb4" +) +target_engine = create_engine( + f"mysql+pymysql://{target_config['user']}:{quote_plus(target_config['password'])}@{target_config['host']}:{target_config['port']}/{target_config['database']}?charset=utf8mb4" +) + + +def safe_convert_to_int(series): + """安全转换为整数""" + return pd.to_numeric(series, errors='coerce').fillna(0).astype(int) + + +def safe_convert_to_float(series): + """安全转换为浮点数""" + return pd.to_numeric(series, errors='coerce').fillna(0.0) + + +def process_tm_task(): + """处理维修任务主表""" + try: + sql = """ + SELECT + tt.ID as task_id, + 4 as task_type, + IF(tt.`STATUS` = 6, 1, 0) AS task_status, + bmr.APPLY_NUMBER AS code, + bmr.CREATOR as create_by, + bmr.CREATE_TIME + FROM + ba_ma_repair bmr + LEFT JOIN tm_task tt on bmr.ID = tt.ID + LEFT JOIN tm_task_status tts on tt.`STATUS` = tts.`CODE` + WHERE bmr.company_id = 1 + """ + df = pd.read_sql(sql, source_engine) + + # 转换状态字段 + df['task_status'] = safe_convert_to_int(df['task_status']) + + df.to_sql('tm_task', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修任务主表数据") + return True + except Exception as e: + print(f"处理维修任务主表时出错: {str(e)}") + return False + + +def process_tm_task_agreement(): + """处理维修任务关联表""" + try: + sql = """ + SELECT + bmr.ID as task_id, + bat.AGREEMENT_ID as agreement_id + FROM + ba_ma_repair bmr + LEFT JOIN ba_agreement_task bat on bmr.ID = bat.TASK_ID + WHERE bmr.company_id = 1 + """ + df = pd.read_sql(sql, source_engine) + + # 过滤空agreement_id + df = df.dropna(subset=['agreement_id']) + + df.to_sql('tm_task_agreement', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修任务关联数据") + return True + except Exception as e: + print(f"处理维修任务关联表时出错: {str(e)}") + return False + + +def process_repair_apply_details(): + """处理维修申请明细（合并三个来源）""" + try: + # 来源1：数量申请维修 + sql1 = """ + SELECT + bmr.ID as task_id, + ttmt.MA_TYPE_ID as type_id, + NULL as ma_id, + ttmt.MACHINES_NUM as repair_num, + ttmt.ACTUAL_NUM as repaired_num, + ttmt2.MACHINES_NUM as scrap_num, + ttmt.IS_SURE as status, + bmr.CREATOR as create_by, + bmr.CREATE_TIME, + bmr.BACK_ID as back_id, + IF(tt.`STATUS` = 6, 0, 1) AS is_ds + FROM + ba_ma_repair bmr + LEFT JOIN tm_task tt on bmr.ID = tt.ID + LEFT JOIN tm_task_ma_type ttmt on bmr.ID = ttmt.TASK_ID + LEFT JOIN ba_ma_scarp bms on bms.REPAIR_ID = bmr.ID + LEFT JOIN tm_task_ma_type ttmt2 on bms.ID = ttmt2.TASK_ID AND ttmt.MA_TYPE_ID = ttmt2.MA_TYPE_ID + WHERE ttmt.IS_COUNT = 1 + GROUP BY ttmt.TASK_ID, ttmt.MA_TYPE_ID + """ + + # 来源2：编码申请维修 + sql2 = """ + SELECT + bmr.ID AS task_id, + mm.type AS type_id, + mm.ID AS ma_id, + 1 AS repair_num, +IF + ( tt.`STATUS` = 6, 1, 0 ) AS repaired_num, + 0 AS scrap_num, +IF + ( tt.`STATUS` = 6, 1, 0 ) AS status, + bmr.CREATOR AS create_by, + bmr.CREATE_TIME, + bmr.BACK_ID AS back_id, +IF + ( tt.`STATUS` = 6, 0, 1 ) AS is_ds +FROM + ba_ma_repair_pass brp + LEFT JOIN ba_ma_repair bmr ON brp.repair_id = bmr.ID + LEFT JOIN tm_task tt ON bmr.ID = tt.ID + LEFT JOIN ma_machines mm ON brp.ma_id = mm.id + """ + + # 来源3：编码申请维修含报废 + sql3 = """ + SELECT + bmr.ID as task_id, + mm.type AS type_id, + tmsr.MA_ID as ma_id, + 1 as repair_num, + 0 as repaired_num, + 1 as scrap_num, + 1 as status, + bmr.CREATOR as create_by, + bmr.CREATE_TIME, + bmr.BACK_ID as back_id, + IF(tt.`STATUS` = 6, 0, 1) AS is_ds + FROM + ba_ma_scarp bms + LEFT JOIN ba_ma_repair bmr on bms.repair_id = bmr.ID + LEFT JOIN tm_task tt on bmr.ID = tt.ID + LEFT JOIN tm_ma_scarp_reason tmsr on bms.ID = tmsr.TASK_ID + LEFT JOIN ma_machines mm on tmsr.ma_id = mm.id + WHERE tmsr.IS_COUNT = 0 + """ + + # 合并三个来源的数据 + df1 = pd.read_sql(sql1, source_engine) + df2 = pd.read_sql(sql2, source_engine) + df3 = pd.read_sql(sql3, source_engine) + df = pd.concat([df1, df2, df3], ignore_index=True) + + # 转换数值字段 + num_cols = ['repair_num', 'repaired_num', 'scrap_num', 'status', 'is_ds'] + for col in num_cols: + df[col] = safe_convert_to_int(df[col]) + + df.to_sql('repair_apply_details', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修申请明细（合并3个来源）") + return True + except Exception as e: + print(f"处理维修申请明细时出错: {str(e)}") + return False + + +def process_repair_apply_record(): + """处理维修申请记录（合并四个来源）""" + try: + # 来源1：数量维修合格 + sql1 = """ + SELECT + bmr.ID as task_id, + ttmt.MA_TYPE_ID as type_id, + NULL as ma_id, + ttmt.ACTUAL_NUM as repair_num, + 0 as scrap_num, + ttmt.IS_SURE as status, + bmr.CREATOR as create_by, + IFNULL(bmr.REPAIR_TIME, bmr.CREATE_TIME) as create_time, + NULL as scrap_type, + NULL as scrap_reason + FROM + ba_ma_repair bmr + LEFT JOIN tm_task_ma_type ttmt on bmr.ID = ttmt.TASK_ID + LEFT JOIN ba_ma_scarp bms on bms.REPAIR_ID = bmr.ID + LEFT JOIN tm_task_ma_type ttmt2 on bms.ID = ttmt2.TASK_ID AND ttmt.MA_TYPE_ID = ttmt2.MA_TYPE_ID + WHERE ttmt.IS_COUNT = 1 + GROUP BY ttmt.TASK_ID, ttmt.MA_TYPE_ID + HAVING repair_num > 0 + """ + + # 来源2：数量维修报废 + sql2 = """ + SELECT + bmr.ID as task_id, + ttmt.MA_TYPE_ID as type_id, + NULL as ma_id, + 0 as repair_num, + ttmt2.MACHINES_NUM as scrap_num, + ttmt.IS_SURE as status, + bmr.CREATOR as create_by, + bmr.REPAIR_TIME as create_time, + tmsr.DAMAGE as scrap_type, + tmsr.SCARP_REASON as scrap_reason + FROM + ba_ma_repair bmr + LEFT JOIN tm_task_ma_type ttmt on bmr.ID = ttmt.TASK_ID + LEFT JOIN ba_ma_scarp bms on bms.REPAIR_ID = bmr.ID + LEFT JOIN tm_ma_scarp_reason tmsr on bms.ID = tmsr.TASK_ID + LEFT JOIN tm_task_ma_type ttmt2 on bms.ID = ttmt2.TASK_ID AND ttmt.MA_TYPE_ID = ttmt2.MA_TYPE_ID + WHERE ttmt.IS_COUNT = 1 + GROUP BY ttmt.TASK_ID, ttmt.MA_TYPE_ID + HAVING scrap_num > 0 + """ + + # 来源3：编码维修合格 + sql3 = """ + SELECT + bmr.ID as task_id, + mm.type as type_id, + mm.ID as ma_id, + 1 as repair_num, + 0 as scrap_num, + 1 as status, + bmr.CREATOR as create_by, + bmr.REPAIR_TIME as create_time, + NULL as scrap_type, + NULL as scrap_reason + FROM + ba_ma_repair_pass brp + LEFT JOIN ba_ma_repair bmr on brp.repair_id = bmr.ID + LEFT JOIN ma_machines mm on brp.ma_id = mm.id + """ + + # 来源4：编码维修报废 + sql4 = """ + SELECT + bmr.ID as task_id, + mm.type as type_id, + tmsr.MA_ID as ma_id, + 0 as repair_num, + 1 as scrap_num, + 1 as status, + bmr.CREATOR as create_by, + bmr.REPAIR_TIME as create_time, + tmsr.DAMAGE as scrap_type, + tmsr.SCARP_REASON as scrap_reason + FROM + ba_ma_scarp bms + LEFT JOIN ba_ma_repair bmr on bms.repair_id = bmr.ID + LEFT JOIN tm_ma_scarp_reason tmsr on bms.ID = tmsr.TASK_ID + LEFT JOIN ma_machines mm on tmsr.ma_id = mm.id + WHERE tmsr.IS_COUNT = 0 + """ + + # 合并四个来源的数据 + df1 = pd.read_sql(sql1, source_engine) + df2 = pd.read_sql(sql2, source_engine) + df3 = pd.read_sql(sql3, source_engine) + df4 = pd.read_sql(sql4, source_engine) + df = pd.concat([df1, df2, df3, df4], ignore_index=True) + + # 转换数值字段 + num_cols = ['repair_num', 'scrap_num', 'status'] + for col in num_cols: + df[col] = safe_convert_to_int(df[col]) + + df.to_sql('repair_apply_record', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修申请记录（合并4个来源）") + return True + except Exception as e: + print(f"处理维修申请记录时出错: {str(e)}") + return False + + +def process_repair_cost(): + """处理维修费用""" + try: + sql = """ + SELECT + rp.TASK_ID as task_id, + rp.MA_TYPE_ID as type_id, + rp.MA_ID as ma_id, + rp.REPAIR_NUM as repair_num, + rp.PRICE as costs, + rp.IS_BUCKLE as part_type, + 1 as status + FROM + repair_part rp + LEFT JOIN ba_ma_repair bmr on rp.TASK_ID = bmr.ID + """ + df = pd.read_sql(sql, source_engine) + + # 转换数值字段 + df['repair_num'] = safe_convert_to_int(df['repair_num']) + df['costs'] = safe_convert_to_float(df['costs']) + df['part_type'] = safe_convert_to_int(df['part_type']) + df['status'] = safe_convert_to_int(df['status']) + + df.to_sql('repair_cost', target_engine, if_exists='append', index=False) + print(f"成功导入 {len(df)} 条维修费用记录") + return True + except Exception as e: + print(f"处理维修费用时出错: {str(e)}") + return False + + +if __name__ == "__main__": + # 执行所有转换流程 + processes = [ + process_tm_task, + process_tm_task_agreement, + process_repair_apply_details, + process_repair_apply_record, + process_repair_cost + ] + + success = all([p() for p in processes]) + + if success: + print("所有维修相关表转换完成！") + else: + print("!!! 部分转换失败，请检查错误日志 !!!") \ No newline at end of file diff --git a/机具/退料.py b/机具/退料.py new file mode 100644 index 0000000..17b5615 --- /dev/null +++ b/机具/退料.py @@ -0,0 +1,224 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus +# 读取配置文件 +config = configparser.ConfigParser() +config.read(r'D:\code\Bonus-Transfer-Machines\machines\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']}" +) + +def process_tm_task(): + try: + # 执行转换SQL + sql = """ + SELECT + tt.ID as task_id, + 3 AS task_type, + IF(tt.`STATUS` = 4,2,0) as task_status, + bmb.APPLY_NUMBER as code, + bmb.CREATOR as create_by, + bmb.CREATE_TIME as create_time + FROM + ba_ma_back bmb + LEFT JOIN tm_task tt on bmb.ID = tt.ID + LEFT JOIN tm_task_status ts on tt.`STATUS` = ts.`CODE` + WHERE bmb.company_id =1 + GROUP BY bmb.ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('tm_task', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条退料任务表数据") + return True + + except Exception as e: + print(f"处理退料任务表数据时出错: {str(e)}") + return False + +def process_tm_task_agreement(): + try: + # 执行转换SQL + sql = """ + SELECT + tt.ID as task_id, + bat.AGREEMENT_ID as agreement_id + FROM + ba_ma_back bmb + LEFT JOIN tm_task tt on bmb.ID = tt.ID + LEFT JOIN ba_agreement_task bat on bmb.ID = bat.TASK_ID + WHERE bmb.company_id =1 + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('tm_task_agreement', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条退料任务协议表数据") + return True + + except Exception as e: + print(f"处理退料任务协议表数据时出错: {str(e)}") + return False + +def process_back_apply_info(): + try: + # 执行转换SQL + sql = """ + SELECT + bmb.ID as id, + bmb.APPLY_NUMBER as code, + tt.ID as task_id, + bmb.BACKER AS back_person, + bmb.PHONE as phone, + bmb.CREATOR as create_by, + bmb.CREATE_TIME as create_time, + IF(tt.`STATUS` = 4,1,0) as status, + bmb.IS_PRINT as print_status + FROM + ba_ma_back bmb + LEFT JOIN tm_task tt on bmb.ID = tt.ID + LEFT JOIN tm_task_status ts on tt.`STATUS` = ts.`CODE` + WHERE bmb.company_id =1 + GROUP BY bmb.ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('back_apply_info', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条退料申请主表数据") + return True + + except Exception as e: + print(f"处理退料申请主表数据时出错: {str(e)}") + return False + +def process_back_apply_details(): + """处理退料申请明细数据""" + try: + # 执行转换SQL + sql = """ + SELECT + bmb.APPLY_NUMBER as code, + bmb.id as parent_id, + ttmt.MA_TYPE_ID as type_id, + ttmt.MACHINES_NUM as pre_num, + ttmt.MACHINES_NUM as audit_num , + ttmt.BACK_BAD_NUM as bad_num, + ttmt.BACK_GOOD_NUM as good_num, + IF(tt.`STATUS` = 4,2,0) as status, + bmb.CREATOR as create_by, + bmb.CREATE_TIME as create_time + FROM + ba_ma_back bmb + LEFT JOIN tm_task tt on bmb.ID = tt.ID + LEFT JOIN tm_task_ma_type ttmt on bmb.ID = ttmt.TASK_ID + WHERE bmb.company_id =1 + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('back_apply_details', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条退料申请明细数据") + return True + + except Exception as e: + print(f"处理退料申请明细时出错: {str(e)}") + return False + + +def process_back_check_details(): + """处理退料明细数据""" + try: + # 执行转换SQL + sql = """ + SELECT + bmb.ID as parent_id, + mm.TYPE as type_id, + mm.ID as ma_id, + ttm.CREATOR as create_by, + ttm.CREATE_TIME as create_time, + 1 as back_num , + 1 as back_status , + 1 as good_num, + 0 as bad_num + FROM + ba_ma_back bmb + LEFT JOIN tm_task_ma ttm on bmb.ID = ttm.TASK_ID + LEFT JOIN ma_machines mm on ttm.MA_ID = mm.ID + LEFT JOIN ma_type mt on mm.TYPE = mt.ID + WHERE mt.IS_COUNT = 0 + UNION + SELECT + bmb.id as parent_id, + ttmt.MA_TYPE_ID as type_id, + null as ma_id, + pu.`NAME` as create_by, + bmb.CREATE_TIME as create_time, + ROUND(ttmt.MACHINES_NUM,3) as back_num, + 1 as back_status , + ttmt.BACK_GOOD_NUM as ood_num, + ttmt.BACK_BAD_NUM as bad_num + FROM + ba_ma_back bmb + LEFT JOIN tm_task_ma_type ttmt on bmb.ID = ttmt.TASK_ID + LEFT JOIN pm_user pu on bmb.CREATOR = pu.ID + WHERE ttmt.IS_COUNT =1 + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('back_check_details', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条退料明细数据") + return True + + except Exception as e: + print(f"处理退料明细时出错: {str(e)}") + return False + + +if __name__ == "__main__": + # 执行5个转换流程 + success1 = process_tm_task() + success2 = process_tm_task_agreement() + success3 = process_back_apply_info() + success4 = process_back_apply_details() + success5 = process_back_check_details() + + if success1 and success2 and success3 and success4 and success5: + print("所有数据转换完成！") + else: + print("数据转换过程中出现错误，请检查日志") \ No newline at end of file diff --git a/机具/领料.py b/机具/领料.py new file mode 100644 index 0000000..8b8c896 --- /dev/null +++ b/机具/领料.py @@ -0,0 +1,217 @@ +import configparser +import pandas as pd +from sqlalchemy import create_engine +from urllib.parse import quote_plus +# 读取配置文件 +config = configparser.ConfigParser() +config.read(r'D:\code\Bonus-Transfer-Machines\machines\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']}" +) + +def process_tm_task(): + try: + # 执行转换SQL + sql = """ + SELECT + tt.ID as task_id, + 2 as task_type, + IF(SUM(ttmt.MACHINES_NUM)- SUM(ttmt.ACTUAL_NUM )>0,3,4) as task_status, + bmc.APPLY_NUMBER as code, + pu.`NAME` as create_by, + bmc.CREATE_TIME as create_time + FROM + ba_ma_collar bmc + LEFT JOIN tm_task tt on bmc.ID = tt.ID + LEFT JOIN tm_task_ma_type ttmt on bmc.ID = ttmt.TASK_ID + LEFT JOIN pm_user pu on bmc.CREATOR = pu.ID + WHERE bmc.company_id =1 + GROUP BY bmc.ID + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('tm_task', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条租赁任务表数据") + return True + + except Exception as e: + print(f"处理租赁任务表数据时出错: {str(e)}") + return False + +def process_tm_task_agreement(): + try: + # 执行转换SQL + sql = """ + SELECT + tt.ID as task_id, + bat.AGREEMENT_ID as agreement_id + FROM + ba_ma_collar bmc + LEFT JOIN tm_task tt on bmc.ID = tt.ID + LEFT JOIN ba_agreement_task bat on bmc.ID = bat.TASK_ID + WHERE bmc.company_id =1 + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('tm_task_agreement', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条租赁任务协议表数据") + return True + + except Exception as e: + print(f"处理租赁任务协议表数据时出错: {str(e)}") + return False + +def process_lease_apply_info(): + try: + # 执行转换SQL + sql = """ + SELECT + bmc.ID as id, + bmc.APPLY_NUMBER as code, + tt.ID as task_id, + bmc.COLLAR_MAN as lease_person, + bmc.COLLAR_PHONE as phone, + pu.`NAME` as create_by, + bmc.CREATE_TIME as create_time, + bmc.direct_id as direct_id, + bma.LEASE_COMPANY as unit_id, + bma.PROJECT as project_id + FROM + ba_ma_collar bmc + LEFT JOIN tm_task tt on bmc.ID = tt.ID + LEFT JOIN ba_agreement_task bat on bmc.ID = bat.TASK_ID + LEFT JOIN ba_ma_agreement bma on bat.AGREEMENT_ID = bma.ID + LEFT JOIN pm_user pu on bmc.CREATOR = pu.ID + WHERE bmc.company_id =1 + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('lease_apply_info', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条租赁申请主表数据") + return True + + except Exception as e: + print(f"处理租赁申请主表数据时出错: {str(e)}") + return False + +def process_lease_apply_details(): + """处理租赁申请明细数据""" + try: + # 执行转换SQL + sql = """ + SELECT + bmc.ID as parent_id, + ttmt.MA_TYPE_ID as type_id, + ttmt.MACHINES_NUM as pre_num, + ttmt.ACTUAL_NUM as al_num, + IF(ttmt.MACHINES_NUM - ttmt.ACTUAL_NUM =0,2,0) AS status + FROM + ba_ma_collar bmc + LEFT JOIN tm_task_ma_type ttmt on bmc.ID = ttmt.TASK_ID + WHERE bmc.company_id =1 + AND ttmt.MA_TYPE_ID IS NOT NULL + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('lease_apply_details', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条租赁申请明细数据") + return True + + except Exception as e: + print(f"处理租赁申请明细时出错: {str(e)}") + return False + + +def process_lease_out_details(): + """处理租赁出库明细数据""" + try: + # 执行转换SQL + sql = """ + SELECT + bmc.ID as parent_id, + mm.TYPE as type_id, + mm.ID as ma_id, + ttm.CREATOR as create_by, + ttm.CREATE_TIME as create_time, + 1 as out_num + FROM + ba_ma_collar bmc + LEFT JOIN tm_task_ma ttm on bmc.ID = ttm.TASK_ID + LEFT JOIN ma_machines mm on ttm.MA_ID = mm.ID + LEFT JOIN ma_type mt on mm.TYPE = mt.ID + WHERE bmc.company_id =1 AND mt.IS_COUNT = 0 + UNION + SELECT + bmc.ID as parent_id, + ttot.MA_TYPE_ID as type_id, + null as ma_id, + pu.`NAME` as create_by, + ttot.CREATE_TIME as create_time, + ttot.NUM as out_num + FROM + ba_ma_collar bmc + LEFT JOIN tm_task_out_type ttot on bmc.ID = ttot.TASK_ID + LEFT JOIN ma_type mt on ttot.MA_TYPE_ID = mt.ID + LEFT JOIN pm_user pu on ttot.CREATOR = pu.ID + WHERE bmc.company_id =1 AND ttot.IS_COUNT =1 + """ + df = pd.read_sql(sql, source_engine) + + # 写入目标表 + df.to_sql('lease_out_details', target_engine, + if_exists='append', index=False) + + print(f"成功导入 {len(df)} 条租赁出库明细数据") + return True + + except Exception as e: + print(f"处理租赁出库明细时出错: {str(e)}") + return False + + +if __name__ == "__main__": + # 执行5个转换流程 + success1 = process_tm_task() + success2 = process_tm_task_agreement() + success3 = process_lease_apply_info() + success4 = process_lease_apply_details() + success5 = process_lease_out_details() + + if success1 and success2 and success3 and success4 and success5: + print("所有数据转换完成！") + else: + print("数据转换过程中出现错误，请检查日志") \ No newline at end of file