Skip to content

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

美团截图增加ip 版本号

parent 09bc0086
Hide whitespace changes
Inline Side-by-side
Showing with 8977 additions and 158 deletions
aidso_geo/config/base_config.py
View file @ d8bd6be6
......@@ -69,6 +69,7 @@ class PlatformType(Enum):
QIANWENANDROID = "TYQWA"
YUANBAOANDROID = "TXYBA"
XIAOHONGSHUANDROID = "XHSA"
BAIDUBAIKAN = "BK"
@classmethod
......@@ -133,6 +134,10 @@ class BaseConfig:
PlatformType.XIAOHONGSHUANDROID.value: {
"url": f"{base_url}hongshu_android",
"storage_path": lambda tid: f"geo/{tid}/XHSA/original.text"
},
PlatformType.BAIDUBAIKAN.value: {
"url": f"{base_url}baikan",
"storage_path": lambda tid: f"geo/{tid}/BK/original.text"
}
}
......@@ -165,6 +170,7 @@ def get_user_info(secUid):
print(response.text)
if __name__ == '__main__':
key_list = ['BDAI:geo:stream_batch:list',
'BDAI:geo:batch:list',
......@@ -198,15 +204,15 @@ if __name__ == '__main__':
# aa = list(aa)
t.delete('mt_third_task')
# for a in aa:
# sec_uid = get_sec_id(a)
# get_user_info(sec_uid)
# print(t.scard('third_hx_user_id'))
while 1:
for k in key_list:
print(f"{k}----{t.llen(k)}")
time.sleep(10)
# while 1:
# for k in key_list:
# print(f"{k}----{t.llen(k)}")
# time.sleep(10)
#
# print(len(t.smembers(key)))
......
aidso_geo/core/down_load_bot.py
View file @ d8bd6be6
This source diff could not be displayed because it is too large. You can view the blob instead.
aidso_geo/core/feishu_snipaste.py
View file @ d8bd6be6
......@@ -13,6 +13,9 @@ from datetime import datetime, timedelta
from openpyxl.workbook import Workbook
import os ,sys
from loguru import logger
from aidso_geo.utils.tos_utils import put_string_to_tos
BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(BASE_DIR)
from aidso_geo.models import spider_save_tos
......@@ -301,6 +304,9 @@ def doubao_process_original_data(file_path, original_content):
rich_media_block = []
think_bool = False
response_bool = False
file_path_result = os.path.dirname(file_path)
print(file_path_result)
content_list = original_content.split("\n")
for i in content_list:
......@@ -316,9 +322,19 @@ def doubao_process_original_data(file_path, original_content):
try:
json_content = json.loads(payload)
except (IndexError, json.JSONDecodeError):
continue
if json_content.get('query_list'):
extra = json_content.get('ack_client_meta').get('conversation_info').get('extra')
if json_content.get('ack_client_meta').get('conversation_info').get('extra'):
extra_json = json.loads(extra)
inner_user_ip = extra_json.get('inner_user_ip')
inner_pc_version = extra_json.get('inner_pc_version')
if inner_user_ip:
put_string_to_tos(f"{file_path_result}/inner_user_ip.txt", inner_user_ip)
if inner_pc_version:
put_string_to_tos(f"{file_path_result}/inner_pc_version.txt", inner_pc_version)
if json_content.get('event_type') == 2001:
even_data = json.loads(json_content.get('event_data'))
......@@ -481,6 +497,8 @@ class ExcelWriter:
self.worksheet.cell(row=1, column=8, value='查询时间')
self.worksheet.cell(row=1, column=9, value='ocr结果')
self.worksheet.cell(row=1, column=10, value='排名')
self.worksheet.cell(row=1, column=11, value='ip')
self.worksheet.cell(row=1, column=12, value='版本号')
......@@ -488,15 +506,11 @@ class ExcelWriter:
def write_batch_rows(self, data_list):
if not isinstance(data_list, list) or len(data_list) == 0:
raise ValueError("data_list必须是非空的列表")
# "taskId": i.get("taskId"),
# "prompt": i.get("prompt"),
# "platform": i.get("platform"),
# "insertime": i.get("insertime"),
# 找到下一个空行
next_row = self.worksheet.max_row + 1
# 批量写入数据
for idx, (prompt, platform, content, think, quote,share, snipaste, insertime,ocr,rank) in enumerate(data_list):
for idx, (prompt, platform, content, think, quote,share, snipaste, insertime,ocr,rank,inner_user_ip,inner_pc_version) in enumerate(data_list):
self.worksheet.cell(row=next_row + idx, column=1, value=prompt)
self.worksheet.cell(row=next_row + idx, column=2, value=platform)
self.worksheet.cell(row=next_row + idx, column=3, value=content)
......@@ -507,6 +521,8 @@ class ExcelWriter:
self.worksheet.cell(row=next_row + idx, column=8, value=insertime)
self.worksheet.cell(row=next_row + idx, column=9, value=ocr)
self.worksheet.cell(row=next_row + idx, column=10, value=rank)
self.worksheet.cell(row=next_row + idx, column=11, value=inner_user_ip)
self.worksheet.cell(row=next_row + idx, column=12, value=inner_pc_version)
# 保存文件
self.workbook.save(self.file_path)
......@@ -625,6 +641,8 @@ def to_excel2(pt, cn):
context_path = f"geo_snipaste/{pt}/doubao/{reqId}/context.txt"
quote_path = f"geo_snipaste/{pt}/doubao/{reqId}/quote.txt"
think = f"geo_snipaste/{pt}/doubao/{reqId}/think.txt"
inner_user_ip= f"geo_snipaste/{pt}/doubao/{reqId}/inner_user_ip.txt"
inner_pc_version= f"geo_snipaste/{pt}/doubao/{reqId}/inner_pc_version.txt"
png_url = f'https://tcdn.aidso.com/geo_snipaste/{pt}/doubao/{reqId}/png.png'
text_json_str = tos_utils.get_string_from_tos(file) or "{}"
......@@ -639,7 +657,10 @@ def to_excel2(pt, cn):
png_url,
timestamp_to_datetime(insertime),
get_context(png_url),
rank
rank,
tos_utils.get_string_from_tos(inner_user_ip),
tos_utils.get_string_from_tos(inner_pc_version)
)
batch_data = [None] * len(task_list)
......@@ -672,37 +693,7 @@ def to_excel2(pt, cn):
return out_path
# batch_data = []
# for i in task_list:
# reqId = i.get("reqId")
# prompt = i.get("prompt")
# platform = i.get("platform")
# insertime = i.get("insertime")
# rank = i.get("rank")
# file = f"geo_snipaste/{pt}/doubao/{reqId}/text.json"
# context_path =f"geo_snipaste/{pt}/doubao/{reqId}/context.txt"
# quote_path = f"geo_snipaste/{pt}/doubao/{reqId}/quote.txt"
# think = f"geo_snipaste/{pt}/doubao/{reqId}/think.txt"
# batch_data.append(
# (
# prompt,
# plat_form_map.get(platform),
# tos_utils.get_string_from_tos(context_path),
# tos_utils.get_string_from_tos(think),
# tos_utils.get_string_from_tos(quote_path),
# json.loads(tos_utils.get_string_from_tos(file)).get('share_url', ''),
# f'https://tcdn.aidso.com/geo_snipaste/{pt}/doubao/{reqId}/png.png',
# timestamp_to_datetime(insertime),
# get_context(f'https://tcdn.aidso.com/geo_snipaste/{pt}/doubao/{reqId}/png.png'),
# rank
# )
# )
# logger.success(f"时间{pt}--{cn}---导出数据{len(batch_data)}条")
# out_path = os.path.join(EXCEL_DIR, f"{pt}美团{cn}.xlsx")
# excel_writer = ExcelWriter(out_path)
# excel_writer.write_batch_rows(batch_data)
# excel_writer.close()
# return out_path
......@@ -1021,6 +1012,8 @@ def webhook_snipaste(promp_list):
context_path = f"geo_snipaste/{tos_pt}/doubao/{reqId}/context.txt"
quote_path = f"geo_snipaste/{tos_pt}/doubao/{reqId}/quote.txt"
think = f"geo_snipaste/{tos_pt}/doubao/{reqId}/think.txt"
inner_user_ip = f"geo_snipaste/{pt}/doubao/{reqId}/inner_user_ip.txt"
inner_pc_version = f"geo_snipaste/{pt}/doubao/{reqId}/inner_pc_version.txt"
png_url = f'https://tcdn.aidso.com/geo_snipaste/{tos_pt}/doubao/{reqId}/png.png'
text_json_str = tos_utils.get_string_from_tos(file) or "{}"
......@@ -1035,7 +1028,9 @@ def webhook_snipaste(promp_list):
png_url,
timestamp_to_datetime(insertime),
get_context(png_url),
rank
rank,
inner_user_ip,
inner_pc_version
)
batch_data = [None] * len(task_list)
......@@ -1078,11 +1073,39 @@ def webhook_snipaste(promp_list):
if __name__ == "__main__":
t1 = threading.Thread(target=scheduler_thread, daemon=True)
t2 = threading.Thread(target=runner_thread, daemon=True)
t1.start()
t2.start()
t1.join()
t2.join()
# redis_client = init_redis4()
# print(redis_client.delete('mt_third_task'))
\ No newline at end of file
# t1 = threading.Thread(target=scheduler_thread, daemon=True)
# t2 = threading.Thread(target=runner_thread, daemon=True)
# t1.start()
# t2.start()
# t1.join()
# t2.join()
pt = 20260630
cn =1
result = bh_utils.query_data(f"select * from geo_third_task_log where cn = {cn} and pt = {pt}") or []
for i in result:
req_id = i.get("reqId")
if not req_id:
continue
file = f"geo_snipaste/{pt}/doubao/{req_id}/text.json"
try:
raw = tos_utils.get_string_from_tos(file)
if not raw:
continue
content = json.loads(raw).get("content")
if content:
doubao_process_original_data(file, content)
except Exception as e:
logger.success(f"[pt={pt} cn={cn}] 处理 {file} 失败: {e}")
# 2) 导出 Excel
out_path = to_excel2(pt, cn)
# 3) 上传 + 发飞书
file_name = os.path.basename(out_path)
file_key = upload_file(out_path, file_name)
if file_key:
chat_message(file_key)
logger.success(f"[pt={pt} cn={cn}] 已发送飞书文件: {file_name}")
else:
logger.success(f"[pt={pt} cn={cn}] 上传失败，未发送飞书消息")
\ No newline at end of file
aidso_geo/core/meituan_process.py
View file @ d8bd6be6
import requests
import json
import time
import os ,sys
from loguru import logger
import os
import sys
from apscheduler.schedulers.blocking import BlockingScheduler
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
KEEP_FIELDS = (
"prompt",
"taskId",
"reqId",
"platform",
"type",
"thinkingEnabled",
"channel",
)
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",'')
task_type = task_data.get("type",'')
url = "http://172.16.1.223:8086/api/geo/task_commit"
keep_fields = ["prompt", "taskId", "reqId", "platform", "type", "thinkingEnabled"]
data = {k: i.get(k) for k in keep_fields}
fields = ["prompt", "taskId", "reqId", "platform", "type", "thinkingEnabled"]
payload = {k: data.get(k) for k in fields}
headers = {
'Content-Type': 'application/json'
}
payload = {k: task_data.get(k) for k in KEEP_FIELDS }
response = requests.request("POST", url, headers=headers, json=payload)
try:
response = requests.post(url, json=payload)
response_data = response.json()
print(response.text)
if response_data.get("code") == 200:
task_data["status"] = 'PROCESSING'
bh_utils.insert_data("geo_third_task_data",[task_data])
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:
print(e)
logger.error(e)
def get_result_api(task_data):
req_id = task_data.get('reqId')
req_id = task_data.get("reqId",'')
prompt = task_data.get("prompt",'')
platform = task_data.get("platform",'')
channel = task_data.get("channel",'')
task_type = task_data.get("type",'')
url = f"http://172.16.1.223:8086/api/geo/task_check?reqId={req_id}"
payload={}
response = requests.request("GET", url, data=payload)
try:
response = requests.get(url)
response_data = response.json()
if response_data.get("code") == 200 and response_data.get('data').get('status') == 'success':
task_data["status"] = 'SUCCESS'
bh_utils.insert_data("geo_third_task_data",[task_data])
print(response.text)
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:
print(e)
logger.error(e)
def query_task_commit():
return bh_utils.query_data(
"select * from geo_third_task_data where status = 'ING'"
)
def query_task_check():
return bh_utils.query_data(
"select * from geo_third_task_data where status = 'PROCESSING'"
)
def task_commit():
data_list = bh_utils.query_data("select * from geo_third_task_data where status = 'ING' ")
return data_list
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()
print(check_tasks)
if check_tasks:
for task in query_task_check():
get_result_api(task)
def get_task():
data_list = bh_utils.query_data(
"select * from geo_third_task_data where status = 'PROCESSING' ")
return data_list
if __name__ == '__main__':
try:
while 1:
commit_data_list = task_commit()
if commit_data_list:
for i in commit_data_list:
task_commit_api(i)
task_data_list = get_task()
if task_data_list:
for i in task_data_list:
get_result_api(i)
# print('sleep')
time.sleep(10)
except Exception as e:
print(e)
if __name__ == "__main__":
logger.info("调度 启动")
scheduler = BlockingScheduler(timezone="Asia/Shanghai")
# task_check()
scheduler.add_job(
task_commit,
trigger="interval",
seconds=30,
id='third_task_commit',
max_instances=1,
coalesce=True,
replace_existing=True
)
# {"prompt": "羽绒服品牌推荐?",
# "taskId": "21fd537f-131b-469a-9d5c-6eaccc00da2c11011d61",
# "reqId": "5f4d87b5-7c9e-4095-81d2-7fb9ccccc44cff111111e1",
# "platform": "TYQWA",
# "type": "stream",
# "thinkingEnabled": "1"}
\ No newline at end of file
scheduler.add_job(
task_check,
trigger="interval",
seconds=30,
id='third_task_check',
max_instances=1,
coalesce=True,
replace_existing=True
)
scheduler.start()
logger.info("third_task_commit 启动")
logger.info("third_task_check 启动")
#
\ No newline at end of file
aidso_geo/models/doubao_android_data_process.py
View file @ d8bd6be6
......@@ -293,7 +293,6 @@ def doubao_mobile_process_original_data(data):
new_url_list.append(item)
url_list = new_url_list
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)
......@@ -326,10 +325,11 @@ if __name__ == '__main__':
# 电商+poi 21fd5xxx37f-131b-4ss69axxxxx-9d5c-6ea00da2c0d6 0提及
# aa = ['667b7528-5ff4-4ccb-bb98-7d0b5972014d']
# https://tcdn.aidso.com/geo/6a5a03b63da04b9d809ea9fd27e266f6/DOUBA/rich_media_block.txt?secret=1104ee98dd68e955a5b45452010a5c54060b570e000d5849524855
# 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 = '798f148f-6896-4853-b589-d968cd2d150e' and platform = 'DOUBA'")
f"select * from geo_commit_task where taskId = '6a5a03b63da04b9d809ea9fd27e266f6' and platform = 'DOUBA'")
......
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