Skip to content

A
aidso-data
Commit 8da85c3c authored by Yaowentong's avatar Yaowentong
Browse files
Options

千问电商数据升级 

豆包深度思考可以出现 没有引用来源
parent 7a1ad569
Hide whitespace changes
Inline Side-by-side
Showing with 236 additions and 159 deletions
aidso_geo/core/down_load_bot.py
View file @ 8da85c3c
......@@ -631,12 +631,12 @@ def zhou_report_suggestion(phone, begin, end, brand_name, platform=None):
f.write(json.dumps(item, ensure_ascii=False) + "\n\n\n")
def qian_report(phone, begin, end, brand_name, platform=None):
def qian_report(phone, begin, end, plan_name, platform=None):
req_list = get_req_id(phone, begin, end)
req_ids = []
req_time_map = {}
for item in req_list:
if item.get('brand_name') == brand_name:
if item.get('plan_name') == plan_name:
req_id = item.get("req_id")
created_at = item.get("created_at")
req_ids.append(req_id)
......@@ -653,7 +653,9 @@ def qian_report(phone, begin, end, brand_name, platform=None):
result = []
if query_list:
for q in query_list:
print( q.get('reqId'))
r = {
"thinkingEnabled":q.get('thinkingEnabled'),
"prompt": q.get('prompt'),
"reqId": q.get('reqId'),
"platform": plat_form_map[q.get('platform')],
......@@ -662,37 +664,37 @@ def qian_report(phone, begin, end, brand_name, platform=None):
"created_at": req_time_map[q.get('reqId')]
}
result.append(r)
all_file = f"/Users/yaowentong/Desktop/{brand_name}_all.txt"
lite_file = f"/Users/yaowentong/Desktop/{brand_name}_lite.txt"
all_file = f"/Users/yaowentong/Desktop/{plan_name}_all.txt"
lite_file = f"/Users/yaowentong/Desktop/{plan_name}_lite.txt"
with open(all_file, "w", encoding="utf-8") as f:
for item in result:
f.write(json.dumps(item, ensure_ascii=False) + "\n\n\n")
remove_keys = {"site_icon", "task_id", "quto_id"}
data = []
with open(all_file, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
item = json.loads(line)
quote = item.get("quote")
if quote:
quote_list = json.loads(quote) # 把 quote 字符串转成 list
for one in quote_list: # one 是每个 dict
if isinstance(one, dict):
for k in remove_keys:
one.pop(k, None)
item["quote"] = json.dumps(quote_list, ensure_ascii=False)
data.append(item)
with open(lite_file, "w", encoding="utf-8") as f:
for item in data:
f.write(json.dumps(item, ensure_ascii=False) + "\n\n\n")
# remove_keys = {"site_icon", "task_id", "quto_id"}
# data = []
# with open(all_file, "r", encoding="utf-8") as f:
# for line in f:
# line = line.strip()
# if not line:
# continue
#
# item = json.loads(line)
#
# quote = item.get("quote")
# if quote:
# quote_list = json.loads(quote) # 把 quote 字符串转成 list
#
# for one in quote_list: # one 是每个 dict
# if isinstance(one, dict):
# for k in remove_keys:
# one.pop(k, None)
#
# item["quote"] = json.dumps(quote_list, ensure_ascii=False)
# data.append(item)
#
# with open(lite_file, "w", encoding="utf-8") as f:
# for item in data:
# f.write(json.dumps(item, ensure_ascii=False) + "\n\n\n")
def zhou_report(phone, begin, end, brand_name):
......@@ -1175,77 +1177,125 @@ def limit_excel_text(text, max_len=32000):
return str(text)[:max_len]
def txt_to_excel(txt_file, excel_file):
data_list = []
# TXT 每个非空行是一条 JSON 数据
with open(txt_file, "r", encoding="utf-8") as f:
for line_number, line in enumerate(f, start=1):
line = line.strip()
if not line:
continue
try:
item = json.loads(line)
if isinstance(item, dict):
data_list.append(item)
else:
print(f"跳过第 {line_number} 行：内容不是 JSON 对象")
except json.JSONDecodeError as e:
print(f"第 {line_number} 行解析失败：{e}")
if not data_list:
print("TXT 中没有可写入的数据")
return
columns = [
"thinkingEnabled",
"prompt",
"reqId",
"platform",
"content",
"quote",
"created_at"
]
wb = Workbook()
ws = wb.active
ws.title = "数据"
# 表头样式
header_font = Font(bold=True, color="FFFFFF")
header_fill = PatternFill(
fill_type="solid",
fgColor="4F81BD"
)
header_alignment = Alignment(
horizontal="center",
vertical="center"
)
# 写入表头
for column_index, column_name in enumerate(columns, start=1):
cell = ws.cell(
row=1,
column=column_index,
value=column_name
)
cell.font = header_font
cell.fill = header_fill
cell.alignment = header_alignment
# 写入数据
for row_index, item in enumerate(data_list, start=2):
for column_index, column_name in enumerate(columns, start=1):
value = item.get(column_name, "")
# list/dict 先转换成 JSON 字符串
if isinstance(value, (list, dict)):
value = json.dumps(value, ensure_ascii=False)
value = limit_excel_text(value)
cell = ws.cell(
row=row_index,
column=column_index,
value=value
)
cell.alignment = Alignment(
vertical="top",
wrap_text=True
)
# 设置列宽
column_widths = {
"A": 18, # thinkingEnabled
"B": 40, # prompt
"C": 40, # reqId
"D": 20, # platform
"E": 100, # content
"F": 100, # quote
"G": 22 # created_at
}
for column_letter, width in column_widths.items():
ws.column_dimensions[column_letter].width = width
# 冻结表头
ws.freeze_panes = "A2"
# 开启筛选
ws.auto_filter.ref = ws.dimensions
wb.save(excel_file)
print(f"转换完成，共写入 {len(data_list)} 条数据")
print(f"Excel 文件：{excel_file}")
# =========================
# 使用示例
# =========================
if __name__ == "__main__":
# result = get_req_id(18156037075,'2026-06-24','2026-06-24')
# req_list = []
# for i in result:
# req_list.append(i.get('req_id'))
# print(req_list)
plan_name = "/Users/yaowentong/Desktop/Oral-B（0714~0716）_all.txt"
plan_name2 = "/Users/yaowentong/Desktop/Oral-B（0714~0716）_all.xlsx"
qian_report(15100000026,'2026-07-14','2026-07-16','Oral-B（0714~0716）')
txt_to_excel(plan_name,plan_name2)
# qian_report(15100000026,'2026-06-21','2026-06-21','oral')
phone = 18521518364
begin = '2026-04-15'
end = '2026-07-13'
req_list = get_req_id(phone, begin, end)
brand_name1 = '王康医生'
brand_name2 = '殷敏毅'
req_list_result1 = []
req_list_result2 = []
for i in req_list :
if i.get('brand_name') == brand_name1:
req_list_result1.append(i.get('req_id'))
if i.get('brand_name') == brand_name2:
req_list_result2.append(i.get('req_id'))
print(f"req_list_result1:{len(req_list_result1)}")
print(req_list_result1)
print(f"req_list_result2:{len(req_list_result2)}")
print(req_list_result2)
# data_list = bh_utils.query_data(
# f"select * from geo_commit_task where platform in ('TYQWA','DOUBA') and insertime >1780243200")
#
#
# def handle_item(i):
# try:
# if i.get('comWordsMap'):
# i['comWordsMap'] = json.loads(i.get('comWordsMap'))
# if i.get('brandWords'):
# i['brandWords'] = json.loads(i.get('brandWords'))
# if i.get('comWords'):
# i['comWords'] = json.loads(i.get('comWords'))
# if i.get('keywords'):
# i['keywords'] = json.loads(i.get('keywords'))
# if i.get('productWordsMap'):
# i['productWordsMap'] = json.loads(i.get('productWordsMap'))
#
# taskId = i.get('taskId')
# reqId = i.get('reqId')
# platform = i.get('platform')
# content_file_path = f'geo/{taskId}/{platform}/context.txt'
# result_file_path = f'geo/{taskId}/{platform}/result.json'
#
# content = tos_utils.get_string_from_tos(content_file_path)
#
# if 'render_ecom_card_widget_' not in content:
# return
# logger.success(f"{reqId}----{platform}更新完成")
# result_content = tos_utils.get_string_from_tos(result_file_path)
# result_content_json = json.loads(result_content)
# result_content_json["hasGoods"] = 1
#
# tos_utils.put_string_to_tos(result_file_path, result_content_json)
#
# if platform == 'TYQWA':
# return qianwen_android_process_original_data(i)
#
# elif platform == 'DOUBA':
# return doubao_mobile_process_original_data(i)
# except Exception as e:
# print(f"{i.get('taskId')} error: {e}")
#
#
# if data_list:
# with ThreadPoolExecutor(max_workers=50) as executor:
# list(executor.map(handle_item, data_list))
aidso_geo/core/meituan_process.py
View file @ 8da85c3c
......@@ -23,7 +23,7 @@ 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",'')
channel = task_data.get("channel") or ""
task_type = task_data.get("type",'')
url = "http://172.16.1.223:8086/api/geo/task_commit"
......@@ -45,7 +45,7 @@ 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",'')
channel = task_data.get("channel") or ""
task_type = task_data.get("type",'')
url = f"http://172.16.1.223:8086/api/geo/task_check?reqId={req_id}"
......@@ -83,7 +83,6 @@ def task_commit():
def task_check():
check_tasks = query_task_check()
print(check_tasks)
if check_tasks:
for task in query_task_check():
get_result_api(task)
......
aidso_geo/core/routes/interface.py
View file @ 8da85c3c
......@@ -5,7 +5,7 @@ import queue
from concurrent.futures import ThreadPoolExecutor
from loguru import logger
BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from datetime import datetime
sys.path.append(BASE_DIR)
from aidso_geo.models.process import process_call_back, commit_task, main_process
from aidso_geo.config.base_config import init_redis, init_redis8
......@@ -158,6 +158,7 @@ def task_commit():
type = data.get('type',"")
platform = data.get('platform',"")
prompt = data.get('prompt',"")
data["pt"] = datetime.now().strftime("%Y%m%d")
required_fields = ["prompt", "taskId", "reqId", "platform", "type"]
missing_fields = [field for field in required_fields if field not in data]
if missing_fields:
......
aidso_geo/models/doubao_android_data_process.py
View file @ 8da85c3c
......@@ -311,7 +311,6 @@ def doubao_mobile_process_original_data(task_data):
suggestions, rich_media_block)
return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
except Exception as e:
# traceback.print_exc()
parts = file_path.split('/')
platform = parts[2]
task_id = parts[1]
......@@ -344,7 +343,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 reqId = 'f6599e25c11544e69f911b53aff98ce7'")
f"select * from geo_commit_task where taskId = '79ac3438-a362-4eaa-b954-070791ba634b' and platform = 'DOUBA'")
......
aidso_geo/models/process.py
View file @ 8da85c3c
......@@ -1696,32 +1696,49 @@ def platform_process(data):
# -------------
process_func = PLATFORM_PROCESS_MAP.get(platform)
if not channel:
if check_file_in_tos(context_path):
response_content = tos_utils.get_string_from_tos(context_path)
elif process_func:
_, _, _, _, response_content, _ = process_func(data)
# -------------
if response_content:
result_v2(response_content, data)
else:
scheduler(data)
else:
if process_func:
_, _, _, _, response_content, _ = process_func(data)
if response_content:
commit_task(data, 'SUCCESS')
resp_cache_key = f"geo:task_check:resp:{reqId}"
resp = {
"code": 200,
"msg": "success",
"data": {
"status": 'ING',
"result": {}
}
}
_cache_set_json(resp_cache_key, resp, 1)
else:
scheduler(data)
# -------------
if response_content:
result_v2(response_content, data)
else:
scheduler(data)
# if check_file_in_tos(context_path):
# response_content = tos_utils.get_string_from_tos(context_path)
# elif process_func:
# _, _, _, _, response_content, _ = process_func(data)
# # -------------
# if response_content:
# result_v2(response_content, data)
# else:
# scheduler(data)
# if not channel:
# if check_file_in_tos(context_path):
# response_content = tos_utils.get_string_from_tos(context_path)
# elif process_func:
# _, _, _, _, response_content, _ = process_func(data)
# # -------------
# if response_content:
# result_v2(response_content, data)
# else:
# scheduler(data)
# else:
# _, _, _, _, response_content, _ = process_func(data)
# if response_content:
# commit_task(data, 'SUCCESS')
# resp_cache_key = f"geo:task_check:resp:{reqId}"
# resp = {
# "code": 200,
# "msg": "success",
# "data": {
# "status": 'ING',
# "result": {}
# }
# }
# _cache_set_json(resp_cache_key, resp, 1)
# else:
# scheduler(data)
return True
except Exception as e:
......@@ -1743,6 +1760,7 @@ def main_process(data):
prompt = data.get('prompt')
data["prompt"] = prompt.strip()
data["search_enabled"] = data.get('searchEnabled', '1')
data["pt"] = datetime.now().strftime("%Y%m%d")
if data.get('thinkingEnabled'):
data["thinking_enabled"] = data.get('thinkingEnabled')
check = bh_utils.query_data(
......@@ -1995,23 +2013,26 @@ def run_data(PAGE_SIZE,MAX_WORKERS):
if __name__ == '__main__':
data_list = bh_utils.query_data(f"select * from geo_commit_task where status ='ING' and type !='stream' and platform = 'TXYB'")
# req_id_sql = ",".join([f"'{req_id}'" for req_id in req_ids])
#
# data_list = bh_utils.query_data(f"select * from geo_commit_task where reqId in ({req_id_sql}) ")
data_list = bh_utils.query_data("select * from geo_commit_task where status = 'ING' ")
def handle_item(i):
if i.get('comWordsMap'):
i['comWordsMap'] = json.loads(i.get('comWordsMap'))
if i.get('brandWords'):
i['brandWords'] = json.loads(i.get('brandWords'))
if i.get('comWords'):
i['comWords'] = json.loads(i.get('comWords'))
if i.get('keywords'):
i['keywords'] = json.loads(i.get('keywords'))
if i.get('productWordsMap'):
i['productWordsMap'] = json.loads(i.get('productWordsMap'))
i["comWordsMap"] = safe_json_loads(i.get("comWordsMap"), [])
i["brandWords"] = safe_json_loads(i.get("brandWords"), [])
i["comWords"] = safe_json_loads(i.get("comWords"), [])
i["keywords"] = safe_json_loads(i.get("keywords"), [])
i["productWordsMap"] = safe_json_loads(i.get("productWordsMap"), [])
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=50) as executor:
......
aidso_geo/models/qianwen_android_data_process.py
View file @ 8da85c3c
import json
import traceback
from asyncio import as_completed
from concurrent.futures import ThreadPoolExecutor
from loguru import logger
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
......@@ -30,19 +26,28 @@ def qianwen_android_process_original_data(task_data):
original_content = get_string_from_tos(file_path)
try:
sse_json_data = json.loads(original_content)
# sse_json_data = json.loads(json_original.get('sse'))
response_content = sse_json_data.get('reply_text','')
think_content = sse_json_data.get('deepthink_text','')
paa_answer = sse_json_data.get('paa_answer','')
bar_sources = sse_json_data.get('bar_sources')
source_content =next(
(item.get('content') for item in bar_sources if item.get('type') == 'source'),
{}
)
source_content = {}
if bar_sources:
source_content =next(
(item.get('content') for item in bar_sources if item.get('type') == 'source'),
{}
)
source_list = source_content.get('list', [])
video_note_list = sse_json_data.get('video_note_list',{}).get('list')
video_note_data = sse_json_data.get('video_note_list') or {}
if isinstance(video_note_data, str):
try:
video_note_data = json.loads(video_note_data)
except json.JSONDecodeError:
video_note_data = {}
video_note_list = video_note_data.get('list', [])
video_list = []
if video_note_list:
for i in video_note_list:
......@@ -58,13 +63,13 @@ def qianwen_android_process_original_data(task_data):
video_list.append(rich_media)
rich_media_block.append({
"utl":video_list,
"url":video_list,
"source_seq":''
})
source_count_list = []
if source_list:
for s in source_list:
source_url = {
source_url = {
'url':s.get('url'),
'title':s.get('title'),
'summary':s.get('summary'),
......@@ -171,7 +176,8 @@ def qianwen_android_process_original_data(task_data):
"source_seq": '',
})
taobao_list = []
if mime_type == 'multi_load/iframe' and (multi_load_type == 'taoassistant_fold_product_feeds' or multi_load_type == 'taoassistant_single_product') and status == 'complete':
if mime_type == 'multi_load/iframe' and (multi_load_type == 'taoassistant_fold_product_feeds' or multi_load_type == 'taoassistant_single_product' or multi_load_type == 'taoassistant_single_product_v2') and status == 'complete':
for mu in multi_load:
if mu.get('type') == 'taoassistant_fold_product_feeds' :
......@@ -226,7 +232,7 @@ def qianwen_android_process_original_data(task_data):
'card_type':'fold_product'
})
if mu.get('type') == 'taoassistant_single_product':
if mu.get('type') in ('taoassistant_single_product','taoassistant_single_product_v2'):
mu_content = mu.get('content',{})
mu_content_cardData = mu_content.get('cardData',{})
mu_content_cardData_data = mu_content_cardData.get('data',{})
......@@ -255,6 +261,7 @@ def qianwen_android_process_original_data(task_data):
'card_type': 'single_product'
})
if taobao_list:
response_content += 'render_ecom_card_widget_taobao_start:'
damai_str = json.dumps(taobao_list, ensure_ascii=False)
......@@ -361,11 +368,11 @@ def qianwen_android_process_original_data(task_data):
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,
suggestions, rich_media_block)
return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
except Exception as e:
parts = file_path.split('/')
platform = parts[2]
task_id = parts[1]
......@@ -400,7 +407,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 = '00117939-99e0-4ea1-97f9-cce0ecc11210' and platform = 'TYQWA'")
data_list = bh_utils.query_data(f"select * from geo_commit_task where taskId = '9dfd84a4-f551-4d74-b1ee-9367aed53a8c' and platform = 'TYQWA'")
def handle_item(i):
......
aidso_geo/models/yuanbao_android_data_process.py
View file @ 8da85c3c
......@@ -178,7 +178,7 @@ if __name__ == '__main__':
# yuanbao_android_process_original_data(file_path2)
data_list = bh_utils.query_data(
f"select * from geo_commit_task where taskId = '4ae7a6f3-c2a4-4804-9dc2-8f2385674c60' and platform = 'TXYBA'")
f"select * from geo_commit_task where taskId = '2499b6c1-d974-4b11-830a-90f0b5e7b9e5' and platform = 'TXYBA'")
# # #
......
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