三方任务调度更新

电商搜索功能

aidso_geo/config/base_config.py

@@ -174,7 +174,19 @@ def get_user_info(secUid):
 
     print(response.text)
 
-
+def init_redis4():
+    try:
+        redis_client = redis.Redis(
+            host='172.16.0.24',
+            port=6379,
+            db=4,
+            password='aiyingli@@123',
+            socket_timeout=5,
+            decode_responses=True  # 自动解码为字符串，避免bytes类型问题
+        )
+        return redis_client
+    except Exception as e:
+        return None
 
 if __name__ == '__main__':
     key_list = ['BDAI:geo:stream_batch:list',
@@ -205,11 +217,11 @@ if __name__ == '__main__':
 'XHSA:geo:batch:list',
 'geo:task_commit:list']
 
-    redis5 = init_redis()
+    redis5 = init_redis4()
     # delete(redis_key)
-
-    print(redis5.delete('mt:snipaste_v3:only_content'))
-    print(redis5.delete('mt:snipaste_v3:with_share'))
+    print(redis5.delete("mt:DP:snipaste_v3:only_content"))
+    #     print(redis5.delete('mt:snipaste_v3:only_content'))
+    # print(redis5.delete('mt:snipaste_v3:with_share'))
 
 
 

aidso_geo/core/feishu_snipaste_v3.py

