豆包助手

aidso_geo/config/base_config.py

@@ -278,10 +278,10 @@ if __name__ == '__main__':
 'XHSA:geo:batch:list',
 'geo:task_commit:list']
     # mt:snipaste_v3:only_content
-    init_redis4_1 = init_redis()
+    init_redis_1 = init_redis()
+    init_redis_1.delete('DB:geo:batch:list')
+    init_redis_1.delete('DB:geo:stream_batch:list')
 
-    init_redis4_1.delete("third_cx_user_id")
-    print(init_redis4_1.scard('third_cx_user_id'))
     # old_count, new_count = deduplicate_redis_list(
     #     redis_client8,
     #     "geo:task_commit:list",

aidso_geo/core/doubao_assistent.py

@@ -227,7 +227,7 @@ if __name__ == "__main__":
     while 1:
         tasks = redis8.rpop(
             "geo:task_commit:JIKE:list",
-            5,
+            1,
         ) or []
 
         if not tasks:
@@ -237,7 +237,7 @@ if __name__ == "__main__":
 
         success_tasks = []
         success_original_tasks = []
-        with ThreadPoolExecutor(max_workers=5) as executor:
+        with ThreadPoolExecutor(max_workers=1) as executor:
             future_task_map = {
                 executor.submit(process_task, task): task
                 for task in tasks

aidso_geo/core/meituan_process.py

@@ -85,36 +85,51 @@ where t2.reqId is null
         logger.info("task_commit_api 没有待提交任务")
         return
 
-    values = []
+    # 普通任务
+    normal_values = []
 
+    channel_one_values = []
     for task_data in query_queue:
         payload = {
             key: task_data.get(key)
             for key in KEEP_FIELDS
         }
 
-        values.append(
-            json.dumps(
-                payload,
-                ensure_ascii=False
-            )
-        )
+        value = json.dumps(payload, ensure_ascii=False)
+
+        if task_data.get("platform") =="DB":
+            channel_one_values.append(value)
+        else:
+            normal_values.append(value)
 
         # 修改数据库记录状态
         task_data["status"] = "PROCESSING"
 
     redis_client8 = init_redis8()
-    redis_client8.lpush(
+    pipeline = redis_client8.pipeline(transaction=True)
+
+    if normal_values:
+        pipeline.lpush(
             "geo:task_commit:list",
-        *values,
+            *normal_values
         )
+
+    if channel_one_values:
+        pipeline.lpush(
+            "geo:task_commit:JIKE:list",
+            *channel_one_values
+        )
+        logger.info(
+            f"令狐豆包 提交任务数量：{len(channel_one_values)}")
+
+    pipeline.execute()
     bh_utils.insert_data(
         "geo_third_task_data",
         query_queue,
     )
 
     logger.info(
-        f"task_commit_api 提交任务数量：{len(values)}")
+        f"task_commit_api 提交任务数量：{len(normal_values)}")
 
 def get_result_api():
     try:
@@ -150,6 +165,77 @@ def get_result_api():
     except Exception as e:
         logger.error(e)
 
+
+def task_commit_api1():
+    query_queue = bh_utils.query_data(
+        """
+select t1.reqId,
+       t1.prompt,
+       t1.taskId,
+       t1.platform,
+       t1.type,
+       t1.insertime,
+       t1.status,
+       t1.thinkingEnabled,
+       t1.channel
+from (SELECT *
+      FROM geo_third_task_data
+      where platform = 'DB' ) t1
+         left join (select * from geo_commit_task where pt > date_format(date_sub(now(), 10), '%Y%m%d')) t2
+                   on t1.reqId = t2.reqId  where t2.status ='ING'
+        """
+    )
+
+    if not query_queue:
+        logger.info("task_commit_api 没有待提交任务")
+        return
+
+    # 普通任务
+    normal_values = []
+
+    channel_one_values = []
+    for task_data in query_queue:
+        payload = {
+            key: task_data.get(key)
+            for key in KEEP_FIELDS
+        }
+
+        value = json.dumps(payload, ensure_ascii=False)
+
+        if task_data.get("platform") =="DB":
+            channel_one_values.append(value)
+        else:
+            normal_values.append(value)
+
+        # 修改数据库记录状态
+        task_data["status"] = "PROCESSING"
+
+    redis_client8 = init_redis8()
+    pipeline = redis_client8.pipeline(transaction=True)
+
+    if normal_values:
+        pipeline.lpush(
+            "geo:task_commit:list",
+            *normal_values
+        )
+
+    if channel_one_values:
+        pipeline.lpush(
+            "geo:task_commit:JIKE:list",
+            *channel_one_values
+        )
+        logger.info(
+            f"令狐豆包 提交任务数量：{len(channel_one_values)}")
+
+    pipeline.execute()
+    bh_utils.insert_data(
+        "geo_third_task_data",
+        query_queue,
+    )
+
+    logger.info(
+        f"task_commit_api 提交任务数量：{len(normal_values)}")
+
 if __name__ == "__main__":
     logger.info("调度 启动")
     scheduler = BlockingScheduler(timezone="Asia/Shanghai")
@@ -178,4 +264,3 @@ if __name__ == "__main__":
 
 
 
-

aidso_geo/core/routes/dashboard.py

@@ -10,12 +10,13 @@ from flask import (
     session,
     url_for,
 )
-from aidso_geo.config.base_config import init_redis
+from aidso_geo.config.base_config import init_redis, init_redis8
 from aidso_geo.utils import bh_utils
 
 dashboard_app = Blueprint("dashboard", __name__)
 
 redis_client = init_redis()
+redis8_client = init_redis8()
 
 DASHBOARD_USERNAME = "aidso"
 DASHBOARD_PASSWORD = "aidso666"
@@ -94,7 +95,10 @@ def queue_status():
         data[stream_key] = redis_client.llen(stream_key)
         data[batch_key] = redis_client.llen(batch_key)
 
-    data["geo:task_commit:list"] = redis_client.llen("geo:task_commit:list")
+    data["geo:task_commit:list"] = redis8_client.llen("geo:task_commit:list")
+    data["geo:task_commit:JIKE:list"] = redis8_client.llen(
+        "geo:task_commit:JIKE:list"
+    )
 
     return jsonify(data)
 

aidso_geo/core/static/dashboard/queue_monitor.css

@@ -127,7 +127,7 @@
 
         .summary {
             display: grid;
-            grid-template-columns: repeat(4, 1fr);
+            grid-template-columns: repeat(5, 1fr);
             gap: 16px;
             margin-bottom: 24px;
         }

aidso_geo/core/static/dashboard/queue_monitor.js

@@ -158,6 +158,9 @@ const API_URL = "/api/queue/status";
         document.getElementById("streamBatchCount").innerText = streamBatchCount;
         document.getElementById("batchCount").innerText = batchCount;
         document.getElementById("commitCount").innerText = Number(data["geo:task_commit:list"] || 0);
+        document.getElementById("jikeCommitCount").innerText = Number(
+            data["geo:task_commit:JIKE:list"] || 0
+        );
         document.getElementById("lastTime").innerText = "最后刷新：" + new Date().toLocaleString();
     }
 

