feat:添加日志

client/AIGCClient.py

@@ -45,6 +45,17 @@ class AIGCClient(object):
         url = f"{self.base_url}/aigc/crawler/plan/save"
         return self.post(url, params)
 
+    def get_produce_plan_by_id(self, produce_plan_id: str) -> dict:
+        url = f"{self.base_url}/aigc/produce/plan/detail"
+        payload = {
+            "id": produce_plan_id,
+        }
+        return self.post(url, params=payload)['data']
+
+    def save_produce_plan(self, produce_plan: dict) -> dict:
+        url = f"{self.base_url}/aigc/produce/plan/save"
+        return self.post(url, params=produce_plan)
+
     def post(self, url: str, params: dict) -> dict:
         request_param = {
             "params": params,

enums/automation_job.py

@@ -15,7 +15,6 @@ class AutomationJobCronInfo(Enum):
     account_tencent_huxuan = ("腾讯互选平台账号供给（其他）", "account_tencent_huxuan", "other", 9)
     workflow_schedule = ("基于需求的供给workflow", "workflow_schedule", "other", 6)
     workflow_schedule_history_homochronous = ("基于需求的供给workflow（历史同期）", "workflow_schedule", "history_homochronous", 9)
-    workflow_schedule_zhu_yue = ("基于需求的供给workflow（逐月）", "workflow_schedule", "zhu_yue", 9)
 
     def __init__(self, task_name, crawler_mode, video_source, task_start_hour):
         self.task_name = task_name

monitor/supply_workflow_monitor.py

@@ -302,58 +302,66 @@ def crawler_and_produce_stat_query(ts: int, workflow_id: str) -> List[Dict[str,
     return mysql_helper.execute_query(sql)
 
 
-def main():
-    global header_template_index
+def workflow_monitor(workflow_id: str, workflow_name: str, status: int):
     today_midnight = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
 
     start_dt_str = today_midnight.strftime('%Y-%m-%d %H:%M:%S')
     end_dt_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
     timestamp_ms = int(today_midnight.timestamp() * 1000)
+
+    task_status_stat = workflow_task_status_stat(timestamp_ms, workflow_id)
+
+    exe_step_stat = task_exe_step_stat_query(timestamp_ms, workflow_id)
+    status_stat = crawler_and_produce_stat_query(timestamp_ms, workflow_id)
+
+    # 构建标题
+
+    text_tag_list = []
+    if status == 0:
+        text_tag_list.append(build_text_tag_json("已关闭", "red"))
+    elif status == 1:
+        text_tag_list.append(build_text_tag_json("开启中", "green"))
+    else:
+        text_tag_list.append(build_text_tag_json("未知", "orange"))
+
+    template = header_template[header_template_index % len(header_template)]
+    sub_title = build_sub_title_json(f"{start_dt_str} - {end_dt_str}")
+    header = build_header_json(f"【workflow策略】{workflow_name}", template, sub_title, text_tag_list)
+
+    # 构建下文内容
+    task_status_df = pd.DataFrame(task_status_stat)
+    task_info = f"**总任务数**: {task_status_df['任务数'].sum()}" + \
+                "".join([
+                    f"\n**{item['任务状态']}任务数**: {item['任务数']}"
+                    for item in task_status_df.to_dict(orient='records')
+                ])
+
+    elements = [
+        build_markdown_element_json(task_info),
+        build_hr_element_json(),
+        build_table_element_json(pd.DataFrame(exe_step_stat)),
+        build_hr_element_json(),
+        build_table_element_json(pd.DataFrame(status_stat)),
+    ]
+
+    config = build_config_json()
+
+    card_json = build_card_json(elements, header, config)
+
+    feishu_inform_util.send_card_msg_to_feishu(
+        webhook=fei_shu_webhook,
+        card_json=card_json
+    )
+
+
+def main():
+    global header_template_index
     workflows = mysql_helper.execute_query('select id, name, status from supply_workflow;')
     for workflow in workflows:
-        task_status_stat = workflow_task_status_stat(timestamp_ms, workflow['id'])
-
-        exe_step_stat = task_exe_step_stat_query(timestamp_ms, workflow['id'])
-        status_stat = crawler_and_produce_stat_query(timestamp_ms, workflow['id'])
-
-        # 构建标题
-
-        text_tag_list = []
-        if workflow['status'] == 0:
-            text_tag_list.append(build_text_tag_json("已关闭", "red"))
-        elif workflow['status'] == 1:
-            text_tag_list.append(build_text_tag_json("开启中", "green"))
-        else:
-            text_tag_list.append(build_text_tag_json("未知", "orange"))
-
-        template = header_template[header_template_index % len(header_template)]
-        sub_title = build_sub_title_json(f"{start_dt_str} - {end_dt_str}")
-        header = build_header_json(f"【workflow策略】{workflow['name']}", template, sub_title, text_tag_list)
-
-        # 构建下文内容
-        task_status_df = pd.DataFrame(task_status_stat)
-        task_info = f"**总任务数**: {task_status_df['任务数'].sum()}" + \
-                    "".join([
-                        f"\n**{item['任务状态']}任务数**: {item['任务数']}"
-                        for item in task_status_df.to_dict(orient='records')
-                    ])
-
-        elements = [
-            build_markdown_element_json(task_info),
-            build_hr_element_json(),
-            build_table_element_json(pd.DataFrame(exe_step_stat)),
-            build_hr_element_json(),
-            build_table_element_json(pd.DataFrame(status_stat)),
-        ]
-
-        config = build_config_json()
-
-        card_json = build_card_json(elements, header, config)
-
-        feishu_inform_util.send_card_msg_to_feishu(
-            webhook=fei_shu_webhook,
-            card_json=card_json
-        )
+        try:
+            workflow_monitor(workflow['id'], workflow['name'], workflow['status'])
+        except Exception as e:
+            print(f"【workflow策略】{workflow['name']} 监控异常: {e}")
 
 
 if __name__ == '__main__':

script/clear_produce_plan_input_source.py

@@ -0,0 +1,30 @@
+from client.AIGCClient import AIGCClient
+
+aigc_client = AIGCClient(token="8bf14f27fc3a486788f3383452422d72", base_url="https://aigc-api.aiddit.com")
+
+
+def main():
+    produce_plan_ids = [
+        "20260509113208212653909",
+        "20260509113221856771441",
+        "20260509113329478111972",
+        "20260509113453897708339",
+        "20260509113509961697272",
+        "20260509113531215757335",
+        "20260509113756891904084",
+        "20260509113817127161918",
+        "20260509113833884726698",
+        "20260509113937229233085"
+    ]
+    for produce_plan_id in produce_plan_ids:
+        # produce_plan = aigc_client.get_produce_plan_by_id(produce_plan_id)
+        # inputSources = produce_plan['inputSourceGroups'][0]['inputSources']
+        # produce_plan['inputSourceGroups'][0]['inputSources'] = [inputSources[0]]
+
+        # aigc_client.save_produce_plan(produce_plan)
+
+        print(produce_plan_id)
+
+
+if __name__ == '__main__':
+    main()