Skip to content

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

飞书截图_v2

parent d9c9e149
Show whitespace changes
Inline Side-by-side
Showing with 1289 additions and 0 deletions
aidso_geo/core/feishu_snipaste_v2.py 0 → 100644
View file @ fbffc4cc
import json
from datetime import datetime, timedelta
from apscheduler.schedulers.blocking import BlockingScheduler
import openpyxl
import requests
from urllib.parse import urlparse, parse_qs
import redis
from loguru import logger
import uuid
from openpyxl.workbook import Workbook
import time
import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from aidso_geo.models import spider_save_tos
from aidso_geo.utils import bh_utils, tos_utils
APP_ID = "cli_aaac8fa146fadbd5"
APP_SECRET = "N5omKg6d3sHi3JEIopQ8lcDAZKOiaNQ4"
SHEET_URL = "https://s12is4u3s19.feishu.cn/sheets/THiJsNAukhTuwjt47kgc0TiZneb"
SHEET_ID = "THiJsNAukhTuwjt47kgc0TiZneb"
POLL_SECONDS = 180
EXCEL_DIR = "/Users/yaowentong/Desktop"
INIT_TASK_LOCK_EXPIRE = 3 * 24 * 3600 # 3天：防止 key 堆积
container_id = 'oc_60a61d199f4a81b673f01b489b40509b'
def parse_spreadsheet_token(sheet_url: str) -> str:
parsed = urlparse(sheet_url)
parts = parsed.path.strip("/").split("/")
if len(parts) >= 2 and parts[0] == "sheets":
return parts[1]
raise ValueError(f"无法解析 spreadsheet_token: {sheet_url}")
def request_json(method, url, **kwargs):
resp = requests.request(method, url, timeout=30, **kwargs)
try:
data = resp.json()
except Exception:
raise RuntimeError(f"返回不是 JSON: {resp.text}")
if resp.status_code >= 400:
raise RuntimeError(f"HTTP 请求失败: status={resp.status_code}, body={data}")
if data.get("code") != 0:
raise RuntimeError(f"飞书业务错误: {data}")
return data
def get_tenant_access_token(app_id: str, app_secret: str) -> str:
url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
data = request_json(
"POST",
url,
json={
"app_id": app_id,
"app_secret": app_secret
}
)
return data["tenant_access_token"]
def get_sheet_meta(spreadsheet_token: str, tenant_access_token: str) -> dict:
url = f"https://open.feishu.cn/open-apis/sheets/v3/spreadsheets/{spreadsheet_token}/sheets/query"
headers = {
"Authorization": f"Bearer {tenant_access_token}",
"Content-Type": "application/json; charset=utf-8"
}
return request_json("GET", url, headers=headers)
def get_first_sheet_id(spreadsheet_token: str, tenant_access_token: str) -> str:
data = get_sheet_meta(spreadsheet_token, tenant_access_token)
sheets = data.get("data", {}).get("sheets", [])
if not sheets:
raise RuntimeError(f"没有获取到 sheets: {data}")
first_sheet = sheets[0]
sheet_id = first_sheet.get("sheet_id")
if not sheet_id:
raise RuntimeError(f"没有 sheet_id: {first_sheet}")
return sheet_id
def batch_get_values(spreadsheet_token: str, tenant_access_token: str, ranges: list):
url = f"https://open.feishu.cn/open-apis/sheets/v2/spreadsheets/{spreadsheet_token}/values_batch_get"
headers = {
"Authorization": f"Bearer {tenant_access_token}",
"Content-Type": "application/json; charset=utf-8"
}
params = []
for r in ranges:
params.append(("ranges", r))
return request_json("GET", url, headers=headers, params=params)
def get_not_null_values(values):
"""
把飞书返回的二维数组，转成一维数组，并过滤 None
"""
result = []
for row in values:
for cell in row:
if cell is not None:
result.append(cell)
return result
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
)
return redis_client
except Exception as e:
return None
def init_task(pt_for_tomorrow: str, platform):
spreadsheet_token = parse_spreadsheet_token(SHEET_URL)
tenant_access_token = get_tenant_access_token(APP_ID, APP_SECRET)
sheet_id = get_first_sheet_id(spreadsheet_token, tenant_access_token)
platform_map = {
'DB': [
f"{sheet_id}!A2:A10000"
],
'DP': [
f"{sheet_id}!B2:B10000"
],
'TXYB': [
f"{sheet_id}!C2:C10000"
],
'TYQW': [
f"{sheet_id}!D2:D10000"
],
}
result = batch_get_values(
spreadsheet_token=spreadsheet_token,
tenant_access_token=tenant_access_token,
ranges=platform_map[platform]
)
value_ranges = result.get("data", {}).get("valueRanges", [])
data_list = []
for value_range in value_ranges:
values = value_range.get("values", [])
data = get_not_null_values(values)
for cn in range(3):
for index,p in enumerate(data):
req_id = str(uuid.uuid4())
data_list.append({
"reqId": req_id,
"prompt": p,
"platform": platform,
"cn": cn,
"pt": pt_for_tomorrow,
"rank": index,
"channel":'MT'
})
bh_utils.insert_data("geo_feishu_snipaste", data_list)
logger.success(f"[init_task] 已写入任务: pt={pt_for_tomorrow},platform={platform}, rows={len(data_list)}")
def init_task_once(pt_for_tomorrow: str, platform):
"""
只对某个 pt 生成一次任务（避免 scheduler 线程重复触发或重启重复写）
"""
r = init_redis4()
if not r:
# redis 不可用时，退化为直接执行（可能重复写入，至少不影响流程）
init_task(pt_for_tomorrow, platform)
return
lock_key = f"mt:init_task_done:{pt_for_tomorrow}:{platform}"
ok = r.set(lock_key, "1", nx=True, ex=INIT_TASK_LOCK_EXPIRE)
if ok:
init_task(pt_for_tomorrow, platform)
else:
logger.success(f"[init_task_once] 已生成过 pt={pt_for_tomorrow}，跳过")
def init_task_scheduler():
tomorrow = datetime.now().date() + timedelta(days=1)
pt_tomorrow = tomorrow.strftime("%Y%m%d")
plate_form =['DB','DP','TXYB','TYQW']
for i in plate_form:
init_task_once(pt_tomorrow,i)
def wait_mt_tasks_done(redis_client,platform):
while True:
try:
size = redis_client.scard(f"mt:feishu_snipaste:{platform}")
logger.success(f"platform:{platform}size:{size}")
except Exception as e:
logger.success(f"redis 读取 mt:feishu_snipaste:{platform} 失败: {e}")
size = -1
if size == 0:
return
time.sleep(POLL_SECONDS)
def doubao_process_original_data(file_path, original_content):
url_list = ""
think_content = ""
response_content = ""
search_keyword = []
suggestions = []
is_think = False
rich_media_block = []
think_bool = False
response_bool = False
content_list = original_content.split("\n")
for i in content_list:
if i != "":
if not i.startswith("data:"):
continue
payload = i[len("data:"):].lstrip()
if not payload:
continue
try:
json_content = json.loads(payload)
except (IndexError, json.JSONDecodeError):
continue
if json_content.get('event_type') == 2001:
even_data = json.loads(json_content.get('event_data'))
message_data = even_data.get('message')
if even_data.get('tts_content') is not None:
response_content = even_data.get('tts_content')
if message_data.get('content_type') == 2007:
for i in json.loads(message_data.get('content')).get("search_result").get("video_card").get(
"card_list"):
rich_media_block.append(i)
if message_data.get('content_type') == 10040 and message_data.get('is_finish') is None:
think_bool = True
continue
if message_data.get('content_type') == 10040 and message_data.get('is_finish') == True:
think_bool = False
continue
if think_bool:
if json.loads(message_data.get('content')).get('text') is not None:
think_content += json.loads(message_data.get('content')).get('text')
content_json = json.loads(message_data.get('content'))
if message_data.get('content_type') == 10025 and content_json.get('results') is not None:
think_content += "\n\n"
think_content += "**搜索"
think_content += str(len(content_json.get('queries')))
think_content += "个关键词，参考"
think_content += str(len(content_json.get('results')))
think_content += "篇文章**"
think_content += "\n\n"
if message_data.get('content_type') == 10025:
content_json = json.loads(message_data.get('content'))
if content_json.get('queries') is not None and content_json.get('results') is not None:
search_keyword = search_keyword + json.loads(message_data.get('content')).get('queries')
if content_json.get('scene') == 2:
url_list = content_json.get('results')
if message_data.get('content_type') == 2002:
suggestions = suggestions + json.loads(message_data.get('content')).get('suggestions')
else:
if json_content.get("patch_op"):
if json_content.get('patch_op')[0].get("patch_object") == 111:
if json_content.get('patch_op')[0].get("patch_value").get("tts_content"):
response_content += json_content.get('patch_op')[0].get("patch_value").get("tts_content")
if json_content.get('patch_op')[0].get("patch_object") == 1:
if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
"content").get("search_query_result_block"):
search_keyword = json_content.get('patch_op')[0].get("patch_value").get("content_block")[
0].get("content").get("search_query_result_block").get("queries")
url_list = json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
"content").get("search_query_result_block").get("results")
if is_think == True and \
json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
"content").get("search_query_result_block").get("results") is not None:
think_content += "\n\n"
think_content += "**"
think_content += \
json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
"content").get("search_query_result_block").get("summary")
think_content += "**"
think_content += "\n\n"
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(
"parent_id") and len(json_content.get('patch_op')) > 1:
is_think = True
if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
"content").get('text_block').get("text"):
think_content += \
json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
"content").get("text_block").get("text")
continue
if json_content.get('patch_op')[0].get("patch_value").get("content_block")[0].get(
"block_type") == 10040:
is_think = False
continue
if json_content.get('patch_op')[0].get("patch_object") == 50:
for sug in json.loads(
json_content.get('patch_op')[0].get("patch_value").get("ext").get("sp_v2")):
suggestions.append(sug.get("content"))
if is_think:
if json_content.get("text"):
think_content += json_content.get("text")
else:
if json_content.get("content"):
content_block = json_content.get("content").get('content_block')
if content_block:
if len(content_block) > 0:
if content_block[0].get('block_type') == 10050:
for i in content_block[0].get('content').get('rich_media_block').get('creations'):
rich_media_block.append(i.get('video'))
if content_block[0].get('block_type') == 10000:
if content_block[0].get("content").get("text_block").get("text"):
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)
return (file_path, search_keyword, url_list, think_content, response_content, suggestions)
def deepseek_process_original_data(file_path, original_content):
url_list = ""
think_content = ""
response_content = ""
search_keyword = []
is_think = False
think_bool = False
suggestions = []
response_bool = False
# 按空行分割内容，过滤空字符串
content_list = [item for item in original_content.split("\n\n") if item]
for item in content_list:
if item.startswith("event: "):
# 可根据需要补充event数据处理逻辑
continue
# 处理data类型数据
if item.startswith("data: "):
try:
# 提取并解析JSON数据
data_str = item.split("data: ")[1]
json_data = json.loads(data_str)
except (IndexError, json.JSONDecodeError):
continue # 跳过格式错误的数据
if isinstance(json_data.get('v'),dict):
if json_data.get('v').get('response').get('thinking_enabled') == True:
is_think=True
if json_data.get('v').get('response').get('search_enabled') == True:
if len(json_data.get('v').get('response').get('fragments'))>0:
query_list = json_data.get('v').get('response').get('fragments')[0].get('queries')
if query_list:
result = [item.get('query', '') for item in query_list]
search_keyword.extend(result)
if is_think:
if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'SEARCH':
search_keyword.append(json_data.get('v')[0].get('queries')[0].get('query'))
if json_data.get('p') == 'response/fragments/0/results' or json_data.get(
'p') == 'response/fragments/-1/results':
url_list = json_data.get('v')
if json_data.get('p') =='response/fragments' and json_data.get('v')[0].get('type') == 'THINK':
think_bool = True
think_content+=json_data.get('v')[0].get('content')
continue
if json_data.get('p') == 'response' :
if isinstance(json_data.get('v')[0].get('v'),list):
if json_data.get('v')[0].get('v')[0].get('type') == 'TOOL_SEARCH':
for q in json_data.get('v')[0].get('v')[0].get('queries'):
search_keyword.append(q.get('query'))
if json_data.get('v')[1].get('p') == 'fragments' or json_data.get('v')[1].get('p') == 'response/fragments':
if json_data.get('v')[1].get('v')[0].get('type') == 'THINK':
think_content+=json_data.get('v')[1].get('v')[0].get('content')
think_bool = True
continue
if json_data.get('v')[0].get('p') == 'fragments' or json_data.get('v')[0].get('p') == 'response/fragments':
if json_data.get('v')[0].get('v')[0].get('type') == 'THINK':
think_content+=json_data.get('v')[0].get('v')[0].get('content')
think_bool = True
continue
if json_data.get('v')[0].get('v') == 'FINISHED':
response_bool = False
if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'RESPONSE':
think_bool = False
response_bool = True
response_content += json_data.get('v')[0].get('content')
continue
if json_data.get('p') == 'response/fragments/1/elapsed_secs':
think_bool = False
if json_data.get('p') == 'response/fragments/-1/elapsed_secs':
think_bool = False
if json_data.get('p') == 'response/fragments':
response_bool = False
if response_bool:
if isinstance(json_data.get('v'), str):
response_content += json_data.get('v')
if think_bool:
if isinstance(json_data.get('v'), str):
think_content+=json_data.get('v')
else:
if json_data.get('p') == 'response/fragments/-1/content':
response_bool = True
if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'SEARCH':
search_keyword.append(json_data.get('v')[0].get('queries')[0].get('query'))
if json_data.get('p') == 'response/fragments/0/results' or json_data.get('p') == 'response/fragments/-1/results':
url_list = json_data.get('v')
if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'RESPONSE':
response_content+=json_data.get('v')[0].get('content')
response_bool = True
continue
if json_data.get('p') == 'response':
if json_data.get('v')[1].get('p') == 'fragments':
if json_data.get('v')[1].get('v')[0].get('type') == 'RESPONSE':
response_content+=json_data.get('v')[1].get('v')[0].get('content')
response_bool = True
continue
if json_data.get('v')[0].get('p') == 'fragments':
if json_data.get('v')[0].get('v')[0].get('type') == 'RESPONSE':
response_content+=json_data.get('v')[0].get('v')[0].get('content')
response_bool = True
continue
if json_data.get('v')[0].get('v') == 'FINISHED':
response_bool = False
if json_data.get('p') == 'response/fragments' and json_data.get('v')[0].get('type') == 'TIP':
response_bool= False
continue
if json_data.get('v') == 'FINISHED':
response_bool = False
if response_bool:
if isinstance(json_data.get('v'),str):
response_content+=json_data.get('v')
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)
def yuanbao_process_original_data(file_path, original_content):
url_list = []
think_content = ""
response_process = ""
response_content = ""
search_keyword = []
suggestions = []
is_think = False
think_bool = False
response_bool = False
rich_media_block =[]
content_list = original_content.split("\n\n")
for i in content_list:
if i.startswith("data: "):
try:
# 提取并解析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')
if json_data.get('type') == 'text' and json_data.get('msg') is not None:
response_content+=json_data.get('msg')
if json_data.get('type') == 'deepSearch':
if json_data.get('contents'):
if json_data.get('contents')[0].get('msg'):
think_content += json_data.get('contents')[0].get('msg')
if not response_process:
response_content += json_data.get('contents')[0].get('msg')
if json_data.get('type') == 'image':
response_content = "生成了图片"
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)
def tongyi_process_original_data(file_path, original_content):
url_list = []
url_list_batch = []
think_content = ""
response_content = ""
search_keyword = []
suggestions = []
is_think = False
think_bool = False
response_bool = False
result = []
content_list = original_content.split("\n")
for i in content_list:
if i.startswith("data:"):
try:
# 提取并解析JSON数据
data_str = i.split("data:")[1]
json_data = json.loads(data_str)
except (IndexError, json.JSONDecodeError):
continue
#
if isinstance(json_data, dict):
if json_data.get('msgStatus'):
if json_data.get("incremental") == False:
if json_data.get('msgStatus') == 'finished':
if json_data.get('contents'):
for cn in json_data.get('contents'):
if cn.get('contentType') == 'plugin':
pluginResult = json.loads(cn.get('content')).get('pluginResult')
if pluginResult:
try:
if isinstance(json.loads(pluginResult), list):
if len(json.loads(pluginResult)) >= 2:
url_list = json.loads(pluginResult)[1].get(
'search_results')
if isinstance(json.loads(pluginResult), dict):
url_list = json.loads(pluginResult).get(
'links')
except Exception as e:
...
if cn.get('contentType') == 'think':
think_content = json.loads(cn.get('content')).get('content')
if cn.get('contentType') == 'text':
response_content = cn.get('content')
if json_data.get("data"):
if json_data.get("data").get('status'):
if json_data.get("data").get('status') == 'complete':
messages = json_data.get('data').get('messages')
for cn in messages:
if cn.get('mime_type') == 'bar/iframe' and cn.get('status') == 'complete':
url_list_batch = cn.get('meta_data').get('sources')[0].get('content').get(
'list')
if cn.get('mime_type') == 'multi_load/iframe' and cn.get('status') == 'complete':
response_content = cn.get('content')
response_content = response_content.replace("[(deep_think)]", "")
multi_load = cn.get('meta_data').get('multi_load')
if multi_load:
for mu in multi_load:
if mu.get('type') == 'deep_think':
if mu.get('content').get('status') == 'complete':
think_content = mu.get('content').get('think_content')
if json_data.get("data").get('messages'):
messages = json_data.get("data").get('messages')
for ms in messages:
if ms.get('mime_type') == 'multi_load/iframe' and ms.get('status') == 'complete':
response_content = ms.get('content')
response_content = response_content.replace("[(deep_think)]", "")
multi_load = ms.get('meta_data').get('multi_load')
if multi_load:
for mu in multi_load:
if mu.get('type') == 'deep_think':
if mu.get('content').get('status') == 'complete':
think_content = mu.get('content').get('think_content')
if ms.get('mime_type') == 'bar/iframe' and ms.get('status') == 'complete':
url_list_batch = ms.get('meta_data').get('sources')[0].get('content').get('list')
if ms.get('mime_type') == 'paa/iframe' and ms.get('status') == 'complete':
paas = ms.get('meta_data').get('paas')
if paas:
for pa in ms.get('meta_data').get('paas'):
suggestions.append(pa.get('show_text'))
if url_list_batch:
for url in url_list_batch:
url_list.append(
{
"url": url.get('url', ''),
"title": url.get('title', ''),
"snippet": url.get('summary', ''),
"host_name": url.get('name', ''),
"host_logo": url.get('icon', ''),
"time": url.get('publish_time', ''),
}
)
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)
PLATFORM_PROCESS_MAP = {
"DB": doubao_process_original_data,
"DP": deepseek_process_original_data,
"TXYB": yuanbao_process_original_data,
"TYQW": tongyi_process_original_data,
}
def timestamp_to_datetime(timestamp, fmt="%Y-%m-%d %H:%M:%S"):
timestamp = int(timestamp)
local_time = datetime.fromtimestamp(timestamp)
return local_time.strftime(fmt)
def get_context(image_url):
url = "https://ark.cn-beijing.volces.com/api/v3/responses"
payload = json.dumps({
"model": "doubao-seed-2-0-lite-260428",
"input": [
{
"role": "user",
"content": [
{
"type": "input_image",
"image_url": image_url
},
{
"type": "input_text",
"text": "提取图中文字"
}
]
}
]
})
headers = {
'Authorization': 'Bearer fcc424e5-58af-494d-9683-5787413a26c9',
'Content-Type': 'application/json'
}
try:
response = requests.request("POST", url, headers=headers, data=payload)
data = response.json()
return next(
(
content.get("text", "")
for item in data.get("output", [])
if item.get("type") == "message"
for content in item.get("content", [])
if content.get("type") == "output_text"
),
""
)
except Exception as e:
return ""
class ExcelWriter:
"""
Excel写入工具类，专门用于写入taskID和response两列数据
"""
def __init__(self, file_path):
"""
初始化Excel写入工具
:param file_path: Excel文件保存路径（如: './output.xlsx'）
"""
self.file_path = file_path
self.workbook = None
self.worksheet = None
# 初始化工作簿和工作表
self._init_workbook()
def _init_workbook(self):
"""初始化工作簿和工作表，若文件已存在则打开，不存在则新建"""
# 检查文件是否存在
if os.path.exists(self.file_path):
self.workbook = openpyxl.load_workbook(self.file_path)
# 取第一个工作表
self.worksheet = self.workbook.active
# 检查表头是否存在，不存在则添加
if self.worksheet.cell(row=1, column=1).value != 'taskID' or \
self.worksheet.cell(row=1, column=2).value != 'response':
# 在第一行插入表头
self.worksheet.insert_rows(1)
self.worksheet.cell(row=1, column=1, value='taskID')
self.worksheet.cell(row=1, column=2, value='response')
else:
# 新建工作簿
self.workbook = Workbook()
self.worksheet = self.workbook.active
# 设置表头
self.worksheet.cell(row=1, column=1, value='问题')
self.worksheet.cell(row=1, column=2, value='平台')
self.worksheet.cell(row=1, column=3, value='回答')
self.worksheet.cell(row=1, column=4, value='思考过程')
self.worksheet.cell(row=1, column=5, value='引用来源')
self.worksheet.cell(row=1, column=6, value='分享链接')
self.worksheet.cell(row=1, column=7, value='截图')
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='排名')
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):
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)
self.worksheet.cell(row=next_row + idx, column=4, value=think)
self.worksheet.cell(row=next_row + idx, column=5, value=quote)
self.worksheet.cell(row=next_row + idx, column=6, value=share)
self.worksheet.cell(row=next_row + idx, column=7, value=snipaste)
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.workbook.save(self.file_path)
def close(self):
"""关闭工作簿，释放资源"""
if self.workbook:
self.workbook.close()
def to_excel2(pt, cn,platform):
query_list_log = bh_utils.query_data(
f"SELECT DISTINCT reqId FROM geo_feishu_snipaste where cn = {cn} and pt = {pt} and platform = '{platform}'"
)
uuid_list = []
for i in query_list_log:
uuid_list.append(i.get('reqId'))
uuid_str = ",".join([f"'{uuid}'" for uuid in uuid_list])
query_list = bh_utils.query_data(
f"SELECT DISTINCT reqId,prompt, platform,insertime,rank FROM geo_third_task WHERE reqId IN ({uuid_str}) order by rank asc"
)
task_list = []
for i in query_list:
task_list.append(
{
"reqId": i.get("reqId"),
"prompt": i.get("prompt"),
"platform": i.get("platform"),
"insertime": i.get("insertime"),
"rank": i.get("rank")
}
)
plat_form_map = {
"DP": "deepseek网页版",
"DB": "豆包网页版",
"TXYB": "腾讯元宝",
"TYQW": "通义千问",
"KIMI": "kimi",
"WXYY": "文心一言",
"BDAI": "百度ai",
"DYAI": "抖音ai",
"DOUBA": "豆包安卓版",
"DPA": "deepseek安卓版",
}
def build_row(i):
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}/{platform}/{reqId}/text.json"
context_path = f"geo_snipaste/{pt}/{platform}/{reqId}/context.txt"
quote_path = f"geo_snipaste/{pt}/{platform}/{reqId}/quote.txt"
think = f"geo_snipaste/{pt}/{platform}/{reqId}/think.txt"
png_url = f'https://tcdn.aidso.com/geo_snipaste/{pt}/{platform}/{reqId}/png.png'
text_json_str = tos_utils.get_string_from_tos(file) or "{}"
return (
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(text_json_str).get('share_url', ''),
png_url,
timestamp_to_datetime(insertime),
get_context(png_url),
rank
)
batch_data = [None] * len(task_list)
with ThreadPoolExecutor(max_workers=50) as executor:
future_map = {
executor.submit(build_row, i): index
for index, i in enumerate(task_list)
}
for future in as_completed(future_map):
index = future_map[future]
try:
batch_data[index] = future.result()
except Exception as e:
logger.exception(f"导出数据失败 index={index}, task={task_list[index]}, err={e}")
batch_data[index] = None
# 去掉失败的数据
batch_data = [row for row in batch_data if row is not None]
logger.success(f"时间{pt}--{cn}---导出数据{len(batch_data)}条")
out_path = os.path.join(EXCEL_DIR, f"美团-{pt}-{platform}-{cn}.xlsx")
excel_writer = ExcelWriter(out_path)
excel_writer.write_batch_rows(batch_data)
excel_writer.close()
return out_path
def upload_file(file_path, file_name):
url = "https://open.feishu.cn/open-apis/im/v1/files"
headers = {"Authorization": f"Bearer {get_tenant_access_token(APP_ID, APP_SECRET)}"}
files = {
"file": (
file_name,
open(file_path, "rb"),
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)
}
data = {
"file_type": "xlsx",
"file_name": file_name
}
try:
resp = requests.post(url=url, headers=headers, files=files, data=data, timeout=30)
rj = resp.json()
return (rj.get("data") or {}).get("file_key")
except FileNotFoundError as e:
logger.success(f"错误：文件不存在 - {e}")
except Exception as e:
logger.success(f"请求失败：{e}")
finally:
try:
f = files.get("file")[1]
if f:
f.close()
except Exception:
pass
return None
def chat_message(file_key):
url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id"
payload = json.dumps({
"content": f'{{"file_key":"{file_key}"}}',
"msg_type": "file",
"receive_id": container_id,
"uuid": str(uuid.uuid4())
})
headers = {
"Authorization": f"Bearer {get_tenant_access_token(APP_ID, APP_SECRET)}",
"Content-Type": "application/json"
}
resp = requests.request("POST", url, headers=headers, data=payload, timeout=30)
logger.success(resp.text)
def process_cn_after_tasks_done(pt: str, cn: int,platform):
redis_client = init_redis4()
logger.success(f"[pt={pt} cn={cn}] 等待 {platform} 跑空...")
wait_mt_tasks_done(redis_client,platform)
logger.success(f"[pt={pt} cn={cn}] {platform} 已为空，初步完成")
for i in range(3):
diff_result = bh_utils.query_data(
f"select a1.reqId,a1.prompt,a1.`rank` from (select reqId,prompt,`rank` from geo_feishu_snipaste where pt = {pt} and cn = {cn} and platform ='{platform}') as a1 left join geo_third_task as a2 on a1.reqId = a2.reqId where a2.reqId is null")
if not diff_result:
logger.success(f"[pt={pt} cn={cn}] {platform} 第{i + 1}轮检查已无缺失")
break
for row in diff_result:
row["thinking_enabled"] = "1"
values = [json.dumps(row, ensure_ascii=False) for row in diff_result]
redis_client.sadd(f"mt:feishu_snipaste:{platform}", *values)
logger.success(f"[pt={pt} cn={cn} {platform}] 缺失数据开始补充")
wait_mt_tasks_done(redis_client, platform)
logger.success(f"[pt={pt} cn={cn} {platform}] 缺失数据补充完成")
time.sleep(120)
logger.success(f"[pt={pt} cn={cn}] {platform} 已为空，开始汇总导出/发送")
# 1) 解析原始 json
result = bh_utils.query_data(f"select * from geo_feishu_snipaste where cn = {cn} and pt = {pt} and platform ='{platform}' ") or []
for i in result:
req_id = i.get("reqId")
if not req_id:
continue
file = f"geo_snipaste/{pt}/{platform}/{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:
process_func = PLATFORM_PROCESS_MAP.get(platform)
process_func(file, 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,platform)
# 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} {platform}] 已发送飞书文件: {file_name}")
else:
logger.success(f"[pt={pt} cn={cn} {platform}] 上传失败，未发送飞书消息")
def send_task(prompt_list,pt,count,platform_list):
data_list = []
for platform in platform_list:
for cn in range(count):
for index,p in enumerate(prompt_list):
req_id = str(uuid.uuid4())
data_list.append({
"reqId": req_id,
"prompt": p,
"platform": platform,
"cn": cn,
"pt": pt,
"rank": index
})
logger.success(f"[init_task] 已写入任务: pt={pt}, prompts={len(prompt_list)}, rows={len(data_list)}")
bh_utils.insert_data("geo_feishu_snipaste", data_list)
def push_platform_tasks(pt_today: str, platform: str):
"""
单个平台投递任务：
cn=0/1/2 顺序写入该平台自己的 Redis 队列
"""
r = init_redis4()
redis_key = f"mt:feishu_snipaste:{platform}"
for cn in (0, 1, 2):
try:
rows = bh_utils.query_data(
f"""
SELECT reqId, prompt, rank
FROM geo_feishu_snipaste
WHERE cn = {cn}
AND pt = {pt_today}
AND platform = '{platform}'
ORDER BY rank ASC
"""
) or []
for row in rows:
row["thinking_enabled"] = "1"
values = [json.dumps(row, ensure_ascii=False) for row in rows]
if values:
r.sadd(redis_key, *values)
process_cn_after_tasks_done(pt_today, cn,platform)
logger.success(
f"[push_platform_tasks] pt={pt_today}, platform={platform}, cn={cn}, count={len(values)}, redis_key={redis_key}"
)
except Exception as e:
logger.exception(
f"[push_platform_tasks] pt={pt_today}, platform={platform}, cn={cn} 执行失败: {e}"
)
continue
def web_hook_alone(count,pt,tos_pt,platform):
redis_client = init_redis4()
redis_key = f"mt:feishu_snipaste:{platform}"
for cn in range(count):
rows = bh_utils.query_data(
f"SELECT reqId, prompt, rank FROM geo_feishu_snipaste WHERE cn = {cn} AND pt = {pt} AND platform = '{platform}'ORDER BY rank ASC"
) or []
for row in rows:
row["thinking_enabled"] = "1"
values = [json.dumps(row, ensure_ascii=False) for row in rows]
if values:
redis_client.sadd(redis_key, *values)
logger.success(f"[pt={pt} cn={cn} {platform}] 缺失数据开始补充")
wait_mt_tasks_done(redis_client, platform)
logger.success(f"[pt={pt} cn={cn} {platform}] 缺失数据补充完成")
for i in range(3):
diff_result = bh_utils.query_data(
f"select a1.reqId,a1.prompt,a1.`rank` from (select reqId,prompt,`rank` from geo_feishu_snipaste where pt = {pt} and cn = {cn} and platform ='{platform}') as a1 left join geo_third_task as a2 on a1.reqId = a2.reqId where a2.reqId is null")
for row in diff_result:
row["thinking_enabled"] = "1"
values = [json.dumps(row, ensure_ascii=False) for row in diff_result]
if not values:
logger.success(f"[pt={pt} cn={cn}] 第{i + 1}轮检查已无缺失")
break
redis_client.sadd(redis_key, *values)
logger.success(f"[pt={pt} cn={cn} {platform}] 缺失数据开始补充")
wait_mt_tasks_done(redis_client, platform)
logger.success(f"[pt={pt} cn={cn} {platform}] 缺失数据补充完成")
time.sleep(180)
logger.success(f"[pt={pt} cn={cn} {platform}] 已为空，开始汇总导出/发送")
result = bh_utils.query_data(f"select * from geo_feishu_snipaste where cn = {cn} and pt = {pt} and platform ='{platform}'") or []
for i in result:
req_id = i.get("reqId")
if not req_id:
continue
file = f"geo_snipaste/{tos_pt}/{platform}/{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:
process_func = PLATFORM_PROCESS_MAP.get(platform)
process_func(file, content)
except Exception as e:
logger.success(f"[pt={pt} cn={cn}] 处理 {file} 失败: {e}")
query_list_log = bh_utils.query_data(
f"select * from geo_feishu_snipaste where cn = {cn} and pt = {pt} and platform ='{platform}'"
)
uuid_list = []
for i in query_list_log:
uuid_list.append(i.get('reqId'))
uuid_str = ",".join([f"'{uuid}'" for uuid in uuid_list])
query_list = bh_utils.query_data(
f"SELECT DISTINCT reqId,prompt, platform,insertime,rank FROM geo_third_task WHERE reqId IN ({uuid_str}) order by rank asc"
)
task_list = []
for i in query_list:
task_list.append(
{
"reqId": i.get("reqId"),
"prompt": i.get("prompt"),
"platform": i.get("platform"),
"insertime": i.get("insertime"),
"rank": i.get("rank")
}
)
plat_form_map = {
"DP": "deepseek网页版",
"DB": "豆包网页版",
"TXYB": "腾讯元宝",
"TYQW": "通义千问",
"KIMI": "kimi",
"WXYY": "文心一言",
"BDAI": "百度ai",
"DYAI": "抖音ai",
"DOUBA": "豆包安卓版",
"DPA": "deepseek安卓版",
}
def build_row(i):
reqId = i.get("reqId")
prompt = i.get("prompt")
platform = i.get("platform")
insertime = i.get("insertime")
rank = i.get("rank")
file = f"geo_snipaste/{tos_pt}/{platform}/{reqId}/text.json"
context_path = f"geo_snipaste/{tos_pt}/{platform}/{reqId}/context.txt"
quote_path = f"geo_snipaste/{tos_pt}/{platform}/{reqId}/quote.txt"
think = f"geo_snipaste/{tos_pt}/{platform}/{reqId}/think.txt"
png_url = f'https://tcdn.aidso.com/geo_snipaste/{tos_pt}/{platform}/{reqId}/png.png'
text_json_str = tos_utils.get_string_from_tos(file) or "{}"
return (
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(text_json_str).get('share_url', ''),
png_url,
timestamp_to_datetime(insertime),
get_context(png_url),
rank
)
batch_data = [None] * len(task_list)
with ThreadPoolExecutor(max_workers=50) as executor:
future_map = {
executor.submit(build_row, i): index
for index, i in enumerate(task_list)
}
for future in as_completed(future_map):
index = future_map[future]
try:
batch_data[index] = future.result()
except Exception as e:
logger.exception(f"导出数据失败 index={index}, task={task_list[index]}, err={e}")
batch_data[index] = None
# 去掉失败的数据
batch_data = [row for row in batch_data if row is not None]
out_path = os.path.join(EXCEL_DIR, f"美团-{pt}-{platform}-{cn}.xlsx")
excel_writer = ExcelWriter(out_path)
excel_writer.write_batch_rows(batch_data)
excel_writer.close()
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}] 上传失败，未发送飞书消息")
def web_hook(promp_list,platform_list):
count = 1
pt = datetime.now().strftime("%Y%m%d%H%M")
tos_pt = datetime.now().strftime("%Y%m%d")
send_task(promp_list, pt, count,platform_list)
with ThreadPoolExecutor(max_workers=len(platform_list)) as executor:
future_map = {
executor.submit(web_hook_alone,count,pt,tos_pt,platform): platform
for platform in platform_list
}
for future in as_completed(future_map):
platform = future_map[future]
try:
future.result()
logger.success(f"[web_hook] 平台处理完成: pt={pt}, platform={platform}")
except Exception as e:
logger.exception(f"[web_hook] 平台处理失败: pt={pt}, platform={platform}, err={e}")
def run_once_for_pt(pt_today: str):
"""
四个平台并发投递：
DB / DP / TXYB / TYQW 同时跑
每个平台内部 cn=0/1/2 顺序跑
"""
platform_list = ["DB", "DP", "TXYB", "TYQW"]
with ThreadPoolExecutor(max_workers=4) as executor:
future_map = {
executor.submit(push_platform_tasks, pt_today, platform): platform
for platform in platform_list
}
for future in as_completed(future_map):
platform = future_map[future]
try:
future.result()
logger.success(f"[run_once_for_pt] 平台投递完成: pt={pt_today}, platform={platform}")
except Exception as e:
logger.exception(f"[run_once_for_pt] 平台投递失败: pt={pt_today}, platform={platform}, err={e}")
def init_tomorrow_tasks():
"""
每天晚上 23:00 执行：
初始化明天各个平台的任务
"""
tomorrow = datetime.now().date() + timedelta(days=1)
pt_tomorrow = tomorrow.strftime("%Y%m%d")
platform_list = ["DB", "DP", "TXYB", "TYQW"]
for p in platform_list:
try:
init_task_once(pt_tomorrow, p)
logger.success(f"[init_tomorrow_tasks] 初始化成功 pt={pt_tomorrow}, platform={p}")
except Exception as e:
logger.exception(f"[init_tomorrow_tasks] 初始化失败 pt={pt_tomorrow}, platform={p}, err={e}")
def run_task():
pt_today = datetime.now().date().strftime("%Y%m%d")
run_once_for_pt(pt_today)
if __name__ == "__main__":
scheduler = BlockingScheduler(timezone="Asia/Shanghai")
# 每天晚上 23:00 执行
scheduler.add_job(
init_tomorrow_tasks,
trigger="cron",
hour=23,
minute=0,
second=0,
id="init_tomorrow_tasks",
replace_existing=True,
max_instances=1,
coalesce=True,
)
# 每天晚上 00:10 执行
scheduler.add_job(
run_task,
trigger="cron",
hour=0,
minute=10,
second=0,
id="run_task",
replace_existing=True,
max_instances=1,
coalesce=True,
)
logger.success(
"定时任务注册完成："
"init_tomorrow_tasks(每天23:00), "
"run_task(每天0:10),"
)
scheduler.start()
# platform_list = ["DB", "DP", "TXYB", "TYQW"]
# for platform in platform_list:
# init_task('20260610', platform)
# run_once_for_pt('20260610')
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