feat:添加重置生成计划状态的逻辑

client/AIGCClient.py

@@ -0,0 +1,29 @@
+import json
+
+import requests
+
+
+class AIGCClient(object):
+    def __init__(self, token, base_url):
+        self.token = token
+        self.base_url = base_url
+
+    def update_produce_plan_status(self, produce_plan_id: str, status: int):
+        params = {
+            "id": produce_plan_id,
+            "status": status,
+        }
+        url = f"{self.base_url}/aigc/produce/plan/updatePlanStatus"
+        self.post(url, params)
+
+    def post(self, url: str, params: dict):
+        request_param = {
+            "params": params,
+            "baseInfo": {
+                "token": self.token,
+            }
+        }
+        print(f"invoke aigc platform url: {url} request: {json.dumps(request_param)}")
+        response = requests.post(url, json=request_param)
+        print(f"invoke aigc platform url: {url} request: {json.dumps(request_param)} response: {response.json()}")
+        return response.json()

client/ApolloClient.py

@@ -5,8 +5,14 @@ class ApolloClient(object):
     def __init__(self, base_url):
         self.base_url = base_url
 
-    def get_config(self, app_id: str, cluster: str, namespace: str):
+    def get_config(self, app_id: str, cluster: str = "default", namespace: str = "application"):
         url = f"{self.base_url}/configs/{app_id}/{cluster}/{namespace}"
         print(f"请求的Apollo地址为: {url}")
         response = requests.get(url)
         return response.json()
+
+    def get_value(self, app_id: str, cluster: str = "default", namespace: str = "application", key: str = None):
+        if not key:
+            raise ValueError("Key not empty")
+        config_json = self.get_config(app_id=app_id, cluster=cluster, namespace=namespace)
+        return config_json['configurations'][key]

monitor/feature_spark_monitor.py

@@ -193,8 +193,8 @@ def _main():
         print("未获取到已完成的任务，跳过")
         return
 
-    j = apollo.get_config("recommend-feature", "default", "application")
-    dts_config = json_util.remove_comments(j['configurations']['dts.config'])
+    j = apollo.get_value("recommend-feature", "default", "application", "dts.config")
+    dts_config = json_util.remove_comments(j)
     table_config_list = json.loads(dts_config)
     for table_config in table_config_list:
         print("\n\n\n")

script/aigc_produce_plan_refresh.py

@@ -0,0 +1,26 @@
+import json
+
+from client.AIGCClient import AIGCClient
+from client.ApolloClient import ApolloClient
+from util import json_util
+
+client = ApolloClient("http://apolloconfig-internal.piaoquantv.com")
+aigc_client = AIGCClient(token="8bf14f27fc3a486788f3383452422d72", base_url="https://aigc-api.aiddit.com")
+
+
+def main():
+    config_str = client.get_value(
+        app_id="crawler-scheduler",
+        cluster="default",
+        namespace="application",
+        key="hot.video.keywords.provide.task.config"
+    )
+    config_json = json.loads(json_util.remove_comments(config_str))
+    for merge_cate2 in config_json['mergeCate2PatternPlanIdMap']:
+        for mode in config_json['mergeCate2PatternPlanIdMap'][merge_cate2]:
+            for produce_plan_id in config_json['mergeCate2PatternPlanIdMap'][merge_cate2][mode]:
+                aigc_client.update_produce_plan_status(produce_plan_id, 0)
+
+
+if __name__ == '__main__':
+    main()