@@ -1968,6 +1968,7 @@ def export_share_rounds_to_excel(
             all_keywords.append(keyword)
 
     keyword_totals_by_task = {}
+    recommended_rounds_by_task = {}
     seen_task_rounds = set()
     for round_row in round_keyword_rows:
         task_id = str(round_row.get("taskId") or "")
@@ -1995,6 +1996,7 @@ def export_share_rounds_to_excel(
             task_id,
             {},
         )
+        has_recommended_keyword = False
         for keyword_item in keywords_count:
             if not isinstance(keyword_item, dict):
                 continue
@@ -2013,6 +2015,14 @@ def export_share_rounds_to_excel(
                 task_keyword_totals.get(keyword, 0)
                 + int(word_count > 0)
             )
+            if keyword != "美团" and word_count > 0:
+                has_recommended_keyword = True
+
+        if has_recommended_keyword:
+            recommended_rounds_by_task.setdefault(
+                task_id,
+                set(),
+            ).add(count)
 
     if output_path is None:
         formatted_date = format_pt_display(
@@ -2042,7 +2052,8 @@ def export_share_rounds_to_excel(
             "问题",
             "是否达标",
             "达标轮次",
-            "达标率",
+            "提及率",
+            "推荐率",
             "本次是否达标",
             *all_keywords,
             "平台",
@@ -2079,6 +2090,10 @@ def export_share_rounds_to_excel(
         hit_round_count = row.get("hit_round_count")
         hit_round_count_value = int(hit_round_count or 0)
         hit_rate = hit_round_count_value / 30
+        recommendation_rate = (
+            len(recommended_rounds_by_task.get(task_id, set()))
+            / 30
+        )
         if not processed_round_count:
             qualified_text = "没处理"
         elif hit_round_count_value >= 24:
@@ -2138,6 +2153,7 @@ def export_share_rounds_to_excel(
                 qualified_text,
                 hit_round_count_value,
                 hit_rate,
+                recommendation_rate,
                 current_qualified_text,
                 *keyword_total_values,
                 PLATFORM_CONFIGS.get(
@@ -2160,8 +2176,9 @@ def export_share_rounds_to_excel(
             ]
         )
 
-    for cell in worksheet["D"][1:]:
-        cell.number_format = "0.00%"
+    for column_name in ("D", "E"):
+        for cell in worksheet[column_name][1:]:
+            cell.number_format = "0.00%"
 
     workbook.save(output_path)
     workbook.close()
@@ -3111,4 +3128,9 @@ def run_daily_pipeline_safely(
 
 
 if __name__ == "__main__":
+    # pt = '20260803'
+    # platform_db = 'DB'
+    # platform_dp = 'DP'
+    # run_daily_pipeline(pt=pt, platform=platform_db)
+    # run_daily_pipeline(pt=pt, platform=platform_dp)
     start_scheduler()

aidso_geo/core/meituan_process.py

-import requests
+
 from loguru import logger
 import os
 import sys
+import json
 from apscheduler.schedulers.blocking import BlockingScheduler
+import redis
+
+
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 sys.path.append(BASE_DIR)
-
 from aidso_geo.utils import bh_utils
 
+
+_redis8_client = None
+_redis8_pool = None
+def init_redis8():
+    global _redis8_client, _redis8_pool
+
+    if _redis8_client is not None:
+        return _redis8_client
+
+    try:
+        _redis8_pool = redis.ConnectionPool(
+            host='redis-cnlfmu7rl14awitrz.redis.ivolces.com',
+            port=6379,
+            db=8,
+            password='aiyingli@@123',
+            socket_timeout=120,
+            socket_connect_timeout=5,
+            decode_responses=True,
+            retry_on_timeout=True,
+            health_check_interval=30,
+            socket_keepalive=True,
+            max_connections=100,
+        )
+
+        _redis8_client = redis.Redis(connection_pool=_redis8_pool)
+        _redis8_client.ping()
+        return _redis8_client
+    except Exception as e:
+        logger.error(f"Redis 初始化失败: {e}")
+        _redis8_client = None
+        _redis8_pool = None
+        return None
+
+
+
+
 KEEP_FIELDS = (
     "prompt",
     "taskId",
@@ -20,95 +59,115 @@ KEEP_FIELDS = (
 )
 
 
-def task_commit_api(task_data):
-    req_id = task_data.get("reqId",'')
-    prompt = task_data.get("prompt",'')
-    platform = task_data.get("platform",'')
-    channel = task_data.get("channel") or ""
-    task_type = task_data.get("type",'')
-    url = "http://172.16.1.223:8086/api/geo/task_commit"
-
-    payload = {k: task_data.get(k) for k in KEEP_FIELDS }
-
-    try:
-        response = requests.post(url, json=payload)
-        response_data = response.json()
-
-        if response_data.get("code") == 200:
-            task_data["status"] = "PROCESSING"
-            bh_utils.insert_data("geo_third_task_data", [task_data])
-            logger.success(f'{req_id}--{prompt}--{platform}--{channel}--{task_type}--提交成功')
-    except Exception as e:
-        logger.error(e)
+def task_commit_api():
+    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,
+ifnull(t1.is_share,0) as is_share
+from (SELECT *
+      FROM geo_third_task_data
+      where status != 'SUCCESS') 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.reqId is null
+        """
+    )
 
+    if not query_queue:
+        logger.info("task_commit_api 没有待提交任务")
+        return
 
-def get_result_api(task_data):
-    req_id = task_data.get("reqId",'')
-    prompt = task_data.get("prompt",'')
-    platform = task_data.get("platform",'')
-    channel = task_data.get("channel") or ""
-    task_type = task_data.get("type",'')
+    values = []
 
-    url = f"http://172.16.1.223:8086/api/geo/task_check?reqId={req_id}"
+    for task_data in query_queue:
+        payload = {
+            key: task_data.get(key)
+            for key in KEEP_FIELDS
+        }
 
-    try:
-        response = requests.get(url)
-        response_data = response.json()
-        data = response_data.get("data") or {}
-
-        if response_data.get("code") == 200 and data.get("status") == "success":
-            task_data["status"] = "SUCCESS"
-            bh_utils.insert_data("geo_third_task_data", [task_data])
-            logger.success(f'{req_id}--{prompt}--{platform}--{channel}--{task_type}--获取成功')
-    except Exception as e:
-        logger.error(e)
+        values.append(
+            json.dumps(
+                payload,
+                ensure_ascii=False
+            )
+        )
 
+        # 修改数据库记录状态
+        task_data["status"] = "PROCESSING"
 
-def query_task_commit():
-    return bh_utils.query_data(
-        "select * from geo_third_task_data where status = 'ING'"
+    redis_client8 = init_redis8()
+    redis_client8.lpush(
+        "geo:task_commit:list",
+        *values,
     )
-
-
-
-def query_task_check():
-    return bh_utils.query_data(
-        "select * from geo_third_task_data where status = 'PROCESSING'"
+    bh_utils.insert_data(
+        "geo_third_task_data",
+        query_queue,
     )
 
-def task_commit():
-    commit_tasks = query_task_commit()
-    if commit_tasks:
-        for task in query_task_commit():
-            task_commit_api(task)
-
-def task_check():
-    check_tasks = query_task_check()
-    if check_tasks:
-        for task in query_task_check():
-            get_result_api(task)
-
+    logger.info(
+        f"task_commit_api 提交任务数量：{len(values)}")
 
+def get_result_api():
+    try:
+        query_result = bh_utils.query_data("""
+            select
+        t1.reqId,
+        t1.prompt,
+        t1.taskId,
+        t1.platform,
+        t1.type,
+        t1.insertime,
+        t2.status,
+        t1.thinkingEnabled,
+        t1.channel,
+        t1.is_share
+         from (
+        SELECT *
+                FROM geo_third_task_data
+                WHERE status = 'PROCESSING'
+                LIMIT 500) 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 = 'SUCCESS'
+            """)
+
+        if not query_result:
+            logger.info("get_result_api 没有完成任务")
+            return
+
+        bh_utils.insert_data(
+            "geo_third_task_data",
+            query_result,
+        )
+        logger.info(
+            f"get_result_api 完成任务数量：{len(query_result)}")
+    except Exception as e:
+        logger.error(e)
 
 if __name__ == "__main__":
-
     logger.info("调度 启动")
     scheduler = BlockingScheduler(timezone="Asia/Shanghai")
     scheduler.add_job(
-        task_commit,
+        task_commit_api,
         trigger="interval",
         seconds=30,
-        id='third_task_commit',
+        id='task_commit_api',
         max_instances=1,
         coalesce=True,
         replace_existing=True
     )
 
     scheduler.add_job(
-        task_check,
+        get_result_api,
         trigger="interval",
         seconds=30,
-        id='third_task_check',
+        id='get_result_api',
         max_instances=1,
         coalesce=True,
         replace_existing=True
@@ -116,4 +175,7 @@ if __name__ == "__main__":
     scheduler.start()
     logger.info("third_task_commit 启动")
     logger.info("third_task_check 启动")
-#
\ No newline at end of file
+
+
+
+

aidso_geo/core/routes/third_interface.py

@@ -195,8 +195,8 @@ def gen_authorization():
     return secrets.token_hex(16).upper()
 
 type_map ={
-    '0':"stream",
-    '1':"stream_batch"
+    '0':"stream_batch",
+    '1':"batch"
 }
 
 
@@ -293,8 +293,8 @@ def create_auth():
         if p not in SYSTEM_ALLOW_PLATFORM:
             return err(400, f"allow_platforms[{idx}] '{p}' not allowed")
 
-        if p in seen:
-            return err(400, f"duplicate allow_platforms value: {p}")
+        # if p in seen:
+        #     return err(400, f"duplicate allow_platforms value: {p}")
 
         seen.add(p)
         normalized_platforms.append(p)

aidso_geo/models/doubao_android_data_process.py

@@ -366,7 +366,7 @@ if __name__ == '__main__':
     # file_path = 'geo/c7eb465e-f385-4aa2-89c4-a7cf11897f45/KIMI/1(1).txt'
 
     data_list = bh_utils.query_data(
-        f"select * from geo_commit_task where taskId = '57e07263-9844-4aaf-b455-db2de2e828d8' and platform =  'DOUBA'")
+        f"select * from geo_commit_task where taskId = '9453321c-0c89-4b6c-802a-68970376a7c4' and platform =  'DOUBA'")
 
 
 

aidso_geo/models/douyinai_data_process.py

@@ -31,7 +31,11 @@ def douyin_ai_process_original_data(data):
                     # 提取并解析JSON数据
                     data_str = item.split("data:")[1]
                     json_data = json.loads(data_str)
-
+                    print(json_data)
+                    print('----')
+                    print('----')
+                    print('----')
+                    print('----')
                 except (IndexError, json.JSONDecodeError):
                     continue
 
@@ -119,7 +123,7 @@ def douyin_ai_process_original_data(data):
 
 if __name__ == '__main__':
     data_list = bh_utils.query_data(
-        f"select * from geo_commit_task where taskId = '1cfb92e0-b0ae-40e5-b198-342d079dea0a' and platform = 'DYAI'")
+        f"select * from geo_commit_task where taskId = '22f06135-0187-43e7-97e1-ed6adba9be16' and platform = 'DYAI'")
 
 
     def handle_item(i):

aidso_geo/models/eco_data_process.py

@@ -33,7 +33,6 @@ def process_eco_product_relation(
             str(item.get("eco_title") or "").strip()
             for item in batch
         ]
-
         try:
             ai_result = (
                 ai_utils.ai_get_product_relation_spu(
@@ -105,11 +104,15 @@ def process_eco_product_relation(
                     )
                 except (TypeError, ValueError):
                     current = 0
+                if brand == spu_name:
+                    spu_name = brand
+                else:
+                    spu_name = brand+" "+spu_name
 
                 relation_data = {
                     "brand": brand,
                     "search_goods_word":keyword,
-                    "spu_name": brand+" "+spu_name,
+                    "spu_name": spu_name,
                     "current": (
                         1 if current == 1 else 0
                     ),
@@ -357,7 +360,7 @@ def yuanbao_android_process_txmap_eco(data,eco_list):
 def save_eco_data_to_bh(data,eco_type,eco_list):
 
     platform = data.get('platform')
-    search_goods_word = data.get('search_goods_word')
+    search_goods_word = data.get('searchGoodsWord')
 
     eco_result = []
     if platform == 'TYQWA':
@@ -386,7 +389,9 @@ def save_eco_data_to_bh(data,eco_type,eco_list):
     if search_goods_word:
         eco_result = process_eco_product_relation(eco_result, search_goods_word)
     bh_utils.insert_data('geo_eco_data',eco_result)
-
-
-
+if __name__ == '__main__':
+    result = bh_utils.query_data("select * from geo_eco_data where req_id in ('794107f1e46b436f81f18aac77043f89')")
+    eco_result = process_eco_product_relation(result, "小米洗地机")
+    bh_utils.insert_data('geo_eco_data', eco_result)
+#
 

aidso_geo/models/process.py

@@ -2032,9 +2032,9 @@ if __name__ == '__main__':
         type_t = i.get('type')
         type_t = 'batch'
         # commit_task(i,'ING')
-        # return task_send_queue(i,type_t)
+        return task_send_queue(i,type_t)
         # return deepseek_data_process.deepseek_process_original_data(i)
-        return platform_process(i)
+        # return platform_process(i)
     #
     if data_list:
         with ThreadPoolExecutor(max_workers=50) as executor:

aidso_geo/utils/ai_utils.py

@@ -3,7 +3,6 @@ import time
 
 import requests
 import json
-from aidso_geo.utils.tos_utils import get_string_from_tos
 
 
 def ai_get_brand_list(content, prompt):
@@ -145,7 +144,7 @@ def ai_get_product_list(content, prompt):
 def ai_get_product_relation_spu(product_list, keyword):
     url = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
     payload = {
-        "model": "doubao-seed-2-0-mini-260428",
+        "model": "doubao-seed-2-1-pro-260628",
         "messages": [
             {
                 "role": "system",
@@ -154,14 +153,14 @@ def ai_get_product_relation_spu(product_list, keyword):
 1. 需为商品列表中的每一件商品单独生成一条匹配记录，每条记录必须固定包含以下4个字段，各字段的取值规则明确如下：
    （1）produce_name：填写对应商品的完整原始名称，即商品列表中给出的该商品全称，不得做任何增删修改；
    （2）brand：填写该商品对应的SPU品牌名称，需精准识别商品所属品牌，参考示例：“北京同仁堂陈皮茯苓茶”的品牌取值为“北京同仁堂”，“红魔11 Pro+”的品牌取值为“红魔”；若商品无明确可识别的品牌信息，则直接返回该商品的完整名称作为brand字段值；
-   （3）spu_name：填写该商品的标准SPU名称，即去除品牌前缀后的商品核心名称，参考示例：“北京同仁堂陈皮茯苓茶”的spu_name取值为“陈皮茯苓茶”，“一加 Ace 6 至尊版”的spu_name取值为“Ace 6 至尊版”；
+   （3）spu_name：填写该商品的标准SPU名称，即去除品牌前缀后的商品核心名称，参考示例：“北京同仁堂陈皮茯苓茶”的spu_name取值为“陈皮茯苓茶”，“一加 Ace 6 至尊版”的spu_name取值为“Ace 6 至尊版”；需注意同一款商品的spu_name表述必须完全统一，禁止出现同一SPU存在空格差异、格式差异的情况，例如不得同时出现“无线洗地机5”和“无线洗地机 5”这类仅空格/格式有区别的命名，需统一为规范表述；
    （4）current：判断该商品与给定查询关键词是否存在关联，关联判定范围包括但不限于：关键词为该商品的品牌名、关键词为该品牌旗下的子品牌/系列名称、商品属于该关键词对应的品牌产品线；只要满足上述任意一种关联情形，该字段取值为1，若不存在任何关联则取值为0。
 2. 一致性校验特别要求：同批次传入的商品中，若商品标题指向的品牌名相同，brand字段的取值必须保持完全统一，禁止出现同一品牌同时标注“小米”和“xiaomi”这类中英文/不同写法混用的情况，需统一为规范名称；同批次商品的spu_name也需保持表述一致，禁止出现同一SPU同时标注“12 promax”和“12promax”这类格式不统一的情况，需统一为规范表述。
    {
         "produce_name": "郎酒 红花郎15",
         "brand": "郎酒",
-        "spu_name": "红花郎15",
-        "current": 0
+        "spu_name": "郎酒 红花郎15",
+        "current": 1
     },
                 """
             },
@@ -173,6 +172,7 @@ def ai_get_product_relation_spu(product_list, keyword):
         "thinking": {
             "type": "disabled"
         },
+        "max_tokens": 16384,
         "temperature": 0,
         "response_format": {
             "type": "json_schema",
@@ -227,7 +227,7 @@ def ai_get_product_relation_spu(product_list, keyword):
         'Authorization': 'Bearer ark-7afc3be2-37a8-47fd-9f02-996258a3d305-27da0',
         'Content-Type': 'application/json'
     }
-
+    content  =""
     try:
         response = requests.post(
             url,
@@ -235,9 +235,6 @@ def ai_get_product_relation_spu(product_list, keyword):
             json=payload,
             timeout=120,
         )
-        if response.status_code != 200:
-            print("status_code:", response.status_code)
-            print("response_text:", response.text)
         response.raise_for_status()
 
         response_data = response.json()
@@ -255,11 +252,11 @@ def ai_get_product_relation_spu(product_list, keyword):
             content = json.loads(content)
 
         product_words = content.get("product_words", [])
+        print(product_words if isinstance(product_words, list) else [])
         return product_words if isinstance(product_words, list) else []
 
     except Exception as e:
-        print(e)
-        return []
+        return  []
 
 def ai_get_product_list_search(product_list):
     url = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"