校验唯一任务添加时间

优化续约代办请求次数

back_ground_module/update_denominator_reporting_adjustment_to_bi.py

@@ -205,11 +205,9 @@ class DenominatorReportingAdjustment:
     def main(self):
         task_start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
         try:
-
             # step1:获取宜搭数据
             self.get_yida_data()
             logger.info("✅ 获取宜搭数据成功")
-
             df = pd.DataFrame(self.denominator_data_list)
 
             # step2:清空BI数据表

logs/error_task.log

@@ -2534,3 +2534,23 @@
 2025-12-31 10:05:48,088 - log_config.py - error_task_logger - ERROR - 任务 经销商新签服务单转BI 超过执行窗口5分钟以上,标记为过期。
 2025-12-31 10:05:48,088 - log_config.py - error_task_logger - ERROR - 任务 高德匹配手机号 超过执行窗口5分钟以上,标记为过期。
 2025-12-31 10:05:48,089 - log_config.py - error_task_logger - ERROR - 任务 省市区人员关系表转BI 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,158 - log_config.py - error_task_logger - ERROR - 任务 NGV新增数据 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,159 - log_config.py - error_task_logger - ERROR - 任务 新签客户回访 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,160 - log_config.py - error_task_logger - ERROR - 任务 续约客户回访 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,161 - log_config.py - error_task_logger - ERROR - 任务 接车宝日常派发 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,161 - log_config.py - error_task_logger - ERROR - 任务 私域小程序数据支撑 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,162 - log_config.py - error_task_logger - ERROR - 任务 小六提成数据支撑 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,163 - log_config.py - error_task_logger - ERROR - 任务 异业合作数据支撑 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,164 - log_config.py - error_task_logger - ERROR - 任务 短信数据支撑 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,164 - log_config.py - error_task_logger - ERROR - 任务 海外邮件推送 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,165 - log_config.py - error_task_logger - ERROR - 任务 异常服务待办派发 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,166 - log_config.py - error_task_logger - ERROR - 任务 简道云海外项目CRM客户档案迁移BI 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,167 - log_config.py - error_task_logger - ERROR - 任务 安装服务历史派发 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,167 - log_config.py - error_task_logger - ERROR - 任务 分母报备调整 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,168 - log_config.py - error_task_logger - ERROR - 任务 分子报备调整 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,169 - log_config.py - error_task_logger - ERROR - 任务 履约表数据支撑 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,169 - log_config.py - error_task_logger - ERROR - 任务 字段监控 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,170 - log_config.py - error_task_logger - ERROR - 任务 经销商新签服务单转BI 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,171 - log_config.py - error_task_logger - ERROR - 任务 高德匹配手机号 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,171 - log_config.py - error_task_logger - ERROR - 任务 省市区人员关系表转BI 超过执行窗口5分钟以上,标记为过期。
+2026-01-04 10:29:26,172 - log_config.py - error_task_logger - ERROR - 任务 续约回访待办 超过执行窗口5分钟以上,标记为过期。

logs/task.log

@@ -187,3 +187,51 @@
 2025-12-31 10:05:48,090 - utils.py - task_logger - INFO - 任务 省市区人员关系表转BI 状态已更新为 过期。
 2025-12-31 10:05:48,090 - utils.py - task_logger - INFO - 启动任务加载完成。
 2025-12-31 10:05:48,090 - main.py - task_logger - INFO - 程序已启动...
