684 lines
32 KiB
Plaintext
684 lines
32 KiB
Plaintext
{
|
|
"cells": [
|
|
{
|
|
"metadata": {},
|
|
"cell_type": "markdown",
|
|
"source": "## 获取数据",
|
|
"id": "1af8678c57a7160"
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"id": "initial_id",
|
|
"metadata": {
|
|
"collapsed": true,
|
|
"ExecuteTime": {
|
|
"end_time": "2025-08-20T08:58:29.047944Z",
|
|
"start_time": "2025-08-20T08:58:27.944621Z"
|
|
}
|
|
},
|
|
"source": [
|
|
"# -*- coding: utf-8 -*-\n",
|
|
"import pandas as pd\n",
|
|
"import datetime\n",
|
|
"from config import Config\n",
|
|
"from api import API\n",
|
|
"import pymysql # 使用 pymysql 替代 mysql.connector\n",
|
|
"from back_ground_module import CommonModule\n",
|
|
"import os\n",
|
|
"import mysql.connector\n",
|
|
"import pandas as pd\n",
|
|
"import json\n",
|
|
"import numpy as np\n",
|
|
"import mysql.connector\n",
|
|
"from mysql.connector import Error\n",
|
|
"from log_config import configure_task_logger, configure_error_task_logger\n",
|
|
"\n",
|
|
"logger = configure_task_logger()\n",
|
|
"error_task_logger = configure_error_task_logger()\n",
|
|
"api_instance = API()\n",
|
|
"common_module = CommonModule()\n",
|
|
"output_dir = \"output\" # 设置输出目录\n",
|
|
"os.makedirs(output_dir, exist_ok=True)\n",
|
|
"\n",
|
|
"\n",
|
|
"class NonStandardPerformanceToBI:\n",
|
|
" def __init__(self):\n",
|
|
" self.dealer_service_data = None\n",
|
|
" self.field_mapping = {\n",
|
|
" \"报备类型\": \"_widget_1753770875899\",\n",
|
|
" \"协作内容\": \"_widget_1753770875915\",\n",
|
|
" \"订单类型\": \"_widget_1753770875966\",\n",
|
|
" \"情况说明\": \"_widget_1753770875944\",\n",
|
|
" \"订单编号\": \"_widget_1753770875887\",\n",
|
|
" \"实付金额\": \"_widget_1753770875889\",\n",
|
|
" \"门店编码\": \"_widget_1753770875890\",\n",
|
|
" \"门店名称\": \"_widget_1753770875888\",\n",
|
|
" \"版本\": \"_widget_1753770875891\",\n",
|
|
" \"年限\": \"_widget_1753948745953\",\n",
|
|
" \"支付日期\": \"_widget_1753770875893\",\n",
|
|
" \"开户/处理日期\": \"_widget_1753770875894\",\n",
|
|
" \"小六业绩金额\": \"_widget_1753770875898\",\n",
|
|
" \"区域业绩金额\": \"_widget_1753770875937\",\n",
|
|
" \"报备业绩归属人\": \"_widget_1753770875901\",\n",
|
|
" \"报备业绩归属区域经理\": \"_widget_1753770875903\",\n",
|
|
" \"报备业绩归属大区\": \"_widget_1753866196486\",\n",
|
|
" \"原业绩归属人\": \"_widget_1753856032683\",\n",
|
|
" \"原业绩归属区域经理\": \"_widget_1753866196485\",\n",
|
|
" \"小六业绩比例\": \"_widget_1753770875917\",\n",
|
|
" \"区域业绩比例\": \"_widget_1753770875921\",\n",
|
|
" \"运营专家\": \"_widget_1753770875902\",\n",
|
|
" \"提成类型\": \"_widget_1753778922504\",\n",
|
|
" \"SaaS新签提成比例\": \"_widget_1753770875949\",\n",
|
|
" \"服务包提成比例\": \"_widget_1753778922567\",\n",
|
|
" \"提成金额\": \"_widget_1753770875948\",\n",
|
|
" \"新签提成比例-首年\": \"_widget_1753778922503\",\n",
|
|
" \"新签提成比例-非首年\": \"_widget_1753778922548\",\n",
|
|
" \"新签阶段及提成比例\": \"_widget_1753778656359\",\n",
|
|
" \"新签阶段及提成比例.选择提成阶段\": \"_widget_1753778656359._widget_1753778656361\",\n",
|
|
" \"新签阶段及提成比例.新签阶段\": \"_widget_1753778656359._widget_1753948745962\",\n",
|
|
" \"新签阶段及提成比例.提成比例\": \"_widget_1753778656359._widget_1753778656362\",\n",
|
|
" }\n",
|
|
"\n",
|
|
" # 定义需要特殊处理的列表字段及其内部字段映射\n",
|
|
" self.list_fields_config = {\n",
|
|
" \"新签阶段及提成比例\": {\n",
|
|
" \"_widget_1753778656361\": \"选择提成阶段\",\n",
|
|
" \"_widget_1753948745962\": \"新签阶段\",\n",
|
|
" \"_widget_1753778656362\": \"提成比例\"\n",
|
|
" },\n",
|
|
" # 可以在这里添加其他列表字段的配置\n",
|
|
" # \"另一个列表字段\": {\n",
|
|
" # \"原始字段名1\": \"映射后字段名1\",\n",
|
|
" # \"原始字段名2\": \"映射后字段名2\"\n",
|
|
" # }\n",
|
|
" }\n",
|
|
"\n",
|
|
" def load_all_data(self):\n",
|
|
" # 获取非标业绩提报数据\n",
|
|
" payload = {\"api_key\": \"66b9678280b37f8a276b1d01\",\n",
|
|
" \"entry_id\": \"68886b7c0382a7249ae0b5d6\",\n",
|
|
" }\n",
|
|
" dealer_service = api_instance.entry_data_list(payload)\n",
|
|
" self.dealer_service_data = dealer_service.get(\"data\") # api请求格式，将数据封装在data字典里\n",
|
|
"\n",
|
|
" def process_list_field(self, field_value, field_config):\n",
|
|
" \"\"\"通用方法：处理列表类型的字段\"\"\"\n",
|
|
" if not isinstance(field_value, (list, np.ndarray)):\n",
|
|
" return field_value\n",
|
|
"\n",
|
|
" processed_list = []\n",
|
|
" for item in field_value:\n",
|
|
" if not isinstance(item, dict):\n",
|
|
" processed_list.append(item)\n",
|
|
" continue\n",
|
|
"\n",
|
|
" processed_item = {}\n",
|
|
" for original_key, mapped_key in field_config.items():\n",
|
|
" if original_key in item:\n",
|
|
" # 处理包含id的字典字段\n",
|
|
" if isinstance(item[original_key], dict) and \"id\" in item[original_key]:\n",
|
|
" processed_item[mapped_key] = item[original_key][\"id\"]\n",
|
|
" else:\n",
|
|
" processed_item[mapped_key] = item[original_key]\n",
|
|
" else:\n",
|
|
" processed_item[mapped_key] = None\n",
|
|
" processed_list.append(processed_item)\n",
|
|
" return processed_list\n",
|
|
"\n",
|
|
" def data_process(self):\n",
|
|
" df = pd.DataFrame(self.dealer_service_data)\n",
|
|
" # 反转映射字典\n",
|
|
" reverse_mapping = {v: k for k, v in self.field_mapping.items()}\n",
|
|
" # 1.列明替换\n",
|
|
" df.columns = [reverse_mapping.get(col, col) for col in df.columns]\n",
|
|
"\n",
|
|
" # 2.成员字段取值\n",
|
|
" user_columns = [\"报备业绩归属人\", \"报备业绩归属区域经理\", \"原业绩归属人\", \"原业绩归属区域经理\", \"运营专家\"]\n",
|
|
"\n",
|
|
" for col in user_columns:\n",
|
|
" df[col] = df[col].map(lambda x: x.get(\"name\", \"\") if isinstance(x, dict) else \"\")\n",
|
|
"\n",
|
|
" # 3.日期字段转为北京时间\n",
|
|
" time_columns = [\"支付日期\", \"开户/处理日期\"]\n",
|
|
"\n",
|
|
" df[time_columns] = df[time_columns].apply(\n",
|
|
" lambda col: pd.to_datetime(col, errors='coerce')\n",
|
|
" .dt.tz_localize(None)\n",
|
|
" .dt.strftime('%Y-%m-%d %H:%M:%S')\n",
|
|
" )\n",
|
|
"\n",
|
|
" # 4.处理所有配置的列表字段\n",
|
|
" if \"新签阶段及提成比例\" in df.columns:\n",
|
|
" # 先处理订单登记表字段\n",
|
|
" df[\"新签阶段及提成比例\"] = df[\"新签阶段及提成比例\"].apply(\n",
|
|
" lambda x: self.process_list_field(x, self.list_fields_config[\"新签阶段及提成比例\"])\n",
|
|
" if x is not None and (isinstance(x, (list, dict, np.ndarray)) or not pd.isna(x))\n",
|
|
" else None\n",
|
|
" )\n",
|
|
"\n",
|
|
" # 拆分行\n",
|
|
" df_exploded = df.explode(\"新签阶段及提成比例\")\n",
|
|
"\n",
|
|
" # 将订单登记表中的字段提取到主表中\n",
|
|
" order_fields = self.list_fields_config[\"新签阶段及提成比例\"].values()\n",
|
|
" for field in order_fields:\n",
|
|
" df_exploded[field] = df_exploded[\"新签阶段及提成比例\"].apply(\n",
|
|
" lambda x: x.get(field) if isinstance(x, dict) else None\n",
|
|
" )\n",
|
|
"\n",
|
|
" # 删除原始的订单登记表列\n",
|
|
" df_exploded = df_exploded.drop(columns=[\"新签阶段及提成比例\"])\n",
|
|
"\n",
|
|
" # 重置索引\n",
|
|
" df = df_exploded.reset_index(drop=True)\n",
|
|
"\n",
|
|
" return df\n",
|
|
"\n",
|
|
" def write_to_bi(self, df):\n",
|
|
" # 数据库连接信息\n",
|
|
" HS_DB_Config = {\n",
|
|
" 'host': \"f6-public.rwlb.rds.aliyuncs.com\",\n",
|
|
" 'user': \"rw_operation_data_relay\",\n",
|
|
" 'password': \"m+q5Z4%IVuF9bf\",\n",
|
|
" 'database': \"f6operation_data_relay\"\n",
|
|
" }\n",
|
|
" table_name = \"non_standard_performance_to_BI\" # 替换为你的实际表名\n",
|
|
"\n",
|
|
" # 建立数据库连接\n",
|
|
" connection = mysql.connector.connect(\n",
|
|
" host=HS_DB_Config[\"host\"],\n",
|
|
" user=HS_DB_Config[\"user\"],\n",
|
|
" password=HS_DB_Config[\"password\"],\n",
|
|
" database=HS_DB_Config[\"database\"]\n",
|
|
" )\n",
|
|
" cursor = connection.cursor()\n",
|
|
"\n",
|
|
" try:\n",
|
|
" # 查询表列名\n",
|
|
" cursor.execute(f\"SHOW COLUMNS FROM {table_name}\")\n",
|
|
" columns_info = cursor.fetchall()\n",
|
|
" db_columns = [col[0] for col in columns_info] # 提取列名\n",
|
|
" df = df.replace([None, np.nan, pd.NA, 'nan', 'NaN', 'NAN', ''], None)\n",
|
|
" # 保留 DataFrame 中与数据库列名匹配的列\n",
|
|
" filtered_df = df[df.columns.intersection(db_columns)]\n",
|
|
"\n",
|
|
" # 如果没有匹配的列，直接返回\n",
|
|
" if filtered_df.empty:\n",
|
|
" print(\"DataFrame 中没有与数据库表结构匹配的列。\")\n",
|
|
" return\n",
|
|
"\n",
|
|
" # 筛选列之后，插入前处理 dict 类型\n",
|
|
" filtered_df = filtered_df.copy()\n",
|
|
" for col in filtered_df.columns:\n",
|
|
" if filtered_df[col].apply(lambda x: isinstance(x, (dict, list)) if x is not None else False).any():\n",
|
|
" filtered_df.loc[:, col] = filtered_df[col].apply(\n",
|
|
" lambda x: json.dumps(x, ensure_ascii=False) if x is not None else x\n",
|
|
" )\n",
|
|
"\n",
|
|
" # 构建插入语句\n",
|
|
" placeholders = ', '.join(['%s'] * len(filtered_df.columns))\n",
|
|
" # 使用反引号避免特殊列明\n",
|
|
" columns = ', '.join([f\"`{col}`\" for col in filtered_df.columns])\n",
|
|
" insert_sql = f\"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})\"\n",
|
|
"\n",
|
|
" # 将 DataFrame 写入数据库\n",
|
|
" for _, row in filtered_df.iterrows():\n",
|
|
" cursor.execute(insert_sql, tuple(row))\n",
|
|
"\n",
|
|
" connection.commit()\n",
|
|
" print(f\"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。\")\n",
|
|
"\n",
|
|
" except Exception as e:\n",
|
|
" print(\"写入数据库时发生错误：\", e)\n",
|
|
" connection.rollback()\n",
|
|
" finally:\n",
|
|
" cursor.close()\n",
|
|
" connection.close()\n",
|
|
"\n",
|
|
" def clear_table_data(self):\n",
|
|
" \"\"\"\n",
|
|
" 清空指定 MySQL 表的数据。\n",
|
|
" 参数已写死在函数内部，直接调用即可。\n",
|
|
" \"\"\"\n",
|
|
" # 数据库连接信息\n",
|
|
" HS_DB_Config = {\n",
|
|
" 'host': \"f6-public.rwlb.rds.aliyuncs.com\",\n",
|
|
" 'user': \"rw_operation_data_relay\",\n",
|
|
" 'password': \"m+q5Z4%IVuF9bf\",\n",
|
|
" 'database': \"f6operation_data_relay\"\n",
|
|
" }\n",
|
|
" table_name = \"non_standard_performance_to_BI\" # 要清空的表名\n",
|
|
"\n",
|
|
" connection = None\n",
|
|
" try:\n",
|
|
" # 建立数据库连接\n",
|
|
" connection = mysql.connector.connect(\n",
|
|
" host=HS_DB_Config[\"host\"],\n",
|
|
" user=HS_DB_Config[\"user\"],\n",
|
|
" password=HS_DB_Config[\"password\"],\n",
|
|
" database=HS_DB_Config[\"database\"]\n",
|
|
" )\n",
|
|
" if connection.is_connected():\n",
|
|
" cursor = connection.cursor()\n",
|
|
"\n",
|
|
" # 使用TRUNCATE清空表数据\n",
|
|
" cursor.execute(f\"TRUNCATE TABLE {table_name}\")\n",
|
|
" connection.commit()\n",
|
|
"\n",
|
|
" print(f\"成功清空表 {table_name} 中的所有数据\")\n",
|
|
"\n",
|
|
" except Error as e:\n",
|
|
" print(f\"清空表时发生错误: {e}\")\n",
|
|
" if connection and connection.is_connected():\n",
|
|
" connection.rollback()\n",
|
|
" finally:\n",
|
|
" if connection and connection.is_connected():\n",
|
|
" cursor.close()\n",
|
|
" connection.close()\n",
|
|
" print(\"数据库连接已关闭\")\n",
|
|
"\n",
|
|
" def main(self):\n",
|
|
" task_start_time = datetime.datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\")\n",
|
|
"\n",
|
|
" # step1: 获取数据\n",
|
|
" self.load_all_data()\n",
|
|
"\n",
|
|
" # step2:数据处理\n",
|
|
" df = self.data_process()\n",
|
|
" # df.to_csv(os.path.join(output_dir, \"new_dealer_service_order_to_bi.csv\"))\n",
|
|
"\n",
|
|
" # step3:数据库删除\n",
|
|
" self.clear_table_data()\n",
|
|
"\n",
|
|
" # step4:数据写入BI\n",
|
|
" self.write_to_bi(df)\n",
|
|
"\n",
|
|
" common_module.send_task_status(task_start_time, \"非标业绩提报转BI\")\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == '__main__':\n",
|
|
" start = NonStandardPerformanceToBI()\n",
|
|
" start.main()"
|
|
],
|
|
"outputs": [
|
|
{
|
|
"name": "stderr",
|
|
"output_type": "stream",
|
|
"text": [
|
|
"\u001B[92m2025-08-20 16:58:28,211 - api.py - task_logger - INFO - 已获取 8 条数据\u001B[0m\n"
|
|
]
|
|
},
|
|
{
|
|
"name": "stdout",
|
|
"output_type": "stream",
|
|
"text": [
|
|
"成功清空表 non_standard_performance_to_BI 中的所有数据\n",
|
|
"数据库连接已关闭\n",
|
|
"成功写入 8 条记录到 non_standard_performance_to_BI 表中。\n"
|
|
]
|
|
},
|
|
{
|
|
"name": "stderr",
|
|
"output_type": "stream",
|
|
"text": [
|
|
"\u001B[92m2025-08-20 16:58:29,045 - common_module.py - task_logger - INFO - 任务状态发送成功: {'data': {'creator': {'name': 'F6汽车科技', 'username': '#admin', 'status': 1, 'type': 0}, 'updater': {'name': 'F6汽车科技', 'username': '#admin', 'status': 1, 'type': 0}, 'deleter': None, 'createTime': '2025-08-20T08:58:26.821Z', 'updateTime': '2025-08-20T08:58:26.821Z', 'deleteTime': None, '_widget_1744873387500': '2025-08-20T00:00:00.000Z', '_widget_1743644977694': '非标业绩提报转BI', '_widget_1744873387501': '2025-08-20T08:58:28.000Z', '_widget_1744873387502': '2025-08-20T08:58:28.000Z', '_widget_1744873387504': '0', '_id': '68a58e326435007d9a859fa2', 'appId': '6694d3c4fcb69ca9a111a6c4', 'entryId': '67ede908eb9c22261016466e'}}\u001B[0m\n"
|
|
]
|
|
}
|
|
],
|
|
"execution_count": 2
|
|
},
|
|
{
|
|
"metadata": {
|
|
"ExecuteTime": {
|
|
"end_time": "2025-08-14T06:21:29.349444Z",
|
|
"start_time": "2025-08-14T06:21:28.784674Z"
|
|
}
|
|
},
|
|
"cell_type": "code",
|
|
"source": [
|
|
"# -*- coding: utf-8 -*-\n",
|
|
"import pandas as pd\n",
|
|
"import datetime\n",
|
|
"from config import Config\n",
|
|
"from api import API\n",
|
|
"import pymysql # 使用 pymysql 替代 mysql.connector\n",
|
|
"from back_ground_module import CommonModule\n",
|
|
"import os\n",
|
|
"import mysql.connector\n",
|
|
"import pandas as pd\n",
|
|
"import json\n",
|
|
"import numpy as np\n",
|
|
"import mysql.connector\n",
|
|
"from mysql.connector import Error\n",
|
|
"\n",
|
|
"start_time = datetime.datetime.now()\n",
|
|
"api_instance = API()\n",
|
|
"common_module = CommonModule()\n",
|
|
"\n",
|
|
"# 保存为CSV文件\n",
|
|
"output_dir = \"output\" # 设置输出目录\n",
|
|
"\n",
|
|
"# 创建输出目录（如果不存在）\n",
|
|
"os.makedirs(output_dir, exist_ok=True)\n",
|
|
"\n",
|
|
"\n",
|
|
"class NewDealerServiceOrderToBI:\n",
|
|
" def __init__(self):\n",
|
|
" self.dealer_service_data = None\n",
|
|
" self.field_mapping = {\n",
|
|
" \"报备类型\": \"_widget_1753770875899\",\n",
|
|
" \"协作内容\": \"_widget_1753770875915\",\n",
|
|
" \"订单类型\": \"_widget_1753770875966\",\n",
|
|
" \"情况说明\": \"_widget_1753770875944\",\n",
|
|
" \"订单编号\": \"_widget_1753770875887\",\n",
|
|
" \"实付金额\": \"_widget_1753770875889\",\n",
|
|
" \"门店编码\": \"_widget_1753770875890\",\n",
|
|
" \"门店名称\": \"_widget_1753770875888\",\n",
|
|
" \"版本\": \"_widget_1753770875891\",\n",
|
|
" \"年限\": \"_widget_1753948745953\",\n",
|
|
" \"支付日期\": \"_widget_1753770875893\",\n",
|
|
" \"开户/处理日期\": \"_widget_1753770875894\",\n",
|
|
" \"小六业绩金额\": \"_widget_1753770875898\",\n",
|
|
" \"区域业绩金额\": \"_widget_1753770875937\",\n",
|
|
" \"报备业绩归属人\": \"_widget_1753770875901\",\n",
|
|
" \"报备业绩归属区域经理\": \"_widget_1753770875903\",\n",
|
|
" \"报备业绩归属大区\": \"_widget_1753866196486\",\n",
|
|
" \"原业绩归属人\": \"_widget_1753856032683\",\n",
|
|
" \"原业绩归属区域经理\": \"_widget_1753866196485\",\n",
|
|
" \"小六业绩比例\": \"_widget_1753770875917\",\n",
|
|
" \"区域业绩比例\": \"_widget_1753770875921\",\n",
|
|
" \"运营专家\": \"_widget_1753770875902\",\n",
|
|
" \"提成类型\": \"_widget_1753778922504\",\n",
|
|
" \"SaaS新签提成比例\": \"_widget_1753770875949\",\n",
|
|
" \"服务包提成比例\": \"_widget_1753778922567\",\n",
|
|
" \"提成金额\": \"_widget_1753770875948\",\n",
|
|
" \"新签提成比例-首年\": \"_widget_1753778922503\",\n",
|
|
" \"新签提成比例-非首年\": \"_widget_1753778922548\",\n",
|
|
" \"新签阶段及提成比例\": \"_widget_1753778656359\",\n",
|
|
" \"新签阶段及提成比例.选择提成阶段\": \"_widget_1753778656359._widget_1753778656361\",\n",
|
|
" \"新签阶段及提成比例.新签阶段\": \"_widget_1753778656359._widget_1753948745962\",\n",
|
|
" \"新签阶段及提成比例.提成比例\": \"_widget_1753778656359._widget_1753778656362\",\n",
|
|
" }\n",
|
|
"\n",
|
|
" # 定义需要特殊处理的列表字段及其内部字段映射\n",
|
|
" self.list_fields_config = {\n",
|
|
" \"新签阶段及提成比例\": {\n",
|
|
" \"_widget_1753778656361\": \"选择提成阶段\",\n",
|
|
" \"_widget_1753948745962\": \"新签阶段\",\n",
|
|
" \"_widget_1753778656362\": \"提成比例\"\n",
|
|
" },\n",
|
|
" # 可以在这里添加其他列表字段的配置\n",
|
|
" # \"另一个列表字段\": {\n",
|
|
" # \"原始字段名1\": \"映射后字段名1\",\n",
|
|
" # \"原始字段名2\": \"映射后字段名2\"\n",
|
|
" # }\n",
|
|
" }\n",
|
|
"\n",
|
|
" def load_all_data(self):\n",
|
|
" # 获取非标业绩提报数据\n",
|
|
" payload = {\"api_key\": \"66b9678280b37f8a276b1d01\",\n",
|
|
" \"entry_id\": \"68886b7c0382a7249ae0b5d6\",\n",
|
|
" }\n",
|
|
" dealer_service = api_instance.entry_data_list(payload)\n",
|
|
" self.dealer_service_data = dealer_service.get(\"data\") # api请求格式，将数据封装在data字典里\n",
|
|
"\n",
|
|
" def process_list_field(self, field_value, field_config):\n",
|
|
" \"\"\"通用方法：处理列表类型的字段\"\"\"\n",
|
|
" if not isinstance(field_value, list):\n",
|
|
" return field_value\n",
|
|
"\n",
|
|
" processed_list = []\n",
|
|
" for item in field_value:\n",
|
|
" if not isinstance(item, dict):\n",
|
|
" processed_list.append(item)\n",
|
|
" continue\n",
|
|
"\n",
|
|
" processed_item = {}\n",
|
|
" for original_key, mapped_key in field_config.items():\n",
|
|
" if original_key in item:\n",
|
|
" # 处理包含id的字典字段\n",
|
|
" if isinstance(item[original_key], dict) and \"id\" in item[original_key]:\n",
|
|
" processed_item[mapped_key] = item[original_key][\"id\"]\n",
|
|
" else:\n",
|
|
" processed_item[mapped_key] = item[original_key]\n",
|
|
" else:\n",
|
|
" processed_item[mapped_key] = None\n",
|
|
" processed_list.append(processed_item)\n",
|
|
" return processed_list\n",
|
|
"\n",
|
|
" def data_process(self):\n",
|
|
" df = pd.DataFrame(self.dealer_service_data)\n",
|
|
" # 反转映射字典\n",
|
|
" reverse_mapping = {v: k for k, v in self.field_mapping.items()}\n",
|
|
" # 1.列明替换\n",
|
|
" df.columns = [reverse_mapping.get(col, col) for col in df.columns]\n",
|
|
"\n",
|
|
" # 2.成员字段取值\n",
|
|
" user_columns = [\"报备业绩归属人\", \"报备业绩归属区域经理\", \"原业绩归属人\", \"原业绩归属区域经理\", \"运营专家\"]\n",
|
|
"\n",
|
|
" for col in user_columns:\n",
|
|
" df[col] = df[col].map(lambda x: x.get(\"name\", \"\") if isinstance(x, dict) else \"\")\n",
|
|
"\n",
|
|
" # 3.日期字段转为北京时间\n",
|
|
" time_columns = [\"支付日期\", \"开户/处理日期\"]\n",
|
|
"\n",
|
|
" df[time_columns] = df[time_columns].apply(\n",
|
|
" lambda col: pd.to_datetime(col, errors='coerce')\n",
|
|
" .dt.tz_localize(None)\n",
|
|
" .dt.strftime('%Y-%m-%d %H:%M:%S')\n",
|
|
" )\n",
|
|
" df.to_csv(\"feibiao.csv\", index=False)\n",
|
|
"\n",
|
|
" # 4.处理所有配置的列表字段\n",
|
|
" for field_name, field_config in self.list_fields_config.items():\n",
|
|
" if field_name in df.columns:\n",
|
|
" df[field_name] = df[field_name].apply(\n",
|
|
" lambda x: (\n",
|
|
" self.process_list_field(x, field_config)\n",
|
|
" if (isinstance(x, np.ndarray) and x.size > 0) # 非空 NumPy 数组\n",
|
|
" or (isinstance(x, list) and len(x) > 0) # 非空列表\n",
|
|
" or (not isinstance(x, (np.ndarray, list)) and x is not None and not pd.isna(x)) # 其他非空值\n",
|
|
" else None # 空数组、空列表、None、NaN 都返回 None\n",
|
|
" )\n",
|
|
" )\n",
|
|
"\n",
|
|
" return df\n",
|
|
"\n",
|
|
" def write_to_bi(self, df):\n",
|
|
" # 数据库连接信息\n",
|
|
" HS_DB_Config = {\n",
|
|
" 'host': \"f6-public.rwlb.rds.aliyuncs.com\",\n",
|
|
" 'user': \"rw_operation_data_relay\",\n",
|
|
" 'password': \"m+q5Z4%IVuF9bf\",\n",
|
|
" 'database': \"f6operation_data_relay\"\n",
|
|
" }\n",
|
|
" table_name = \"new_dealer_service_order_to_bi\" # 替换为你的实际表名\n",
|
|
"\n",
|
|
" # 建立数据库连接\n",
|
|
" connection = mysql.connector.connect(\n",
|
|
" host=HS_DB_Config[\"host\"],\n",
|
|
" user=HS_DB_Config[\"user\"],\n",
|
|
" password=HS_DB_Config[\"password\"],\n",
|
|
" database=HS_DB_Config[\"database\"]\n",
|
|
" )\n",
|
|
" cursor = connection.cursor()\n",
|
|
"\n",
|
|
" try:\n",
|
|
" # 查询表列名\n",
|
|
" cursor.execute(f\"SHOW COLUMNS FROM {table_name}\")\n",
|
|
" columns_info = cursor.fetchall()\n",
|
|
" db_columns = [col[0] for col in columns_info] # 提取列名\n",
|
|
" df = df.replace([None, np.nan, pd.NA, 'nan', 'NaN', 'NAN', ''], None)\n",
|
|
" # 保留 DataFrame 中与数据库列名匹配的列\n",
|
|
" filtered_df = df[df.columns.intersection(db_columns)]\n",
|
|
"\n",
|
|
" # 如果没有匹配的列，直接返回\n",
|
|
" if filtered_df.empty:\n",
|
|
" print(\"DataFrame 中没有与数据库表结构匹配的列。\")\n",
|
|
" return\n",
|
|
"\n",
|
|
" # 筛选列之后，插入前处理 dict 类型\n",
|
|
" filtered_df = filtered_df.copy()\n",
|
|
" for col in filtered_df.columns:\n",
|
|
" if filtered_df[col].apply(lambda x: isinstance(x, (dict, list)) if x is not None else False).any():\n",
|
|
" filtered_df.loc[:, col] = filtered_df[col].apply(\n",
|
|
" lambda x: json.dumps(x, ensure_ascii=False) if x is not None else x\n",
|
|
" )\n",
|
|
"\n",
|
|
" # 构建插入语句\n",
|
|
" placeholders = ', '.join(['%s'] * len(filtered_df.columns))\n",
|
|
" # 使用反引号避免特殊列明\n",
|
|
" columns = ', '.join([f\"`{col}`\" for col in filtered_df.columns])\n",
|
|
" insert_sql = f\"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})\"\n",
|
|
"\n",
|
|
" # 将 DataFrame 写入数据库\n",
|
|
" for _, row in filtered_df.iterrows():\n",
|
|
" cursor.execute(insert_sql, tuple(row))\n",
|
|
"\n",
|
|
" connection.commit()\n",
|
|
" print(f\"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。\")\n",
|
|
"\n",
|
|
" except Exception as e:\n",
|
|
" print(\"写入数据库时发生错误：\", e)\n",
|
|
" connection.rollback()\n",
|
|
" finally:\n",
|
|
" cursor.close()\n",
|
|
" connection.close()\n",
|
|
"\n",
|
|
" def clear_table_data(self):\n",
|
|
" \"\"\"\n",
|
|
" 清空指定 MySQL 表的数据。\n",
|
|
" 参数已写死在函数内部，直接调用即可。\n",
|
|
" \"\"\"\n",
|
|
" # 数据库连接信息\n",
|
|
" HS_DB_Config = {\n",
|
|
" 'host': \"f6-public.rwlb.rds.aliyuncs.com\",\n",
|
|
" 'user': \"rw_operation_data_relay\",\n",
|
|
" 'password': \"m+q5Z4%IVuF9bf\",\n",
|
|
" 'database': \"f6operation_data_relay\"\n",
|
|
" }\n",
|
|
" table_name = \"new_dealer_service_order_to_bi\" # 要清空的表名\n",
|
|
"\n",
|
|
" connection = None\n",
|
|
" try:\n",
|
|
" # 建立数据库连接\n",
|
|
" connection = mysql.connector.connect(\n",
|
|
" host=HS_DB_Config[\"host\"],\n",
|
|
" user=HS_DB_Config[\"user\"],\n",
|
|
" password=HS_DB_Config[\"password\"],\n",
|
|
" database=HS_DB_Config[\"database\"]\n",
|
|
" )\n",
|
|
" if connection.is_connected():\n",
|
|
" cursor = connection.cursor()\n",
|
|
"\n",
|
|
" # 使用TRUNCATE清空表数据\n",
|
|
" cursor.execute(f\"TRUNCATE TABLE {table_name}\")\n",
|
|
" connection.commit()\n",
|
|
"\n",
|
|
" print(f\"成功清空表 {table_name} 中的所有数据\")\n",
|
|
"\n",
|
|
" except Error as e:\n",
|
|
" print(f\"清空表时发生错误: {e}\")\n",
|
|
" if connection and connection.is_connected():\n",
|
|
" connection.rollback()\n",
|
|
" finally:\n",
|
|
" if connection and connection.is_connected():\n",
|
|
" cursor.close()\n",
|
|
" connection.close()\n",
|
|
" print(\"数据库连接已关闭\")\n",
|
|
"\n",
|
|
" def main(self):\n",
|
|
" task_start_time = datetime.datetime.now().strftime(\"%Y-%m-%d %H:%M:%S\")\n",
|
|
"\n",
|
|
" # step1: 获取数据\n",
|
|
" self.load_all_data()\n",
|
|
"\n",
|
|
" # step2:数据处理\n",
|
|
" df = self.data_process()\n",
|
|
" print(df)\n",
|
|
" df.to_csv(os.path.join(output_dir, \"new_dealer_service_order_to_bi.csv\"))\n",
|
|
"\n",
|
|
" # step3:数据库删除\n",
|
|
" # self.clear_table_data()\n",
|
|
"\n",
|
|
" # step4:数据写入BI\n",
|
|
" # self.write_to_bi(df)\n",
|
|
"\n",
|
|
" # common_module.send_task_status(task_start_time, \"非标业绩提报转BI\")\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == '__main__':\n",
|
|
" start = NewDealerServiceOrderToBI()\n",
|
|
" start.main()"
|
|
],
|
|
"id": "c63f1a4b217a1576",
|
|
"outputs": [
|
|
{
|
|
"name": "stderr",
|
|
"output_type": "stream",
|
|
"text": [
|
|
"\u001B[92m2025-08-14 14:21:29,218 - api.py - task_logger - INFO - 已获取 2 条数据\u001B[0m\n"
|
|
]
|
|
},
|
|
{
|
|
"name": "stdout",
|
|
"output_type": "stream",
|
|
"text": [
|
|
" creator \\\n",
|
|
"0 {'name': 'F6汽车科技', 'username': '#admin', 'stat... \n",
|
|
"1 {'name': 'F6汽车科技', 'username': '#admin', 'stat... \n",
|
|
"\n",
|
|
" updater deleter \\\n",
|
|
"0 {'name': '张阳', 'username': '4210192048793363',... None \n",
|
|
"1 {'name': '张阳', 'username': '4210192048793363',... None \n",
|
|
"\n",
|
|
" createTime updateTime deleteTime flowState \\\n",
|
|
"0 2025-08-07T02:24:56.648Z 2025-08-07T02:25:35.002Z None 1 \n",
|
|
"1 2025-08-07T02:27:43.775Z 2025-08-07T02:28:39.026Z None 1 \n",
|
|
"\n",
|
|
" 报备类型 协作内容 订单类型 ... 提成类型 SaaS新签提成比例 服务包提成比例 提成金额 新签提成比例-首年 \\\n",
|
|
"0 新签超3年 SaaS新签 ... 0.00 None NaN NaN \n",
|
|
"1 大数报表漏统计 服务包 ... 新签 0.25 None 555.0 0.25 \n",
|
|
"\n",
|
|
" 新签提成比例-非首年 新签阶段及提成比例 \\\n",
|
|
"0 0 None \n",
|
|
"1 0 [{'选择提成阶段': '688889a36f36b252847d811e', '新签阶段'... \n",
|
|
"\n",
|
|
" _id appId \\\n",
|
|
"0 68940e78593f4b6b032ff3dd 66b9678280b37f8a276b1d01 \n",
|
|
"1 68940f1f6226ebc6c7b57f65 66b9678280b37f8a276b1d01 \n",
|
|
"\n",
|
|
" entryId \n",
|
|
"0 68886b7c0382a7249ae0b5d6 \n",
|
|
"1 68886b7c0382a7249ae0b5d6 \n",
|
|
"\n",
|
|
"[2 rows x 39 columns]\n"
|
|
]
|
|
}
|
|
],
|
|
"execution_count": 11
|
|
}
|
|
],
|
|
"metadata": {
|
|
"kernelspec": {
|
|
"display_name": "saas",
|
|
"language": "python",
|
|
"name": "saas"
|
|
},
|
|
"language_info": {
|
|
"codemirror_mode": {
|
|
"name": "ipython",
|
|
"version": 2
|
|
},
|
|
"file_extension": ".py",
|
|
"mimetype": "text/x-python",
|
|
"name": "python",
|
|
"nbconvert_exporter": "python",
|
|
"pygments_lexer": "ipython2",
|
|
"version": "2.7.6"
|
|
}
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 5
|
|
}
|