338 lines
14 KiB
Python
338 lines
14 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
|
|
|
|
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 NonStandardPerformanceToBI:
|
|
""" 非标业绩提报转BI"""
|
|
|
|
def __init__(self):
|
|
self.dealer_service_data = None
|
|
self.field_mapping = {
|
|
"报备类型": "_widget_1753770875899",
|
|
"协作内容": "_widget_1753770875915",
|
|
"情况说明": "_widget_1753770875944",
|
|
"订单编号": "_widget_1753770875887",
|
|
"实付金额": "_widget_1753770875889",
|
|
"门店编码": "_widget_1753770875890",
|
|
"门店名称": "_widget_1753770875888",
|
|
"版本": "_widget_1753770875891",
|
|
"年限": "_widget_1753948745953",
|
|
"支付日期": "_widget_1753770875893",
|
|
"开户/处理日期": "_widget_1753770875894",
|
|
"小六业绩金额": "_widget_1753770875898",
|
|
"区域业绩金额": "_widget_1753770875937",
|
|
"报备业绩归属区域经理": "_widget_1753770875903",
|
|
"报备业绩归属大区": "_widget_1753866196486",
|
|
"原业绩归属人": "_widget_1753856032683",
|
|
"原业绩归属区域经理": "_widget_1753866196485",
|
|
"小六业绩比例": "_widget_1753770875917",
|
|
"区域业绩比例": "_widget_1753770875921",
|
|
"运营专家": "_widget_1753770875902",
|
|
"提成类型": "_widget_1753778922504",
|
|
"SaaS新签提成比例": "_widget_1753770875949",
|
|
"服务包提成比例": "_widget_1753778922567",
|
|
"提成金额": "_widget_1753770875948",
|
|
"新签提成比例-首年": "_widget_1753778922503",
|
|
"新签提成比例-非首年": "_widget_1753778922548",
|
|
"新签阶段及提成比例": "_widget_1753778656359",
|
|
"业绩动作": "_widget_1756708722933",
|
|
"提成动作": "_widget_1756708722932",
|
|
"新签阶段及提成比例.选择提成阶段": "_widget_1753778656359._widget_1753778656361",
|
|
"新签阶段及提成比例.新签阶段": "_widget_1753778656359._widget_1753948745962",
|
|
"新签阶段及提成比例.提成比例": "_widget_1753778656359._widget_1753778656362",
|
|
"业绩类型": "_widget_1753770875966",
|
|
"报备业绩归属小六": "_widget_1753770875901",
|
|
"原业绩归属大区": "_widget_1755159216098",
|
|
"业绩分类": "_widget_1758706882564",
|
|
"流程是否结束": "_widget_1761633418013",
|
|
"业绩类型-聚合": "_widget_1758706882564",
|
|
"业绩分组": "_widget_1762417447169",
|
|
"商品名称": "_widget_1762219744898",
|
|
"履约金额": "_widget_1762220516367",
|
|
"业绩归属日期": "_widget_1762417447127",
|
|
"公司名称": "_widget_1762420723743",
|
|
"公司ID": "_widget_1762420723744",
|
|
"报备业绩金额-区域提交": "_widget_1766375035236",
|
|
"业绩归属小六-区域提交": "_widget_1766461143813",
|
|
"业绩归属月": "_widget_1766375035265",
|
|
"是否同步衡石": "_widget_1766484337844",
|
|
"提交人": "creator",
|
|
"提交时间": "createTime",
|
|
"更新时间": "updateTime"
|
|
}
|
|
|
|
# 定义需要特殊处理的列表字段及其内部字段映射
|
|
self.list_fields_config = {
|
|
"新签阶段及提成比例": {
|
|
"_widget_1753778656361": "选择提成阶段",
|
|
"_widget_1753948745962": "新签阶段",
|
|
"_widget_1753778656362": "提成比例"
|
|
},
|
|
# 可以在这里添加其他列表字段的配置
|
|
# "另一个列表字段": {
|
|
# "原始字段名1": "映射后字段名1",
|
|
# "原始字段名2": "映射后字段名2"
|
|
# }
|
|
}
|
|
|
|
def load_all_data(self):
|
|
# 获取非标业绩提报数据
|
|
payload = {"api_key": "66b9678280b37f8a276b1d01",
|
|
"entry_id": "68886b7c0382a7249ae0b5d6",
|
|
}
|
|
dealer_service = api_instance.entry_data_list(payload)
|
|
self.dealer_service_data = dealer_service.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):
|
|
df = pd.DataFrame(self.dealer_service_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]
|
|
|
|
# 只保留流程是否结束为是的内容
|
|
target_col = "流程是否结束"
|
|
if target_col in df.columns:
|
|
# 只有当列存在时才进行过滤，且 pandas 会自动处理 NaN != "是" 的情况
|
|
df = df[df[target_col] == "是"]
|
|
else:
|
|
logger.warning(f"字段 '{target_col}' 不存在，跳过过滤步骤，保留所有数据或根据业务需求处理。")
|
|
|
|
if df.empty:
|
|
logger.info("过滤后数据为空，无需后续处理。")
|
|
return df
|
|
|
|
# 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 = ["支付日期", "开户/处理日期", "提交时间", "更新时间", "业绩归属月", "业绩归属日期"]
|
|
|
|
for col in time_columns:
|
|
# 1. 解析为 datetime，并明确指定为 UTC（即使原始字符串无时区）
|
|
dt_utc = pd.to_datetime(df[col], errors='coerce', utc=True)
|
|
|
|
# 2. 转换为北京时间
|
|
dt_beijing = dt_utc.dt.tz_convert('Asia/Shanghai')
|
|
|
|
# 3. 去掉时区信息（变成 naive datetime），然后格式化为字符串
|
|
df[col] = dt_beijing.dt.tz_localize(None).dt.strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
# 4.业绩动作等于拆单做复制
|
|
|
|
# 4.1. 定义条件
|
|
mask = df['业绩动作'] == '拆单'
|
|
|
|
# 4.2. 复制满足条件的行
|
|
new_rows = df[mask].copy() # ⚠️ 一定要用 .copy() 避免 SettingWithCopyWarning
|
|
|
|
# 3. 修改新行中的某些列
|
|
new_rows['小六业绩金额'] = -new_rows['小六业绩金额']
|
|
new_rows['区域业绩金额'] = -new_rows['区域业绩金额']
|
|
new_rows['报备业绩归属小六'] = new_rows['原业绩归属人']
|
|
new_rows['报备业绩归属区域经理'] = new_rows['原业绩归属区域经理']
|
|
new_rows['报备业绩归属大区'] = new_rows['原业绩归属大区']
|
|
|
|
# 4. 合并回原 DataFrame
|
|
df = pd.concat([df, new_rows], ignore_index=True)
|
|
|
|
# 5.处理所有配置的列表字段
|
|
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 = Config.HS_DB_Config
|
|
table_name = "non_standard_performance_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 = "non_standard_performance_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()
|
|
|
|
logger.info(f"成功清空表 {table_name} 中的所有数据")
|
|
|
|
except Error as e:
|
|
error_task_logger.error(f"清空表时发生错误: {e}")
|
|
if connection and connection.is_connected():
|
|
connection.rollback()
|
|
finally:
|
|
if connection and connection.is_connected():
|
|
cursor.close()
|
|
connection.close()
|
|
logger.info("数据库连接已关闭")
|
|
|
|
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 __name__ == '__main__':
|
|
start = NonStandardPerformanceToBI()
|
|
start.main()
|