+2026-01-04 10:29:25,326 - utils.py - task_logger - INFO - 任务队列已从磁盘加载。
+2026-01-04 10:29:26,143 - api.py - task_logger - INFO - 获取了35条数据
+2026-01-04 10:29:26,157 - sample_cloud_modules.py - task_logger - INFO - 任务已从云端获取并保存到 tasks.csv 文件。
+2026-01-04 10:29:26,157 - main.py - task_logger - INFO - 任务列表已保存到 csv 文件中。
+2026-01-04 10:29:26,157 - utils.py - task_logger - INFO - 启动时加载并执行任务...
+2026-01-04 10:29:26,158 - utils.py - task_logger - INFO - 任务已从磁盘加载到全局任务字典。
+2026-01-04 10:29:26,159 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,159 - utils.py - task_logger - INFO - 任务 NGV新增数据 状态已更新为 过期。
+2026-01-04 10:29:26,160 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,160 - utils.py - task_logger - INFO - 任务 新签客户回访 状态已更新为 过期。
+2026-01-04 10:29:26,160 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,161 - utils.py - task_logger - INFO - 任务 续约客户回访 状态已更新为 过期。
+2026-01-04 10:29:26,161 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,161 - utils.py - task_logger - INFO - 任务 接车宝日常派发 状态已更新为 过期。
+2026-01-04 10:29:26,162 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,162 - utils.py - task_logger - INFO - 任务 私域小程序数据支撑 状态已更新为 过期。
+2026-01-04 10:29:26,163 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,163 - utils.py - task_logger - INFO - 任务 小六提成数据支撑 状态已更新为 过期。
+2026-01-04 10:29:26,163 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,163 - utils.py - task_logger - INFO - 任务 异业合作数据支撑 状态已更新为 过期。
+2026-01-04 10:29:26,164 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,164 - utils.py - task_logger - INFO - 任务 短信数据支撑 状态已更新为 过期。
+2026-01-04 10:29:26,165 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,165 - utils.py - task_logger - INFO - 任务 海外邮件推送 状态已更新为 过期。
+2026-01-04 10:29:26,166 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,166 - utils.py - task_logger - INFO - 任务 异常服务待办派发 状态已更新为 过期。
+2026-01-04 10:29:26,166 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,166 - utils.py - task_logger - INFO - 任务 简道云海外项目CRM客户档案迁移BI 状态已更新为 过期。
+2026-01-04 10:29:26,167 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,167 - utils.py - task_logger - INFO - 任务 安装服务历史派发 状态已更新为 过期。
+2026-01-04 10:29:26,168 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,168 - utils.py - task_logger - INFO - 任务 分母报备调整 状态已更新为 过期。
+2026-01-04 10:29:26,169 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,169 - utils.py - task_logger - INFO - 任务 分子报备调整 状态已更新为 过期。
+2026-01-04 10:29:26,169 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,169 - utils.py - task_logger - INFO - 任务 履约表数据支撑 状态已更新为 过期。
+2026-01-04 10:29:26,170 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,170 - utils.py - task_logger - INFO - 任务 字段监控 状态已更新为 过期。
+2026-01-04 10:29:26,171 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,171 - utils.py - task_logger - INFO - 任务 经销商新签服务单转BI 状态已更新为 过期。
+2026-01-04 10:29:26,171 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,171 - utils.py - task_logger - INFO - 任务 高德匹配手机号 状态已更新为 过期。
+2026-01-04 10:29:26,172 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,172 - utils.py - task_logger - INFO - 任务 省市区人员关系表转BI 状态已更新为 过期。
+2026-01-04 10:29:26,173 - utils.py - task_logger - INFO - 所有任务状态已保存到磁盘。
+2026-01-04 10:29:26,173 - utils.py - task_logger - INFO - 任务 续约回访待办 状态已更新为 过期。
+2026-01-04 10:29:26,173 - utils.py - task_logger - INFO - 启动任务加载完成。
+2026-01-04 10:29:26,173 - main.py - task_logger - INFO - 程序已启动...

task_executor.py

@@ -37,7 +37,6 @@ def execute_task(task_id) -> bool:
         "分母报备调整": Module.update_molecule_reporting_adjustment_to_bi,
         "履约表数据支撑": Module.import_performance_data,
         "字段监控": Module.data_monitor,
-        "测试3": Module.text3,
         "经销商新签服务单转BI": Module.new_dealer_service_order_to_bi,
         "合伙人结算登记同步到BI": Module.partner_settlement_to_BI,
         "非标业绩提报转BI": Module.non_standar_performance_to_BI,

tasks.csv

@@ -29,3 +29,4 @@ NGV更新数据,12:30,True,待执行
 合伙人结算登记同步到BI,17:02,True,待执行
 高德匹配手机号,05:00,True,过期
 省市区人员关系表转BI,08:00,True,过期
+续约回访待办,09:35,True,过期

test/续约待办宜搭传给简道云.py

@@ -572,10 +572,35 @@ class YDToJDYRenewalToDo(object):
                 print(f"获取流程实例数据时出错: {e}")
                 continue
 
-        df = pd.DataFrame(all_process_list)
+        df_current = pd.DataFrame(all_process_list)
-        df.to_csv(f"{output_dir}/{start_time}_{end_time}_all_process_list.csv", index=False)
+        current_file = f"{output_dir}/{start_time}_{end_time}_all_process_list.csv"
 
