Skip to content

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

美团截图 

电商词搜索
parent 27ac512a
Show whitespace changes
Inline Side-by-side
Showing with 1431 additions and 103 deletions
aidso_geo/core/feishu_snipaste_v3.py
View file @ 4b1f9b3a
"""每日把飞书 Wiki 文档表格转换为截图任务并写入数据库。"""
from __future__ import annotations
import re
import json
import os
import sys
import tempfile
import uuid
from datetime import datetime
import time
......@@ -12,7 +13,16 @@ import redis
import requests
from loguru import logger
from openpyxl import Workbook
from openpyxl.styles import (
Alignment,
Border,
Font,
PatternFill,
Side,
)
from openpyxl.utils import get_column_letter
from aidso_geo.clients.tos_client import tos_client
from aidso_geo.models import spider_save_tos
POLL_SECONDS = 120
BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
......@@ -27,12 +37,34 @@ FEISHU_CHAT_ID = "oc_41bc8e496a0d9bb93498a6a6cf6f30c1"
WIKI_TOKEN = "RndAwboqZiuCAFkTGkScaWHPnmg"
TABLE_NAME = "douchacha_data.geo_feishu_snipaste_v3"
LOG_TABLE_NAME = "douchacha_data.geo_feishu_snipaste_log_v3"
RESULT_TABLE_NAME = "douchacha_data.geo_feishu_snipaste_result_v3"
ROUND_COUNT = 30
MAX_ROUND_COUNT = 30
REDIS_QUEUE_KEY = "mt:snipaste_v3:only_content"
REDIS_QUEUE_KEY_SHARE = "mt:snipaste_v3:with_share"
SYNC_HOUR = 0
SYNC_MINUTE = 1
PLATFORM_CONFIGS = {
"DB": {
"name": "豆包网页版",
"sheet_name": "豆包网页版",
"tos_directory": "doubao",
"content_redis_key": REDIS_QUEUE_KEY,
"share_redis_key": REDIS_QUEUE_KEY_SHARE,
},
"DP": {
"name": "DeepSeek网页版",
"sheet_name": "deepseek网页版",
"tos_directory": "DP",
"content_redis_key": "mt:DP:snipaste_v3:only_content",
"share_redis_key": "mt:DP:snipaste_v3:with_share",
},
}
THIRTY_ROUND_EXPORT_WORKERS = 20
EXCEL_CELL_MAX_LENGTH = 32767
ILLEGAL_EXCEL_CHARACTERS = re.compile(
r"[\x00-\x08\x0B-\x0C\x0E-\x1F]"
)
PROMPT_HEADERS = {"问题", "prompt", "问题词", "问题内容"}
......@@ -45,6 +77,16 @@ class TaskResultUnavailableError(RuntimeError):
pass
def normalize_platform(platform: str = "DB") -> str:
normalized_platform = str(platform or "").strip().upper()
if normalized_platform not in PLATFORM_CONFIGS:
raise ValueError(
f"不支持的平台: {platform!r}，"
f"可选值={list(PLATFORM_CONFIGS)}"
)
return normalized_platform
def normalize_pt(pt=None) -> str:
"""
规范化运行批次。
......@@ -157,7 +199,11 @@ def get_wiki_node(access_token: str) -> dict:
return node
def get_first_sheet(spreadsheet_token: str, access_token: str) -> dict:
def get_sheet_by_name(
spreadsheet_token: str,
access_token: str,
sheet_name: str,
) -> dict:
data = request_json(
"GET",
(
......@@ -170,9 +216,27 @@ def get_first_sheet(spreadsheet_token: str, access_token: str) -> dict:
if not sheets:
raise RuntimeError(f"飞书电子表格中没有工作表: {data}")
sheet = sheets[0]
normalized_sheet_name = str(sheet_name or "").strip().lower()
sheet = next(
(
item
for item in sheets
if str(item.get("title") or "").strip().lower()
== normalized_sheet_name
),
None,
)
if sheet is None:
available_sheet_names = [
str(item.get("title") or "")
for item in sheets
]
raise RuntimeError(
f"飞书电子表格中没有名为 {sheet_name!r} 的工作表，"
f"现有工作表={available_sheet_names}"
)
if not sheet.get("sheet_id"):
raise RuntimeError(f"飞书返回中没有 sheet_id: {sheets[0]}")
raise RuntimeError(f"飞书返回中没有 sheet_id: {sheet}")
return sheet
......@@ -187,8 +251,16 @@ def column_name(column_count: int) -> str:
return "".join(reversed(result))
def get_sheet_rows(spreadsheet_token: str, access_token: str) -> list[list]:
sheet = get_first_sheet(spreadsheet_token, access_token)
def get_sheet_rows(
spreadsheet_token: str,
access_token: str,
sheet_name: str,
) -> list[list]:
sheet = get_sheet_by_name(
spreadsheet_token,
access_token,
sheet_name,
)
sheet_id = sheet["sheet_id"]
grid_properties = sheet.get("grid_properties") or {}
row_count = int(grid_properties.get("row_count") or 1000)
......@@ -228,7 +300,21 @@ def cell_text(value) -> str:
return str(value).strip()
def build_tasks(rows: list[list], pt: str) -> list[dict]:
def build_tasks(
rows: list[list],
pt: str,
platform: str,
) -> list[dict]:
platform = normalize_platform(platform)
business_line_column_index = None
for row in rows:
for column_index, cell in enumerate(row):
if cell_text(cell) == "业务线":
business_line_column_index = column_index
break
if business_line_column_index is not None:
break
tasks = []
for row in rows:
if not row:
......@@ -239,10 +325,23 @@ def build_tasks(rows: list[list], pt: str) -> list[dict]:
continue
keywords = []
for cell in row[1:]:
for column_index, cell in enumerate(
row[1:],
start=1,
):
if column_index == business_line_column_index:
continue
keyword = cell_text(cell)
if keyword:
keywords.append(keyword)
business_line = ""
if (
business_line_column_index is not None
and business_line_column_index < len(row)
):
business_line = cell_text(
row[business_line_column_index]
)
tasks.append(
{
"taskId": str(uuid.uuid4()),
......@@ -250,31 +349,43 @@ def build_tasks(rows: list[list], pt: str) -> list[dict]:
"keywords": json.dumps(keywords, ensure_ascii=False),
"pt": pt,
"sheet_order": len(tasks) + 1,
"platform": platform,
"business_line": business_line,
}
)
return tasks
def get_tasks(pt: str) -> list[dict]:
def get_tasks(pt: str, platform: str = "DB") -> list[dict]:
platform = normalize_platform(platform)
rows = bh_utils.query_data(
(
f"SELECT taskId, prompt, keywords, pt, sheet_order "
f"FROM {TABLE_NAME} WHERE pt = %s"
f"SELECT taskId, prompt, keywords, pt, sheet_order, "
f"business_line, "
f"ifNull(platform, 'DB') AS platform "
f"FROM {TABLE_NAME} WHERE pt = %s "
f"AND ifNull(platform, 'DB') = %s"
),
(pt,),
(pt, platform),
)
if rows is None:
raise RuntimeError(f"查询 {TABLE_NAME} 失败")
return rows
def create_round_logs(tasks: list[dict], pt: str) -> int:
def create_round_logs(
tasks: list[dict],
pt: str,
platform: str = "DB",
) -> int:
platform = normalize_platform(platform)
existing_rows = bh_utils.query_data(
(
f"SELECT taskId, `count` FROM {LOG_TABLE_NAME} "
f"WHERE pt = %s"
f"WHERE pt = %s "
f"AND ifNull(platform, 'DB') = %s"
),
(pt,),
(pt, platform),
)
if existing_rows is None:
raise RuntimeError(f"查询 {LOG_TABLE_NAME} 失败")
......@@ -298,6 +409,7 @@ def create_round_logs(tasks: list[dict], pt: str) -> int:
"keywords": task.get("keywords"),
"`count`": count,
"pt": pt,
"platform": platform,
}
)
......@@ -318,7 +430,9 @@ def push_rounds_to_redis(
redis_key: str,
pt: str = None,
task_ids: list = None,
platform: str = "DB",
) -> int:
platform = normalize_platform(platform)
is_share_round = (
start_count == end_count
and start_count in (0, 99)
......@@ -367,11 +481,12 @@ def push_rounds_to_redis(
(
f"SELECT reqId, prompt, pt FROM {LOG_TABLE_NAME} "
f"WHERE pt = %s "
f"AND ifNull(platform, 'DB') = %s "
f"AND toInt32(`count`) BETWEEN %s AND %s "
f"{task_filter_sql}"
f"ORDER BY toInt32(`count`), taskId"
),
tuple(query_params),
tuple([query_params[0], platform, *query_params[1:]]),
)
if rows is None:
raise RuntimeError(
......@@ -416,10 +531,15 @@ def push_rounds_to_redis(
return len(values)
def sync_feishu_to_database(pt: str = None) -> None:
def sync_feishu_to_database(
pt: str = None,
platform: str = "DB",
) -> None:
validate_config()
platform = normalize_platform(platform)
platform_config = PLATFORM_CONFIGS[platform]
pt = normalize_pt(pt)
tasks = get_tasks(pt)
tasks = get_tasks(pt, platform=platform)
title = ""
if not tasks:
......@@ -431,9 +551,13 @@ def sync_feishu_to_database(pt: str = None) -> None:
raise RuntimeError(
f"当前 Wiki 节点类型是 {obj_type!r}，仅支持 sheet"
)
table_rows = get_sheet_rows(node["obj_token"], access_token)
table_rows = get_sheet_rows(
node["obj_token"],
access_token,
platform_config["sheet_name"],
)
tasks = build_tasks(table_rows, pt)
tasks = build_tasks(table_rows, pt, platform)
if not tasks:
raise RuntimeError("飞书文档中没有解析到可写入的表格任务")
if not bh_utils.insert_data(TABLE_NAME, tasks):
......@@ -445,10 +569,14 @@ def sync_feishu_to_database(pt: str = None) -> None:
f"rows={len(tasks)}"
)
inserted_log_count = create_round_logs(tasks, pt)
inserted_log_count = create_round_logs(
tasks,
pt,
platform=platform,
)
logger.success(
f"[第一阶段 pt={pt} count=1-30] "
f"[第一阶段 platform={platform} pt={pt} count=1-30] "
f"飞书数据库同步完成: title={title}, "
f"tasks={len(tasks)}, inserted_logs={inserted_log_count}"
)
......@@ -458,21 +586,25 @@ def start_scheduler() -> None:
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler(timezone="Asia/Shanghai")
for platform in PLATFORM_CONFIGS:
scheduler.add_job(
run_daily_pipeline_safely,
trigger="cron",
hour=SYNC_HOUR,
minute=SYNC_MINUTE,
second=0,
id="feishu_snipaste_v3_daily_pipeline",
kwargs={"platform": platform},
id=f"feishu_snipaste_v3_daily_pipeline_{platform}",
replace_existing=True,
max_instances=1,
coalesce=True,
misfire_grace_time=3600,
)
logger.success(
f"[调度 count=1-99] 七阶段每日任务已注册: "
f"{SYNC_HOUR:02d}:{SYNC_MINUTE:02d}"
f"[调度 platforms={list(PLATFORM_CONFIGS)} count=1-99] "
f"七阶段每日任务已注册: "
f"{SYNC_HOUR:02d}:{SYNC_MINUTE:02d}, "
f"jobs={len(PLATFORM_CONFIGS)}"
)
scheduler.start()
......@@ -636,6 +768,214 @@ def doubao_process_original_data(file_path, original_content):
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
url_id_map = {}
# 按空行分割内容，过滤空字符串
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
fragments = json_data.get('v').get('response').get('fragments')
if fragments:
if isinstance(fragments, list):
if fragments[0].get('type') == 'RESPONSE':
response_content += fragments[0].get('content')
if fragments[0].get('type') == 'THINK':
think_content += fragments[0].get('content')
think_bool = True
if fragments[0].get('type') == 'SEARCH':
if isinstance(fragments[0].get('queries'), list):
for sea in fragments[0].get('queries'):
search_keyword.append(sea.get('query'))
# 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 isinstance(json_data.get('v'),list):
v_json_data = json_data.get('v')
has_tool_search = False
tool_search_id = None
for item in v_json_data:
if item.get('p') == 'references':
references = item.get('v') or []
if isinstance(references, list):
for reference in references:
if (
isinstance(reference, dict)
and reference.get('type') == 'TOOL_OPEN'
):
has_tool_search = True
tool_search_id = reference.get('id')
break
for v in v_json_data:
if v.get('p') == 'content' and has_tool_search:
p_content = v.get('v') or ''
p_content = re.sub(
r'\[reference:\d+\]',
f'[reference:{tool_search_id}]',
p_content
)
response_content+=p_content
if v.get('p') == 'fragments':
if v.get('p') == 'fragments':
fragments = v.get('v') or []
if isinstance(fragments, list):
for fragment in fragments:
if (
isinstance(fragment, dict)
and fragment.get('type') == 'TOOL_OPEN'
and fragment.get('result') is not None
):
url_id_map[fragment.get('result').get('url')] = fragment.get('id')
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')
used_ids = set(url_id_map.values())
next_id = 1
for item in url_list:
url = item.get('url')
if url in url_id_map:
item['cite_index'] = url_id_map[url]
elif item.get('cite_index') is None:
while next_id in used_ids:
next_id += 1
item['cite_index'] = next_id
used_ids.add(next_id)
next_id += 1
think_content = think_content.replace('DEEP_SEARCH','')
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,
}
def get_platform_tos_directory(platform: str) -> str:
platform = normalize_platform(platform)
return PLATFORM_CONFIGS[platform]["tos_directory"]
def wait_mt_tasks_done(
redis_key: str,
pt: str = None,
......@@ -704,13 +1044,17 @@ def process_rounds_to_result(
end_count: int,
redis_key: str,
task_ids: list = None,
platform: str = "DB",
) -> int:
"""
发送指定轮次，等待并补发未完成任务，
最后解析 TOS 正文、统计关键词并写入结果表。
"""
platform = normalize_platform(platform)
pt = normalize_pt(pt)
redis_pt = pt[:8]
tos_directory = get_platform_tos_directory(platform)
process_original_data = PLATFORM_PROCESS_MAP[platform]
normalized_task_ids = None
task_filter_sql = ""
task_filter_params = []
......@@ -740,6 +1084,7 @@ def process_rounds_to_result(
redis_key,
pt=pt,
task_ids=normalized_task_ids,
platform=platform,
)
wait_mt_tasks_done(
redis_key,
......@@ -756,6 +1101,7 @@ def process_rounds_to_result(
SELECT reqId, prompt, pt
FROM {LOG_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND toInt32(`count`) BETWEEN %s AND %s
{task_filter_sql}
) AS a1
......@@ -764,6 +1110,7 @@ def process_rounds_to_result(
""",
(
pt,
platform,
start_count,
end_count,
*task_filter_params,
......@@ -826,11 +1173,13 @@ def process_rounds_to_result(
SELECT *
FROM {LOG_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND toInt32(`count`) BETWEEN %s AND %s
{task_filter_sql}
""",
(
pt,
platform,
start_count,
end_count,
*task_filter_params,
......@@ -850,7 +1199,7 @@ def process_rounds_to_result(
return None
file = (
f"geo_snipaste/{redis_pt}/doubao/"
f"geo_snipaste/{redis_pt}/{tos_directory}/"
f"{req_id}/text.json"
)
try:
......@@ -893,7 +1242,7 @@ def process_rounds_to_result(
_,
response_content,
_,
) = doubao_process_original_data(
) = process_original_data(
file,
content,
)
......@@ -931,6 +1280,7 @@ def process_rounds_to_result(
"pt": int(row.get("pt")),
"has_keyword": has_keyword,
"keywords_count": keywords_count,
"platform": platform,
}
except Exception as exc:
logger.warning(
......@@ -1081,8 +1431,7 @@ def process_rounds_to_result(
for attempt in range(1, 4):
try:
success = bh_utils.insert_data(
"douchacha_data."
"geo_feishu_snipaste_result_v3",
RESULT_TABLE_NAME,
batch_items,
retry_times=0,
)
......@@ -1233,16 +1582,19 @@ def create_and_process_share_rounds(
hit_start_count: int = 1,
hit_end_count: int = 30,
hit_threshold: int = 24,
platform: str = "DB",
) -> int:
"""
第 1-30 轮中关键词命中轮数达到阈值的 taskId，
生成 count=0 的分享任务并完成 Redis、TOS 和结果入库流程。
"""
platform = normalize_platform(platform)
qualified_rows = bh_utils.query_data(
"""
f"""
SELECT taskId
FROM douchacha_data.geo_feishu_snipaste_result_v3
FROM {RESULT_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND `count` BETWEEN %s AND %s
AND has_keyword = 1
GROUP BY taskId
......@@ -1250,6 +1602,7 @@ def create_and_process_share_rounds(
""",
(
int(pt),
platform,
hit_start_count,
hit_end_count,
hit_threshold,
......@@ -1276,10 +1629,12 @@ def create_and_process_share_rounds(
SELECT taskId, prompt, keywords, pt
FROM {TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND taskId IN ({task_id_placeholders})
""",
(
pt,
platform,
*task_ids,
),
)
......@@ -1291,11 +1646,13 @@ def create_and_process_share_rounds(
SELECT taskId
FROM {LOG_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND toInt32(`count`) = 0
AND taskId IN ({task_id_placeholders})
""",
(
pt,
platform,
*task_ids,
),
)
......@@ -1320,6 +1677,7 @@ def create_and_process_share_rounds(
"keywords": task.get("keywords"),
"`count`": "0",
"pt": str(task.get("pt") or pt),
"platform": platform,
}
)
......@@ -1342,6 +1700,7 @@ def create_and_process_share_rounds(
start_count=0,
end_count=0,
redis_key=redis_key,
platform=platform,
)
return inserted_log_count
......@@ -1352,16 +1711,19 @@ def create_and_process_unqualified_share_rounds(
hit_start_count: int = 1,
hit_end_count: int = 30,
hit_threshold: int = 24,
platform: str = "DB",
) -> int:
"""
第 1-30 轮中关键词命中轮数不足阈值的 taskId，
生成 count=99 的分享任务并完成 Redis、TOS 和结果入库流程。
"""
platform = normalize_platform(platform)
qualified_rows = bh_utils.query_data(
"""
f"""
SELECT taskId
FROM douchacha_data.geo_feishu_snipaste_result_v3
FROM {RESULT_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND `count` BETWEEN %s AND %s
AND has_keyword = 1
GROUP BY taskId
......@@ -1369,6 +1731,7 @@ def create_and_process_unqualified_share_rounds(
""",
(
int(pt),
platform,
hit_start_count,
hit_end_count,
hit_threshold,
......@@ -1387,8 +1750,9 @@ def create_and_process_unqualified_share_rounds(
SELECT taskId, prompt, keywords, pt
FROM {TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
""",
(pt,),
(pt, platform),
)
if tasks is None:
raise RuntimeError("第五阶段从主表查询未达标任务失败")
......@@ -1411,9 +1775,10 @@ def create_and_process_unqualified_share_rounds(
SELECT taskId
FROM {LOG_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND toInt32(`count`) = 99
""",
(pt,),
(pt, platform),
)
if existing_rows is None:
raise RuntimeError("第五阶段查询已生成 count=99 任务失败")
......@@ -1436,6 +1801,7 @@ def create_and_process_unqualified_share_rounds(
"keywords": task.get("keywords"),
"`count`": "99",
"pt": str(task.get("pt") or pt),
"platform": platform,
}
)
......@@ -1460,6 +1826,7 @@ def create_and_process_unqualified_share_rounds(
start_count=99,
end_count=99,
redis_key=redis_key,
platform=platform,
)
return inserted_log_count
......@@ -1468,14 +1835,18 @@ def export_share_rounds_to_excel(
pt: str,
output_path: str = None,
share_count: int = 0,
platform: str = "DB",
) -> str:
"""
导出当天指定分享轮次的任务及其 TOS 结果到 Excel。
"""
if share_count not in (0, 99):
raise ValueError("Excel 导出仅支持分享轮次 count=0 或 count=99")
platform = normalize_platform(platform)
platform_config = PLATFORM_CONFIGS[platform]
pt = normalize_pt(pt)
tos_pt = pt[:8]
tos_directory = get_platform_tos_directory(platform)
stage_name = "第四阶段" if share_count == 0 else "第六阶段"
file_label = "达标数据" if share_count == 0 else "未达标"
qualification_filter_sql = (
......@@ -1501,6 +1872,8 @@ def export_share_rounds_to_excel(
LEFT JOIN {TABLE_NAME} AS a5
ON a1.taskId = a5.taskId
AND a1.pt = a5.pt
AND ifNull(a1.platform, 'DB')
= ifNull(a5.platform, 'DB')
LEFT JOIN (
SELECT
taskId,
......@@ -1511,22 +1884,32 @@ def export_share_rounds_to_excel(
taskId,
`count`,
max(has_keyword) AS round_has_keyword
FROM douchacha_data.geo_feishu_snipaste_result_v3
FROM {RESULT_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND `count` BETWEEN 1 AND 30
GROUP BY taskId, `count`
) AS round_result
GROUP BY taskId
) AS a3 ON a1.taskId = a3.taskId
LEFT JOIN douchacha_data.geo_feishu_snipaste_result_v3 AS a4
LEFT JOIN {RESULT_TABLE_NAME} AS a4
ON a1.reqId = a4.reqId
AND toInt32(a1.`count`) = a4.`count`
AND ifNull(a1.platform, 'DB')
= ifNull(a4.platform, 'DB')
WHERE a1.pt = %s
AND ifNull(a1.platform, 'DB') = %s
AND toInt32(a1.`count`) = %s
{qualification_filter_sql}
ORDER BY a5.sheet_order, a1.reqId
""",
(int(pt), pt, share_count),
(
int(pt),
platform,
pt,
platform,
share_count,
),
)
if rows is None:
raise RuntimeError(
......@@ -1538,9 +1921,10 @@ def export_share_rounds_to_excel(
SELECT taskId, keywords
FROM {TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
ORDER BY sheet_order, taskId
""",
(pt,),
(pt, platform),
)
if task_keyword_rows is None:
raise RuntimeError(
......@@ -1548,13 +1932,14 @@ def export_share_rounds_to_excel(
)
round_keyword_rows = bh_utils.query_data(
"""
f"""
SELECT taskId, `count`, keywords_count
FROM douchacha_data.geo_feishu_snipaste_result_v3
FROM {RESULT_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND `count` BETWEEN 1 AND 30
""",
(int(pt),),
(int(pt), platform),
)
if round_keyword_rows is None:
raise RuntimeError(
......@@ -1636,7 +2021,10 @@ def export_share_rounds_to_excel(
)
output_path = os.path.join(
"/Users/yaowentong/Desktop",
f"{formatted_date}_{file_label}.xlsx",
(
f"{formatted_date}_"
f"{platform_config['name']}_{file_label}.xlsx"
),
)
output_dir = os.path.dirname(os.path.abspath(output_path))
......@@ -1653,6 +2041,8 @@ def export_share_rounds_to_excel(
[
"问题",
"是否达标",
"达标轮次",
"达标率",
"本次是否达标",
*all_keywords,
"平台",
......@@ -1668,10 +2058,6 @@ def export_share_rounds_to_excel(
]
)
plat_form_map = {
"DB": "豆包网页版",
}
def timestamp_to_datetime(timestamp):
if timestamp in (None, ""):
return ""
......@@ -1686,14 +2072,16 @@ def export_share_rounds_to_excel(
task_id = str(row.get("taskId") or "")
prompt = row.get("prompt")
platform = row.get("platform")
row_platform = str(row.get("platform") or platform)
insertime = row.get("insertime")
sheet_order = row.get("sheet_order")
processed_round_count = row.get("processed_round_count")
hit_round_count = row.get("hit_round_count")
hit_round_count_value = int(hit_round_count or 0)
hit_rate = hit_round_count_value / 30
if not processed_round_count:
qualified_text = "没处理"
elif int(hit_round_count or 0) >= 24:
elif hit_round_count_value >= 24:
qualified_text = "是"
else:
qualified_text = "否"
......@@ -1715,7 +2103,10 @@ def export_share_rounds_to_excel(
for keyword in all_keywords
]
base_path = f"geo_snipaste/{tos_pt}/doubao/{req_id}"
base_path = (
f"geo_snipaste/{tos_pt}/"
f"{tos_directory}/{req_id}"
)
text_json_path = f"{base_path}/text.json"
context_path = f"{base_path}/context.txt"
think_path = f"{base_path}/think.txt"
......@@ -1726,7 +2117,8 @@ def export_share_rounds_to_excel(
)
png_url = (
f"https://tcdn.aidso.com/"
f"geo_snipaste/{tos_pt}/doubao/{req_id}/png.png"
f"geo_snipaste/{tos_pt}/"
f"{tos_directory}/{req_id}/png.png"
)
text_json_str = (
......@@ -1744,9 +2136,14 @@ def export_share_rounds_to_excel(
[
prompt,
qualified_text,
hit_round_count_value,
hit_rate,
current_qualified_text,
*keyword_total_values,
plat_form_map.get(platform, "豆包网页版"),
PLATFORM_CONFIGS.get(
row_platform,
platform_config,
).get("name", row_platform),
tos_utils.get_string_from_tos(context_path),
tos_utils.get_string_from_tos(think_path),
tos_utils.get_string_from_tos(quote_path),
......@@ -1763,6 +2160,9 @@ def export_share_rounds_to_excel(
]
)
for cell in worksheet["D"][1:]:
cell.number_format = "0.00%"
workbook.save(output_path)
workbook.close()
logger.success(
......@@ -1776,6 +2176,7 @@ def export_share_rounds_to_excel(
def export_unqualified_share_rounds_to_excel(
pt: str,
output_path: str = None,
platform: str = "DB",
) -> str:
"""
第六阶段导出当天 count=99 的未达标分享结果。
......@@ -1784,6 +2185,7 @@ def export_unqualified_share_rounds_to_excel(
pt=pt,
output_path=output_path,
share_count=99,
platform=platform,
)
......@@ -1867,30 +2269,549 @@ def upload_excel_to_feishu_chat(
return message_id
def build_daily_run_report(pt: str) -> str:
def _parse_json_list(value) -> list:
if isinstance(value, list):
return value
if not isinstance(value, str) or not value.strip():
return []
try:
parsed = json.loads(value)
except json.JSONDecodeError:
return [value]
return parsed if isinstance(parsed, list) else []
def _parse_task_keywords(value) -> list[str]:
keywords = []
seen_keywords = set()
for item in _parse_json_list(value):
keyword = str(item or "").strip()
if not keyword or keyword in seen_keywords:
continue
seen_keywords.add(keyword)
keywords.append(keyword)
return keywords
def _parse_round_keywords_count(value) -> dict:
keyword_counts = {}
for item in _parse_json_list(value):
if not isinstance(item, dict):
continue
keyword = str(item.get("keyword") or "").strip()
if not keyword:
continue
try:
word_count = int(item.get("word_count") or 0)
except (TypeError, ValueError):
word_count = 0
keyword_counts[keyword] = max(
keyword_counts.get(keyword, 0),
word_count,
)
return keyword_counts
def _excel_safe_text(value) -> str:
text = str(value or "")
text = ILLEGAL_EXCEL_CHARACTERS.sub("", text)
return text[:EXCEL_CELL_MAX_LENGTH]
def _safe_get_tos_text(path: str) -> str:
try:
return tos_utils.get_string_from_tos(path) or ""
except Exception as exc:
logger.warning(
f"30轮明细读取 TOS 失败: path={path}, error={exc}"
)
return ""
def _timestamp_to_text(timestamp) -> str:
if timestamp in (None, ""):
return ""
try:
timestamp_number = int(timestamp)
if timestamp_number > 10 ** 12:
timestamp_number //= 1000
return datetime.fromtimestamp(timestamp_number).strftime(
"%Y-%m-%d %H:%M:%S"
)
except (TypeError, ValueError, OSError, OverflowError):
return str(timestamp)
def _sheet_order_sort_key(row: dict) -> tuple:
try:
sheet_order = int(row.get("sheet_order"))
return (
0,
sheet_order,
str(row.get("prompt") or ""),
)
except (TypeError, ValueError):
return (
1,
0,
str(row.get("prompt") or ""),
)
def _chinese_round_number(number: int) -> str:
digits = "零一二三四五六七八九"
if number < 10:
return digits[number]
if number == 10:
return "十"
if number < 20:
return f"十{digits[number % 10]}"
if number % 10 == 0:
return f"{digits[number // 10]}十"
return f"{digits[number // 10]}十{digits[number % 10]}"
def _safe_file_name_part(value: str) -> str:
normalized_value = str(value or "").strip() or "未分类"
normalized_value = re.sub(
r'[\\/:*?"<>|]',
"_",
normalized_value,
)
return normalized_value.strip(". ") or "未分类"
def _style_thirty_round_worksheet(
worksheet,
total_columns: int,
) -> None:
worksheet.freeze_panes = "A2"
worksheet.auto_filter.ref = worksheet.dimensions
worksheet.sheet_view.showGridLines = False
worksheet.row_dimensions[1].height = 32
header_fill = PatternFill("solid", fgColor="6D28D9")
header_font = Font(color="FFFFFF", bold=True)
header_alignment = Alignment(
horizontal="center",
vertical="center",
wrap_text=True,
)
row_border = Border(
bottom=Side(style="thin", color="D8DCE6")
)
for cell in worksheet[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = header_alignment
for row in worksheet.iter_rows(
min_row=2,
max_row=worksheet.max_row,
max_col=total_columns,
):
for cell in row:
cell.alignment = Alignment(
vertical="top",
wrap_text=True,
)
cell.border = row_border
column_widths = {
"问题": 42,
"批次": 16,
"30次是否达标": 16,
"本次是否达标": 16,
"平台": 18,
"回答": 70,
"思考过程": 55,
"引用来源": 55,
"查询时间": 22,
"IP": 18,
"版本号": 20,
}
for column_index, cell in enumerate(
worksheet[1],
start=1,
):
worksheet.column_dimensions[
get_column_letter(column_index)
].width = column_widths.get(str(cell.value), 16)
def _upload_thirty_round_excel_to_tos(
file_path: str,
object_key: str,
retry_times: int = 3,
) -> str:
client = tos_client.get_client()
bucket = tos_client.bucket_name
with open(file_path, "rb") as file_object:
file_content = file_object.read()
last_error = None
for attempt in range(1, retry_times + 1):
try:
client.put_object(
bucket,
object_key,
content=file_content,
content_type=(
"application/vnd.openxmlformats-officedocument."
"spreadsheetml.sheet"
),
)
logger.success(
f"[第七阶段 count=1-30] 30轮明细已上传 TOS: "
f"tos://{bucket}/{object_key}, "
f"attempt={attempt}/{retry_times}"
)
return f"tos://{bucket}/{object_key}"
except Exception as exc:
last_error = exc
logger.warning(
f"[第七阶段 count=1-30] 30轮明细上传失败: "
f"object_key={object_key}, "
f"attempt={attempt}/{retry_times}, error={exc}"
)
raise RuntimeError(
f"30轮明细上传 TOS 连续失败 {retry_times} 次: "
f"object_key={object_key}"
) from last_error
def export_thirty_round_details(
pt: str,
platform: str = "DB",
) -> list[dict]:
"""
按业务线导出当前 pt、当前平台的 1-30 轮明细并上传到 TOS。
返回可直接写入飞书日报的业务线及 CDN 下载地址列表。
"""
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
platform = normalize_platform(platform)
pt = normalize_pt(pt)
platform_config = PLATFORM_CONFIGS[platform]
tos_directory = get_platform_tos_directory(platform)
log_rows = bh_utils.query_data(
f"""
SELECT
a1.reqId,
a1.taskId,
a1.prompt,
a1.pt,
toInt32(a1.`count`) AS round_count,
a2.sheet_order,
a2.business_line,
a3.insertime
FROM {LOG_TABLE_NAME} AS a1
LEFT JOIN {TABLE_NAME} AS a2
ON a1.taskId = a2.taskId
AND a1.pt = a2.pt
AND ifNull(a1.platform, 'DB')
= ifNull(a2.platform, 'DB')
LEFT JOIN geo_third_task AS a3
ON a1.reqId = a3.reqId
WHERE toString(a1.pt) = %s
AND ifNull(a1.platform, 'DB') = %s
AND toInt32(a1.`count`) BETWEEN 1 AND 30
""",
(pt, platform),
)
if log_rows is None:
raise RuntimeError("第七阶段查询 1-30 轮日志数据失败")
result_rows = bh_utils.query_data(
f"""
SELECT
reqId,
taskId,
pt,
toInt32(`count`) AS round_count,
has_keyword
FROM {RESULT_TABLE_NAME}
WHERE toString(pt) = %s
AND ifNull(platform, 'DB') = %s
AND toInt32(`count`) BETWEEN 1 AND 30
""",
(pt, platform),
)
if result_rows is None:
raise RuntimeError("第七阶段查询 1-30 轮结果数据失败")
result_by_req_id = {}
task_round_hits = defaultdict(dict)
for result_row in result_rows:
req_id = str(result_row.get("reqId") or "")
task_id = str(result_row.get("taskId") or "")
result_pt = str(result_row.get("pt") or "")
try:
round_count = int(
result_row.get("round_count") or 0
)
has_keyword = int(
result_row.get("has_keyword") or 0
)
except (TypeError, ValueError):
continue
if (
not req_id
or not task_id
or not 1 <= round_count <= 30
):
continue
current_result = result_by_req_id.setdefault(
req_id,
{
"has_keyword": 0,
},
)
current_result["has_keyword"] = max(
current_result["has_keyword"],
has_keyword,
)
task_key = (result_pt, task_id)
task_round_hits[task_key][round_count] = max(
task_round_hits[task_key].get(round_count, 0),
has_keyword,
)
rows_by_business_line = defaultdict(list)
for log_row in log_rows:
business_line = str(
log_row.get("business_line") or ""
).strip() or "未分类"
rows_by_business_line[business_line].append(log_row)
downloads = []
for business_line in sorted(rows_by_business_line):
business_line_rows = rows_by_business_line[business_line]
rows_by_round = defaultdict(list)
for log_row in business_line_rows:
try:
round_count = int(
log_row.get("round_count") or 0
)
except (TypeError, ValueError):
continue
if 1 <= round_count <= 30:
rows_by_round[round_count].append(log_row)
safe_business_line = _safe_file_name_part(business_line)
file_name = (
f"{pt}_{platform_config['name']}_"
f"{safe_business_line}_30轮明细.xlsx"
)
local_path = os.path.join(
tempfile.gettempdir(),
file_name,
)
headers = [
"问题",
"批次",
"30次是否达标",
"本次是否达标",
"平台",
"回答",
"思考过程",
"引用来源",
"查询时间",
"IP",
"版本号",
]
def build_excel_row(log_row: dict) -> list:
req_id = str(log_row.get("reqId") or "")
task_id = str(log_row.get("taskId") or "")
row_pt = str(log_row.get("pt") or "")
result_item = result_by_req_id.get(req_id)
round_hits = task_round_hits.get(
(row_pt, task_id),
{},
)
if not round_hits:
thirty_round_qualified = "没处理"
else:
hit_round_count = sum(
1
for value in round_hits.values()
if value == 1
)
thirty_round_qualified = (
"是" if hit_round_count >= 24 else "否"
)
if result_item is None:
current_qualified = "没处理"
else:
current_qualified = (
"是"
if int(
result_item.get("has_keyword") or 0
) == 1
else "否"
)
base_path = (
f"geo_snipaste/{row_pt[:8]}/"
f"{tos_directory}/{req_id}"
)
return [
_excel_safe_text(log_row.get("prompt")),
row_pt,
thirty_round_qualified,
current_qualified,
platform_config["name"],
_excel_safe_text(
_safe_get_tos_text(
f"{base_path}/context.txt"
)
),
_excel_safe_text(
_safe_get_tos_text(
f"{base_path}/think.txt"
)
),
_excel_safe_text(
_safe_get_tos_text(
f"{base_path}/quote.txt"
)
),
_timestamp_to_text(log_row.get("insertime")),
_excel_safe_text(
_safe_get_tos_text(
f"{base_path}/inner_user_ip.txt"
)
),
_excel_safe_text(
_safe_get_tos_text(
f"{base_path}/inner_pc_version.txt"
)
),
]
workbook = Workbook()
workbook.remove(workbook.active)
total_rows = 0
with ThreadPoolExecutor(
max_workers=THIRTY_ROUND_EXPORT_WORKERS
) as executor:
for round_count in range(1, 31):
worksheet = workbook.create_sheet(
title=(
f"第"
f"{_chinese_round_number(round_count)}轮"
)
)
worksheet.append(headers)
round_rows = [
row
for row in sorted(
rows_by_round.get(round_count, []),
key=_sheet_order_sort_key,
)
if row.get("reqId")
]
for excel_row in executor.map(
build_excel_row,
round_rows,
):
worksheet.append(excel_row)
total_rows += 1
_style_thirty_round_worksheet(
worksheet,
len(headers),
)
logger.info(
f"[第七阶段 platform={platform} pt={pt} "
f"business_line={business_line} "
f"count={round_count}] 30轮明细 Sheet 完成: "
f"rows={len(round_rows)}"
)
workbook.save(local_path)
workbook.close()
object_key = (
f"geo_snipaste/{pt[:8]}/"
f"{tos_directory}/{file_name}"
)
try:
_upload_thirty_round_excel_to_tos(
file_path=local_path,
object_key=object_key,
)
finally:
try:
os.remove(local_path)
except OSError:
pass
cdn_url = (
f"https://tcdn.aidso.com/"
f"geo_snipaste/{pt[:8]}/"
f"{tos_directory}/{file_name}"
)
downloads.append(
{
"business_line": business_line,
"url": cdn_url,
}
)
logger.success(
f"[第七阶段 platform={platform} pt={pt} "
f"business_line={business_line} count=1-30] "
f"30轮明细导出完成: rows={total_rows}, "
f"url={cdn_url}"
)
if not downloads:
logger.warning(
f"[第七阶段 platform={platform} pt={pt} count=1-30] "
f"没有可导出的30轮业务线数据"
)
return downloads
def build_daily_run_report(
pt: str,
platform: str = "DB",
thirty_round_downloads: list = None,
) -> str:
"""
汇总当天各阶段运行结果，生成 Markdown 报告。
"""
platform = normalize_platform(platform)
platform_name = PLATFORM_CONFIGS[platform]["name"]
pt = normalize_pt(pt)
tasks = bh_utils.query_data(
f"""
SELECT taskId, prompt, sheet_order
FROM {TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
""",
(pt,),
(pt, platform),
)
if tasks is None:
raise RuntimeError("最后阶段查询当天主任务失败")
result_rows = bh_utils.query_data(
"""
f"""
SELECT taskId, `count`, has_keyword
FROM douchacha_data.geo_feishu_snipaste_result_v3
FROM {RESULT_TABLE_NAME}
WHERE pt = %s
AND ifNull(platform, 'DB') = %s
AND `count` BETWEEN 1 AND 30
""",
(int(pt),),
(int(pt), platform),
)
if result_rows is None:
raise RuntimeError("最后阶段查询关键词统计结果失败")
......@@ -1928,7 +2849,7 @@ def build_daily_run_report(pt: str) -> str:
thirty_round_failed = set(task_map) - thirty_round_qualified
markdown_lines = [
"## 美团 GEO 今日任务运行总结",
f"## {platform_name} GEO 今日任务运行总结",
"",
f"**统计日期：** {format_pt_display(pt)}",
"",
......@@ -1940,10 +2861,33 @@ def build_daily_run_report(pt: str) -> str:
f"- 今日问题任务总数：{len(task_map)} 个",
f"- 达标问题：{len(thirty_round_qualified)} 个",
f"- 未达标问题：{len(thirty_round_failed)} 个",
]
if thirty_round_downloads:
markdown_lines.extend(
[
"",
"**30轮数据如下：**",
"",
]
)
for download in thirty_round_downloads:
business_line = str(
download.get("business_line") or "未分类"
)
download_url = str(download.get("url") or "")
if not download_url:
continue
markdown_lines.append(
f"- {business_line}："
f"[点击下载30轮明细]({download_url})"
)
markdown_lines.extend(
[
"",
"### 未达标问题 Top 10",
"",
]
)
def failed_task_sort_key(task):
task_id = str(task.get("taskId") or "")
......@@ -1985,8 +2929,13 @@ def build_daily_run_report(pt: str) -> str:
def send_markdown_report_to_feishu_chat(
markdown_text: str,
chat_id: str,
platform: str = "DB",
) -> str:
card_title = "美团 GEO 今日任务运行总结"
platform = normalize_platform(platform)
card_title = (
f"{PLATFORM_CONFIGS[platform]['name']} "
f"GEO 今日任务运行总结"
)
card_markdown_lines = []
for line in markdown_text.splitlines():
stripped_line = line.strip()
......@@ -2050,40 +2999,51 @@ def send_markdown_report_to_feishu_chat(
return message_id
def run_daily_pipeline(pt: str = None) -> None:
def run_daily_pipeline(
pt: str = None,
platform: str = "DB",
) -> None:
"""
按顺序执行飞书截图任务的第一至第七阶段。
"""
platform = normalize_platform(platform)
platform_config = PLATFORM_CONFIGS[platform]
pt = normalize_pt(pt)
start_count = 1
end_count = 30
logger.success(
f"[每日任务 pt={pt} count=1-99] 七阶段任务开始执行"
f"[每日任务 platform={platform} pt={pt} count=1-99] "
f"七阶段任务开始执行"
)
# 第一阶段：飞书同步到主表和 30 轮日志表。
sync_feishu_to_database(pt=pt)
sync_feishu_to_database(pt=pt, platform=platform)
# 第二阶段：所有任务发送 1-30 轮，完成三轮补发、解析和结果入库。
process_rounds_to_result(
pt=pt,
start_count=start_count,
end_count=end_count,
redis_key=REDIS_QUEUE_KEY,
redis_key=platform_config["content_redis_key"],
platform=platform,
)
# 第三阶段：1-30 轮命中不少于 24 轮的任务生成 count=0 分享任务。
create_and_process_share_rounds(
pt=pt,
redis_key=REDIS_QUEUE_KEY_SHARE,
redis_key=platform_config["share_redis_key"],
hit_start_count=1,
hit_end_count=30,
hit_threshold=24,
platform=platform,
)
# 第四阶段：导出 count=0 的达标分享结果。
excel_path = export_share_rounds_to_excel(pt=pt)
excel_path = export_share_rounds_to_excel(
pt=pt,
platform=platform,
)
upload_excel_to_feishu_chat(
file_path=excel_path,
chat_id=FEISHU_CHAT_ID,
......@@ -2094,15 +3054,19 @@ def run_daily_pipeline(pt: str = None) -> None:
# 第五阶段：1-30 轮命中不足 24 轮的任务生成 count=99 分享任务。
create_and_process_unqualified_share_rounds(
pt=pt,
redis_key=REDIS_QUEUE_KEY_SHARE,
redis_key=platform_config["share_redis_key"],
hit_start_count=1,
hit_end_count=30,
hit_threshold=24,
platform=platform,
)
# 第六阶段：导出 count=99 的未达标分享结果。
unqualified_excel_path = (
export_unqualified_share_rounds_to_excel(pt=pt)
export_unqualified_share_rounds_to_excel(
pt=pt,
platform=platform,
)
)
upload_excel_to_feishu_chat(
file_path=unqualified_excel_path,
......@@ -2111,28 +3075,40 @@ def run_daily_pipeline(pt: str = None) -> None:
count_label="99",
)
# 第七阶段：生成并发送今日运行总结。
markdown_report = build_daily_run_report(pt=pt)
# 第七阶段：导出30轮明细，并在今日运行总结中发送下载链接。
thirty_round_downloads = export_thirty_round_details(
pt=pt,
platform=platform,
)
markdown_report = build_daily_run_report(
pt=pt,
platform=platform,
thirty_round_downloads=thirty_round_downloads,
)
send_markdown_report_to_feishu_chat(
markdown_text=markdown_report,
chat_id=FEISHU_CHAT_ID,
platform=platform,
)
logger.success(
f"[每日任务 pt={pt} count=1-99] 七阶段任务执行完成"
f"[每日任务 platform={platform} pt={pt} count=1-99] "
f"七阶段任务执行完成"
)
def run_daily_pipeline_safely(pt: str = None) -> None:
def run_daily_pipeline_safely(
pt: str = None,
platform: str = "DB",
) -> None:
try:
run_daily_pipeline(pt=pt)
run_daily_pipeline(pt=pt, platform=platform)
except Exception:
logger.exception(
"[每日任务 count=1-99] 七阶段任务执行失败"
f"[每日任务 platform={platform} count=1-99] "
f"七阶段任务执行失败"
)
if __name__ == "__main__":
# run_daily_pipeline('20260729')
start_scheduler()
# run_daily_pipeline(pt="2026072702")
aidso_geo/core/routes/interface.py
View file @ 4b1f9b3a
......@@ -4,12 +4,16 @@ import json
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
from aidso_geo.utils import bh_utils, tos_utils, url_utils
from aidso_geo.models.eco_data_process import process_eco_product_relation
line_app = Blueprint("line_app", __name__)
redis_client = init_redis()
redis_client8 = init_redis8()
......@@ -323,3 +327,95 @@ def check_quto():
"reqId": req_id
})
@line_app.route('/api/geo/check_eco_keyword', methods=['POST'])
def check_eco():
try:
body = request.get_json(silent=True) or {}
reqids = body.get("req_ids")
keyword = body.get("keyword")
if not isinstance(reqids, list):
return jsonify({
"code": 400,
"message": "reqids必须是数组",
"data": None,
})
# 过滤空值并去重，同时保持原顺序。
reqids = list(dict.fromkeys(
str(req_id).strip()
for req_id in reqids
if str(req_id or "").strip()
))
if not reqids:
return jsonify({
"code": 400,
"message": "reqids不能为空",
"data": None,
})
if not isinstance(keyword, str) or not keyword.strip():
return jsonify({
"code": 400,
"message": "keyword必须是非空字符串",
"data": None,
})
keyword = keyword.strip()
placeholders = ", ".join(
["%s"] * len(reqids)
)
result = bh_utils.query_data(
f"""
SELECT *
FROM geo_eco_data
WHERE req_id IN ({placeholders})
""",
tuple(reqids),
)
if not result:
return jsonify({
"code": 200,
"message": "处理成功",
"data": {
"reqids": reqids,
"keyword": keyword,
"query_rows": len(result),
"insert_rows": len(result),
},
})
eco_result = process_eco_product_relation(
result,
keyword,
)
bh_utils.insert_data("geo_eco_data",eco_result)
return jsonify({
"code": 200,
"message": "处理成功",
"data": {
"reqids": reqids,
"keyword": keyword,
"query_rows": len(result),
"insert_rows": len(eco_result),
},
})
except Exception as exc:
return jsonify({
"code": 500,
"message": f"商品关系处理失败: {exc}",
"data": None,
})
aidso_geo/models/eco_data_process.py
View file @ 4b1f9b3a
import re
from aidso_geo.utils import bh_utils
from aidso_geo.utils import bh_utils,ai_utils
def process_eco_product_relation(
eco_result,
keyword,
batch_size=500,
):
"""
每100条商品调用一次AI，并将识别结果合并到原商品数据中。
produce_name只用于匹配，不写入最终结果。
"""
def normalize_title(value):
return " ".join(
str(value or "").strip().split()
)
final_result = []
for start_index in range(
0,
len(eco_result),
batch_size,
):
batch = eco_result[
start_index:start_index + batch_size
]
product_list = [
str(item.get("eco_title") or "").strip()
for item in batch
]
try:
ai_result = (
ai_utils.ai_get_product_relation_spu(
product_list,
keyword,
)
)
except Exception as exc:
print(
f"AI商品识别失败: "
f"batch={start_index // batch_size + 1}, "
f"error={exc}"
)
ai_result = []
if not isinstance(ai_result, list):
ai_result = []
# produce_name只用于建立匹配关系。
ai_result_map = {}
for ai_item in ai_result:
if not isinstance(ai_item, dict):
continue
produce_name = str(
ai_item.get("produce_name") or ""
).strip()
normalized_name = normalize_title(
produce_name
)
if not normalized_name:
continue
if normalized_name not in ai_result_map:
ai_result_map[normalized_name] = ai_item
for original_item in batch:
eco_title = str(
original_item.get("eco_title") or ""
).strip()
matched_item = ai_result_map.get(
normalize_title(eco_title)
)
# 未匹配时的默认值。
relation_data = {
"brand": eco_title,
"spu_name": "",
"current": 0,
}
if isinstance(matched_item, dict):
brand = str(
matched_item.get("brand")
or eco_title
).strip()
spu_name = str(
matched_item.get("spu_name")
or ""
).strip()
try:
current = int(
matched_item.get("current", 0)
)
except (TypeError, ValueError):
current = 0
relation_data = {
"brand": brand,
"search_goods_word":keyword,
"spu_name": brand+" "+spu_name,
"current": (
1 if current == 1 else 0
),
}
final_result.append(
{
**original_item,
**relation_data,
}
)
return final_result
def extract_price(price):
if not price:
......@@ -234,6 +357,7 @@ def yuanbao_android_process_txmap_eco(data,eco_list):
def save_eco_data_to_bh(data,eco_type,eco_list):
platform = data.get('platform')
search_goods_word = data.get('search_goods_word')
eco_result = []
if platform == 'TYQWA':
......@@ -259,6 +383,10 @@ def save_eco_data_to_bh(data,eco_type,eco_list):
if eco_type == 'txmap':
eco_result = yuanbao_process_txmap_eco(data,eco_list)
if search_goods_word:
eco_result = process_eco_product_relation(eco_result, search_goods_word)
bh_utils.insert_data('geo_eco_data',eco_result)
aidso_geo/models/process.py
View file @ 4b1f9b3a
......@@ -2032,9 +2032,9 @@ if __name__ == '__main__':
type_t = i.get('type')
type_t = 'batch'
# commit_task(i,'ING')
return task_send_queue(i,type_t)
# return task_send_queue(i,type_t)
# return deepseek_data_process.deepseek_process_original_data(i)
# return platform_process(i)
return platform_process(i)
#
if data_list:
with ThreadPoolExecutor(max_workers=50) as executor:
......
aidso_geo/models/qianwen_data_process.py
View file @ 4b1f9b3a
......@@ -33,7 +33,6 @@ def qianwen_process_original_data(data):
data_str = i.split("data:")[1]
json_data = json.loads(data_str)
except (IndexError, json.JSONDecodeError):
continue
......@@ -146,6 +145,19 @@ def qianwen_process_original_data(data):
if paas:
for pa in ms.get('meta_data').get('paas'):
suggestions.append(pa.get('show_text'))
if ms.get('mime_type') == 'bar/workflow' and ms.get('status') == 'complete':
ms_meta_data_multi_load = ms.get('meta_data').get('multi_load')
if isinstance(ms_meta_data_multi_load,list):
for i in ms_meta_data_multi_load:
ms_meta_data_multi_load_type =i.get('type')
if ms_meta_data_multi_load_type == 'bar_thinking':
think_content+=i.get('content').get('body')
if ms_meta_data_multi_load_type == 'bar_ref_source_inline':
if i.get('content'):
if i.get('content').get('query_list'):
search_keyword.extend(i.get('content').get('query_list'))
if i.get('content').get('docs'):
url_list_batch.extend(i.get('content').get('docs'))
if url_list_batch:
for url in url_list_batch:
......@@ -163,7 +175,6 @@ def qianwen_process_original_data(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]
......@@ -195,7 +206,7 @@ if __name__ == '__main__':
# # file_path3 = 'geo/51a7ee04-711c-4cf0-9d4c-4b523fba7037/TYQW/original.text'
# qianwen_process_original_data(file_path2)
# for i in task_id:
data_list = bh_utils.query_data(f"select * from geo_commit_task where taskId = '13e740b9-955c-4f9b-a03e-92e656ffaf43' and platform = 'TYQW'")
data_list = bh_utils.query_data(f"select * from geo_commit_task where taskId = '9fd93209-01db-4dd2-84e3-02eeccbedb16' and platform = 'TYQW'")
# # #
# # #
......
aidso_geo/utils/ai_utils.py
View file @ 4b1f9b3a
......@@ -3,9 +3,6 @@ import time
import requests
import json
from aidso_geo.core.down_load_bot import get_req_id
from aidso_geo.utils import bh_utils
from aidso_geo.utils.tos_utils import get_string_from_tos
......@@ -144,6 +141,126 @@ def ai_get_product_list(content, prompt):
except Exception as e:
return []
def ai_get_product_relation_spu(product_list, keyword):
url = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
payload = {
"model": "doubao-seed-2-0-mini-260428",
"messages": [
{
"role": "system",
"content": """
你的核心任务为：基于给定的商品列表与查询关键词，为列表内的每一件商品匹配对应信息，最终输出符合规范要求的JSON结果，执行过程需严格遵循以下规则：
1. 需为商品列表中的每一件商品单独生成一条匹配记录，每条记录必须固定包含以下4个字段，各字段的取值规则明确如下：
（1）produce_name：填写对应商品的完整原始名称，即商品列表中给出的该商品全称，不得做任何增删修改；
（2）brand：填写该商品对应的SPU品牌名称，需精准识别商品所属品牌，参考示例：“北京同仁堂陈皮茯苓茶”的品牌取值为“北京同仁堂”，“红魔11 Pro+”的品牌取值为“红魔”；若商品无明确可识别的品牌信息，则直接返回该商品的完整名称作为brand字段值；
（3）spu_name：填写该商品的标准SPU名称，即去除品牌前缀后的商品核心名称，参考示例：“北京同仁堂陈皮茯苓茶”的spu_name取值为“陈皮茯苓茶”，“一加 Ace 6 至尊版”的spu_name取值为“Ace 6 至尊版”；
（4）current：判断该商品与给定查询关键词是否存在关联，关联判定范围包括但不限于：关键词为该商品的品牌名、关键词为该品牌旗下的子品牌/系列名称、商品属于该关键词对应的品牌产品线；只要满足上述任意一种关联情形，该字段取值为1，若不存在任何关联则取值为0。
2. 一致性校验特别要求：同批次传入的商品中，若商品标题指向的品牌名相同，brand字段的取值必须保持完全统一，禁止出现同一品牌同时标注“小米”和“xiaomi”这类中英文/不同写法混用的情况，需统一为规范名称；同批次商品的spu_name也需保持表述一致，禁止出现同一SPU同时标注“12 promax”和“12promax”这类格式不统一的情况，需统一为规范表述。
{
"produce_name": "郎酒 红花郎15",
"brand": "郎酒",
"spu_name": "红花郎15",
"current": 0
},
"""
},
{
"role": "user",
"content": f"""需要处理的商品列表：{product_list} 本次查询的关键词为：{keyword}"""
}
],
"thinking": {
"type": "disabled"
},
"temperature": 0,
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "product_relation_result",
"strict": True,
"schema": {
"type": "object",
"properties": {
"product_words": {
"type": "array",
"description": "商品识别和关联判断结果",
"items": {
"type": "object",
"properties": {
"produce_name": {
"type": "string",
"description": "商品完整原始名称",
},
"brand": {
"type": "string",
"description": "商品所属品牌",
},
"spu_name": {
"type": "string",
"description": "去除品牌后的标准SPU名称",
},
"current": {
"type": "integer",
"enum": [0, 1],
"description": "与查询关键词有关为1，否则为0",
},
},
"required": [
"produce_name",
"brand",
"spu_name",
"current",
],
"additionalProperties": False,
},
},
},
"required": ["product_words"],
"additionalProperties": False,
},
},
},
}
headers = {
'Authorization': 'Bearer ark-7afc3be2-37a8-47fd-9f02-996258a3d305-27da0',
'Content-Type': 'application/json'
}
try:
response = requests.post(
url,
headers=headers,
json=payload,
timeout=120,
)
if response.status_code != 200:
print("status_code:", response.status_code)
print("response_text:", response.text)
response.raise_for_status()
response_data = response.json()
content = (
response_data
.get("choices", [{}])[0]
.get("message", {})
.get("content")
)
if not content:
return []
if isinstance(content, str):
content = json.loads(content)
product_words = content.get("product_words", [])
return product_words if isinstance(product_words, list) else []
except Exception as e:
print(e)
return []
def ai_get_product_list_search(product_list):
url = "https://ark.cn-beijing.volces.com/api/v3/chat/completions"
payload = json.dumps({
......@@ -989,16 +1106,16 @@ render_ecom_card_widget_jd_start:
render_ecom_card_widget_jd_end:
"""
brand_list =[
product_list =[
"瓜子二手车","二手车之家"
# fcc424e5-58af-494d-9683-5787413a26c9
]
promp = """
空调选购
keyword = """
瓜子
"""
print(ai_get_product_list(content,promp))
print(ai_get_product_relation_spu(product_list,keyword))
# print(ai_result)
# pro =
# ai_get_product_list()
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