280 lines
11 KiB
Python
280 lines
11 KiB
Python
## 获取数据
|
|
# -*- coding: utf-8 -*-
|
|
import pandas as pd
|
|
import datetime
|
|
from config import Config
|
|
from api import API
|
|
import pymysql # 使用 pymysql 替代 mysql.connector
|
|
from back_ground_module import CommonModule
|
|
import os
|
|
import mysql.connector
|
|
import pandas as pd
|
|
import json
|
|
import numpy as np
|
|
import mysql.connector
|
|
from mysql.connector import Error
|
|
from log_config import configure_task_logger, configure_error_task_logger
|
|
import sys
|
|
|
|
logger = configure_task_logger()
|
|
error_task_logger = configure_error_task_logger()
|
|
api_instance = API()
|
|
common_module = CommonModule()
|
|
output_dir = "output" # 设置输出目录
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
|
|
|
|
class PartnerSettlementToBI:
|
|
def __init__(self):
|
|
self.partner_settlement_data = None
|
|
self.field_mapping = {
|
|
"选择合伙人": "_widget_1753930627469",
|
|
"合伙人姓名": "_widget_1712801992726",
|
|
"手机号": "_widget_1712803222895",
|
|
"合伙人身份": "_widget_1712803222894",
|
|
"合伙人所在省市": "_widget_1712803222896",
|
|
"合伙人登记人": "_widget_1712803222900",
|
|
"战区经理": "_widget_1712803222901",
|
|
"提交人": "_widget_1753941892609",
|
|
"合伙人分类": "_widget_1753943042503",
|
|
"战区": "_widget_1754530653275",
|
|
"订单登记表": "_widget_1712803222905",
|
|
"订单登记表.订单编号": "_widget_1712803222905._widget_1712803222907",
|
|
"订单登记表.销售阶段": "_widget_1712803222905._widget_1712805391009",
|
|
"订单登记表.版本": "_widget_1712803222905._widget_1712803222908",
|
|
"订单登记表.年限": "_widget_1712803222905._widget_1712815331264",
|
|
"订单登记表.成交金额": "_widget_1712803222905._widget_1712805391002",
|
|
"订单登记表.佣金": "_widget_1712803222905._widget_1753952737266",
|
|
"订单登记表.理论佣金": "_widget_1712803222905._widget_1753952737267",
|
|
"订单登记表.佣金比例": "_widget_1712803222905._widget_1712807001396",
|
|
"合计佣金": "_widget_1753948415171",
|
|
"理论合计佣金": "_widget_1753952737280",
|
|
"特殊情况备注": "_widget_1712805391035",
|
|
"合伙人介绍证明（微信聊天截图等）": "_widget_1712815331256",
|
|
"合伙人类型": "_widget_1753957844818",
|
|
}
|
|
|
|
# 定义需要特殊处理的列表字段及其内部字段映射
|
|
self.list_fields_config = {
|
|
"订单登记表": {
|
|
"_widget_1712803222907": "订单编号",
|
|
"_widget_1712805391009": "销售阶段",
|
|
"_widget_1712803222908": "版本",
|
|
"_widget_1712815331264": "年限",
|
|
"_widget_1712805391002": "成交金额",
|
|
"_widget_1753952737266": "佣金",
|
|
"_widget_1753952737267": "理论佣金",
|
|
"_widget_1712807001396": "佣金比例",
|
|
},
|
|
# 可以在这里添加其他列表字段的配置
|
|
# "另一个列表字段": {
|
|
# "原始字段名1": "映射后字段名1",
|
|
# "原始字段名2": "映射后字段名2"
|
|
# }
|
|
}
|
|
|
|
def load_all_data(self):
|
|
payload = {"api_key": "66b9678280b37f8a276b1d01",
|
|
# "entry_id": "68a57e3a0bc339d3384d1b0c", # 测试
|
|
"entry_id": "661748c7c727764d79557674",
|
|
}
|
|
partner_settlement = api_instance.entry_data_list(payload)
|
|
self.partner_settlement_data = partner_settlement.get("data") # api请求格式，将数据封装在data字典里
|
|
|
|
def process_list_field(self, field_value, field_config):
|
|
"""通用方法：处理列表类型的字段"""
|
|
if not isinstance(field_value, (list, np.ndarray)):
|
|
return field_value
|
|
|
|
processed_list = []
|
|
for item in field_value:
|
|
if not isinstance(item, dict):
|
|
processed_list.append(item)
|
|
continue
|
|
|
|
processed_item = {}
|
|
for original_key, mapped_key in field_config.items():
|
|
if original_key in item:
|
|
# 处理包含id的字典字段
|
|
if isinstance(item[original_key], dict) and "id" in item[original_key]:
|
|
processed_item[mapped_key] = item[original_key]["id"]
|
|
else:
|
|
processed_item[mapped_key] = item[original_key]
|
|
else:
|
|
processed_item[mapped_key] = None
|
|
processed_list.append(processed_item)
|
|
return processed_list
|
|
|
|
def data_process(self):
|
|
if not self.partner_settlement_data:
|
|
print("数据为空终止程序")
|
|
sys.exit(1)
|
|
df = pd.DataFrame(self.partner_settlement_data)
|
|
# 反转映射字典
|
|
reverse_mapping = {v: k for k, v in self.field_mapping.items()}
|
|
# 1.列明替换
|
|
df.columns = [reverse_mapping.get(col, col) for col in df.columns]
|
|
|
|
# 2.成员字段取值
|
|
user_columns = ["合伙人登记人", "提交人", "战区经理"]
|
|
|
|
for col in user_columns:
|
|
df[col] = df[col].map(lambda x: x.get("name", "") if isinstance(x, dict) else "")
|
|
|
|
# 3.处理订单登记表列表字段，将其拆分成多行
|
|
if "订单登记表" in df.columns:
|
|
# 先处理订单登记表字段
|
|
df["订单登记表"] = df["订单登记表"].apply(
|
|
lambda x: self.process_list_field(x, self.list_fields_config["订单登记表"])
|
|
if x is not None and (isinstance(x, (list, dict, np.ndarray)) or not pd.isna(x))
|
|
else None
|
|
)
|
|
|
|
# 拆分行
|
|
df_exploded = df.explode("订单登记表")
|
|
|
|
# 将订单登记表中的字段提取到主表中
|
|
order_fields = self.list_fields_config["订单登记表"].values()
|
|
for field in order_fields:
|
|
df_exploded[field] = df_exploded["订单登记表"].apply(
|
|
lambda x: x.get(field) if isinstance(x, dict) else None
|
|
)
|
|
|
|
# 删除原始的订单登记表列
|
|
df_exploded = df_exploded.drop(columns=["订单登记表"])
|
|
|
|
# 重置索引
|
|
df = df_exploded.reset_index(drop=True)
|
|
|
|
return df
|
|
|
|
def write_to_bi(self, df):
|
|
# 数据库连接信息
|
|
HS_DB_Config = {
|
|
'host': "f6-public.rwlb.rds.aliyuncs.com",
|
|
'user': "rw_operation_data_relay",
|
|
'password': "m+q5Z4%IVuF9bf",
|
|
'database': "f6operation_data_relay"
|
|
}
|
|
table_name = "partner_settlement_to_BI" # 替换为你的实际表名
|
|
|
|
# 建立数据库连接
|
|
connection = mysql.connector.connect(
|
|
host=HS_DB_Config["host"],
|
|
user=HS_DB_Config["user"],
|
|
password=HS_DB_Config["password"],
|
|
database=HS_DB_Config["database"]
|
|
)
|
|
cursor = connection.cursor()
|
|
|
|
try:
|
|
# 查询表列名
|
|
cursor.execute(f"SHOW COLUMNS FROM {table_name}")
|
|
columns_info = cursor.fetchall()
|
|
db_columns = [col[0] for col in columns_info] # 提取列名
|
|
df = df.replace([None, np.nan, pd.NA, 'nan', 'NaN', 'NAN', ''], None)
|
|
# 保留 DataFrame 中与数据库列名匹配的列
|
|
filtered_df = df[df.columns.intersection(db_columns)]
|
|
|
|
# 如果没有匹配的列，直接返回
|
|
if filtered_df.empty:
|
|
print("DataFrame 中没有与数据库表结构匹配的列。")
|
|
return
|
|
|
|
# 筛选列之后，插入前处理 dict 类型
|
|
filtered_df = filtered_df.copy()
|
|
for col in filtered_df.columns:
|
|
if filtered_df[col].apply(lambda x: isinstance(x, (dict, list)) if x is not None else False).any():
|
|
filtered_df.loc[:, col] = filtered_df[col].apply(
|
|
lambda x: json.dumps(x, ensure_ascii=False) if x is not None else x
|
|
)
|
|
|
|
# 构建插入语句
|
|
placeholders = ', '.join(['%s'] * len(filtered_df.columns))
|
|
# 使用反引号避免特殊列明
|
|
columns = ', '.join([f"`{col}`" for col in filtered_df.columns])
|
|
insert_sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
|
|
|
|
# 将 DataFrame 写入数据库
|
|
for _, row in filtered_df.iterrows():
|
|
cursor.execute(insert_sql, tuple(row))
|
|
|
|
connection.commit()
|
|
logger.info(f"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。")
|
|
|
|
except Exception as e:
|
|
error_task_logger.error(f"写入数据库时发生错误: {e}")
|
|
connection.rollback()
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
|
|
def clear_table_data(self):
|
|
"""
|
|
清空指定 MySQL 表的数据。
|
|
参数已写死在函数内部，直接调用即可。
|
|
"""
|
|
# 数据库连接信息
|
|
HS_DB_Config = {
|
|
'host': "f6-public.rwlb.rds.aliyuncs.com",
|
|
'user': "rw_operation_data_relay",
|
|
'password': "m+q5Z4%IVuF9bf",
|
|
'database': "f6operation_data_relay"
|
|
}
|
|
table_name = "partner_settlement_to_BI" # 要清空的表名
|
|
|
|
connection = None
|
|
try:
|
|
# 建立数据库连接
|
|
connection = mysql.connector.connect(
|
|
host=HS_DB_Config["host"],
|
|
user=HS_DB_Config["user"],
|
|
password=HS_DB_Config["password"],
|
|
database=HS_DB_Config["database"]
|
|
)
|
|
if connection.is_connected():
|
|
cursor = connection.cursor()
|
|
|
|
# 使用TRUNCATE清空表数据
|
|
cursor.execute(f"TRUNCATE TABLE {table_name}")
|
|
connection.commit()
|
|
|
|
print(f"成功清空表 {table_name} 中的所有数据")
|
|
|
|
except Error as e:
|
|
print(f"清空表时发生错误: {e}")
|
|
if connection and connection.is_connected():
|
|
connection.rollback()
|
|
finally:
|
|
if connection and connection.is_connected():
|
|
cursor.close()
|
|
connection.close()
|
|
print("数据库连接已关闭")
|
|
|
|
def main(self):
|
|
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
try:
|
|
logger.info("任务开始")
|
|
# step1: 获取数据
|
|
self.load_all_data()
|
|
logger.info("加载数据完成")
|
|
# step2:数据处理
|
|
df = self.data_process()
|
|
# df.to_csv(os.path.join(output_dir, "new_dealer_service_order_to_bi.csv"))
|
|
logger.info("数据处理完成")
|
|
# step3:数据库删除
|
|
self.clear_table_data()
|
|
logger.info("目标数据库已清空")
|
|
# step4:数据写入BI
|
|
self.write_to_bi(df)
|
|
logger.info("数据已写入数据库中")
|
|
|
|
common_module.send_task_status(task_start_time, "合伙人结算登记同步到BI")
|
|
except Exception as e:
|
|
error_task_logger.error(f"合伙人结算登记同步到BI发生错误：{e}")
|
|
common_module.send_task_error(task_start_time, "合伙人结算登记同步到BI", str(e))
|
|
|
|
if "__main__" == __name__:
|
|
partnerSettlementToBI = PartnerSettlementToBI()
|
|
partnerSettlementToBI.main()
|