-        return all_process_list
+        # === 新增：读取上次文件并计算差值 ===
+        if os.path.exists(current_file):
+            try:
+                df_last = pd.read_csv(current_file)
+            except Exception as e:
+                print(f"读取历史文件失败: {e}")
+                df_last = pd.DataFrame()
+        else:
+            df_last = pd.DataFrame()
+
+        # 假设唯一标识字段为 'instanceId'，请根据实际字段名调整
+        id_col = 'processInstanceId'
+        if not df_last.empty and not df_current.empty and id_col in df_current.columns and id_col in df_last.columns:
+            # 转为字符串确保可比
+            last_ids = set(df_last[id_col].astype(str))
+            current_ids = set(df_current[id_col].astype(str))
+            new_ids = current_ids - last_ids
+            diff_records = df_current[df_current[id_col].astype(str).isin(new_ids)].to_dict('records')
+        else:
+            # 没有历史文件 或 无唯一ID → 返回空 list（按你要求）
+            diff_records = []
+
+        # 保存当前全量数据（覆盖）
+        df_current.to_csv(current_file, index=False)
+
+        return diff_records
 
     def filter_renewal_data(self, all_process_list):
         update_data_list = []
@@ -680,10 +705,13 @@ class YDToJDYRenewalToDo(object):
         try:
             # step1 获取简道云与宜搭数据
             jd_ydy_data = self.load_all_data()
-            # step2 过滤已经续约的单子
+            if jd_ydy_data:
-            update_data_list = self.filter_renewal_data(jd_ydy_data)
+                # step2 过滤已经续约的单子
-            # step3 校验简道云是否有进行中的单子并关闭
+                update_data_list = self.filter_renewal_data(jd_ydy_data)
-            self.check_jd_ydy_data(update_data_list)
+                # step3 校验简道云是否有进行中的单子并关闭
+                self.check_jd_ydy_data(update_data_list)
+            else:
+                print("本次执行无处理数据")
         except Exception as e:
             print(e)
 

utils.py

@@ -15,7 +15,7 @@ logger = configure_task_logger()
 # 获取错误任务日志记录器
 error_task_logger = configure_error_task_logger()
 
-# 全局任务字典，使用 unique_id 作为键
+# 全局任务字典，使用 (unique_id, exec_time) 作为组合键，支持一个任务多个执行时间
 all_tasks = {}
 
 
@@ -86,14 +86,14 @@ class PersistentQueue:
         """将任务加入队列并保存到磁盘"""
         self.queue.put(task)
         self.save_to_disk()
-        logger.info(f"任务 {task['unique_id']} 已加入队列。")
+        logger.info(f"任务 {task['unique_id']} ({task['exec_time']}) 已加入队列。")
 
     def get(self):
         """从队列中获取任务并保存到磁盘"""
         if not self.queue.empty():
             task = self.queue.get()
             self.save_to_disk()
-            logger.info(f"任务 {task['unique_id']} 已从队列中取出。")
+            logger.info(f"任务 {task['unique_id']} ({task['exec_time']}) 已从队列中取出。")
             return task
         else:
             logger.info("任务队列为空。")
@@ -120,19 +120,20 @@ task_queue = PersistentQueue()
 def load_tasks_and_execute():
     """
     从 tasks.csv 文件中加载任务到持久化队列中，仅包含待执行任务。
-    避免重复添加任务。
+    支持一个任务多个执行时间，避免重复添加任务。
     """
     try:
         load_tasks_from_csv()  # 从 CSV 文件加载所有任务
 
         now = datetime.now()
-        # 获取队列中已有任务ID（避免重复添加）
+        # 获取队列中已有任务标识（使用 (unique_id, exec_time) 避免重复添加）
-        existing_task_ids = set()
+        existing_task_keys = set()
         # 创建一个临时队列来遍历任务
         temp_queue = Queue()
         while not task_queue.empty():
             task = task_queue.get()
-            existing_task_ids.add(task['unique_id'])
+            task_key = (task['unique_id'], task['exec_time'])
+            existing_task_keys.add(task_key)
             temp_queue.put(task)
         # 将任务放回原队列
         while not temp_queue.empty():
@@ -144,18 +145,21 @@ def load_tasks_and_execute():
 
             if (now - exec_time_today) > timedelta(minutes=5):
                 if task['status'] == '待执行':
-                    error_task_logger.error(f"任务 {task['unique_id']} 超过执行窗口5分钟以上,标记为过期。")
+                    error_task_logger.error(f"任务 {task['unique_id']} ({task['exec_time']}) 超过执行窗口5分钟以上,标记为过期。")
-                    update_task_status(task['unique_id'], '过期')
+                    update_task_status(task['unique_id'], task['exec_time'], '过期')
                 continue  # 不处理过期任务
 
+            # 使用组合键判断任务是否已在队列中
+            task_key = (task['unique_id'], task['exec_time'])
+            
             # 如果任务应该执行、状态是待执行且不在队列中
             if (task['should_execute'](now) and
                     task['status'] == '待执行' and
-                    task['unique_id'] not in existing_task_ids):
+                    task_key not in existing_task_keys):
 
                 task_queue.put(task)
