首次提交
This commit is contained in:
jiang
parent
9ef5eaa08f
commit
5e7926835c
|
|
@ -0,0 +1,6 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="VcsDirectoryMappings">
|
||||
<mapping directory="$PROJECT_DIR$" vcs="Git" />
|
||||
</component>
|
||||
</project>
|
||||
|
|
@ -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("!!! 部分转换失败，请检查错误日志 !!!")
|
||||
|
|
@ -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("!!! 部分转换失败，请检查错误日志 !!!")
|
||||
|
|
@ -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=== 迁移中止 ===")
|
||||
|
|
@ -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=== 迁移中止 ===")
|
||||
|
|
@ -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("数据转换过程中出现错误，请检查日志")
|
||||
|
|
@ -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("数据转换过程中出现错误，请检查日志")
|
||||
|
|
@ -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("数据转换过程中出现错误，请检查日志")
|
||||
|
|
@ -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("!!! 部分转换失败，请检查错误日志 !!!")
|
||||
|
|
@ -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("!!! 部分转换失败，请检查错误日志 !!!")
|
||||
|
|
@ -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("数据转换过程中出现错误，请检查日志")
|
||||
|
|
@ -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("数据转换过程中出现错误，请检查日志")
|
||||
Loading…
Reference in New Issue