saas1.6日志更新

This commit is contained in:
z66
2025-08-14 11:55:03 +08:00
parent d5e60e9014
commit 3bffc6946b
34 changed files with 2999 additions and 2907 deletions
+102 -108
View File
@@ -1,11 +1,8 @@
import datetime
import os
import time
import requests
from api import API
import re
from back_ground_module import CommonModule
import pandas as pd
from log_config import configure_task_logger, configure_error_task_logger
@@ -179,152 +176,149 @@ class NewExceptionTask:
def main(self):
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
global png_url, key, upload_key, province_name, city_name, area_name
self.load_all_data()
global png_url, key, upload_key
self.load_all_data()
self.data_yichang_S = common_module.get_yichang_details(days_back=1).astype(str) # 获取data_NGV 并转为str
self.index = self.build_index(self.json_list)
self.data_yichang_S = common_module.get_yichang_details(days_back=1).astype(str) # 获取data_NGV 并转为str
self.index = self.build_index(self.json_list)
logger.info("开始运行SaaS异常回访")
print("开始运行main")
print(self.date_one)
data_yichang = self.data_yichang_S.copy()
data_yichang = self.data_yichang_S.copy()
def replace_values(series):
# 使用条件判断来进行替换
return series.apply(lambda x: '' if pd.isna(x) or x in ['NA', 'None', ''] else x)
def replace_values(series):
# 使用条件判断来进行替换
return series.apply(lambda x: '' if pd.isna(x) or x in ['NA', 'None', ''] else x)
# 对整个DataFrame的所有列应用替换函数
data_yichang = data_yichang.apply(replace_values)
# 对整个DataFrame的所有列应用替换函数
data_yichang = data_yichang.apply(replace_values)
for index_num, row in data_yichang.iterrows(): # 对过滤后的每一条进行派发
try:
payload_dict = {}
print(data_yichang.iterrows())
for index_num, row in data_yichang.iterrows(): # 对过滤后的每一条进行派发
try:
print("1111")
payload_dict = {}
distribution_date = datetime.datetime.now(datetime.timezone.utc)
distribution_date = distribution_date.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
distribution_date = datetime.datetime.now(datetime.timezone.utc)
distribution_date = distribution_date.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
date_obj1 = datetime.datetime.strptime(row["init_day"], "%Y%m%d").strftime("%Y-%m-%d")
date_obj2 = datetime.datetime.strptime(row["push_day"], "%Y%m%d").strftime("%Y-%m-%d")
date_obj1 = datetime.datetime.strptime(row["init_day"], "%Y%m%d").strftime("%Y-%m-%d")
date_obj2 = datetime.datetime.strptime(row["push_day"], "%Y%m%d").strftime("%Y-%m-%d")
NGV_roles = {
'service_impl_principal': row['service_impl_principal'], # 运营负责人
'area_manager': row['area_manager'], # 区域经理
'technician': row['technician'], # 运营专家
}
for role, name in NGV_roles.items(): # 寻找对应的员工ID
for row_item in self.staff_id_list:
staff_id = self.get_staff_id(row_item, name)
if staff_id:
NGV_roles[role] = staff_id
break # 找到后退出循环
else:
NGV_roles[role] = None # 如果没有找到对应的员工ID
relationship_manager, area_manager, technician = [NGV_roles[role] for role in
['service_impl_principal',
'area_manager',
'technician']]
NGV_roles = {
'service_impl_principal': row['service_impl_principal'], # 运营负责人
'area_manager': row['area_manager'], # 区域经理
'technician': row['technician'], # 运营专家
}
print('NGV_roles', NGV_roles)
for role, name in NGV_roles.items(): # 寻找对应的员工ID
for row_item in self.staff_id_list:
staff_id = self.get_staff_id(row_item, name)
if staff_id:
NGV_roles[role] = staff_id
break # 找到后退出循环
else:
NGV_roles[role] = None # 如果没有找到对应的员工ID
print('NGV_roles[role]', NGV_roles)
relationship_manager, area_manager, technician = [NGV_roles[role] for role in
['service_impl_principal',
'area_manager',
'technician']]
UUid = time.strftime("%Y%m%d%H%M%S", time.localtime())
UUid = time.strftime("%Y%m%d%H%M%S", time.localtime())
NGV_data_id = None
# 获取关联数据
for NGV_Data in self.NGV_data_list:
# NGV_Data = NGV_Data.get("data")
if row["org_code"] == NGV_Data.get("_widget_1734062123071"): # 门店编码
NGV_data_id = NGV_Data.get("_id")
province_name = NGV_Data.get("_widget_1734062123090")
city_name = NGV_Data.get("_widget_1734062123092")
area_name = NGV_Data.get("_widget_1734062123094")
logger.info(f"获取关联数据成功:{NGV_data_id}, {province_name}, {city_name}, {area_name}")
NGV_data_id = None
# 获取关联数据
for NGV_Data in self.NGV_data_list:
# NGV_Data = NGV_Data.get("data")
if row["org_code"] == NGV_Data.get("_widget_1734062123071"): # 门店编码
NGV_data_id = NGV_Data.get("_id")
province_name = NGV_Data.get("_widget_1734062123090")
city_name = NGV_Data.get("_widget_1734062123092")
area_name = NGV_Data.get("_widget_1734062123094")
print('111111', NGV_data_id, province_name, city_name, area_name)
if not NGV_data_id:
logger.warning(f"未找到关联数据,请检查门店编码: {row['org_code']}")
if not NGV_data_id:
print("未找到数据ID")
# 根据省市区派发给异常回访客服
customer_service = self.assign_customer_service(province_name, city_name, area_name, self.index)
# 根据省市区派发给异常回访客服
print('333', province_name, city_name, area_name)
customer_service = self.assign_customer_service(province_name, city_name, area_name, self.index)
payload_dict.update({
"_widget_1748241895829": {"value": row["health_warning_info"]}, # 活跃健康状态变化
payload_dict.update({
"_widget_1748241895829": {"value": row["health_warning_info"]}, # 活跃健康状态变化
"_widget_1748241895830": {"value": row["org_name"]}, # 门店名称
"_widget_1748241895830": {"value": row["org_name"]}, # 门店名称
"_widget_1748241895831": {"value": row["contacts"]}, # 联系人
"_widget_1748241895831": {"value": row["contacts"]}, # 联系
"_widget_1748241895832": {"value": row['contact_mobile']}, # 联系方式
"_widget_1748241895832": {"value": row['contact_mobile']}, # 联系方式
"_widget_1748241895833": {
"value": int(time.mktime(time.strptime(date_obj1, "%Y-%m-%d")) * 1000) if row[
"init_day"] != '' else ''},
# 初始日
"_widget_1748241895833": {"value": int(time.mktime(time.strptime(date_obj1, "%Y-%m-%d")) * 1000) if row[
"init_day"] != '' else ''},
# 初始日
"_widget_1748241895834": {
"value": int(time.mktime(time.strptime(date_obj2, "%Y-%m-%d")) * 1000) if row[
"push_day"] != '' else ''},
# 推进日
"_widget_1748241895834": {"value": int(time.mktime(time.strptime(date_obj2, "%Y-%m-%d")) * 1000) if row[
"push_day"] != '' else ''},
# 推进日
"_widget_1748246808678": {"value": customer_service}, # 当前跟进人
"_widget_1748246808678": {"value": customer_service}, # 当前跟进
"_widget_1748246808679": {"value": relationship_manager}, # 运营负责
"_widget_1748246808679": {"value": relationship_manager}, # 运营负责人
"_widget_1748246808680": {"value": customer_service}, # 区域客服
"_widget_1748246808680": {"value": customer_service}, # 区域客服
"_widget_1748241895839": {
"value": int(time.mktime(time.strptime(row["saas_create_time"], "%Y-%m-%d")) * 1000) if row[
"saas_create_time"] != '' else ''},
# 开户时间
"_widget_1748241895839": {
"value": int(time.mktime(time.strptime(row["saas_create_time"], "%Y-%m-%d")) * 1000) if row[
"saas_create_time"] != '' else ''},
# 开户时间
"_widget_1748246808681": {"value": technician}, # 技术专家
"_widget_1748246808681": {"value": technician}, # 技术专家
"_widget_1748246808682": {"value": area_manager}, # 区域经理
"_widget_1748246808682": {"value": area_manager}, # 区域经理
"_widget_1748241895842": {"value": row['org_code']}, # 门店编码
"_widget_1748241895842": {"value": row['org_code']}, # 门店编码
"_widget_1748241895844": {"value": row['group_name']}, # 公司名称
"_widget_1748241895844": {"value": row['group_name']}, # 公司名称
"_widget_1748241895846": {"value": row['group_grade']}, # 公司等级
"_widget_1748241895846": {"value": row['group_grade']}, # 公司等级
"_widget_1748241895847": {"value": row['region_name']}, # 大区
"_widget_1748241895847": {"value": row['region_name']}, # 大区
"_widget_1748241895848": {"value": row['province_name']}, #
"_widget_1748241895848": {"value": row['province_name']}, #
"_widget_1748241895849": {"value": row['org_type']}, # 门店类型
"_widget_1748241895849": {"value": row['org_type']}, # 门店类型
"_widget_1748241895850": {"value": row['saas_edition_fmt']}, # 系统版本
"_widget_1748241895850": {"value": row['saas_edition_fmt']}, # 系统版本
"_widget_1748241895851": {"value": row['saas_customer_type']}, # saas客户类型
"_widget_1748241895851": {"value": row['saas_customer_type']}, # saas客户类型
"_widget_1748241895852": {"value": row['org_stage']}, # 门店阶段
"_widget_1748241895852": {"value": row['org_stage']}, # 门店阶段
"_widget_1748241895853": {"value": row['contact_mobile']}, # 操作模式E.L/E.S
"_widget_1748241895853": {"value": row['contact_mobile']}, # 操作模式E.L/E.S
"_widget_1748241895855": {"value": row['city_name']}, # 城市
"_widget_1748241895855": {"value": row['city_name']}, # 城市
"_widget_1748247754304": {"value": NGV_data_id}, # 数据id
"_widget_1748247754304": {"value": NGV_data_id}, # 数据id
"_widget_1748512176655": {"value": "未处理"}, # 跟进状态
"_widget_1748512176655": {"value": "未处理"}, # 跟进状态
})
})
routine_follow_up_payload = {
"api_key": "675b900991ad2491c69389ca",
"entry_id": "68340de79f116c0b66b6b0cc", # 异常服务跟进待办
"is_start_workflow": "true",
"data": payload_dict,
"transaction_id": UUid
}
routine_follow_up_payload = {
"api_key": "675b900991ad2491c69389ca",
"entry_id": "68340de79f116c0b66b6b0cc", # 异常服务跟进待办
"is_start_workflow": "true",
"data": payload_dict,
"transaction_id": UUid
}
# print(routine_follow_up_payload)
res = api_instance.data_batch_create(routine_follow_up_payload)
logger.info(f"创建结果:{res}")
except:
pass
common_module.send_task_status(task_start_time, "异常服务待办派发")
res = api_instance.data_batch_create(routine_follow_up_payload)
logger.info(f"创建结果:{res}")
except:
pass
common_module.send_task_status(task_start_time, "异常服务待办派发")
except Exception as e:
error_task_logger.error(f"异常服务待办派发执行时发生异常: {e}")
common_module.send_task_error(task_start_time, "异常服务待办派发", str(e))
if __name__ == '__main__':
+165 -160
View File
@@ -1,18 +1,29 @@
from datetime import date, timedelta, datetime
import holidays
from datetime import timedelta, datetime
from config import Config
import pandas as pd
import pymysql # 使用 pymysql 替代 mysql.connector
from log_config import configure_task_logger, configure_error_task_logger
from api import API
from back_ground_module import CommonModule
from log_config import configure_task_logger, configure_error_task_logger
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
# 获取已经配置好的错误任务日志记录器
error_task_logger = configure_error_task_logger()
common_module = CommonModule()
api_instance = API()
# 保存为CSV文件
output_dir = "output" # 设置输出目录
# 创建输出目录(如果不存在)
import os
os.makedirs(output_dir, exist_ok=True)
global last_day_end_customer_service, is_customer_service_data_id, customer_service_data_id
class JCBAbnormalRevisit:
"""接车宝异常回访"""
def __init__(self):
# 使用 pymysql 连接数据库
self.daily_revisit_list = None
@@ -47,6 +58,7 @@ class JCBAbnormalRevisit:
def today_customer_service_list(self):
# 获取今日接车宝派发客服顺序
global is_customer_service_data_id
today_customer_service_list = []
all_customer_service_list = []
today_customer_service_start_list = []
@@ -112,7 +124,7 @@ class JCBAbnormalRevisit:
df.iterrows()]
data = {'api_key': Config.EFFICIENT_CAR_PICKUP_APP_ID, 'entry_id': "67174710da507490d8ac12c1",
"data_list": new_sign_abnormal_data} # 派发数据
"data_list": new_sign_abnormal_data} # 派发数据
api_instance.entry_data_batch_create(data)
@@ -133,197 +145,190 @@ class JCBAbnormalRevisit:
"entry_id": Config.EFFICIENT_CAR_PICKUP_CUSTOMER_SERVICE_ID,
"data_id": next_customer_service_data_id,
"data":
{"_widget_1740042824216": {"value": ""}, }}# 明日派发起点人员
{"_widget_1740042824216": {"value": ""}, }} # 明日派发起点人员
api_instance.entry_data_update(data1)
api_instance.entry_data_update(data2)
def main(self):
self.load_all_data()
task_start_time =datetime.now().strftime("%Y-%m-%d %H:%M:%S")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
logger.info(f"开始执行")
self.load_all_data()
data_JCB = common_module.get_jcb_details()
logger.info(f"数据加载完成")
# data_JCB.to_csv(os.path.join(output_dir, 'JCB_all_data.csv'), index=False)
self.fields()
data_JCB = common_module.get_jcb_details()
# 异常待办回访 近1个月开单为0客户
# 当前日期
current_date = datetime.now()
current_date = current_date + timedelta(days=1)
current_date_str = current_date.strftime("%Y-%m-%d")
# current_date = datetime.now()
thirty_days_ago = current_date - timedelta(days=30)
thirty_days_ago = thirty_days_ago.date()
abnormal_data = []
JDY_abnormal_data = []
JDY_revisit_data = []
# df = pd.read_csv(os.path.join(output_dir, "JCB_异常待办.csv")) # 读取异常待办表
# print(df)
for index, row in data_JCB.iterrows():
new_row = row.copy()
new_row['开户日'] = datetime.strptime(new_row['开户日'], "%Y-%m-%d").date()
if new_row['开户日'] < thirty_days_ago and row['近30天开单天数'] == 0 and row['客户状态'] == "留存":
# print(row['账号'], row['开户日'], row['近30天开单天数'], row["客户状态"])
row["日期"] = datetime.strptime(row['开户日'], "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
abnormal_data.append(row)
# 推送给客服
abnormal_data = pd.DataFrame(abnormal_data)
abnormal_data["表单类型"] = "异常待办"
abnormal_data["派发日期"] = current_date_str
# abnormal_data.to_excel(os.path.join(output_dir, 'JCB_异常待办.xlsx'), index=False) # 派发B(所有异常待办)
# 保存为CSV文件
output_dir = "output" # 设置输出目录
for abnormal_items in self.abnormal_list:
last_send_date = abnormal_items.get("_widget_1740723898405", {}) # 派发日期
last_30_days_orders = abnormal_items.get("_widget_1740723898401", {}) # 近30天开单数
phone = abnormal_items.get("_widget_1740723898391", {}) # 手机号
account = abnormal_items.get("_widget_1740723898390", {}) # 账号
data_id = abnormal_items.get("_id", {}) # 数据id
JDY_abnormal_data.append([data_id, account, phone, last_send_date, last_30_days_orders])
# 创建输出目录(如果不存在)
import os
os.makedirs(output_dir, exist_ok=True)
JDY_abnormal_data = pd.DataFrame(JDY_abnormal_data,
columns=["数据id", "账号", "联系手机号", "派发日期",
"近30天开单天数"]) # 派发A(简道云上异常待办)
# JDY_abnormal_data.columns = ["数据id", "账号", "联系手机号", "派发日期", "近30天开单天数"]
# JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_云端异常待办.xlsx'), index=False) # 派发A
# data_JCB.to_csv(os.path.join(output_dir, 'JCB_all_data.csv'), index=False)
self.fields()
# 将 '联系手机号' 列转换为字符串类型
JDY_abnormal_data['联系手机号'] = JDY_abnormal_data['联系手机号'].astype(str).str.replace('.0', '')
abnormal_data['联系手机号'] = abnormal_data['联系手机号'].astype(str)
# JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_云端异常待办.xlsx'), index=False) # 派发A
# abnormal_data.to_excel(os.path.join(output_dir, 'JCB_今日异常待办.xlsx'), index=False) # 派发B
# 异常待办回访 近1个月开单为0客户
# 当前日期
current_date = datetime.now()
current_date = current_date + timedelta(days=1)
current_date_str = current_date.strftime("%Y-%m-%d")
# current_date = datetime.now()
thirty_days_ago = current_date - timedelta(days=30)
thirty_days_ago = thirty_days_ago.date()
abnormal_data = []
JDY_abnormal_data = []
JDY_revisit_data = []
# df = pd.read_csv(os.path.join(output_dir, "JCB_异常待办.csv")) # 读取异常待办表
# print(df)
for index, row in data_JCB.iterrows():
new_row = row.copy()
new_row['开户日'] = datetime.strptime(new_row['开户日'], "%Y-%m-%d").date()
if new_row['开户日'] < thirty_days_ago and row['近30天开单天数'] == 0 and row['客户状态'] == "留存":
# print(row['账号'], row['开户日'], row['近30天开单天数'], row["客户状态"])
row["日期"] = datetime.strptime(row['开户日'], "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
abnormal_data.append(row)
# 推送给客服
abnormal_data = pd.DataFrame(abnormal_data)
abnormal_data["表单类型"] = "异常待办"
abnormal_data["派发日期"] = current_date_str
# abnormal_data.to_excel(os.path.join(output_dir, 'JCB_异常待办.xlsx'), index=False) # 派发B(所有异常待办)
today = datetime.now().weekday()
for abnormal_items in self.abnormal_list:
last_send_date = abnormal_items.get("_widget_1740723898405", {}) # 派发日期
last_30_days_orders = abnormal_items.get("_widget_1740723898401", {}) # 近30天开单数
phone = abnormal_items.get("_widget_1740723898391", {}) # 手机号
account = abnormal_items.get("_widget_1740723898390", {}) # 账号
data_id = abnormal_items.get("_id", {}) # 数据id
JDY_abnormal_data.append([data_id, account, phone, last_send_date, last_30_days_orders])
# 随机抽40条派发
df_40 = pd.DataFrame()
if 0 <= today <= 4:
# if 1>2:
# 假设 JDY_abnormal_data 和 abnormal_data 都有重复列 '重复列'
df3 = pd.merge(JDY_abnormal_data, abnormal_data, on=["联系手机号", "账号"], how='inner',
suffixes=('', '_y'))
# 删除以 _y 结尾的列(即来自右侧 DataFrame 的重复列)
df3 = df3.loc[:, ~df3.columns.str.endswith('_y')]
df3['派发日期'] = pd.to_datetime(df3['派发日期']).dt.strftime("%Y-%m-%d")
# df3.to_excel(os.path.join(output_dir, 'JCB_异常待办情况1.xlsx'),
# index=False, ) # B存在,A存在 ,今日派发与历史派发都存在,派发并删历史
JDY_abnormal_data = pd.DataFrame(JDY_abnormal_data,
columns=["数据id", "账号", "联系手机号", "派发日期",
"近30天开单天数"]) # 派发A(简道云上异常待办)
# JDY_abnormal_data.columns = ["数据id", "账号", "联系手机号", "派发日期", "近30天开单天数"]
# JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_云端异常待办.xlsx'), index=False) # 派发A
df_40 = df3[df3.index < 40]
# df_40.to_excel(os.path.join(output_dir, 'JCB_异常待办情况2.xlsx'), index=False, )
# 将 '联系手机号' 列转换为字符串类型
JDY_abnormal_data['联系手机号'] = JDY_abnormal_data['联系手机号'].astype(str).str.replace('.0', '')
abnormal_data['联系手机号'] = abnormal_data['联系手机号'].astype(str)
# JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_云端异常待办.xlsx'), index=False) # 派发A
# abnormal_data.to_excel(os.path.join(output_dir, 'JCB_今日异常待办.xlsx'), index=False) # 派发B
for index, row in df_40.iterrows(): # 删除已推送的数据
delete_data = {"api_key": Config.EFFICIENT_CAR_PICKUP_APP_ID,
"entry_id": Config.EFFICIENT_CAR_PICKUP_CUSTOMER_HISTORY_ID,
"data_id": row["数据id"]}
# print(delete_data)
api_instance.entry_data_delete(delete_data)
logger.info(f"已删除随机抽取40条数据")
today = datetime.now().weekday()
# 随机抽40条派发
df_40 = pd.DataFrame()
if 0 <= today <= 4:
# if 1>2:
# 假设 JDY_abnormal_data 和 abnormal_data 都有重复列 '重复列'
df3 = pd.merge(JDY_abnormal_data, abnormal_data, on=["联系手机号", "账号"], how='inner',
suffixes=('', '_y'))
# B不存在A存在 今日派发不存在,历史存在,删历史
# 使用 outer 合并,并添加指示器列 _merge
df_merged = pd.merge(JDY_abnormal_data, abnormal_data, on=["联系手机号", "账号"], how='outer',
indicator=True,
suffixes=('', '_y')) # outer保留所有数据,indicator标注来源
# 筛选出只存在于 JDY_abnormal_data 中的行
df_a_not_in_b = df_merged[df_merged['_merge'] == 'left_only']
# 删除以 _y 结尾的列(即来自右侧 DataFrame 的重复列)
df3 = df3.loc[:, ~df3.columns.str.endswith('_y')]
df3['派发日期'] = pd.to_datetime(df3['派发日期']).dt.strftime("%Y-%m-%d")
# df3.to_excel(os.path.join(output_dir, 'JCB_异常待办情况1.xlsx'),
# index=False, ) # B存在,A存在 ,今日派发与历史派发都存在,派发并删历史
df_40 = df3[df3.index < 40]
# df_40.to_excel(os.path.join(output_dir, 'JCB_异常待办情况2.xlsx'), index=False, )
for index, row in df_40.iterrows(): # 删除已推送的数据
df_a_not_in_b = df_a_not_in_b.loc[:, ~df_a_not_in_b.columns.str.endswith('_y')]
df_a_not_in_b['派发日期'] = pd.to_datetime(df_a_not_in_b['派发日期']).dt.strftime("%Y-%m-%d")
# 保存到 Excel 文件
# df_a_not_in_b.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_A存在B不存在.xlsx'), index=False)
for index, row in df_a_not_in_b.iterrows(): # 删除已推送的数据
delete_data = {"api_key": Config.EFFICIENT_CAR_PICKUP_APP_ID,
"entry_id": Config.EFFICIENT_CAR_PICKUP_CUSTOMER_HISTORY_ID,
"data_id": row["数据id"]}
# print(delete_data)
api_instance.entry_data_delete(delete_data)
logger.info("已删除派发后数据")
# B存在A存在 今日派发存在,历史存在,删历史
# 使用 outer 合并,并添加指示器列 _merge
df_merged = pd.merge(JDY_abnormal_data, abnormal_data, on=["联系手机号", "账号"], how='outer', indicator=True,
suffixes=('', '_y')) # outer保留所有数据,indicator标注来源
# 筛选出只存在于 JDY_abnormal_data 中的行
df_a_not_in_b = df_merged[df_merged['_merge'] == 'left_only']
# 删除以 _y 结尾的列(即来自右侧 DataFrame 的重复列)
df_a_not_in_b = df_a_not_in_b.loc[:, ~df_a_not_in_b.columns.str.endswith('_y')]
df_a_not_in_b['派发日期'] = pd.to_datetime(df_a_not_in_b['派发日期']).dt.strftime("%Y-%m-%d")
# 保存到 Excel 文件
# df_a_not_in_b.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_A存在B不存在.xlsx'), index=False)
for index, row in df_a_not_in_b.iterrows(): # 删除已推送的数据
delete_data = {"api_key": Config.EFFICIENT_CAR_PICKUP_APP_ID,
"entry_id": Config.EFFICIENT_CAR_PICKUP_CUSTOMER_HISTORY_ID,
"data_id": row["数据id"]}
# print(delete_data)
api_instance.entry_data_delete(delete_data)
# B存在A存在 今日派发存在,历史存在,为新增异常,直接派发
df_merged = pd.merge(JDY_abnormal_data, abnormal_data, on=["联系手机号", "账号"], how='outer',
indicator=True,
suffixes=('_x', '')) # outer保留所有数据,indicator标注来源
# df_merged.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在_134434.xlsx'), index=False)
# 筛选出只存在于 JDY_abnormal_data 中的行
df_b_not_in_a = df_merged[df_merged['_merge'] == 'right_only']
# df_b_not_in_a.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在_111.xlsx'), index=False)
# 删除以 _y 结尾的列(即来自右侧 DataFrame 的重复列)
df_b_not_in_a = df_b_not_in_a.loc[:, ~df_b_not_in_a.columns.str.endswith('_x')]
# df_b_not_in_a.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在_122.xlsx'), index=False)
df_b_not_in_a['派发日期'] = pd.to_datetime(df_b_not_in_a['派发日期']).dt.strftime("%Y-%m-%d")
# 保存到 Excel 文件
# df_b_not_in_a.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在.xlsx'), index=False)
# B存在A不存在 今日派发存在,历史不存在,为新增异常,直接派发
df_merged = pd.merge(JDY_abnormal_data, abnormal_data, on=["联系手机号", "账号"], how='outer', indicator=True,
suffixes=('_x', '')) # outer保留所有数据,indicator标注来源
# df_merged.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在_134434.xlsx'), index=False)
# 筛选出只存在于 JDY_abnormal_data 中的行
df_b_not_in_a = df_merged[df_merged['_merge'] == 'right_only']
# df_b_not_in_a.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在_111.xlsx'), index=False)
# 删除以 _y 结尾的列(即来自右侧 DataFrame 的重复列)
df_b_not_in_a = df_b_not_in_a.loc[:, ~df_b_not_in_a.columns.str.endswith('_x')]
# df_b_not_in_a.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在_122.xlsx'), index=False)
df_b_not_in_a['派发日期'] = pd.to_datetime(df_b_not_in_a['派发日期']).dt.strftime("%Y-%m-%d")
# 保存到 Excel 文件
# df_b_not_in_a.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_B存在A不存在.xlsx'), index=False)
# 合并两个当日派发的df
df_abnormal_data = pd.concat([df_40, df_b_not_in_a], ignore_index=True)
# df_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_合并当日派发.xlsx'), index=False)
# 合并两个当日派发的df
df_abnormal_data = pd.concat([df_40, df_b_not_in_a], ignore_index=True)
# df_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_合并当日派发.xlsx'), index=False)
for abnormal_items in self.daily_revisit_list: # 遍历云端已经派发的数据
account = abnormal_items.get("_widget_1739258942667", {}) # 账号
sub_date = abnormal_items.get("createTime", {}) # 提交时间
update_date = abnormal_items.get("updateTime", {}) # 更新时间
entry_style = abnormal_items.get("_widget_1739951204545", {}) # 表单类型
entry_type = abnormal_items.get("flowState", {}) # 表单状态 0流转中 1流转完成 2 手动结束
for abnormal_items in self.daily_revisit_list: # 遍历云端已经派发的数据
account = abnormal_items.get("_widget_1739258942667", {}) # 账号
sub_date = abnormal_items.get("createTime", {}) # 提交时间
update_date = abnormal_items.get("updateTime", {}) # 更新时间
entry_style = abnormal_items.get("_widget_1739951204545", {}) # 表单类型
entry_type = abnormal_items.get("flowState", {}) # 表单状态 0流转中 1流转完成 2 手动结束
data_id = abnormal_items.get("_id", {}) # 数据id
JDY_revisit_data.append([data_id, account, sub_date, update_date, entry_style, entry_type])
data_id = abnormal_items.get("_id", {}) # 数据id
JDY_revisit_data.append([data_id, account, sub_date, update_date, entry_style, entry_type])
JDY_revisit_data = pd.DataFrame(JDY_revisit_data)
JDY_revisit_data.columns = ["数据id", "账号", "提交时间", "更新时间", "表单类型", "表单状态"]
# JDY_revisit_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_原始数据.xlsx'), index=False)
JDY_revisit_data = pd.DataFrame(JDY_revisit_data)
JDY_revisit_data.columns = ["数据id", "账号", "提交时间", "更新时间", "表单类型", "表单状态"]
# JDY_revisit_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_原始数据.xlsx'), index=False)
filtered_data = JDY_revisit_data[JDY_revisit_data['表单类型'] == '异常待办'] # 过滤表单类型
# filtered_data = filtered_data[filtered_data['表单状态'] == 1] # 过滤表单状态
# filtered_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_过滤数据.xlsx'), index=False)
filtered_data = JDY_revisit_data[JDY_revisit_data['表单类型'] == '异常待办'] # 过滤表单类型
# filtered_data = filtered_data[filtered_data['表单状态'] == 1] # 过滤表单状态
# filtered_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_过滤数据.xlsx'), index=False)
filtered_data['提交时间'] = pd.to_datetime(filtered_data['提交时间']).dt.strftime("%Y-%m-%d")
latest_update_time = filtered_data.groupby('账号')['提交时间'].max().reset_index()
latest_update_time.rename(columns={'提交时间': '最新提交时间'}, inplace=True)
filtered_data['提交时间'] = pd.to_datetime(filtered_data['提交时间']).dt.strftime("%Y-%m-%d")
latest_update_time = filtered_data.groupby('账号')['提交时间'].max().reset_index()
latest_update_time.rename(columns={'提交时间': '最新提交时间'}, inplace=True)
filtered_data_with_latest = pd.merge(
filtered_data,
latest_update_time,
left_on=['账号', '提交时间'],
right_on=['账号', '最新提交时间']
)
# 过滤出每个账号中提交时间为最新的记录
latest_JDY_abnormal_data = filtered_data_with_latest[
filtered_data_with_latest['提交时间'] == filtered_data_with_latest['最新提交时间']
]
# latest_JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_最新数据_1.xlsx'), index=False)
filtered_data_with_latest = pd.merge(
filtered_data,
latest_update_time,
left_on=['账号', '提交时间'],
right_on=['账号', '最新提交时间']
)
latest_JDY_abnormal_data['提交时间'] = pd.to_datetime(latest_JDY_abnormal_data['提交时间']).dt.strftime(
"%Y-%m-%d")
# 过滤出每个账号中提交时间为最新的记录
latest_JDY_abnormal_data = filtered_data_with_latest[
filtered_data_with_latest['提交时间'] == filtered_data_with_latest['最新提交时间']
]
# latest_JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_最新数据_1.xlsx'), index=False)
thirty_days_ago = (current_date - timedelta(days=30)).strftime("%Y-%m-%d")
final_JDY_abnormal_data = latest_JDY_abnormal_data[
latest_JDY_abnormal_data['提交时间'] > thirty_days_ago] # 筛选出提交时间为近30天的数据
latest_JDY_abnormal_data['提交时间'] = pd.to_datetime(latest_JDY_abnormal_data['提交时间']).dt.strftime("%Y-%m-%d")
# final_JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_最新数据.xlsx'), index=False)
thirty_days_ago = (current_date - timedelta(days=30)).strftime("%Y-%m-%d")
df_abnormal_data = df_abnormal_data[~df_abnormal_data['账号'].isin(final_JDY_abnormal_data['账号'])]
# empty_num = df_abnormal_data['手机号'].isnull().sum()
df_abnormal_data = df_abnormal_data[df_abnormal_data["联系手机号"] != "None"]
# df_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_派发数据.xlsx'), index=False)
final_JDY_abnormal_data = latest_JDY_abnormal_data[latest_JDY_abnormal_data['提交时间'] > thirty_days_ago] # 筛选出提交时间为近30天的数据
self.send_request(df_abnormal_data)
common_module.send_task_status(task_start_time, "接车宝异常派发")
logger.info("接车宝异常派发完成")
# final_JDY_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_日常回访_最新数据.xlsx'), index=False)
df_abnormal_data = df_abnormal_data[~df_abnormal_data['账号'].isin(final_JDY_abnormal_data['账号'])]
# empty_num = df_abnormal_data['手机号'].isnull().sum()
df_abnormal_data = df_abnormal_data[df_abnormal_data["联系手机号"] != "None"]
# df_abnormal_data.to_excel(os.path.join(output_dir, 'JCB_异常待办情况_派发数据.xlsx'), index=False)
self.send_request(df_abnormal_data)
common_module.send_task_status(task_start_time, "接车宝异常派发")
# df_abnormal_data = [self.row_to_dict(row, self.field_mapping) for index, row in
# df_abnormal_data.iterrows()]
#
# data = {'api_key': Config.EFFICIENT_CAR_PICKUP_APP_ID, 'entry_id':"67174710da507490d8ac12c1",
# "data_list": df_abnormal_data}
#
#
# result = api_instance.entry_data_batch_create(data)
except Exception as e:
common_module.send_task_error(task_start_time, "接车宝异常派发", str(e))
error_task_logger.error(f"接车宝异常派发执行异常: {e}")
@staticmethod
def row_to_dict(row, field_mapping):
+106 -92
View File
@@ -1,18 +1,27 @@
from datetime import date, timedelta, datetime
import holidays
from datetime import timedelta, datetime
from config import Config
import pandas as pd
import pymysql # 使用 pymysql 替代 mysql.connector
from back_ground_module import CommonModule
from log_config import configure_task_logger, configure_error_task_logger
from api import API
from log_config import configure_task_logger, configure_error_task_logger
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
# 获取已经配置好的错误任务日志记录器
error_task_logger = configure_error_task_logger()
# 保存为CSV文件
output_dir = "output" # 设置输出目录
# 创建输出目录(如果不存在)
import os
os.makedirs(output_dir, exist_ok=True)
common_module = CommonModule()
api_instance = API()
class JCBEfficientCarPickup:
"""接车宝日常回访"""
def __init__(self):
# 使用 pymysql 连接数据库
self.field_mapping = {}
@@ -30,6 +39,7 @@ class JCBEfficientCarPickup:
def today_customer_service_list(self):
# 获取今日接车宝派发客服顺序
global is_customer_service_data_id
today_customer_service_list = []
all_customer_service_list = []
today_customer_service_start_list = []
@@ -98,6 +108,7 @@ class JCBEfficientCarPickup:
"data_list": new_sign_abnormal_data}
result = api_instance.entry_data_batch_create(data)
logger.info(f"数据发送成功:{result}")
data1 = {"api_key": Config.EFFICIENT_CAR_PICKUP_APP_ID,
"entry_id": Config.EFFICIENT_CAR_PICKUP_CUSTOMER_SERVICE_ID,
@@ -119,108 +130,111 @@ class JCBEfficientCarPickup:
{"_widget_1740042824216": {"value": ""}, }}
api_instance.entry_data_update(data1)
api_instance.entry_data_update(data2)
result2 = api_instance.entry_data_update(data2)
logger.info(f"明日派发人员信息已修改:{result2}")
def main(self):
task_start_time =datetime.now().strftime("%Y-%m-%d %H:%M:%S")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
logger.info(f"接车宝日常回访开始执行")
data_JCB = common_module.get_jcb_details()
logger.info(f"数据加载完成")
data_JCB = common_module.get_jcb_details()
print(data_JCB)
# data_JCB.to_csv(os.path.join(output_dir, 'JCB_all_data.csv'), index=False)
self.fields()
# 保存为CSV文件
output_dir = "output" # 设置输出目录
# 新签异常待办回访。
# 当前日期
current_date = datetime.now()
current_date = current_date + timedelta(days=0)
current_date_str = current_date.strftime("%Y-%m-%d")
# 创建输出目录(如果不存在)
import os
os.makedirs(output_dir, exist_ok=True)
seven_days_ago = current_date - timedelta(days=7)
seven_days_ago = seven_days_ago.date()
# print(three_days_ago)
new_sign_abnormal = []
# data_JCB.to_csv(os.path.join(output_dir, 'JCB_all_data.csv'), index=False)
self.fields()
for index, row in data_JCB.iterrows():
new_row = row.copy()
# 先转成字符串,再解析回 date 对象
new_row['开户日'] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
if new_row['开户日'] == seven_days_ago and row['当月开单天数'] == 0:
# print(row['账号'], row['开户日'], row['当月开单天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
new_sign_abnormal.append(row)
# 新签异常待办回访。
# 当前日期
current_date = datetime.now()
current_date = current_date + timedelta(days=0)
current_date_str = current_date.strftime("%Y-%m-%d")
new_sign_abnormal = pd.DataFrame(new_sign_abnormal)
new_sign_abnormal["表单类型"] = "新签异常待办"
new_sign_abnormal["派发日期"] = current_date_str
seven_days_ago = current_date - timedelta(days=7)
seven_days_ago = seven_days_ago.date()
# print(three_days_ago)
new_sign_abnormal = []
self.send_request(new_sign_abnormal) # 发送请求
logger.info(f"新签异常待办回访完成")
for index, row in data_JCB.iterrows():
new_row = row.copy()
# 先转成字符串,再解析回 date 对象
new_row['开户日'] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
if new_row['开户日'] == seven_days_ago and row['当月开单天数'] == 0:
# print(row['账号'], row['开户日'], row['当月开单天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
new_sign_abnormal.append(row)
# 优质客户转商机
# current_date = datetime.now()
thirty_days_ago = current_date - timedelta(days=30)
sixty_days_ago = current_date - timedelta(days=60)
thirty_days_ago = thirty_days_ago.date()
sixty_days_ago = sixty_days_ago.date()
customer_to_opportunity = []
for index, row in data_JCB.iterrows():
new_row = row.copy()
# 先转成字符串,再解析回 date 对象
new_row['到期日'] = datetime.strptime(str(row['到期日']), "%Y-%m-%d").date()
if new_row['到期日'] == thirty_days_ago and row['近一周开单量'] >= 3 and row[
'G状态:近30天开单大于等于10天'] == 1:
print(row['账号'], row['到期日'], row['当月开单天数'], row['当月G天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
customer_to_opportunity.append(row)
# 推送给客服
pass
if new_row['到期日'] == sixty_days_ago and row['近一周开单量'] >= 3 and row[
'G状态:近30天开单大于等于10天'] == 1:
print(row['账号'], row['到期日'], row['当月开单天数'], row['当月G天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
customer_to_opportunity.append(row)
# 推送给客服
pass
new_sign_abnormal = pd.DataFrame(new_sign_abnormal)
new_sign_abnormal["表单类型"] = "新签异常待办"
new_sign_abnormal["派发日期"] = current_date_str
customer_to_opportunity = pd.DataFrame(customer_to_opportunity)
customer_to_opportunity["表单类型"] = "续约优质客户转商机"
customer_to_opportunity["派发日期"] = current_date_str
self.send_request(new_sign_abnormal) # 发送请求
self.send_request(customer_to_opportunity)
logger.info(f"优质客户转商机完成")
# 优质客户转商机
# current_date = datetime.now()
thirty_days_ago = current_date - timedelta(days=30)
sixty_days_ago = current_date - timedelta(days=60)
thirty_days_ago = thirty_days_ago.date()
sixty_days_ago = sixty_days_ago.date()
customer_to_opportunity = []
for index, row in data_JCB.iterrows():
new_row = row.copy()
# 先转成字符串,再解析回 date 对象
new_row['到期日'] = datetime.strptime(str(row['到期日']), "%Y-%m-%d").date()
if new_row['到期日'] == thirty_days_ago and row['近一周开单量'] >= 3 and row[
'G状态:近30天开单大于等于10天'] == 1:
print(row['账号'], row['到期'], row['当月开单天数'], row['当月G天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
customer_to_opportunity.append(row)
# 推送给客服
pass
if new_row['到期日'] == sixty_days_ago and row['近一周开单量'] >= 3 and row[
'G状态:近30天开单大于等于10天'] == 1:
print(row['账号'], row['到期日'], row['当月开单天数'], row['当月G天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
customer_to_opportunity.append(row)
# 推送给客服
pass
# 过期7天客服回访
# current_date = datetime.now()
seven_days_ago = current_date - timedelta(days=7)
seven_days_ago = seven_days_ago.date()
outdated_30 = []
for index, row in data_JCB.iterrows():
new_row = row.copy()
new_row['到期日'] = datetime.strptime(str(row['到期日']), "%Y-%m-%d").date()
# seven_days_ago = seven_days_ago.date()
# print(row['到期日'], seven_days_ago)
if new_row['到期日'] == seven_days_ago and row['客户状态'] == "过期":
print(row['账号'], row['到期日'], row['当月开单天数'], row['当月G天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row[''] = row["日期"].strftime("%Y-%m-%d")
outdated_30.append(row)
# 推送给客服
pass
customer_to_opportunity = pd.DataFrame(customer_to_opportunity)
customer_to_opportunity["表单类型"] = "续约优质客户转商机"
customer_to_opportunity["派发日期"] = current_date_str
outdated_30 = pd.DataFrame(outdated_30)
outdated_30["表单类型"] = "过期7天回访"
outdated_30["派发日期"] = current_date_str
self.send_request(outdated_30)
logger.info(f"过期7天客服回访完成")
self.send_request(customer_to_opportunity)
# 过期7天客服回访
# current_date = datetime.now()
seven_days_ago = current_date - timedelta(days=7)
seven_days_ago = seven_days_ago.date()
outdated_30 = []
for index, row in data_JCB.iterrows():
new_row = row.copy()
new_row['到期日'] = datetime.strptime(str(row['到期日']), "%Y-%m-%d").date()
# seven_days_ago = seven_days_ago.date()
# print(row['到期日'], seven_days_ago)
if new_row['到期日'] == seven_days_ago and row['客户状态'] == "过期":
print(row['账号'], row['到期日'], row['当月开单天数'], row['当月G天数'])
row["日期"] = datetime.strptime(str(row['开户日']), "%Y-%m-%d").date()
row['日期'] = row["日期"].strftime("%Y-%m-%d")
outdated_30.append(row)
# 推送给客服
pass
outdated_30 = pd.DataFrame(outdated_30)
outdated_30["表单类型"] = "过期7天回访"
outdated_30["派发日期"] = current_date_str
self.send_request(outdated_30)
common_module.send_task_status(task_start_time, "接车宝日常派发")
common_module.send_task_status(task_start_time, "接车宝日常派发")
logger.info(f"接车宝日常派发执行完成")
except Exception as e:
common_module.send_task_error(task_start_time, "接车宝日常派发", str(e))
error_task_logger.error(f"接车宝日常派发执行出错:{e}")
@staticmethod
def row_to_dict(row, field_mapping):
+52 -5
View File
@@ -249,7 +249,7 @@ class CommonModule:
]
data_NGV.columns = headers
cols_to_str =["日期","开户日","续约日","到期日"]
cols_to_str = ["日期", "开户日", "续约日", "到期日"]
data_NGV[cols_to_str] = data_NGV[cols_to_str].apply(lambda x: x.astype(str))
data_NGV.to_csv(os.path.join(output_dir, f"{target_date_id}.csv"), index=False)
@@ -576,7 +576,6 @@ class CommonModule:
:param task_start_time: 任务开始时间(字符串格式:"%Y-%m-%d %H:%M:%S",表示北京时间 UTC+8
:param task_name: 任务名称
"""
print(1)
try:
# 1. 获取当前 UTC 时间(时区感知对象)
end_time_utc = datetime.now(UTC) # ✅ 替代 utcnow()
@@ -596,8 +595,6 @@ class CommonModule:
today_utc = end_time_utc.strftime("%Y-%m-%d")
task_end_iso = end_time_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
task_start_iso = task_start_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
print(task_end_iso)
print(task_start_iso)
# 6. 构造请求数据(所有时间以 UTC 格式发送)
payload = {
@@ -617,4 +614,54 @@ class CommonModule:
logger.info(f"任务状态发送成功: {response}")
except Exception as e:
logger.error(f"任务状态发送失败: {e}")
error_task_logger.error(f"任务状态发送失败: {e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
CommonModule.send_task_error(task_start_time, "发送任务状态", e)
def send_task_error(self, task_start_time: str, task_name: str, error_message: str) -> None:
"""
将任务失败情况发送到简道云(影响业务数据时调用)
:param task_start_time: 任务开始时间(字符串格式:"%Y-%m-%d %H:%M:%S",表示北京时间 UTC+8
:param task_name: 任务名称
:param error_message: 失败详情
"""
try:
# 1. 获取当前 UTC 时间(时区感知对象)
end_time_utc = datetime.now(UTC) # ✅ 替代 utcnow()
# 2. 解析传入的北京时间(UTC+8)
task_start_naive = datetime.strptime(task_start_time, "%Y-%m-%d %H:%M:%S")
# 3. 转换为 UTC 时间(减去 8 小时,并附加 UTC 时区)
task_start_utc = task_start_naive - timedelta(hours=8)
task_start_utc = task_start_utc.replace(tzinfo=timezone.utc) # 显式标记为 UTC
# 4. 计算运行时间(时区感知对象可直接相减)
run_time = end_time_utc - task_start_utc
run_time_sec = int(run_time.total_seconds())
# 5. 格式化时间为 UTC 的 ISO 8601 格式(带 "Z"
today_utc = end_time_utc.strftime("%Y-%m-%d")
task_end_iso = end_time_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
task_start_iso = task_start_utc.strftime("%Y-%m-%dT%H:%M:%SZ")
# 6. 构造请求数据(所有时间以 UTC 格式发送)
payload = {
"api_key": Config.SCHEDULED_TASKS_APP_ID,
"entry_id": Config.JDY_TASKS_ERROR_ENTRY_ID,
"data": {
"_widget_1744873387500": {"value": today_utc}, # UTC 日期
"_widget_1743644977694": {"value": task_name},
"_widget_1744873387501": {"value": task_start_iso}, # UTC 开始时间
"_widget_1744873387502": {"value": task_end_iso}, # UTC 结束时间
"_widget_1744873387504": {"value": run_time_sec},
"_widget_1754981992215": {"value": error_message}, # 错误信息
}
}
# 7. 发送请求
response = api_instance.data_batch_create(payload)
logger.info(f"任务错误发生成功: {response}")
except Exception as e:
error_task_logger.error(f"任务错误发送失败: {e}")
+45 -40
View File
@@ -4,8 +4,15 @@ import datetime
from config import Config
from api import API
import pymysql # 使用 pymysql 替代 mysql.connector
from log_config import configure_task_logger, configure_error_task_logger
from back_ground_module import CommonModule
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
# 获取已经配置好的错误任务日志记录器
error_task_logger = configure_error_task_logger()
start_time = datetime.datetime.now()
api_instance = API()
common_module = CommonModule()
@@ -15,47 +22,43 @@ class importCommissionData:
"""
小六提成数据支撑
"""
def __init__(self):
self.field_mapping = {}
self.fields()
def main(self):
task_start_time =datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.Commission_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
for i in range(0,len(abnormal_list)): # 删除历史数据
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.Commission_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
delete_id_list = []
try:
delete_data = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.Commission_form_ID,
"data_id": abnormal_list[i]['_id']}
api_instance.entry_data_delete(delete_data)
except:
pass
for i in range(0, len(abnormal_list)): # 删除历史数据
delete_id_list.append(abnormal_list[i]['_id'])
delete_payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.Commission_form_ID,
"data_id": delete_id_list}
api_instance.entry_data_batch_delete(delete_payload)
data_commission = common_module.get_commission_details()
data_commission = common_module.get_commission_details()
except Exception as e:
error_task_logger.error(f"小六提成数据支撑任务执行出错:{str(e)}")
common_module.send_task_error(task_start_time, "小六提成数据支撑", str(e))
# 生成包含所有行转换后的字典列表
data_commission = [self.row_to_dict(row, self.field_mapping) for index, row in data_commission.iterrows()]
data_commission = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.Commission_form_ID, "data_list": data_commission}
result = api_instance.entry_data_batch_create(data_commission)
result_str = str(result)
# print(result_str[:500])
# 生成包含所有行转换后的字典列表
data_commission = [self.row_to_dict(row, self.field_mapping) for index, row in data_commission.iterrows()]
data_commission = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.Commission_form_ID,
"data_list": data_commission}
# 保存到Excel文件
# output_path = r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细1.xlsx'
# data_NGV_j.to_excel(output_path, index=False)
end_time = datetime.datetime.now()
time_diff = end_time - start_time
# 打印天数、秒数和微秒数
print(f"执行时间: {time_diff.days} 天, {time_diff.seconds} 秒, {time_diff.microseconds} 微秒")
common_module.send_task_status(task_start_time, "小六提成数据支撑")
api_instance.entry_data_batch_create(data_commission)
common_module.send_task_status(task_start_time, "小六提成数据支撑")
except Exception as e:
error_task_logger.error(f"小六提成数据支撑任务执行出错:{str(e)}")
common_module.send_task_error(task_start_time, "小六提成数据支撑", str(e))
@staticmethod
def row_to_dict(row, field_mapping):
@@ -70,14 +73,16 @@ class importCommissionData:
def fields(self):
self.field_mapping = {
"门店id": "_widget_1742884710674",
"提成类型_二级分类": "_widget_1742884710675",
"提成基数(本月)": "_widget_1742884710676",
"提成基数(上月)": "_widget_1742884710677",
"公司id": "_widget_1748930441629",
"门店编码": "_widget_1748931089356",
"门店名称": "_widget_1748931089357"
}
"门店id": "_widget_1742884710674",
"提成类型_二级分类": "_widget_1742884710675",
"提成基数(本月)": "_widget_1742884710676",
"提成基数(上月)": "_widget_1742884710677",
"公司id": "_widget_1748930441629",
"门店编码": "_widget_1748931089356",
"门店名称": "_widget_1748931089357"
}
if __name__ == '__main__':
start = importCommissionData()
start.main()
@@ -5,6 +5,11 @@ from config import Config
from api import API
import pymysql # 使用 pymysql 替代 mysql.connector
from back_ground_module import CommonModule
from log_config import configure_task_logger, configure_error_task_logger
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
error_task_logger = configure_error_task_logger()
start_time = datetime.datetime.now()
api_instance = API()
@@ -13,48 +18,45 @@ common_module = CommonModule()
class importDifferentIndustriesData:
"""异业合作数据支撑"""
def __init__(self):
self.field_mapping = {}
self.fields()
def main(self):
task_start_time =datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.DifferentIndustries_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
for i in range(0,len(abnormal_list)): # 删除历史数据
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.DifferentIndustries_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
delete_id_list = []
try:
delete_data = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.DifferentIndustries_form_ID,
"data_id": abnormal_list[i]['_id']}
api_instance.entry_data_delete(delete_data)
except:
pass
for i in range(0, len(abnormal_list)): # 删除历史数据
delete_id_list.append(abnormal_list[i]['_id'])
delete_payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.DifferentIndustries_form_ID,
"data_id": delete_id_list}
api_instance.entry_data_batch_delete(delete_payload)
except Exception as e:
error_task_logger.error(f"异业合作数据支撑任务执行时发生异常: {e}")
common_module.send_task_error(task_start_time, "异业合作数据支撑", str(e))
data_commission = common_module.get_differentindustries_details()
data_commission = common_module.get_differentindustries_details()
# 生成包含所有行转换后的字典列表
data_commission = [self.row_to_dict(row, self.field_mapping) for index, row in data_commission.iterrows()]
data_commission = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.DifferentIndustries_form_ID,
"data_list": data_commission}
# 生成包含所有行转换后的字典列表
data_commission = [self.row_to_dict(row, self.field_mapping) for index, row in data_commission.iterrows()]
data_commission = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.DifferentIndustries_form_ID, "data_list": data_commission}
result = api_instance.entry_data_batch_create(data_commission)
result_str = str(result)
# print(result_str[:500])
api_instance.entry_data_batch_create(data_commission)
# 保存到Excel文件
# output_path = r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细1.xlsx'
# data_NGV_j.to_excel(output_path, index=False)
end_time = datetime.datetime.now()
time_diff = end_time - start_time
# 打印天数、秒数和微秒数
print(f"执行时间: {time_diff.days} 天, {time_diff.seconds} 秒, {time_diff.microseconds} 微秒")
common_module.send_task_status(task_start_time, "异业合作数据支撑")
common_module.send_task_status(task_start_time, "异业合作数据支撑")
except Exception as e:
error_task_logger.error(f"异业合作数据支撑任务执行时发生异常: {e}")
common_module.send_task_error(task_start_time, "异业合作数据支撑", str(e))
@staticmethod
def row_to_dict(row, field_mapping):
@@ -69,13 +71,15 @@ class importDifferentIndustriesData:
def fields(self):
self.field_mapping = {
"门店id": "_widget_1742884829007",
"商品名称": "_widget_1742884829008",
"服务期结束时间": "_widget_1742884829009",
"门店名称": "_widget_1748931208851",
"公司id": "_widget_1748930826642",
"门店编码": "_widget_1748931208852"
}
"门店id": "_widget_1742884829007",
"商品名称": "_widget_1742884829008",
"服务期结束时间": "_widget_1742884829009",
"门店名称": "_widget_1748931208851",
"公司id": "_widget_1748930826642",
"门店编码": "_widget_1748931208852"
}
if __name__ == '__main__':
start = importDifferentIndustriesData()
start.main()
@@ -5,7 +5,11 @@ from config import Config
from api import API
import pymysql # 使用 pymysql 替代 mysql.connector
from back_ground_module import CommonModule
from log_config import configure_task_logger, configure_error_task_logger
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
error_task_logger = configure_error_task_logger()
start_time = datetime.datetime.now()
api_instance = API()
common_module = CommonModule()
@@ -13,59 +17,45 @@ common_module = CommonModule()
class importGroupNotificationData:
"""短信数据支撑"""
def __init__(self):
self.field_mapping = {}
self.fields()
def main(self):
task_start_time =datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.GroupNotification_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
# print(abnormal_list)
delete_id_list = []
for i in range(0,len(abnormal_list)): # 删除历史数据
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.GroupNotification_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
delete_id_list = []
try:
# delete_data = {"api_key": Config.SaaS_Tasks_APP_ID,
# "entry_id": Config.GroupNotification_form_ID,
# "data_id": abnormal_list[i]['_id']}
# api_instance.entry_data_delete(delete_data)
delete_id_list.append(abnormal_list[i]['_id'])
for i in range(0, len(abnormal_list)): # 删除历史数据
delete_id_list.append(abnormal_list[i]['_id'])
delete_payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.GroupNotification_form_ID,
"data_ids": delete_id_list}
api_instance.entry_data_batch_delete(delete_payload)
except Exception as e:
print(e)
error_task_logger.error(f"删除历史数据失败:{e}")
common_module.send_task_error(task_start_time, "短信数据支撑", str(e))
# print(delete_id_list)
delete_payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.GroupNotification_form_ID,
"data_ids": delete_id_list}
data_commission = common_module.get_GroupNotification_details()
api_instance.entry_data_batch_delete(delete_payload)
# 生成包含所有行转换后的字典列表
data_commission = [self.row_to_dict(row, self.field_mapping) for index, row in data_commission.iterrows()]
data_commission = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.GroupNotification_form_ID,
"data_list": data_commission}
data_commission = common_module.get_GroupNotification_details()
# 生成包含所有行转换后的字典列表
data_commission = [self.row_to_dict(row, self.field_mapping) for index, row in data_commission.iterrows()]
data_commission = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.GroupNotification_form_ID, "data_list": data_commission}
result = api_instance.entry_data_batch_create(data_commission)
# result_str = str(result)
# print(result_str[:500])
# 保存到Excel文件
# output_path = r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细1.xlsx'
# data_NGV_j.to_excel(output_path, index=False)
end_time = datetime.datetime.now()
time_diff = end_time - start_time
# 打印天数、秒数和微秒数
print(f"执行时间: {time_diff.days} 天, {time_diff.seconds} 秒, {time_diff.microseconds} 微秒")
common_module.send_task_status(task_start_time, "短信数据支撑")
api_instance.entry_data_batch_create(data_commission)
common_module.send_task_status(task_start_time, "短信数据支撑")
except Exception as e:
error_task_logger.error(f"短信数据支撑失败:{e}")
common_module.send_task_error(task_start_time, "短信数据支撑", str(e))
@staticmethod
def row_to_dict(row, field_mapping):
@@ -80,17 +70,18 @@ class importGroupNotificationData:
def fields(self):
self.field_mapping = {
"公司id": "_widget_1743065201885",
"是否启动短信功能": "_widget_1743065201886",
"是否购买短信包": "_widget_1743065201887",
"累计购买总数": "_widget_1743065201888",
"累计发送成功总人数": "_widget_1743065201889",
"剩余短信条数": "_widget_1743065201890",
"第一次短信购买时间": "_widget_1743065201891",
"最近一次短信购买时间": "_widget_1743065201892",
"实付总金额": "_widget_1743065201893",
"短信剩余量是否小于20%": "_widget_1743065201894"
}
"公司id": "_widget_1743065201885",
"是否启动短信功能": "_widget_1743065201886",
"是否购买短信包": "_widget_1743065201887",
"累计购买总数": "_widget_1743065201888",
"累计发送成功总人数": "_widget_1743065201889",
"剩余短信条数": "_widget_1743065201890",
"第一次短信购买时间": "_widget_1743065201891",
"最近一次短信购买时间": "_widget_1743065201892",
"实付总金额": "_widget_1743065201893",
"短信剩余量是否小于20%": "_widget_1743065201894"
}
if __name__ == '__main__':
start = importGroupNotificationData()
@@ -5,7 +5,11 @@ from config import Config
from api import API
import pymysql # 使用 pymysql 替代 mysql.connector
from back_ground_module import CommonModule
from log_config import configure_task_logger, configure_error_task_logger
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
error_task_logger = configure_error_task_logger()
start_time = datetime.datetime.now()
api_instance = API()
common_module = CommonModule()
@@ -19,43 +23,38 @@ class importSYXCXData:
def main(self):
task_start_time =datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.SYXCX_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
for i in range(0,len(abnormal_list)): # 删除历史数据
try:
payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.SYXCX_form_ID,
}
abnormal_service = api_instance.entry_data_list(payload)
abnormal_list = abnormal_service.get("data") # api请求格式,将数据封装在data字典里
delete_id_list = []
try:
delete_data = {"api_key": Config.SaaS_Tasks_APP_ID,
for i in range(0, len(abnormal_list)): # 删除历史数据
delete_id_list.append(abnormal_list[i]['_id'])
delete_payload = {"api_key": Config.SaaS_Tasks_APP_ID,
"entry_id": Config.SYXCX_form_ID,
"data_id": abnormal_list[i]['_id']}
api_instance.entry_data_delete(delete_data)
except:
pass
"data_id": delete_id_list}
api_instance.entry_data_batch_delete(delete_payload)
except Exception as e:
error_task_logger.error(f"删除私域小程序数据时出错: {e}")
common_module.send_task_error(task_start_time, "私域小程序数据支撑", str(e))
data_SY = common_module.get_syxcx_details()
data_SY = common_module.get_syxcx_details()
# 生成包含所有行转换后的字典列表
data_SY = [self.row_to_dict(row, self.field_mapping) for index, row in data_SY.iterrows()]
data_SY = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.SYXCX_form_ID, "data_list": data_SY}
result = api_instance.entry_data_batch_create(data_SY)
result_str = str(result)
# print(result_str[:500])
# 生成包含所有行转换后的字典列表
data_SY = [self.row_to_dict(row, self.field_mapping) for index, row in data_SY.iterrows()]
data_SY = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.SYXCX_form_ID, "data_list": data_SY}
# 保存到Excel文件
# output_path = r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细1.xlsx'
# data_NGV_j.to_excel(output_path, index=False)
end_time = datetime.datetime.now()
time_diff = end_time - start_time
# 打印天数、秒数和微秒数
print(f"执行时间: {time_diff.days} 天, {time_diff.seconds} 秒, {time_diff.microseconds} 微秒")
common_module.send_task_status(task_start_time, "私域小程序数据支撑")
api_instance.entry_data_batch_create(data_SY)
common_module.send_task_status(task_start_time, "私域小程序数据支撑")
except Exception as e:
error_task_logger.error(f"私域小程序数据支撑执行时出错: {e}")
common_module.send_task_error(task_start_time, "私域小程序数据支撑", str(e))
@staticmethod
def row_to_dict(row, field_mapping):
+14 -27
View File
@@ -34,24 +34,6 @@ class Config:
RETRY_DELAY = 0.5
# ---------------------- 日志配置 -----------------------
# class Logger:
# @staticmethod
# def setup():
# logging.basicConfig(
# level=logging.INFO,
# format='%(asctime)s - %(levelname)s - %(message)s',
# handlers=[
# logging.StreamHandler(),
# logging.FileHandler(Config.LOG_FILE)
# ]
# )
# return logging.getLogger(__name__)
#
#
# logger = Logger.setup()
# ---------------------- 工具函数 -----------------------
class Utils:
@staticmethod
@@ -379,18 +361,23 @@ class DataMonitor(DataHandler):
def main(self):
import datetime
task_start_time =datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
logger.info(f"=== 开始数据监控任务 ({self.execution_time}) ===")
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
logger.info(f"=== 开始数据监控任务 ({self.execution_time}) ===")
if Utils.is_first_run_today():
success = self.run_daily_snapshot()
else:
success = self.run_hourly_check()
if Utils.is_first_run_today():
success = self.run_daily_snapshot()
else:
success = self.run_hourly_check()
common_tools.send_task_status(task_start_time, "字段监控")
common_tools.send_task_status(task_start_time, "字段监控")
logger.info("=== 数据监控任务完成 ===")
return success
logger.info("=== 数据监控任务完成 ===")
return success
except Exception as e:
error_task_logger.error(f"数据监控任务发生异常: {e}")
common_tools.send_task_error(task_start_time, "字段监控", str(e))
return False
if __name__ == "__main__":
+36 -22
View File
@@ -1,16 +1,20 @@
from yd_api import YDAPI
import pandas as pd
from tqdm import tqdm
import hashlib
from datetime import datetime, timedelta
import pandas as pd
import mysql.connector
from mysql.connector import Error
import json
from back_ground_module import CommonModule
import numpy as np
from config import Config
from log_config import configure_task_logger, configure_error_task_logger
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
# 获取已经配置好的错误任务日志记录器
error_task_logger = configure_error_task_logger()
common_module = CommonModule()
# 初始化 API 实例和 Token
@@ -44,8 +48,6 @@ class TimeConsumingProcess():
PAGES_two = form_data_two.get('totalCount') // 100 + 1
# # 手动控制小于3w
# PAGES_two = 290
for a in tqdm(range(1, PAGES_two + 1)):
try:
form_data_two = api_instance.read_processes_instances(
@@ -55,12 +57,12 @@ class TimeConsumingProcess():
)
all_process_list = all_process_list + form_data_two.get("data")
except Exception as e:
print(f"Error fetching page {a}: {e}")
logger.warning(f"获取流程实例数据时出错: {e}")
continue
return all_process_list
def extract_approval_records(self, process_instances):
def extract_approval_records(self, process_instances: list):
"""提取每条流程的审批记录"""
all_data_list = []
for data in tqdm(process_instances, desc="处理流程实例"):
@@ -332,7 +334,6 @@ class TimeConsumingProcess():
# 4. 额外检查:确保时间格式正确
df = df[df['审批时间'].str.match(r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$') | df['审批时间'].isnull()]
# 生成插入语句
try:
columns = ', '.join(df.columns)
@@ -358,28 +359,41 @@ class TimeConsumingProcess():
def main(self):
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# Step 1: 获取流程实例
process_instances = self.fetch_process_data()
try:
logger.info("开始执行宜搭流程耗时写入BI任务。")
# Step 1: 获取流程实例
process_instances = self.fetch_process_data()
logger.info("获取流程实例成功。")
# Step 2: 提取审批记录
all_data_list = self.extract_approval_records(process_instances)
# Step 2: 提取审批记录
all_data_list = self.extract_approval_records(process_instances)
logger.info("提取审批记录成功。")
# Step 3: 按 '提交申请' 分组
result_groups = self.group_by_process(all_data_list)
# Step 3: 按 '提交申请' 分组
result_groups = self.group_by_process(all_data_list)
logger.info("'提交申请' 分组成功。")
# Step 4: 转换为宽表
df_final, max_steps = self.transform_to_wide_table(result_groups)
# Step 4: 转换为宽表
df_final, max_steps = self.transform_to_wide_table(result_groups)
logger.info("转换为宽表成功。")
# Step 5: 对流程进行分类并保存结果
df_final1 = self.classify_flows(df_final, max_steps)
# Step 5: 对流程进行分类并保存结果
df_final1 = self.classify_flows(df_final, max_steps)
logger.info("对流程进行分类并保存结果成功。")
# Step 6: 耗时计算
df_final2 = self.time_calculate(df_final1)
# Step 6: 耗时计算
df_final2 = self.time_calculate(df_final1)
logger.info("耗时计算成功。")
# Step 7: 向BI写入数据
self.write_to_bi(df_final2)
# Step 7: 向BI写入数据
self.write_to_bi(df_final2)
logger.info("向BI写入数据成功。")
common_module.send_task_status(task_start_time, "宜搭流程耗时写入BI")
common_module.send_task_status(task_start_time, "宜搭流程耗时写入BI")
logger.info("宜搭流程耗时写入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__':
+39 -29
View File
@@ -6,7 +6,11 @@ from api import API
import pymysql # 使用 pymysql 替代 mysql.connector
from back_ground_module import CommonModule
from tqdm import tqdm
from log_config import configure_task_logger, configure_error_task_logger
logger = configure_task_logger()
error_task_logger = configure_error_task_logger()
start_time = datetime.datetime.now()
api_instance = API()
common_module = CommonModule()
@@ -107,40 +111,46 @@ class ImportPerformanceData:
def main(self):
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.load_all_data()
# Step1:获取履约表数据
df = common_module.get_perforamnce_details()
print(df)
try:
self.load_all_data()
# Step1:获取履约表数据
df = common_module.get_perforamnce_details()
logger.info("数据获取完成")
print("数据获取完成")
# Step2:清空现有数据
try:
id_list = [item["_id"] for item in self.performance_data_list]
# Step2:清空现有数据
id_list = [item["_id"] for item in self.performance_data_list]
delete_payload = {
"api_key": "675b900991ad2491c69389ca",
"entry_id": "68637c9818bc333fc14c30ad",
"data_ids": id_list
}
api_instance.entry_data_batch_delete(delete_payload)
logger.info("数据删除完成")
except Exception as e:
error_task_logger.error(f"数据删除失败: {e}")
common_module.send_task_error(task_start_time, "履约表数据支撑", str(e))
delete_payload = {
"api_key": "675b900991ad2491c69389ca",
"entry_id": "68637c9818bc333fc14c30ad",
"data_ids": id_list
}
api_instance.entry_data_batch_delete(delete_payload)
print("数据删除完成")
# Step3:将数据写入简道云中
all_data = self.process_data(df)
# Step3:将数据写入简道云中
all_data = self.process_data(df)
# 分批处理,每批1000条
batch_size = 1000
for i in tqdm(range(0, len(all_data), batch_size)):
batch = all_data[i:i + batch_size]
payload = {
"api_key": "675b900991ad2491c69389ca",
"entry_id": "68637c9818bc333fc14c30ad",
"data_list": batch
}
api_instance.entry_data_batch_create(payload)
# 分批处理,每批1000条
batch_size = 1000
for i in tqdm(range(0, len(all_data), batch_size)):
batch = all_data[i:i + batch_size]
payload = {
"api_key": "675b900991ad2491c69389ca",
"entry_id": "68637c9818bc333fc14c30ad",
"data_list": batch
}
api_instance.entry_data_batch_create(payload)
print("数据写入完成")
common_module.send_task_status(task_start_time, "履约表数据支撑")
logger.info("简道云数据写入完成")
common_module.send_task_status(task_start_time, "履约表数据支撑")
except Exception as e:
error_task_logger.error(f"履约表数据支撑执行失败: {e}")
common_module.send_task_error(task_start_time, "履约表数据支撑", str(e))
@staticmethod
def row_to_dict(row, field_mapping):
+61 -49
View File
@@ -81,67 +81,79 @@ class InstallEventDispatcher:
def main(self):
"""主函数"""
start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 1.加载所有数据
self.load_all_data()
install_service_lead_list = self.install_service_lead
try:
# 1.加载所有数据
self.load_all_data()
install_service_lead_list = self.install_service_lead
# 将list的字段映射为中文
new_sign_abnormal_data = [
self.reversed_dict(old_dict, self.reversed_field_mapping)
for old_dict in install_service_lead_list
]
# 将list的字段映射为中文
new_sign_abnormal_data = [
self.reversed_dict(old_dict, self.reversed_field_mapping)
for old_dict in install_service_lead_list
]
logger.info(f"加载数据完成")
# 2.获取今日值班客服
today_duty_staff = []
for item in self.services_list:
if item.get("_widget_1740117343937") == "":
today_duty_staff.append(item.get("_widget_1740042824214").get("username"))
# 2.获取今日值班客服
today_duty_staff = []
for item in self.services_list:
if item.get("_widget_1740117343937") == "":
today_duty_staff.append(item.get("_widget_1740042824214").get("username"))
count = len(today_duty_staff)
if count == 0:
print("今日值班客服为空,请检查数据")
return
count = len(today_duty_staff)
if count == 0:
logger.warning(f"今日值班客服为空,请检查数据")
common_module.send_task_error(start_time, "安装服务历史派发", "今日值班客服为空")
return
logger.info(f"今日值班客服为:{today_duty_staff}")
# 3.数据准备
new_sign_abnormal_data = [item for item in new_sign_abnormal_data if item["线索状态"] != "已派发"]
# 3.数据准备
new_sign_abnormal_data = [item for item in new_sign_abnormal_data if item["线索状态"] != "已派发"]
# 截取今日需要派发的数据
new_sign_abnormal_data = new_sign_abnormal_data[:count]
# 截取今日需要派发的数据
new_sign_abnormal_data = new_sign_abnormal_data[:count]
# 获取今日要派发数据的id
id_list = [item["_id"] for item in new_sign_abnormal_data]
# 获取今日要派发数据的id
id_list = [item["_id"] for item in new_sign_abnormal_data]
new_sign_abnormal_data = [
self.row_to_dict(row, self.field_mapping)
for row in new_sign_abnormal_data]
new_sign_abnormal_data = [
self.row_to_dict(row, self.field_mapping)
for row in new_sign_abnormal_data]
logger.info(f"数据准备完成")
# 4.派发今日数据
i = 0
for item in new_sign_abnormal_data:
item.update({"_widget_1744182647149": {"value":today_duty_staff[i]}})
# 4.派发今日数据
i = 0
for item in new_sign_abnormal_data:
item.update({"_widget_1744182647149": {"value": today_duty_staff[i]}})
data = {
'api_key': "66f3a68c6e56814df2c6b1af",
'entry_id': "67f5dc467a9f5b2710da965a", # 安装服务意向表
# 'entry_id': "6853c7cc512ffef038917440", # 测试表
"data": item
}
data = {
'api_key': "66f3a68c6e56814df2c6b1af",
'entry_id': "67f5dc467a9f5b2710da965a", # 安装服务意向表
# 'entry_id': "6853c7cc512ffef038917440", # 测试表
"data": item
}
api_instance.data_batch_create(data)
res = api_instance.data_batch_create(data)
logger.info(f"数据派发:{res}")
i += 1
logger.info(f"数据派发完成")
i += 1
# 5.修改原数据状态为已派发
for id in id_list:
data = {
'api_key': "66f3a68c6e56814df2c6b1af",
'entry_id': "68537b5e60a6295c6c09b464",
"data_id": id,
"data": {"_widget_1750301534577": {"value": "已派发"}}
}
res = api_instance.entry_data_update(data)
logger.info(f"数据状态修改:{res}")
logger.info(f"数据状态修改完成")
# 5.修改原数据状态为已派发
for id in id_list:
data = {
'api_key': "66f3a68c6e56814df2c6b1af",
'entry_id': "68537b5e60a6295c6c09b464",
"data_id": id,
"data": {"_widget_1750301534577": {"value": "已派发"}}
}
api_instance.entry_data_update(data)
common_module.send_task_status(start_time, "安装服务历史派发")
common_module.send_task_status(start_time, "安装服务历史派发")
logger.info("安装服务历史任务完成")
except Exception as e:
common_module.send_task_error(start_time, "安装服务历史派发", str(e))
error_task_logger.error(f"安装服务历史派发任务执行失败: {e}")
if __name__ == "__main__":
@@ -1,9 +1,7 @@
# -*- 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
@@ -28,6 +26,8 @@ output_dir = "output" # 设置输出目录
# 创建输出目录(如果不存在)
os.makedirs(output_dir, exist_ok=True)
HS_DB_Config = Config.HS_DB_Config
class NewDealerServiceOrderToBI:
# 经销商新签服务单转BI
@@ -55,7 +55,7 @@ class NewDealerServiceOrderToBI:
'服务是否满意': '_widget_1743148999298', '服务不满意原因': '_widget_1743148999308',
'产品是否满意': '_widget_1743148999300', '产品不满意原因': '_widget_1743148999309',
# '上传评价图片': '_widget_1743148999310',
'培训完成时间':'_widget_1754472835261',
'培训完成时间': '_widget_1754472835261',
'审核备注': '_widget_1743500862664',
'完成日期时间': '_widget_1753162835213', '流水号': '_widget_1753163217437',
'提交人': 'creator', '提交时间': 'createTime', '更新时间': 'updateTime'}
@@ -82,7 +82,8 @@ class NewDealerServiceOrderToBI:
df[col] = df[col].map(lambda x: x.get("name", "") if isinstance(x, dict) else "")
# 3.日期字段转为北京时间
time_columns = ["订单支付时间", "开通时间", "系统到期时间", "完成日期时间", "提交时间", "更新时间","培训完成时间"]
time_columns = ["订单支付时间", "开通时间", "系统到期时间", "完成日期时间", "提交时间", "更新时间",
"培训完成时间"]
df[time_columns] = df[time_columns].apply(
lambda col: pd.to_datetime(col, errors='coerce')
@@ -93,13 +94,6 @@ class NewDealerServiceOrderToBI:
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 = "new_dealer_service_order_to_bi" # 替换为你的实际表名
# 建立数据库连接
@@ -122,7 +116,7 @@ class NewDealerServiceOrderToBI:
# 如果没有匹配的列,直接返回
if filtered_df.empty:
print("DataFrame 中没有与数据库表结构匹配的列。")
logger.warning("DataFrame 中没有与数据库表结构匹配的列。")
return
# 筛选列之后,插入前处理 dict 类型
@@ -144,10 +138,12 @@ class NewDealerServiceOrderToBI:
cursor.execute(insert_sql, tuple(row))
connection.commit()
print(f"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。")
logger.info(f"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。")
except Exception as e:
print("写入数据库时发生错误:", e)
error_task_logger.error(f"写入数据库时发生错误:, {e}")
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "经销商新签服务单转BI", str(e))
connection.rollback()
finally:
cursor.close()
@@ -158,13 +154,6 @@ class NewDealerServiceOrderToBI:
清空指定 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 = "new_dealer_service_order_to_bi" # 要清空的表名
connection = None
@@ -182,36 +171,44 @@ class NewDealerServiceOrderToBI:
# 使用TRUNCATE清空表数据
cursor.execute(f"TRUNCATE TABLE {table_name}")
connection.commit()
print(f"成功清空表 {table_name} 中的所有数据")
logger.info(f"成功清空表 {table_name} 中的所有数据")
except Error as e:
print(f"清空表时发生错误: {e}")
logger.error(f"清空表时发生错误{e}")
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "经销商新签服务单转BI", str(e))
if connection and connection.is_connected():
connection.rollback()
finally:
if connection and connection.is_connected():
cursor.close()
connection.close()
print("数据库连接已关闭")
logger.info("数据库连接已关闭")
def main(self):
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
# step1: 获取数据
self.load_all_data()
# step1: 获取数据
self.load_all_data()
logger.info(f"数据加载完成")
# step2:数据处理
df = self.data_process()
# step2:数据处理
df = self.data_process()
logger.info(f"数据处理完成")
# # step3:数据库删除
self.clear_table_data()
logger.info(f"数据库删除完成")
#
# # step4:数据写入BI
self.write_to_bi(df)
logger.info(f"数据写入BI完成")
# # step3:数据库删除
self.clear_table_data()
#
# # step4:数据写入BI
self.write_to_bi(df)
common_module.send_task_status(task_start_time, "经销商新签服务单转BI")
common_module.send_task_status(task_start_time, "经销商新签服务单转BI")
except Exception as e:
common_module.send_task_error(task_start_time, "经销商新签服务单转BI", str(e))
error_task_logger.error(f"经销商新签服务单转BI任务执行失败: {e}")
if __name__ == '__main__':
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+27 -26
View File
@@ -8,12 +8,20 @@ from dateutil.parser import parse
from back_ground_module import CommonModule
import numpy as np
from config import Config
from log_config import configure_task_logger, configure_error_task_logger
common_module = CommonModule()
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
# 获取已经配置好的错误任务日志记录器
error_task_logger = configure_error_task_logger()
class CRMDataProcessor:
"""泰国CRM数据迁移到BI"""
def __init__(self):
"""
初始化CRM数据处理类
@@ -178,13 +186,6 @@ class CRMDataProcessor:
# 去掉前六列和后两列
df = df.iloc[:, 6:-2]
# del df["creator"]
# del df["createTime"]
# del df["updateTime"]
# del df["updater"]
# del df["deleter"]
# del df["deleteTime"]
# 生成URL
base_url = f"https://www.jiandaoyun.com/dashboard/app/{self.api_key}/form/{self.entry_id}/data/"
df['url'] = base_url + df['_id'].astype(str) + "/qr_link"
@@ -219,7 +220,7 @@ class CRMDataProcessor:
# 只保留映射后的列和URL字段
mapped_columns = list(self.id_to_name_mapping.values()) + ['url']
df = df[[col for col in mapped_columns if col in df.columns]]
#df.replace([np.nan, None, r'^\s*$'], "", regex=True, inplace=True)
# df.replace([np.nan, None, r'^\s*$'], "", regex=True, inplace=True)
# 修改替换空值的实现方式
df = df.fillna("") # 先替换NaN和None
df = df.replace(r'^\s*$', "", regex=True) # 再替换空字符串
@@ -284,9 +285,11 @@ class CRMDataProcessor:
self.connect_db()
self.cursor.execute(f"TRUNCATE TABLE {table_name}")
self.connection.commit()
print(f"成功清空表 {table_name} 中的所有数据")
logger.info(f"成功清空表 {table_name} 中的所有数据")
except Error as e:
print(f"清空表时发生错误: {e}")
error_task_logger.error(f"清空表时发生错误: {e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "简道云海外项目CRM客户档案迁移BI", str(e))
if self.connection and self.connection.is_connected():
self.connection.rollback()
raise
@@ -319,48 +322,46 @@ class CRMDataProcessor:
records = [tuple(row) for row in df.values]
self.cursor.executemany(insert_query, records)
self.connection.commit()
print(f"成功导入 {self.cursor.rowcount} 条记录到 {table_name}")
logger.info(f"成功导入 {self.cursor.rowcount} 条记录到 {table_name}")
except Error as e:
print(f"导入数据时发生错误: {e}")
error_task_logger.error(f"导入数据时发生错误: {e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "简道云海外项目CRM客户档案迁移BI", str(e))
if self.connection and self.connection.is_connected():
self.connection.rollback()
raise
finally:
self.close_db()
# # 批量插入数据
# records = [tuple(None if pd.isna(x) else x for x in row) for row in df.values]
# self.cursor.executemany(insert_query, records)
# self.connection.commit()
# print(f"成功导入 {self.cursor.rowcount} 条记录到 {table_name} 表")
# except Error as e:
# print(f"导入数据时发生错误: {e}")
# raise
# finally:
# self.close_db()
def main(self):
"""运行完整的数据处理流程"""
table_name = "jiandaoyun_crm_customer_profile"
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
try:
logger.info(f"开始处理任务")
# 获取数据
raw_data = self.fetch_crm_data()
logger.info("数据获取完成")
# 处理数据
processed_data = self.process_data(raw_data)
logger.info("数据处理完成")
# 清空表
self.clear_table(table_name)
logger.info("表清空完成")
# 导入数据
self.import_data(processed_data, table_name)
logger.info("数据导入完成")
print("数据处理流程完成")
logger.info("数据处理流程完成")
except Exception as e:
print(f"数据处理流程出错: {e}")
common_module.send_task_error(task_start_time, "简道云海外项目CRM客户档案迁移BI", str(e))
error_task_logger.error(f"任务简道云海外项目CRM客户档案迁移BI执行失败。")
raise
common_module.send_task_status(task_start_time, "简道云海外项目CRM客户档案迁移BI")
+24 -10
View File
@@ -18,8 +18,10 @@ error_task_logger = configure_error_task_logger()
common_module = CommonModule()
class update_ID_form:
"""更新简道云员工ID表"""
def __init__(self):
self.headers = {
'Authorization': Config.JIANDAOYUN_API_TOKEN, # 曹伟应用api测试 app_key
@@ -54,6 +56,8 @@ class update_ID_form:
return df1
except Exception as e:
error_task_logger.error(f"获取部门成员及ID表失败:{e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "简道云员工ID表更新", str(e))
return None
def get_existing_id_form(self):
@@ -65,6 +69,8 @@ class update_ID_form:
return df
except Exception as e:
error_task_logger.error(f"读取现有的ID表失败:{e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "简道云员工ID表更新", str(e))
return None
def delete_existing_data(self, df):
@@ -78,6 +84,8 @@ class update_ID_form:
logger.info("现有数据已成功删除")
except Exception as e:
error_task_logger.error(f"批量删除现有数据失败:{e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "简道云员工ID表更新", str(e))
def update_data(self, df1):
"""批量写入新数据"""
@@ -93,21 +101,27 @@ class update_ID_form:
logger.info("新数据已成功写入")
except Exception as e:
error_task_logger.error(f"批量写入新数据失败:{e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "简道云员工ID表更新", str(e))
def main(self):
"""主函数"""
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
logger.info("每日任务开始执行")
df1 = self.get_department_members()
if df1 is not None:
df = self.get_existing_id_form()
if df is not None:
self.delete_existing_data(df)
self.update_data(df1)
logger.info("每日任务执行完成")
common_module.send_task_status(task_start_time, "简道云员工ID表更新")
try:
logger.info("每日任务开始执行")
df1 = self.get_department_members()
if df1 is not None:
df = self.get_existing_id_form()
if df is not None:
self.delete_existing_data(df)
self.update_data(df1)
logger.info("每日任务执行完成")
common_module.send_task_status(task_start_time, "简道云员工ID表更新")
except Exception as e:
error_task_logger.error(f"简道云员工ID表更新任务执行失败:{e}")
common_module.send_task_error(task_start_time, "简道云员工ID表更新", str(e))
if __name__ == '__main__':
daily_task = update_ID_form()
daily_task.main()
daily_task.main()
+77 -68
View File
@@ -4,7 +4,12 @@ import datetime
from config import Config
from api import API
from back_ground_module import CommonModule
from log_config import configure_task_logger, configure_error_task_logger
logger = configure_task_logger()
# 获取已经配置好的错误任务日志记录器
error_task_logger = configure_error_task_logger()
start_time = datetime.datetime.now()
api_instance = API()
common_module = CommonModule()
@@ -12,6 +17,7 @@ common_module = CommonModule()
class UpdateNGVData:
"""NGV数据每日新增"""
def __init__(self):
self.staff_id_list = None
self.field_mapping = {}
@@ -33,88 +39,91 @@ class UpdateNGVData:
return None
def main(self):
self.load_all_data()
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
data_NGV_j = common_module.get_ngv_details(days_back=1)
data_NGV_j1 = common_module.get_ngv_details(days_back=2)
try:
self.load_all_data()
logger.info(f"数据加载完成")
# 找出在 data_NGV_j 中存在但在 data_NGV_j1 中不存在的 data_id
unique_data_ids = data_NGV_j[~data_NGV_j['org_code'].isin(data_NGV_j1['org_code'])]
data_NGV_j = common_module.get_ngv_details(days_back=1)
data_NGV_j1 = common_module.get_ngv_details(days_back=2)
# 创建一个新的 DataFrame 保存这些唯一的 data_id 及其对应的数据
new_df = data_NGV_j[data_NGV_j['org_code'].isin(unique_data_ids['org_code'])]
# 找出在 data_NGV_j 中存在但在 data_NGV_j1 中不存在的 data_id
unique_data_ids = data_NGV_j[~data_NGV_j['org_code'].isin(data_NGV_j1['org_code'])]
# 对 new_df 进行进一步的过滤,只保留 org_type 为 "一般" 的记录
data_NGV_j = data_NGV_j[data_NGV_j['org_type'] == '一般']
data_NGV_j1 = data_NGV_j1[data_NGV_j1['org_type'] == '一般']
filtered_df = new_df[new_df['org_type'] == '一般']
# 创建一个新的 DataFrame 保存这些唯一的 data_id 及其对应的数据
new_df = data_NGV_j[data_NGV_j['org_code'].isin(unique_data_ids['org_code'])]
# 日期字段转换为日期格式
time_columns = ['date_fmt', 'saas_create_time', 'expiry_time', 'install_create_time', "last_end_date",
"renew_date"]
new_filtered_df = filtered_df.copy() # 复制df,以调整时间
for col in time_columns:
# 1. 转换为datetime类型(带错误处理)
# 使用.loc安全赋值
new_filtered_df[col] = pd.to_datetime(filtered_df[col], errors='coerce', utc=False)
# 对 new_df 进行进一步的过滤,只保留 org_type 为 "一般" 的记录
data_NGV_j = data_NGV_j[data_NGV_j['org_type'] == '一般']
data_NGV_j1 = data_NGV_j1[data_NGV_j1['org_type'] == '一般']
filtered_df = new_df[new_df['org_type'] == '一般']
# 2. 优化后的时区转换(高效向量化操作)
filtered_df[col + '_date'] = (
new_filtered_df[col]
# 本地化为北京时间(东八区)
.dt.tz_localize('Asia/Shanghai', ambiguous='infer', nonexistent='NaT')
# 转换为UTC时区
.dt.tz_convert('UTC')
# 格式化为ISO8601字符串
.dt.strftime('%Y-%m-%dT%H:%M:%SZ')
)
# 日期字段转换为日期格式
time_columns = ['date_fmt', 'saas_create_time', 'expiry_time', 'install_create_time', "last_end_date",
"renew_date"]
new_filtered_df = filtered_df.copy() # 复制df,以调整时间
for col in time_columns:
# 1. 转换为datetime类型(带错误处理)
# 使用.loc安全赋值
new_filtered_df[col] = pd.to_datetime(filtered_df[col], errors='coerce', utc=False)
# 人员字段转换为人员字段
staff_columns = ['area_manager', 'service_impl_principal', "service_salesmen","technician"]
# 将员工列表转为DataFrame
# 三重循环临时方案(确保可写入
for col in staff_columns:
staff_ids = []
for _, row in filtered_df.iterrows():
matched = False
for staff in self.staff_id_list:
if str(staff['_widget_1734942794144']) == str(row[col]):
staff_ids.append(staff['_widget_1734942794145'])
matched = True
break
if not matched:
staff_ids.append(None)
filtered_df[col + "_staff_id"] = staff_ids
# 2. 优化后的时区转换(高效向量化操作)
filtered_df[col + '_date'] = (
new_filtered_df[col]
# 本地化为北京时间(东八区
.dt.tz_localize('Asia/Shanghai', ambiguous='infer', nonexistent='NaT')
# 转换为UTC时区
.dt.tz_convert('UTC')
# 格式化为ISO8601字符串
.dt.strftime('%Y-%m-%dT%H:%M:%SZ')
)
logger.info(f"时间转换完成")
# filtered_df.to_csv(r"D:\Idea Project\SaaS_V1.3\back_ground_module\output\NGV.csv")
# 人员字段转换为人员字段
staff_columns = ['area_manager', 'service_impl_principal', "service_salesmen", "technician"]
# 将员工列表转为DataFrame
# 三重循环临时方案(确保可写入)
for col in staff_columns:
staff_ids = []
for _, row in filtered_df.iterrows():
matched = False
for staff in self.staff_id_list:
if str(staff['_widget_1734942794144']) == str(row[col]):
staff_ids.append(staff['_widget_1734942794145'])
matched = True
break
if not matched:
staff_ids.append(None)
filtered_df[col + "_staff_id"] = staff_ids
logger.info(f"人员转换完成")
# 生成包含所有行转换后的字典列表
# all_data = [self.row_to_dict(row, self.field_mapping) for index, row in data_NGV_j1.iterrows()] # 前两天的全部数据
# all_data = [self.row_to_dict(row, self.field_mapping) for index, row in data_NGV_j.iterrows()] # 前一天的全部数据
all_data = [self.row_to_dict(row, self.field_mapping) for index, row in filtered_df.iterrows()] # 增量数据
# filtered_df.to_csv(r"D:\Idea Project\SaaS_V1.3\back_ground_module\output\NGV.csv")
#
data = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.NGV_TASKS_ENTRY_ID, "data_list": all_data}
# 生成包含所有行转换后的字典列表
# all_data = [self.row_to_dict(row, self.field_mapping) for index, row in data_NGV_j1.iterrows()] # 前两天的全部数据
# all_data = [self.row_to_dict(row, self.field_mapping) for index, row in data_NGV_j.iterrows()] # 前一天的全部数据
all_data = [self.row_to_dict(row, self.field_mapping) for index, row in filtered_df.iterrows()] # 增量数据
result = api_instance.entry_data_batch_create(data)
# result_str = str(result)
# print(result_str[:500])
#
data = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.NGV_TASKS_ENTRY_ID, "data_list": all_data}
# 保存到Excel文件
# output_path = r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细1.xlsx'
# filtered_df.to_excel(output_path, index=False)
# data_NGV_j1.to_excel( r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细j1.xlsx', index=False)
# data_NGV_j.to_excel( r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细j.xlsx', index=False)
# new_df.to_excel(r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细ndf.xlsx', index=False)
result = api_instance.entry_data_batch_create(data)
logger.info(f"数据已推送:{result}")
# result_str = str(result)
# print(result_str[:500])
end_time = datetime.datetime.now()
# 保存到Excel文件
# output_path = r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细1.xlsx'
# filtered_df.to_excel(output_path, index=False)
# data_NGV_j1.to_excel( r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细j1.xlsx', index=False)
# data_NGV_j.to_excel( r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细j.xlsx', index=False)
# new_df.to_excel(r'D:\Idea Project\F6+宜搭+其它(1)\new\文件输出\ngv明细ndf.xlsx', index=False)
time_diff = end_time - start_time
# 打印天数、秒数和微秒数
print(f"执行时: {time_diff.days} 天, {time_diff.seconds} 秒, {time_diff.microseconds} 微秒")
common_module.send_task_status(task_start_time, "NGV新增数据")
common_module.send_task_status(task_start_time, "NGV新增数据")
logger.info(f"任务完成。")
except Exception as e:
error_task_logger.error(f"任务执行时发生异常: {e}")
common_module.send_task_error(task_start_time, "NGV新增数据", str(e))
@staticmethod
def row_to_dict(row, field_mapping):
+183 -180
View File
@@ -16,6 +16,13 @@ error_task_logger = configure_error_task_logger()
start_time = datetime.datetime.now()
api_instance = API()
common_module = CommonModule()
# 保存为CSV文件
output_dir = "output" # 设置输出目录
# 创建输出目录(如果不存在)
import os
os.makedirs(output_dir, exist_ok=True)
class UpdateAllNGVDataDaily:
@@ -26,225 +33,221 @@ class UpdateAllNGVDataDaily:
self.fields()
def main(self):
# 保存为CSV文件
output_dir = "output" # 设置输出目录
# 创建输出目录(如果不存在)
import os
os.makedirs(output_dir, exist_ok=True)
task_start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 获取NGV数据
payload = {"api_key": "675b900991ad2491c69389ca", "entry_id": "675bb02bd2d53c2034c665e4"}
NGV_data_list = api_instance.entry_data_list(payload).get("data", [])
jdy_NGV_data = pd.DataFrame(NGV_data_list)
try:
logger.info("开始执行任务:{}".format(task_start_time))
# 获取NGV数据
payload = {"api_key": "675b900991ad2491c69389ca", "entry_id": "675bb02bd2d53c2034c665e4"}
NGV_data_list = api_instance.entry_data_list(payload).get("data", [])
jdy_NGV_data = pd.DataFrame(NGV_data_list)
payload = {"api_key": "6694d3c4fcb69ca9a111a6c4",
"entry_id": "6769204a1902c9341340a1bc",
}
staff_id = api_instance.entry_data_list(payload)
staff_id_list = staff_id.get("data") # api请求格式,将数据封装在data字典里
payload = {"api_key": "6694d3c4fcb69ca9a111a6c4",
"entry_id": "6769204a1902c9341340a1bc",
}
staff_id = api_instance.entry_data_list(payload)
staff_id_list = staff_id.get("data") # api请求格式,将数据封装在data字典里
logger.info("已获取数据")
# for i in range(1,2):
data_NGV_j = common_module.get_ngv_details(days_back=1)
data_NGV_j.to_csv(os.path.join(output_dir, f"data_NGV_j.csv"), index=False)
data_NGV_j1 = common_module.get_ngv_details(days_back=2)
# for i in range(1,2):
data_NGV_j = common_module.get_ngv_details(days_back=1)
data_NGV_j.to_csv(os.path.join(output_dir, f"data_NGV_j.csv"), index=False)
data_NGV_j1 = common_module.get_ngv_details(days_back=2)
# 对 data_NGV 进行进一步的过滤,只保留 org_type 为 "一般" 的记录
data_NGV_j = data_NGV_j[data_NGV_j['org_type'] == '一般']
data_NGV_j1 = data_NGV_j1[data_NGV_j1['org_type'] == '一般']
# 对 data_NGV 进行进一步的过滤,只保留 org_type 为 "一般" 的记录
data_NGV_j = data_NGV_j[data_NGV_j['org_type'] == '一般']
data_NGV_j1 = data_NGV_j1[data_NGV_j1['org_type'] == '一般']
# 去除不需要的列
columns_to_remove = {'date_id', 'date_fmt', 'pt', 'etl_time'}
# 去除不需要的列
columns_to_remove = {'date_id', 'date_fmt', 'pt', 'etl_time'}
# 获取所有列名并计算要保留的列
columns_to_keep_df1 = list(set(data_NGV_j.columns) - columns_to_remove)
columns_to_keep_df2 = list(set(data_NGV_j1.columns) - columns_to_remove)
# 获取所有列名并计算要保留的列
columns_to_keep_df1 = list(set(data_NGV_j.columns) - columns_to_remove)
columns_to_keep_df2 = list(set(data_NGV_j1.columns) - columns_to_remove)
# 过滤DataFrame以去除指定列
df1_filtered = data_NGV_j[columns_to_keep_df1]
df2_filtered = data_NGV_j1[columns_to_keep_df2]
# 过滤DataFrame以去除指定列
df1_filtered = data_NGV_j[columns_to_keep_df1]
df2_filtered = data_NGV_j1[columns_to_keep_df2]
# 设置唯一标识列作为索引
df1_set_index = df1_filtered.set_index('id_own_org')
df2_set_index = df2_filtered.set_index('id_own_org')
# 设置唯一标识列作为索引
df1_set_index = df1_filtered.set_index('id_own_org')
df2_set_index = df2_filtered.set_index('id_own_org')
df1_set_index = df1_set_index.astype(str).replace(['nan', 'None'], '', ).fillna("")
df2_set_index = df2_set_index.astype(str).replace(['nan', 'None'], '', ).fillna("")
df1_set_index = df1_set_index.astype(str).replace(['nan', 'None'], '', ).fillna("")
df2_set_index = df2_set_index.astype(str).replace(['nan', 'None'], '', ).fillna("")
# 找到两个DataFrame共有的索引
common_index = df1_set_index.index.intersection(df2_set_index.index)
# 找到两个DataFrame共有的索引
common_index = df1_set_index.index.intersection(df2_set_index.index)
# 使用共同的索引来重新索引两个DataFrame
df1_common = df1_set_index.reindex(common_index).fillna('')
df2_common = df2_set_index.reindex(common_index).fillna('')
# 使用共同的索引来重新索引两个DataFrame
df1_common = df1_set_index.reindex(common_index).fillna('')
df2_common = df2_set_index.reindex(common_index).fillna('')
# 确保两个DataFrame有相同的列顺序
common_columns = df1_common.columns.intersection(df2_common.columns)
df1_common = df1_common[common_columns]
df2_common = df2_common[common_columns]
# 确保两个DataFrame有相同的列顺序
common_columns = df1_common.columns.intersection(df2_common.columns)
df1_common = df1_common[common_columns]
df2_common = df2_common[common_columns]
# 比较两个DataFrame的内容
comparison_column = 'match_status'
# 比较两个DataFrame的内容
comparison_column = 'match_status'
# 创建一个布尔Series,指示每一行是否完全相同
matches = (df1_common == df2_common).all(axis=1)
# 创建一个布尔Series,指示每一行是否完全相同
matches = (df1_common == df2_common).all(axis=1)
# 添加新列到第一个DataFrame,标记是否匹配
df1_common[comparison_column] = matches.map({True: '一致', False: '不一致'})
# df1_common.to_csv(os.path.join(output_dir, f"df1_common.csv"))
# 添加新列到第一个DataFrame,标记是否匹配
df1_common[comparison_column] = matches.map({True: '一致', False: '不一致'})
# df1_common.to_csv(os.path.join(output_dir, f"df1_common.csv"))
# 如果需要也可以添加到第二个DataFrame(这里假设只需要处理df1_common)
# df2_common[comparison_column] = matches.map({True: '一致', False: '不一致'})
# 如果需要也可以添加到第二个DataFrame(这里假设只需要处理df1_common)
# df2_common[comparison_column] = matches.map({True: '一致', False: '不一致'})
# 提取只在一个DataFrame中存在的索引对应的行
df1_only_index = df1_set_index.index.difference(df2_set_index.index)
df2_only_index = df2_set_index.index.difference(df1_set_index.index)
# 提取只在一个DataFrame中存在的索引对应的行
df1_only_index = df1_set_index.index.difference(df2_set_index.index)
df2_only_index = df2_set_index.index.difference(df1_set_index.index)
df1_only_rows = df1_set_index.loc[df1_only_index].copy()
df2_only_rows = df2_set_index.loc[df2_only_index].copy()
df1_only_rows = df1_set_index.loc[df1_only_index].copy()
df2_only_rows = df2_set_index.loc[df2_only_index].copy()
# 保存匹配结果
# df1_common.to_csv(os.path.join(output_dir, 'matched_results.csv'), index_label='id_own_org')
# 保存匹配结果
# df1_common.to_csv(os.path.join(output_dir, 'matched_results.csv'), index_label='id_own_org')
# 保存仅在df1中的行
# df1_only_rows.to_csv(os.path.join(output_dir, 'df1_only_rows.csv'), index_label='id_own_org')
# 保存仅在df1中的行
# df1_only_rows.to_csv(os.path.join(output_dir, 'df1_only_rows.csv'), index_label='id_own_org')
# 保存仅在df2中的行
# df2_only_rows.to_csv(os.path.join(output_dir, 'df2_only_rows.csv'), index_label='id_own_org')
# data_NGV_j.to_csv(os.path.join(output_dir, 'data_NGV_j.csv'), index_label='id_own_org')
# data_NGV_j1.to_csv(os.path.join(output_dir, 'data_NGV_j1.csv'), index_label='id_own_org')
# jdy_NGV_data.to_csv(os.path.join(output_dir, 'jdy_NGV_data.csv'), index_label='id_own_org')
# 保存仅在df2中的行
# df2_only_rows.to_csv(os.path.join(output_dir, 'df2_only_rows.csv'), index_label='id_own_org')
# data_NGV_j.to_csv(os.path.join(output_dir, 'data_NGV_j.csv'), index_label='id_own_org')
# data_NGV_j1.to_csv(os.path.join(output_dir, 'data_NGV_j1.csv'), index_label='id_own_org')
# jdy_NGV_data.to_csv(os.path.join(output_dir, 'jdy_NGV_data.csv'), index_label='id_own_org')
# print(f"\nCSV文件已保存到目录: {output_dir}")
# print(f"\nCSV文件已保存到目录: {output_dir}")
temp_jdy_NGV_data = jdy_NGV_data.copy()
temp_jdy_NGV_data = jdy_NGV_data.copy()
# temp_jdy_NGV_data.to_csv(os.path.join(output_dir, 'jdy_NGV_data.csv'), index=False)
temp_jdy_NGV_data.reset_index(inplace=True) # 如果 '门店id' 是索引,则先将其转换为普通列
# temp_jdy_NGV_data.to_csv(os.path.join(output_dir, 'jdy_NGV_data1.csv'), index=False)
if '_widget_1734062123069' not in temp_jdy_NGV_data.columns:
print("'门店id' 不存在")
temp_jdy_NGV_data.rename(columns={'_widget_1734062123069': 'id_own_org'}, inplace=True)
temp_jdy_NGV_data.set_index('id_own_org', inplace=True)
# temp_jdy_NGV_data.to_csv(os.path.join(output_dir, 'jdy_NGV_data.csv'), index=False)
temp_jdy_NGV_data.reset_index(inplace=True) # 如果 '门店id' 是索引,则先将其转换为普通列
# temp_jdy_NGV_data.to_csv(os.path.join(output_dir, 'jdy_NGV_data1.csv'), index=False)
if '_widget_1734062123069' not in temp_jdy_NGV_data.columns:
error_task_logger.error("'门店id' 不存在")
temp_jdy_NGV_data.rename(columns={'_widget_1734062123069': 'id_own_org'}, inplace=True)
temp_jdy_NGV_data.set_index('id_own_org', inplace=True)
# 如果简道云存在,NGV不存在则标记NGV已删除
# 找出在 temp_jdy_NGV_data 中存在,但在 df1_common 中不存在的索引
ids_in_jdy_not_in_df1 = temp_jdy_NGV_data.index[~temp_jdy_NGV_data.index.isin(df1_common.index)]
# 提取这些行,形成新的 DataFrame
only_in_temp_jdy = temp_jdy_NGV_data.loc[ids_in_jdy_not_in_df1]
# 对数据源已经去掉的门店进行标记
for index, only_row in only_in_temp_jdy.iterrows():
result = {}
if '_id' in only_in_temp_jdy.columns:
_id_value = str(only_row['_id']) if not pd.isna(only_row['_id']) else None
result["_id"] = _id_value
# 如果简道云存在,NGV不存在则标记NGV已删除
# 找出在 temp_jdy_NGV_data 中存在,但在 df1_common 中不存在的索引
ids_in_jdy_not_in_df1 = temp_jdy_NGV_data.index[~temp_jdy_NGV_data.index.isin(df1_common.index)]
# 提取这些行,形成新的 DataFrame
only_in_temp_jdy = temp_jdy_NGV_data.loc[ids_in_jdy_not_in_df1]
# 对数据源已经去掉的门店进行标记
for index, only_row in only_in_temp_jdy.iterrows():
result = {}
if '_id' in only_in_temp_jdy.columns:
_id_value = str(only_row['_id']) if not pd.isna(only_row['_id']) else None
result["_id"] = _id_value
if result["_id"]:
data = {
'api_key': Config.SaaS_Tasks_APP_ID,
'entry_id': Config.NGV_TASKS_ENTRY_ID,
"data_id": result["_id"],
"data": {"_widget_1754285499851": {"value": "已删除"}}
}
if result["_id"]:
data = {
'api_key': Config.SaaS_Tasks_APP_ID,
'entry_id': Config.NGV_TASKS_ENTRY_ID,
"data_id": result["_id"],
"data": {"_widget_1754285499851": {"value": "已删除"}}
}
api_instance.entry_data_update(data=data, max_retries=20)
api_instance.entry_data_update(data=data, max_retries=20)
# 简道云与ngv不一致的数据做关联
df1_common = df1_common.join(temp_jdy_NGV_data["_id"], how='left')
df1_common = df1_common[df1_common['match_status'] == '不一致']
# 简道云与ngv不一致的数据做关联
df1_common = df1_common.join(temp_jdy_NGV_data["_id"], how='left')
df1_common = df1_common[df1_common['match_status'] == '不一致']
# 日期字段转换为日期格式
time_columns = ['saas_create_time', 'expiry_time', 'install_create_time', "last_end_date",
"renew_date"]
new_filtered_df = df1_common.copy() # 复制df,以调整时间
for col in time_columns:
# 1. 转换为datetime类型(带错误处理)
# 使用.loc安全赋值
new_filtered_df[col] = pd.to_datetime(df1_common[col], errors='coerce', utc=False)
# 日期字段转换为日期格式
time_columns = ['saas_create_time', 'expiry_time', 'install_create_time', "last_end_date",
"renew_date"]
new_filtered_df = df1_common.copy() # 复制df,以调整时间
for col in time_columns:
# 1. 转换为datetime类型(带错误处理)
# 使用.loc安全赋值
new_filtered_df[col] = pd.to_datetime(df1_common[col], errors='coerce', utc=False)
# 2. 优化后的时区转换(高效向量化操作)
df1_common[col + '_date'] = (
new_filtered_df[col]
# 本地化为北京时间(东八区)
.dt.tz_localize('Asia/Shanghai', ambiguous='infer', nonexistent='NaT')
# 转换为UTC时区
.dt.tz_convert('UTC')
# 格式化为ISO8601字符串
.dt.strftime('%Y-%m-%dT%H:%M:%SZ')
)
# 2. 优化后的时区转换(高效向量化操作)
df1_common[col + '_date'] = (
new_filtered_df[col]
# 本地化为北京时间(东八区)
.dt.tz_localize('Asia/Shanghai', ambiguous='infer', nonexistent='NaT')
# 转换为UTC时区
.dt.tz_convert('UTC')
# 格式化为ISO8601字符串
.dt.strftime('%Y-%m-%dT%H:%M:%SZ')
)
logger.info("日期已转换为UTC格式")
# 人员字段转换为人员字段
staff_columns = ['area_manager', 'service_impl_principal', "service_salesmen", "technician"]
# 将员工列表转为DataFrame
# 三重循环临时方案(确保可写入)
for col in staff_columns:
staff_ids = []
for _, row in df1_common.iterrows():
matched = False
for staff in staff_id_list:
if str(staff['_widget_1734942794144']) == str(row[col]):
staff_ids.append(staff['_widget_1734942794145'])
matched = True
break
if not matched:
staff_ids.append(None)
df1_common[col + "_staff_id"] = staff_ids
# 人员字段转换为人员字段
staff_columns = ['area_manager', 'service_impl_principal', "service_salesmen", "technician"]
# 将员工列表转为DataFrame
# 三重循环临时方案(确保可写入)
for col in staff_columns:
staff_ids = []
for _, row in df1_common.iterrows():
matched = False
for staff in staff_id_list:
if str(staff['_widget_1734942794144']) == str(row[col]):
staff_ids.append(staff['_widget_1734942794145'])
matched = True
break
if not matched:
staff_ids.append(None)
df1_common[col + "_staff_id"] = staff_ids
logger.info("人员字段已替换")
# 并发请求
futures = []
all_data = []
# 并发请求
futures = []
all_data = []
logger.info(f"今日更新数据量为:{len(df1_common)}")
for idx, row in df1_common.iterrows():
result = {}
data_dict = {}
for idx, row in df1_common.iterrows():
result = {}
data_dict = {}
# 根据 field_mapping 进行字段替换
for col_name, widget_id in self.field_mapping.items():
if col_name in df1_common.columns:
value = row[col_name]
clean_value = None if pd.isna(value) else value
data_dict[widget_id] = {"value": clean_value}
# 根据 field_mapping 进行字段替换
for col_name, widget_id in self.field_mapping.items():
if col_name in df1_common.columns:
value = row[col_name]
clean_value = None if pd.isna(value) else value
data_dict[widget_id] = {"value": clean_value}
# 单独处理 _id 列,并将其转换为字符串
if '_id' in df1_common.columns:
_id_value = str(row['_id']) if not pd.isna(row['_id']) else None
result["_id"] = _id_value
# 单独处理 _id 列,并将其转换为字符串
if '_id' in df1_common.columns:
_id_value = str(row['_id']) if not pd.isna(row['_id']) else None
result["_id"] = _id_value
# 组装最终结果
if result["_id"]:
data = {
'api_key': Config.SaaS_Tasks_APP_ID,
'entry_id': Config.NGV_TASKS_ENTRY_ID,
"data_id": result["_id"],
"data": data_dict
}
# 组装最终结果
if result["_id"]:
data = {
'api_key': Config.SaaS_Tasks_APP_ID,
'entry_id': Config.NGV_TASKS_ENTRY_ID,
"data_id": result["_id"],
"data": data_dict
}
api_instance.entry_data_update(data=data, max_retries=20)
else:
# continue
data1 = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.NGV_TASKS_ENTRY_ID,
"data": data_dict}
api_instance.data_batch_create(data=data1, max_retries=20)
api_instance.entry_data_update(data=data, max_retries=20)
else:
# continue
data1 = {'api_key': Config.SaaS_Tasks_APP_ID, 'entry_id': Config.NGV_TASKS_ENTRY_ID,
"data": data_dict}
res = api_instance.data_batch_create(data=data1, max_retries=20)
logger.info(f"补派数据:{res}")
# all_data.append(data_dict)
# all_data.append(data_dict)
# 收集所有结果
for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
logger.info(f"所有请求结果:{result}")
except Exception as exc:
error_task_logger.error(f"请求发生异常: {exc}")
# 收集所有结果
for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
print("请求结果:", result)
except Exception as exc:
print(f"请求发生异常: {exc}")
end_time = datetime.datetime.now()
# df11 = pd.DataFrame(all_data)
# df11.to_csv(f"all_data.csv")
time_diff = end_time - start_time
# 打印天数、秒数和微秒数
print(f"执行时间: {time_diff.days} 天, {time_diff.seconds} 秒, {time_diff.microseconds} 微秒")
common_module.send_task_status(task_start_time, "NGV更新数据")
common_module.send_task_status(task_start_time, "NGV更新数据")
logger.info("NGV更新数据任务已完成。")
except Exception as e:
error_task_logger.error(f"NGV更新数据执行时发生异常: {e}")
common_module.send_task_error(task_start_time, "NGV更新数据", str(e))
@staticmethod
def row_to_dict(row, field_mapping):
@@ -1,16 +1,11 @@
import mysql.connector
from mysql.connector import Error
import numpy as np
import pandas as pd
from yd_api import YDAPI
from api import API
import pandas as pd
from tqdm import tqdm
import time
from datetime import datetime, timedelta
from datetime import datetime
from config import Config
from back_ground_module import CommonModule
import logging
from log_config import configure_task_logger, configure_error_task_logger
import mysql.connector
from mysql.connector import Error
@@ -33,11 +28,11 @@ FORMID = "FORM-WV866IC119W8BZC7AKHAR7VT3FI52W4Q1VBFLD1" # FPO需求提交
appType = "APP_UYZ0KG6L0CCNV80GZ66O" # F6客户服务
systemToken = "XA966F81JAJOFCVVVKO64E9MIIZV1EWE5SFMKJ2" # 密钥
BASE_URL = "https://f6car.aliwork.com" # 基础URL
print(TOKEN)
DB_CONFIG = Config.HS_DB_Config
# 数据库配置
# 数据库配置
class DenominatorReportingAdjustment:
"""分母报备调整"""
@@ -45,7 +40,6 @@ class DenominatorReportingAdjustment:
def __init__(self):
self.structures = None
self.denominator_data_list = None
self.field_map = {
"门店编码": "textField_pl5p5a3",
"门店名称": "textField_fcl5xg6",
@@ -77,7 +71,6 @@ class DenominatorReportingAdjustment:
denominator_data = yd_api_instance.read_processes(token=TOKEN, formUuid=FORMID, page=1, n=100,
appType=appType, systemToken=systemToken)
self.denominator_data_list = []
print(denominator_data)
PAGES_two = denominator_data.get('totalCount') // 100 + 1
parentheses_pattern = re.compile(r'$[^)]*$')
@@ -89,7 +82,8 @@ class DenominatorReportingAdjustment:
# Transform the keys using field_map
transformed_data = {}
for field_key in ['employeeField_mca5shp1', 'employeeField_mca5shp0']:
if field_key in form_data and isinstance(form_data[field_key], list) and len(form_data[field_key]) > 0:
if field_key in form_data and isinstance(form_data[field_key], list) and len(
form_data[field_key]) > 0:
# 取第一个元素
raw_value = form_data[field_key][0]
# 去除括号及其中的内容
@@ -108,10 +102,9 @@ class DenominatorReportingAdjustment:
# continue
self.denominator_data_list.append(transformed_data)
def execute_sql(self, sql, params=None, fetch=False, many=False):
"""执行SQL语句"""
global cursor
conn = None
try:
conn = mysql.connector.connect(**DB_CONFIG)
@@ -123,7 +116,7 @@ class DenominatorReportingAdjustment:
conn.commit()
return cursor.fetchall() if fetch else cursor
except Error as e:
print(f"执行失败: {sql}\n错误: {e}")
error_task_logger.error(f"执行失败: {sql}\n错误: {e}")
if conn: conn.rollback()
return None
finally:
@@ -153,10 +146,9 @@ class DenominatorReportingAdjustment:
# 保留 DataFrame 中与数据库列名匹配的列
filtered_df = df[df.columns.intersection(db_columns)]
# 如果没有匹配的列,直接返回
if filtered_df.empty:
print("DataFrame 中没有与数据库表结构匹配的列。")
logger.warning("DataFrame 中没有与数据库表结构匹配的列。")
return
# 筛选列之后,插入前处理 dict 类型
@@ -167,7 +159,6 @@ class DenominatorReportingAdjustment:
lambda x: json.dumps(x, ensure_ascii=False) if x is not None else x
)
# 构建插入语句
placeholders = ', '.join(['%s'] * len(filtered_df.columns))
# 使用反引号避免特殊列明
@@ -178,12 +169,14 @@ class DenominatorReportingAdjustment:
for _, row in filtered_df.iterrows():
cursor.execute(insert_sql, tuple(row))
connection.commit()
print(f"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。")
logger.info(f"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。")
except Exception as e:
print("写入数据时发生错误", e)
error_task_logger.error(f"写入数据时发生错误: {e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "分母报备调整", str(e))
connection.rollback()
finally:
cursor.close()
@@ -192,31 +185,37 @@ class DenominatorReportingAdjustment:
def clear_table(self):
"""清空表数据"""
if self.execute_sql("TRUNCATE TABLE f6_denominator_adjustment"):
print("成功清空表数据")
logger.info("✅ 清空表数据成功")
def main(self):
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# step1:获取宜搭数据
self.get_yida_data()
try:
# step1:获取宜搭数据
self.get_yida_data()
logger.info("✅ 获取宜搭数据成功")
df = pd.DataFrame(self.denominator_data_list)
df['开户日期'] = df['开户日期'].astype('Int64')
df['开户日期'] = pd.to_datetime(df['开户日期'], unit='ms')
df = pd.DataFrame(self.denominator_data_list)
df['开户日期'] = df['开户日期'].astype('Int64')
df['开户日期'] = pd.to_datetime(df['开户日期'], unit='ms')
df['结束时间'] = df['结束时间'].astype('Int64')
df['结束时间'] = pd.to_datetime(df['结束时间'], unit='ms')
df['结束时间'] = df['结束时间'].astype('Int64')
df['结束时间'] = pd.to_datetime(df['结束时间'], unit='ms')
df['开始时间'] = df['开始时间'].astype('Int64')
df['开始时间'] = pd.to_datetime(df['开始时间'], unit='ms')
df['开始时间'] = df['开始时间'].astype('Int64')
df['开始时间'] = pd.to_datetime(df['开始时间'], unit='ms')
# step2:清空BI数据表
self.clear_table()
logger.info("✅ 清空表数据成功")
# step2:清空BI数据
self.clear_table()
# # step3:写入BI数据
self.write_to_bi(df)
logger.info("✅ 写入BI数据库成功")
# # step3:写入BI数据库
self.write_to_bi(df)
common_module.send_task_status(task_start_time, "分母报备调整")
common_module.send_task_status(task_start_time, "分母报备调整")
except Exception as e:
error_task_logger.error("分母报备调整失败")
common_module.send_task_error(task_start_time, "分母报备调整", str(e))
if __name__ == '__main__':
@@ -1,7 +1,5 @@
import os
import poplib
import time
import pandas as pd
from email.parser import Parser
from email.header import decode_header
from email.utils import parseaddr
@@ -12,26 +10,29 @@ from back_ground_module import CommonModule
import pandas as pd
import pymysql
from pymysql 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()
class EmailProcessor:
"""泰国CRM每日邮件写入简道云与BI"""
def __init__(self):
# 配置信息
self.user_email_address = 'caowei@f6car.cn'
self.user_password = 'Cw@340826'
self.pop_server_host = 'pop.qiye.aliyun.com'
self.pop_server_port = '995'
self.pop_server_port = 995
self.send_name = "f6car"
self.send_addr = 'noreplay@notice.f6car.com'
# 创建输出目录(如果不存在)
output_dir = "email"
os.makedirs(output_dir, exist_ok=True)
nowtime = datetime.now().strftime("%Y%m%d%H%M%S")
self.write_path = os.path.join(output_dir, f'email_data.xlsx')
@@ -47,7 +48,6 @@ class EmailProcessor:
"指标类型": "_widget_1742091963880",
"指标值": "_widget_1742091963882",
"指标子类型": "_widget_1742091963881",
"指标值": "_widget_1742091963882"
}
def connect_email_by_pop3(self):
@@ -120,7 +120,7 @@ class EmailProcessor:
# 打印邮件接收时间
mail_time_str = datetime.strftime(mail_datetime, '%Y-%m-%d %H:%M:%S')
print(f"邮件接收时间: {mail_time_str}")
logger.info(f"邮件接收时间: {mail_time_str}")
# 处理邮件内容
self.parser_content(msg, 0)
@@ -134,7 +134,6 @@ class EmailProcessor:
email_server.quit()
def parser_content(self, msg, indent):
print("邮件处理")
if indent == 0:
self.parser_email_header(msg)
@@ -143,7 +142,6 @@ class EmailProcessor:
name, charset = decode_header(hdr)[0]
if charset:
name = name.decode(charset)
print(f'发件人姓名: {name}, 发件人邮箱: {addr}')
if name == self.send_name:
# 下载附件
@@ -157,9 +155,9 @@ class EmailProcessor:
try:
with open(self.write_path, 'wb') as att_file:
att_file.write(data)
print(f"附件保存成功: {self.write_path}+{filename}")
logger.info(f"附件保存成功: {self.write_path}+{filename}")
except Exception as e:
print(f"附件保存失败: {str(e)}")
error_task_logger.error(f"保存附件时出错: {e}")
if msg.is_multipart():
parts = msg.get_payload()
@@ -173,7 +171,7 @@ class EmailProcessor:
charset = self.guess_charset(msg)
if charset:
content = content.decode(charset)
print(f"{' ' * indent}邮件内容: {content}")
logger.info(f"邮件内容: {content}")
def parser_email_header(self, msg):
# 解析邮件主题
@@ -181,21 +179,22 @@ class EmailProcessor:
value, charset = decode_header(subject)[0]
if charset:
value = value.decode(charset)
print(f'邮件主题: {value}')
# 解析发件人信息
hdr, addr = parseaddr(msg['From'])
name, charset = decode_header(hdr)[0]
name1, charset = decode_header(hdr)[0]
if charset:
name = name.decode(charset)
print(f'发件人姓名: {name}, 发件人邮箱: {addr}')
name1 = name1.decode(charset)
# 解析收件人信息
hdr, addr = parseaddr(msg['To'])
name, charset = decode_header(hdr)[0]
if charset:
name = name.decode(charset)
print(f'收件人姓名: {name}, 收件人邮箱: {addr}')
logger.info(f"邮件主题: {value}, 发件人: {name1}, 收件人: {name}")
@staticmethod
def decode_str(s):
@@ -229,6 +228,7 @@ class EmailProcessor:
mail_datetime = datetime.strptime(mail_datetime, ft)
return mail_datetime
except:
error_task_logger.error(f"邮件时间格式解析错误: {mail_datetime}")
pass
raise Exception("邮件时间格式解析错误")
@@ -244,8 +244,7 @@ class EmailProcessor:
return result
def update_email(self):
# try:
print(self.write_path)
logger.info("开始处理邮件数据")
email_df = pd.read_excel(self.write_path, sheet_name="Sheet0")
print(email_df.head())
@@ -269,6 +268,7 @@ class EmailProcessor:
def up_to_BI(self, df):
# 连接信息
global connection
HS_DB_Config = Config.HS_DB_Config
table_name = "thailand_store_data_email"
@@ -282,7 +282,6 @@ class EmailProcessor:
charset='utf8mb4',
)
print(f"成功连接 {HS_DB_Config["database"]}")
with connection.cursor() as cursor:
# 处理数据
@@ -298,10 +297,12 @@ class EmailProcessor:
cursor.executemany(insert_query, records)
connection.commit()
print(f"成功导入 {cursor.rowcount} 条记录到 {table_name}")
logger.info(f"成功导入 {cursor.rowcount} 条记录到 {table_name}")
except Error as e:
print(f"数据库操作出错: {e}")
error_task_logger.error(f"数据库写入数据时发生异常: {e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "海外邮件推送", str(e))
if connection:
connection.rollback()
finally:
@@ -312,11 +313,16 @@ class EmailProcessor:
def main(cls):
"""邮件处理器的主入口点"""
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
processor = cls()
processor.connect_email_by_pop3()
email_df = processor.update_email()
processor.up_to_BI(email_df) # 发送到BI
common_module.send_task_status(task_start_time, "海外邮件推送")
try:
processor = cls()
processor.connect_email_by_pop3()
logger.info("邮件获取完成")
email_df = processor.update_email()
processor.up_to_BI(email_df) # 发送到BI
common_module.send_task_status(task_start_time, "海外邮件推送")
except Exception as e:
common_module.send_task_error(task_start_time, "海外邮件推送", "失败")
error_task_logger.error(f"任务执行时发生异常: {e}")
if __name__ == "__main__":
@@ -84,14 +84,13 @@ class MoleculeReportingAdjustment:
for field_id, value in form_data.items():
# Find the display name in field_map
for display_name, id_in_map in self.field_map.items():
if id_in_map == field_id :
if id_in_map == field_id:
transformed_data[display_name] = value
break
# if transformed_data.get("是否上传衡石") == "否" or transformed_data.get("是否上传衡石") is None:
# continue
self.molecule_data_list.append(transformed_data)
def execute_sql(self, sql, params=None, fetch=False, many=False):
"""执行SQL语句"""
conn = None
@@ -105,7 +104,10 @@ class MoleculeReportingAdjustment:
conn.commit()
return cursor.fetchall() if fetch else cursor
except Error as e:
print(f"执行失败: {sql}\n错误: {e}")
error_task_logger.error(f"执行SQL语句时发生错误: {sql}\n错误: {e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "分母报备调整", str(e))
if conn: conn.rollback()
return None
finally:
@@ -137,7 +139,7 @@ class MoleculeReportingAdjustment:
# 如果没有匹配的列,直接返回
if filtered_df.empty:
print("DataFrame 中没有与数据库表结构匹配的列。")
logger.warning("DataFrame 中没有与数据库表结构匹配的列。")
return
# 筛选列之后,插入前处理 dict 类型
@@ -159,10 +161,14 @@ class MoleculeReportingAdjustment:
cursor.execute(insert_sql, tuple(row))
connection.commit()
print(f"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。")
logger.info(f"成功写入 {len(filtered_df)} 条记录到 {table_name} 表中。")
except Exception as e:
print("写入数据时发生错误", e)
error_task_logger.error(f"写入数据时发生错误: {e}")
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
common_module.send_task_error(task_start_time, "分母报备调整", str(e))
connection.rollback()
finally:
cursor.close()
@@ -171,24 +177,32 @@ class MoleculeReportingAdjustment:
def clear_table(self):
"""清空表数据"""
if self.execute_sql("TRUNCATE TABLE f6_molecule_adjustment"):
print("成功清空表数据")
logger.info(f"成功清空表数据")
def main(self):
task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# step1:获取宜搭数据
self.get_yida_data()
try:
logger.info(f"开始执行任务")
# step1:获取宜搭数据
self.get_yida_data()
logger.info(f"获取宜搭数据成功")
df = pd.DataFrame(self.molecule_data_list)
df['归属月份'] = df['归属月份'].astype('Int64')
df['归属月份'] = pd.to_datetime(df['归属月份'], unit='ms')
df = pd.DataFrame(self.molecule_data_list)
df['归属月份'] = df['归属月份'].astype('Int64')
df['归属月份'] = pd.to_datetime(df['归属月份'], unit='ms')
# step2:清空BI数据表
self.clear_table()
# step2:清空BI数据表
self.clear_table()
logger.info(f"清空表成功")
# # step3:写入BI数据库
self.write_to_bi(df)
# # step3:写入BI数据库
self.write_to_bi(df)
logger.info(f"写入BI数据库成功")
common_module.send_task_status(task_start_time, "分子报备调整")
common_module.send_task_status(task_start_time, "分子报备调整")
except Exception as e:
error_task_logger.error(f"任务执行失败: {e}")
common_module.send_task_error(task_start_time, "分子报备调整", str(e))
if __name__ == '__main__':
+2 -14
View File
@@ -9,23 +9,10 @@ from back_ground_module import CommonModule
import logging
from log_config import configure_task_logger, configure_error_task_logger
# 配置日志
# logging.basicConfig(
# level=logging.INFO,
# format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
# handlers=[
# logging.FileHandler("process.log"),
# logging.StreamHandler()
# ]
# )
# logger = logging.getLogger(__name__)
# 获取已经配置好的常规日志记录器
logger = configure_task_logger()
# 获取已经配置好的错误任务日志记录器
error_task_logger = configure_error_task_logger()
# 初始化 API 实例和 Token
api_instanceyd = YDAPI()
api_instance = API()
@@ -303,7 +290,7 @@ class YDFpoJiandaoyun:
except Exception as e:
error_task_logger.error(f"清除现有数据时发生错误: {e}", exc_info=True)
def batch_create_entries(self,task_start_time):
def batch_create_entries(self, task_start_time):
"""批量创建条目"""
try:
if not self.all_data:
@@ -381,6 +368,7 @@ class YDFpoJiandaoyun:
except Exception as e:
error_task_logger.error(f"执行过程中发生错误: {e}", exc_info=True)
common_module.send_task_error(task_start_time, "宜搭FPO实例同步简道云", str(e))
if __name__ == '__main__':