aidso_geo/core/templates/queue_monitor.html

@@ -62,6 +62,10 @@
                     <div class="card-name">提交队列</div>
                     <div class="card-value" id="commitCount">0</div>
                 </div>
+                <div class="card">
+                    <div class="card-name">三方豆包</div>
+                    <div class="card-value" id="jikeCommitCount">0</div>
+                </div>
             </div>
 
             <div class="table-wrap">

aidso_geo/models/process.py

@@ -2019,7 +2019,7 @@ if __name__ == '__main__':
 
     # data_list = bh_utils.query_data("select * from geo_commit_task where platform = 'DP' and thinking_enabled = 1 and insertime > 1777564800 and type!='success' order by  insertime asc")
 
-    data_list = bh_utils.query_data("select * from geo_commit_task where  reqId in ('fd51b424-e5d0-4423-837b-e56fe8d4fa0f')")
+    data_list = bh_utils.query_data("select * from geo_commit_task where  status= 'ING' and channel is null")
     # data_list = bh_utils.query_data("select * from geo_commit_task where prompt = '飞鹤和君乐宝奶粉的异同点对比' and platform = 'DB'")
     # data_list = bh_utils.query_data("select * from geo_commit_task where pt = '20260720' and platform = 'TYQW' and type = 'success' ")
 
@@ -2032,7 +2032,7 @@ if __name__ == '__main__':
         i["productWordsMap"] = safe_json_loads(i.get("productWordsMap"), [])
         type_t = i.get('type')
         type_t = 'stream_batch'
-        commit_task(i,'ING')
+        # commit_task(i,'ING')
         return task_send_queue(i,type_t)
         # return deepseek_data_process.deepseek_process_original_data(i)
         # return platform_process(i)