Skip to content

A
aidso-data
Commit 66670d85 authored by Yaowentong's avatar Yaowentong
Browse files
Options

元宝京东元宝小程序高德地图

parent 94534a08
Hide whitespace changes
Inline Side-by-side
Showing with 233 additions and 56 deletions
aidso_geo/config/base_config.py
View file @ 66670d85
......@@ -4,12 +4,12 @@ import os
import time
from enum import Enum
import redis
from loguru import logger
def init_redis():
try:
redis_client = redis.Redis(
host='172.16.0.24',
# host='redis-cnlfmu7rl14awitrz.redis.ivolces.com',
port=6379,
db=0,
password='aiyingli@@123',
......@@ -48,7 +48,7 @@ def init_redis8():
_redis8_client.ping()
return _redis8_client
except Exception as e:
print(f"Redis 初始化失败: {e}")
logger.error(f"Redis 初始化失败: {e}")
_redis8_client = None
_redis8_pool = None
return None
......
aidso_geo/core/commit_process.py
View file @ 66670d85
......@@ -9,8 +9,8 @@ from aidso_geo.config.base_config import init_redis, init_redis8
from aidso_geo.models.process import main_process
redis_client = init_redis()
# redis_client = init_redis8()
# redis_client = init_redis()
redis_client = init_redis8()
QUEUE_KEY = "geo:task_commit:list"
......
aidso_geo/core/routes/interface.py
View file @ 66670d85
......@@ -12,7 +12,7 @@ from aidso_geo.config.base_config import init_redis, init_redis8
from aidso_geo.utils import bh_utils, tos_utils, url_utils
line_app = Blueprint("line_app", __name__)
redis_client = init_redis()
redis_client8 = init_redis8()
MAX_WORKERS = 30
MAX_PENDING_TASKS = 200
......@@ -169,8 +169,8 @@ def task_commit():
logger.success(f"{data['reqId']}--{platform}--{prompt}--任务提交--{type}")
# ok = submit_background_task(main_process, data)
ret = redis_client.lpush("geo:task_commit:list",json.dumps(data, ensure_ascii=False))
# ret = redis_client8.lpush("geo:task_commit:list",json.dumps(data, ensure_ascii=False))
# ret = redis_client.lpush("geo:task_commit:list",json.dumps(data, ensure_ascii=False))
ret = redis_client8.lpush("geo:task_commit:list",json.dumps(data, ensure_ascii=False))
if ret and ret > 0:
resp_cache_key = f"geo:task_check:resp:{req_id}"
resp = {
......@@ -219,6 +219,8 @@ def data_call_back():
logger.success(f"{req_id}--{platform}-------CALL_BACK")
ok = submit_background_task(process_call_back, task_data, result)
# ret = redis_client8.lpush("geo:call_back:list", json.dumps(data, ensure_ascii=False))
# if not ret:
# if ret and ret > 0:
if not ok:
return jsonify({
"code": 503,
......
aidso_geo/core/routes/third_interface.py
View file @ 66670d85
......@@ -530,7 +530,7 @@ def mt_task_commit():
if daily_used_count + cost > daily_limit:
return err(
400,
f"超过单日最大调用次数 {daily_limit}次/天，"
f"超过单日最大调用 {daily_limit}分/天，"
f"本次提交消耗 {cost}，当前已使用 {daily_used_count}，剩余可用 {max(daily_limit - daily_used_count, 0)}"
)
......@@ -538,8 +538,8 @@ def mt_task_commit():
if total_used_count + cost > total_limit:
return err(
400,
f"超过渠道总最大调用次数 {total_limit}次，"
f"本本次提交消耗 {cost}，当前已使用 {total_used_count}，剩余可用 {max(total_limit - total_used_count, 0)}"
f"超过渠道总最大调用 {total_limit}分，"
f"本次提交消耗 {cost}，当前已使用 {total_used_count}，剩余可用 {max(total_limit - total_used_count, 0)}"
)
task_id = str(uuid.uuid4())
......
aidso_geo/core/task_process.py
View file @ 66670d85
......@@ -17,7 +17,7 @@ QUEUE_KEY = "geo:call_back:list"
BATCH_SIZE = 50
# 每 30 秒执行一次
INTERVAL_SECONDS = 30
INTERVAL_SECONDS = 20
# 每轮内部并发数
CONCURRENT_WORKERS = 50
......
aidso_geo/models/baidu_baikan_data_process.py
View file @ 66670d85
......@@ -105,7 +105,7 @@ if __name__ == '__main__':
# data_list = bh_utils.query_data(f"select * from geo_commit_task where platform = 'BK' and taskId = '21fd537f-131b-4wwwww69a-9d5c-6ea00da2c0d6'")
data_list = bh_utils.query_data(f"select * from geo_commit_task where platform = 'BK' and taskId = '21sssfd537f-131b-4wwwww69a-9dsss5c-6ea00da2c0d6'")
data_list = bh_utils.query_data(f"select * from geo_commit_task where platform = 'BK' and taskId = 'ac2ebe76-0311-4587-935a-8733cb154d17'")
# # #
# # #
......
aidso_geo/models/doubao_android_data_process.py
View file @ 66670d85
......@@ -37,7 +37,12 @@ def doubao_mobile_process_original_data(task_data):
try:
json_content = json.loads(payload)
print(json_content)
print('-----')
print('-----')
print('-----')
print('-----')
print('-----')
except (IndexError, json.JSONDecodeError):
continue
......@@ -343,7 +348,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 req_id = '000804db-804b-428d-b5d8-f3d0b717bc3b'")
f"select * from geo_commit_task where reqId = '0d408a82-c850-4796-9fa3-de5af109b2e4'")
......
aidso_geo/models/eco_data_process.py
View file @ 66670d85
......@@ -74,6 +74,34 @@ def doubao_android_process_douyin_eco(data,eco_list):
eco_result.append(raw_data)
return eco_result
def yuanbao_android_process_jd_eco(data,eco_list):
eco_result = []
task_id = data.get('taskId')
platform = data.get('platform')
req_id = data.get('reqId')
prompt = data.get('prompt')
for index, item in enumerate(eco_list):
raw_data = {
"eco_id": item.get('sku', ''),
"eco_url": item.get('pc_url', '') ,
"eco_title": item.get('sku_name', ''),
"eco_index": index+1,
"eco_pic": item.get('image_url', ''),
"eco_price": item.get('price', ''),
"eco_source": 'jdeco',
"platform": platform,
"task_id": task_id,
"req_id": req_id,
"prompt": prompt
}
eco_result.append(raw_data)
return eco_result
def save_eco_data_to_bh(data,eco_type,eco_list):
platform = data.get('platform')
......@@ -89,6 +117,9 @@ def save_eco_data_to_bh(data,eco_type,eco_list):
...
if eco_type == 'dyeco':
eco_result =doubao_android_process_douyin_eco(data,eco_list)
if platform == 'TXYBA':
if eco_type == 'jdeco':
eco_result =yuanbao_android_process_jd_eco(data,eco_list)
bh_utils.insert_data('geo_eco_data',eco_result)
aidso_geo/models/process.py
View file @ 66670d85
......@@ -959,21 +959,30 @@ def replace_word_by_brand(data_list):
def keep_min_rank_by_brand(data_list):
"""
相同 brand 只保留一条，保留 rank 最小的那条
"""
brand_map = {}
def rank_better(new_rank, old_rank):
new_rank = int(new_rank or 0)
old_rank = int(old_rank or 0)
if old_rank == 0:
return new_rank > 0
if new_rank == 0:
return False
return new_rank < old_rank
for item in data_list or []:
brand = item.get("brand")
rank = item.get("rank", 0)
word = item.get("word")
if brand not in brand_map:
brand_map[brand] = item
else:
old_rank = brand_map[brand].get("rank", 0)
if rank < old_rank:
brand_map[brand] = item
key = word if brand == "没有品牌" else brand
if key not in brand_map:
brand_map[key] = item
continue
if rank_better(item.get("rank", 0), brand_map[key].get("rank", 0)):
brand_map[key] = item
return list(brand_map.values())
......@@ -1289,7 +1298,37 @@ def reset_rank(all_vos, brand_vos, competitor_vos):
return all_vos, brand_vos, competitor_vos
def safe_json_loads(value, default=None):
if value is None:
return default
# 已经是 list/dict 了，直接返回
if isinstance(value, (list, dict)):
return value
# Redis 读出来有时是 bytes
if isinstance(value, bytes):
value = value.decode("utf-8")
# 字符串才尝试 json.loads
if isinstance(value, str):
value = value.strip()
if value == "":
return default
try:
return json.loads(value)
except Exception as e:
return default
return default
def result_v2(response_content, data):
data["comWordsMap"] = safe_json_loads(data.get("comWordsMap"), [])
data["brandWords"] = safe_json_loads(data.get("brandWords"), [])
data["comWords"] = safe_json_loads(data.get("comWords"), [])
data["keywords"] = safe_json_loads(data.get("keywords"), [])
data["productWordsMap"] = safe_json_loads(data.get("productWordsMap"), [])
data = process_com_map(data)
reqId = data.get('reqId')
type = data.get('type')
......@@ -1331,7 +1370,6 @@ def result_v2(response_content, data):
all_word_set_list = [str(x).strip() for x in all_word_set_list if str(x).strip()]
# 获取所有词的排名 word + rank_list
all_word_rank_list = get_keyword_ranks(response_content, all_word_set_list)
# 获取所有词的排名 word + rank + count
all_keyword_with_rank = convert_rank_data(all_word_rank_list)
......@@ -1340,7 +1378,6 @@ def result_v2(response_content, data):
all_keyword_with_brand = keyword_map_brand(all_word_list)
# 追加所有词的品牌
all_keyword_with_rank_sentiment_brand = keyword_with_brand(all_keyword_with_rank, all_keyword_with_brand)
# 拆分为 all_词 品牌词 竞品词
ai_word_list_with_rank, brand_words_list_with_rank, com_word_list_with_rank = split_word_rank(ai_word_list,
brand_words,
......@@ -1348,6 +1385,7 @@ def result_v2(response_content, data):
all_keyword_with_rank_sentiment_brand)
# ---------------------------------调整排名-----------------------------
brand_words_list_with_rank, ai_word_list_with_rank = merge_brand_rank_by_subset(brand_words_list_with_rank,
......@@ -1371,9 +1409,9 @@ def result_v2(response_content, data):
ai_word_list_with_rank)
# ---------------------------------调整排名-----------------------------
# 去除all
ai_word_list_with_rank = filter_ai_word_list(ai_word_list_with_rank, brand_words_list_with_rank,
com_word_list_with_rank)
# ---------------------------------处理品牌-----------------------------
ai_word_list_with_rank = replace_word_by_brand(ai_word_list_with_rank)
......@@ -1795,9 +1833,11 @@ def process_success(data):
if think_content:
put_string_to_tos(f"geo/{data.get('taskId')}/{data.get('platform')}/think.txt", think_content)
elif platform in ('TYQW','TYQWA'):
search_keywords, url_list, think_content, response_content = ai_interface.get_qianwewn_message(prompt,think)
search_keywords, url_list, think_content, response_content = ai_interface.get_yuanbao_message(prompt,'0')
elif platform == 'KIMI':
search_keywords, url_list, think_content, response_content = ai_interface.get_kimi_message(prompt,think)
elif platform == 'XHSA':
search_keywords, url_list, think_content, response_content = ai_interface.get_kimi_message(prompt,think)
else:
search_keywords, url_list, think_content, response_content = ai_interface.get_doubao_message(data.get('prompt'),think)
put_string_to_tos(f"geo/{data.get('taskId')}/{data.get('platform')}/context.txt", response_content)
......@@ -1955,7 +1995,7 @@ def run_data(PAGE_SIZE,MAX_WORKERS):
if __name__ == '__main__':
#
data_list = bh_utils.query_data(f"select * from geo_commit_task where status !='SUCCESS'")
data_list = bh_utils.query_data(f"select * from geo_commit_task where reqId='df6b3f7850eb4a538843952f566f0b58'")
def handle_item(i):
if i.get('comWordsMap'):
......@@ -1971,10 +2011,9 @@ if __name__ == '__main__':
type_t = i.get('type')
type_t = 'batch'
# commit_task(i,'ING')
return task_send_queue(i,type_t)
# return platform_process(i)
# return task_send_queue(i,type_t)
return platform_process(i)
#
if data_list:
with ThreadPoolExecutor(max_workers=30) as executor:
futures = [executor.submit(handle_item, i) for i in data_list]
......
aidso_geo/models/qianwen_android_data_process.py
View file @ 66670d85
......@@ -289,12 +289,42 @@ def qianwen_android_process_original_data(task_data):
'id': damai.get('data').get('id')
})
damai_data_list.extend(damai_list)
gaode_list = []
if mime_type == 'multi_load/iframe' and multi_load_type == 'new_amap_poi_list' and multi_load_status == 'complete':
for mu in multi_load:
mu_content = mu.get('content')
mu_source_seq = mu.get('source_seq')
mu_content_model_input = mu_content.get('modelInput')
mu_content_pois = mu_content.get('pois')
for poi in mu_content_pois:
poi_summary = poi.get('summary1')
distance_formatted = poi.get('distance_formatted')
photos = poi.get('photos')
distance_address = poi.get('address')
gaode_result = {
"source_seq":mu_source_seq,
"name":mu_content_model_input.get("name"),
"poi_id":mu_content_model_input.get("poi_id"),
"poi_summary":poi_summary,
"distance_formatted":distance_formatted,
"photos":photos[0].get("url", "") if isinstance(photos, list) and photos and isinstance(photos[0], dict) else "",
"address":distance_address,
}
gaode_list.append(gaode_result)
if damai_list:
response_content += 'render_ecom_card_widget_damai_start:'
damai_str = json.dumps(damai_list, ensure_ascii=False)
response_content += damai_str
response_content += 'render_ecom_card_widget_damai_end:\n'
if gaode_list:
response_content += 'render_ecom_card_widget_gaode_start:'
gaode_str = json.dumps(gaode_list, ensure_ascii=False)
response_content += gaode_str
response_content += 'render_ecom_card_widget_gaod_end:\n'
if mime_type == 'multi_load/iframe' and multi_load_type == 'deep_think' and multi_load_status == 'complete':
think_content += multi_load_content.get('think_content')
# 引用来源 回答 视频列表
......@@ -326,6 +356,8 @@ def qianwen_android_process_original_data(task_data):
url_list = url_list_batch
if damai_data_list:
save_eco_data_to_bh(task_data, 'damai', damai_data_list)
response_content = response_content.replace("[(deep_think)]", "")
response_content = response_content.replace("[(multimodal_chat_think_1)]", "")
spider_save_tos.process_and_save_files(file_path, search_keyword, url_list, think_content, response_content,
......@@ -366,7 +398,7 @@ if __name__ == '__main__':
# 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 = '374593fc-ac18-4eb3-9815-866704ce02ca' and platform = 'TYQWA'")
data_list = bh_utils.query_data(f"select * from geo_commit_task where taskId = '00117939-99e0-4ea1-97f9-cce0ecc11210' and platform = 'TYQWA'")
def handle_item(i):
......
aidso_geo/models/yuanbao_android_data_process.py
View file @ 66670d85
import json
import re
from aidso_geo.models import spider_save_tos
from aidso_geo.models.eco_data_process import save_eco_data_to_bh
from aidso_geo.utils import robot_utils, bh_utils
from aidso_geo.utils.ai_interface import get_parse_sse_result
from aidso_geo.utils.tos_utils import get_string_from_tos
def process_price(ori_price):
price_match = re.search(
r"\d+(?:\.\d+)?",
str(ori_price).replace(",", "")
)
if price_match:
price_number = float(price_match.group())
price = int(price_number) if price_number.is_integer() else price_number
else:
price = 0
return price
def yuanbao_android_process_original_data(data):
file_path = f'geo/{data["taskId"]}/{data["platform"]}/original.text'
def yuanbao_android_process_original_data(task_data):
file_path = f'geo/{task_data["taskId"]}/{task_data["platform"]}/original.text'
"""处理文件内容，提取思考内容、响应内容和URL列表"""
url_list =[]
url_list = []
think_content = ""
response_content = ""
response_process = ""
......@@ -19,6 +34,7 @@ def yuanbao_android_process_original_data(data):
suggestions = []
rich_media_block = []
mini_program = []
jd_goods = []
response_bool = False
try:
original_content = get_string_from_tos(file_path)
......@@ -31,11 +47,9 @@ def yuanbao_android_process_original_data(data):
# 提取并解析JSON数据
data_str = i.split("data: ")[1]
json_data = json.loads(data_str)
except (IndexError, json.JSONDecodeError):
continue # 跳过格式错误的数据
if json_data.get('type') == 'searchGuid':
url_list = json_data.get('docs')
if json_data.get('type') == 'think':
think_content += json_data.get('content')
......@@ -53,30 +67,83 @@ def yuanbao_android_process_original_data(data):
if json_data.get('type') == 'image':
response_content = "生成了图片"
if json_data.get('type') == 'replace':
if json_data.get('replace').get('display') =='videoBoxV2':
replace_data = json_data.get("replace") or {}
if json_data.get('replace').get('display') == 'videoBoxV2':
rich_media_block.extend(json_data.get('replace').get('multimedias'))
if json_data.get('replace').get('display') =='miniProgramBox':
if json_data.get('replace').get('display') == 'miniProgramBox':
response_content+=json_data.get('replace').get('title')
response_content += json_data.get('replace').get('title')
response_content += '\n\nrender_ecom_card_widget_miniprogram_start:\n\n'
for pro in json_data.get('replace').get('multimedias'):
title = pro.get('title','')
icon = pro.get('linkInfo', {}).get('icon','') or pro.get('linkInfo', {}).get('iconDark','')
wx_app_id = pro.get('linkInfo', {}).get('wxAppId','')
title = pro.get('title', '')
icon = pro.get('linkInfo', {}).get('icon', '') or pro.get('linkInfo', {}).get('iconDark',
'')
wx_app_id = pro.get('linkInfo', {}).get('wxAppId', '')
response_mini = {
"title":title,
"icon":icon,
"wx_app_id":wx_app_id,
"title": title,
"icon": icon,
"wx_app_id": wx_app_id,
}
mini_program.append(response_mini)
mini_program_str = json.dumps(mini_program, ensure_ascii=False)
response_content += mini_program_str
response_content += '\n\nrender_ecom_card_widget_miniprogram_end:'
if replace_data.get("display") == "goods":
replace_media_id = replace_data.get('id')
for media in replace_data.get("multimedias") or []:
if not isinstance(media, dict):
continue
goods_content = media.get("goodsContent") or {}
goods = goods_content.get("goods") or {}
# 没有真实商品数据，说明只是占位/加载事件
if not isinstance(goods, dict):
continue
sku_id = goods.get("skuId")
sku_name = goods.get("skuName")
# 有商品 ID 和商品名才认为是有效商品
if not sku_id or not sku_name:
continue
raw_content = {}
try:
raw_content = json.loads(goods.get("rawContent") or "{}")
except (TypeError, json.JSONDecodeError):
raw_content = {}
endpoint_url = goods_content.get("endpointUrl") or {}
shop_id = goods.get("shopId") or raw_content.get("shopId", "")
goods_data = {
"replace_media_id":replace_media_id,
"sku": str(sku_id),
"sku_name": sku_name,
"query": goods.get("query", ""),
"image_url": (
goods.get("imagePath")
or raw_content.get("whiteImage")
or raw_content.get("imgPath")
or ""
),
"price": process_price(goods.get("purchasePrice", "")),
"shop_name": goods.get("shopName", ""),
"shop_id": shop_id,
"source": goods.get("source", ""),
"pc_url": endpoint_url.get("pcUrl", "")
}
jd_goods.append(goods_data)
if jd_goods:
response_content += '\n\nrender_ecom_card_widget_jd_start:\n\n'
jd_goods_str = json.dumps(jd_goods, ensure_ascii=False)
response_content += jd_goods_str
response_content += '\n\nrender_ecom_card_widget_jd_end:'
save_eco_data_to_bh(task_data,'jdeco',jd_goods)
spider_save_tos.process_and_save_files(file_path, search_keyword, url_list, think_content, response_content,
suggestions,rich_media_block)
suggestions, rich_media_block)
return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
except Exception as e:
......@@ -91,24 +158,25 @@ def yuanbao_android_process_original_data(data):
think_content = think
search_keyword = search_word
spider_save_tos.process_and_save_files_ai(file_path, search_keyword, url_list, think_content,
response_content,suggestions)
response_content, suggestions)
return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
else:
response_content = "对话信息获取失败:-200"
robot_utils.feishu_tobot(file_path)
spider_save_tos.process_and_save_files(file_path, search_keyword, url_list, think_content, response_content,
suggestions)
return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
if __name__ == '__main__':
# file_path2 = 'geo/6f49084e-f191-447d-849b-bc6c60fd5d20/TXYBA/original.text'
# yuanbao_android_process_original_data(file_path2)
data_list = bh_utils.query_data(
f"select * from geo_commit_task where taskId = '2e35cfd9-6c36-417c-9384-9140d9a7dcd1' and platform = 'TXYBA'")
data_list = bh_utils.query_data(f"select * from geo_commit_task where taskId = '30fde623-70a3-4ca7-86fb-393b60841402' and platform = 'TXYBA'")
# # #
# # #
......
aidso_geo/utils/bh_utils.py
View file @ 66670d85
......@@ -18,7 +18,7 @@ DB_CONFIG = {
POOL = PooledDB(
creator=pymysql,
maxconnections=30,
maxconnections=50,
mincached=1,
maxcached=10,
blocking=True,
......@@ -34,7 +34,7 @@ POOL = PooledDB(
def is_connection_error(e):
if isinstance(e, pymysql.err.OperationalError):
code = e.args[0] if e.args else None
return code in (0, 2006, 2013)
return code in (0, 2006, 2013,2014)
return False
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment