电商搜索接口更新

aidso_geo/config/base_config.py

@@ -187,6 +187,67 @@ def init_redis4():
         return redis_client
     except Exception as e:
         return None
+import uuid
+def deduplicate_redis_list(redis_client, key, batch_size=500):
+    old_count = redis_client.llen(key)
+
+    if old_count == 0:
+        return 0, 0
+
+    temp_key = f"{key}:dedup:{uuid.uuid4().hex}"
+    seen_req_ids = set()
+    new_count = 0
+
+    try:
+        for start in range(0, old_count, batch_size):
+            end = min(start + batch_size - 1, old_count - 1)
+
+            values = redis_client.lrange(key, start, end)
+            unique_values = []
+
+            for value in values:
+                if isinstance(value, bytes):
+                    text = value.decode("utf-8")
+                else:
+                    text = value
+
+                try:
+                    data = json.loads(text)
+                except (TypeError, json.JSONDecodeError):
+                    # 非法 JSON 暂时保留
+                    unique_values.append(value)
+                    continue
+
+                req_id = data.get("reqId")
+
+                # 没有 reqId 的数据保留
+                if not req_id:
+                    unique_values.append(value)
+                    continue
+
+                if req_id in seen_req_ids:
+                    continue
+
+                seen_req_ids.add(req_id)
+                unique_values.append(value)
+
+            if unique_values:
+                # RPUSH 保持原 List 的顺序
+                redis_client.rpush(temp_key, *unique_values)
+                new_count += len(unique_values)
+
+        if new_count > 0:
+            # 原子地用临时 List 覆盖原 List
+            redis_client.rename(temp_key, key)
+        else:
+            redis_client.delete(key)
+
+        return old_count, new_count
+
+    except Exception:
+        # 发生异常时保留原 List，只清理临时数据
+        redis_client.delete(temp_key)
+        raise
 
 if __name__ == '__main__':
     key_list = ['BDAI:geo:stream_batch:list',
@@ -217,11 +278,17 @@ if __name__ == '__main__':
 'XHSA:geo:batch:list',
 'geo:task_commit:list']
 
-    redis5 = init_redis4()
-    # delete(redis_key)
-    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'))
+    redis_client8 = init_redis8()
+
+    # old_count, new_count = deduplicate_redis_list(
+    #     redis_client8,
+    #     "geo:task_commit:list",
+    #     batch_size=500,
+    # )
+    print(redis_client8.llen("geo:task_commit:list"))
+
+
+
 
 
 

aidso_geo/core/meituan_process.py

@@ -74,7 +74,7 @@ t1.channel,
 ifnull(t1.is_share,0) as is_share
 from (SELECT *
       FROM geo_third_task_data
-      where status != 'SUCCESS') t1
+      where status = 'ING') 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 
@@ -134,7 +134,7 @@ def get_result_api():
         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'
+                ) 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:

aidso_geo/core/routes/interface.py

@@ -342,8 +342,6 @@ def check_eco():
                 "message": "reqids必须是数组",
                 "data": None,
             })
-
-        # 过滤空值并去重，同时保持原顺序。
         reqids = list(dict.fromkeys(
             str(req_id).strip()
             for req_id in reqids
@@ -380,6 +378,18 @@ def check_eco():
         )
 
         if not result:
+            return jsonify({
+                "code": 200,
+                "message": "没有查询到数据",
+                "data": None,
+            })
+
+        statuses = [
+            str(item.get("isCalculateSearchWord"))
+            for item in result
+        ]
+
+        if all(status == "1" for status in statuses):
             return jsonify({
                 "code": 200,
                 "message": "处理成功",
@@ -387,25 +397,57 @@ def check_eco():
                     "reqids": reqids,
                     "keyword": keyword,
                     "query_rows": len(result),
-                    "insert_rows": len(result),
+                    "insert_rows": 0,
+                },
+            })
+
+        if any(status == "0" for status in statuses):
+            return jsonify({
+                "code": 400,
+                "message": "处理中",
+                "data": {
+                    "reqids": reqids,
+                    "keyword": keyword,
+                    "query_rows": len(result),
+                    "insert_rows": 0,
                 },
             })
+        for item in result:
+            item["isCalculateSearchWord"] = 0
+
+        bh_utils.insert_data(
+            "geo_eco_data",
+            result,
+        )
 
-        eco_result = process_eco_product_relation(
+        ok = submit_background_task(
+            process_eco_product_relation,
             result,
             keyword,
         )
 
-        bh_utils.insert_data("geo_eco_data",eco_result)
+        if not ok:
+            for item in result:
+                item["isCalculateSearchWord"] = None
 
+            bh_utils.insert_data(
+                "geo_eco_data",
+                result,
+            )
             return jsonify({
-            "code": 200,
-            "message": "处理成功",
+                "code": 500,
+                "message": "后台任务提交失败",
+                "data": None,
+            })
+
+        return jsonify({
+            "code": 400,
+            "message": "任务已提交，正在处理",
             "data": {
                 "reqids": reqids,
                 "keyword": keyword,
                 "query_rows": len(result),
-                "insert_rows": len(eco_result),
+                "insert_rows": 0,
             },
         })
 

aidso_geo/models/doubao_data_process.py

@@ -116,13 +116,13 @@ def doubao_process_original_data(data):
                                         index = next(
                                             (
                                                 item["text_card"]["index"]
-                                                for item in url_list
+                                                for item in (url_list or [])
                                                 if item.get("text_card", {}).get("title") == target_title
                                             ),
                                             None
                                         )
-                                        if index:
-                                            response_content+=f"[reference:{index}]"
+                                        if index is not None:
+                                            response_content += f"[reference:{index}]"
 
                                 if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
                                         "block_type") ==10000 and json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
