## 获取数据 # -*- 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: """合伙人结算登记同步到BI""" 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_1756870421570", "订单登记表.销售阶段": "_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", "提交时间": "createTime", "更新时间": "updateTime" } # 定义需要特殊处理的列表字段及其内部字段映射 self.list_fields_config = { "订单登记表": { "_widget_1756870421570": "订单编号", "_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: logger.warning("数据为空终止程序") task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") common_module.send_task_status(task_start_time, "合伙人结算登记同步到BI") 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.日期字段转为北京时间 time_columns = ["提交时间", "更新时间"] df[time_columns] = df[time_columns].apply( lambda col: pd.to_datetime(col, errors='coerce') .dt.tz_localize(None) .dt.strftime('%Y-%m-%d %H:%M:%S') ) # 4.处理订单登记表列表字段，将其拆分成多行 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()