-                existing_task_ids.add(task['unique_id'])
+                existing_task_keys.add(task_key)
-                logger.info(f"任务 {task['unique_id']} 加入队列。")
+                logger.info(f"任务 {task['unique_id']} ({task['exec_time']}) 加入队列。")
 
     except Exception as e:
         error_task_logger.error(f"加载任务时出错: {e}")
@@ -165,26 +169,26 @@ def load_tasks_and_execute():
 
 
 def execute_single_task():
-    """从队列中取出并执行单个任务"""
+    """从队列中取出并执行单个任务，支持一个任务多个执行时间"""
     if not task_queue.empty():
         try:
             task = task_queue.get()
             if task:
                 # 标记任务为正在执行
-                update_task_status(task['unique_id'], '正在执行')
+                update_task_status(task['unique_id'], task['exec_time'], '正在执行')
 
                 try:
                     success = execute_task(task['unique_id'])
                     if success:
-                        update_task_status(task['unique_id'], '已完成')
+                        update_task_status(task['unique_id'], task['exec_time'], '已完成')
-                        logger.info(f"任务 {task['unique_id']} 执行成功。")
+                        logger.info(f"任务 {task['unique_id']} ({task['exec_time']}) 执行成功。")
                     else:
-                        update_task_status(task['unique_id'], '失败')
+                        update_task_status(task['unique_id'], task['exec_time'], '失败')
-                        error_task_logger.error(f"任务 {task['unique_id']} 执行失败。")
+                        error_task_logger.error(f"任务 {task['unique_id']} ({task['exec_time']}) 执行失败。")
 
                 except Exception as e:
-                    error_task_logger.error(f"任务 {task['unique_id']} 执行时发生异常: {e}")
+                    error_task_logger.error(f"任务 {task['unique_id']} ({task['exec_time']}) 执行时发生异常: {e}")
-                    update_task_status(task['unique_id'], '失败')
+                    update_task_status(task['unique_id'], task['exec_time'], '失败')
 
                 task_queue.task_done()
 
@@ -193,7 +197,7 @@ def execute_single_task():
 
 
 def load_tasks_from_csv():
-    """从 tasks.csv 文件中加载任务到全局任务字典"""
+    """从 tasks.csv 文件中加载任务到全局任务字典，支持一个任务多个执行时间"""
     try:
         csv_file = Path('tasks.csv')
         if csv_file.is_file():
@@ -212,7 +216,9 @@ def load_tasks_from_csv():
                     exec_time = datetime.strptime(task['exec_time'], '%H:%M').time()
                     task['should_execute'] = lambda now, et=exec_time: now.time() >= et
 
-                    all_tasks[task['unique_id']] = task
+                    # 使用 (unique_id, exec_time) 作为组合键，支持一个任务多个执行时间
+                    task_key = (task['unique_id'], task['exec_time'])
+                    all_tasks[task_key] = task
             logger.info("任务已从磁盘加载到全局任务字典。")
         else:
             logger.warning("tasks.csv 文件未找到，初始化为空任务字典。")
@@ -221,11 +227,12 @@ def load_tasks_from_csv():
 
 
 def save_tasks_to_csv():
-    """将所有任务的状态保存到 tasks.csv 文件"""
+    """将所有任务的状态保存到 tasks.csv 文件，支持一个任务多个执行时间"""
     try:
         with open('tasks.csv', 'w', newline='', encoding='utf-8') as f:
             writer = csv.DictWriter(f, fieldnames=['unique_id', 'exec_time', 'is_switch_on', 'status'])
             writer.writeheader()
+            # 遍历所有任务实例（包括相同unique_id的不同exec_time）
             for task in all_tasks.values():
                 task_to_save = {
                     'unique_id': task['unique_id'],
@@ -240,14 +247,15 @@ def save_tasks_to_csv():
         error_task_logger.error(f"保存任务状态时出错: {e}")
 
 
-def update_task_status(task_id, new_status):
+def update_task_status(task_id, exec_time, new_status):
-    """更新指定任务的状态，并保存到磁盘"""
+    """更新指定任务的状态，并保存到磁盘，支持一个任务多个执行时间"""
-    if task_id in all_tasks:
+    task_key = (task_id, exec_time)
-        all_tasks[task_id]['status'] = new_status
+    if task_key in all_tasks:
+        all_tasks[task_key]['status'] = new_status
         save_tasks_to_csv()
-        logger.info(f"任务 {task_id} 状态已更新为 {new_status}。")
+        logger.info(f"任务 {task_id} ({exec_time}) 状态已更新为 {new_status}。")
     else:
-        error_task_logger.error(f"尝试更新不存在的任务 {task_id} 的状态。")
+        error_task_logger.error(f"尝试更新不存在的任务 {task_id} ({exec_time}) 的状态。")
 
 
 def load_tasks_on_start():