@@ -137,6 +137,7 @@ def doubao_process_original_data(data):
                                     is_think = False
                                     continue
                         if json_content.get('patch_op')[0].get("patch_object") == 50:
+                            if json_content.get('patch_op')[0].get("patch_value").get("ext").get("sp_v2"):
                                 for sug in json.loads(
                                     json_content.get('patch_op')[0].get("patch_value").get("ext").get("sp_v2")):
                                     suggestions.append(sug.get("content"))
@@ -157,6 +158,7 @@ def doubao_process_original_data(data):
                                             response_content = content_block[0].get("content").get("text_block").get("text")
 
         suggestions = list(set(suggestions))
+
         spider_save_tos.process_and_save_files(file_path, search_keyword, url_list, think_content, response_content,
                                                suggestions,rich_media_block)
         return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
@@ -204,7 +206,7 @@ if __name__ == '__main__':
     #     'e9b27490-6ff0-48da-91a5-3dbbb8494c1d', 'b92b318c71d54c399ed033a722c06a35'
     # ]
     # for task in task_id_list:
-    data_list = bh_utils.query_data(f"select * from geo_commit_task  where taskId = 'e5b63f95-f713-4c51-b282-78c8c1c130ad' and  platform  = 'DB'")
+    data_list = bh_utils.query_data(f"select * from geo_commit_task  where taskId = '9bd06279-57d5-4da7-b8a7-f533fb7365f9' and  platform  = 'DB'")
 
     # # #
     # # #

aidso_geo/models/douyinai_data_process.py

@@ -31,11 +31,7 @@ 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
 

aidso_geo/models/eco_data_process.py

@@ -124,6 +124,9 @@ def process_eco_product_relation(
                     **relation_data,
                 }
             )
+    for item in final_result:
+        item["isCalculateSearchWord"] = 1
+    bh_utils.insert_data('geo_eco_data', final_result)
     return final_result
 
 
@@ -388,10 +391,13 @@ 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)
+        for item in eco_result:
+            item["isCalculateSearchWord"] = None
     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)
-#
+    result = bh_utils.query_data("select * from geo_eco_data where req_id in ('3a9edb7d-c98b-4ad2-8ae5-f53c4803ba99')")
+    eco_result = process_eco_product_relation(result, "忠芝")
+    for item in result:
+        item["isCalculateSearchWord"] = None
+    bh_utils.insert_data('geo_eco_data', result)
 

aidso_geo/models/process.py

@@ -2018,7 +2018,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 status = 'ING'")
+    data_list = bh_utils.query_data("select * from geo_commit_task where pt = 20260806 and  status = 'ING' and platform !='DB'")
     # 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' ")
 
@@ -2030,7 +2030,7 @@ if __name__ == '__main__':
         i["keywords"] = safe_json_loads(i.get("keywords"), [])
         i["productWordsMap"] = safe_json_loads(i.get("productWordsMap"), [])
         type_t = i.get('type')
-        type_t = 'batch'
+        type_t = 'stream_batch'
         # commit_task(i,'ING')
         return task_send_queue(i,type_t)
         # return deepseek_data_process.deepseek_process_original_data(i)

aidso_geo/models/qianwen_android_data_process.py

@@ -394,7 +394,7 @@ def qianwen_android_process_original_data(task_data):
 
                                         mu_content_model_input = mu_content.get('modelInput')
                                         mu_content_pois = mu_content.get('pois')
-
+                                        if mu_content_pois:
                                             for poi in mu_content_pois:
                                                 poi_summary = poi.get('summary1')
                                                 distance_formatted = poi.get('distance_formatted')
@@ -557,7 +557,7 @@ if __name__ == '__main__':
     # 04b2b061-a5c0-4429-8d3b-d2b6504f1e8b
     # data_list = bh_utils.query_data(f"select * from geo_commit_task  where taskId = '0001e01d-a8ea-405e-acef-93e4f55abbff' and  platform  = 'TYQWA'")
     data_list = bh_utils.query_data(
-        f"select * from geo_commit_task where taskId = '9b82af12-97ba-47e0-aaca-4d88b93c70e9' and platform = 'TYQWA'")
+        f"select * from geo_commit_task where taskId = '8e38e641-38fa-4105-9a73-0031132c7c28' and platform = 'TYQWA'")
 
 
     def handle_item(i):

aidso_geo/utils/ai_utils.py

@@ -252,7 +252,6 @@ 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: