Commit 3fc3b1a8 authored by yangruiqiang's avatar yangruiqiang

加字段

parent 916b7b37
from venv import logger
import tldextract as tldextract
from openpyxl import load_workbook
......@@ -9,6 +11,705 @@ from aidso_geo.utils import bh_utils, tos_utils
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment
import json
import os
import re
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
LLM_MODEL = os.getenv("GEO_SEMANTIC_ROLE_MODEL", "doubao-seed-2-0-lite-260215")
INFLUENCE_TYPE_MODEL = os.getenv("GEO_INFLUENCE_TYPE_MODEL", "doubao-seed-2-0-pro-260215")
LLM_API_URL = os.getenv("GEO_LLM_API_URL", "https://ark.cn-beijing.volces.com/api/v3/chat/completions")
LLM_API_KEY = os.getenv("GEO_LLM_API_KEY", "Bearer fcc424e5-58af-494d-9683-5787413a26c9")
LLM_MAX_RETRIES = 3
BATCH_QUOTE_SIZE = int(os.getenv("GEO_BATCH_QUOTE_SIZE", "5"))
ROW_MAX_WORKERS = int(os.getenv("GEO_ROW_MAX_WORKERS", "3"))
HTTP_TIMEOUT = 10
USER_AGENT = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"
)
DEFAULT_5118_EXTERNAL_LINK_URL = "https://openapi.chinaz.net/v1/1001/baidupc_domaininclude"
DEFAULT_5118_WEIGHT_URL = "http://apis.5118.com/weight"
DEFAULT_5118_ICP_URL = "http://apis.5118.com/icp/instant"
DEFAULT_EXTERNAL_LINK_API_KEY = ""
DEFAULT_5118_WEIGHT_API_KEY = "40D1FBB274CF47029F8C446AD2BCA2AD"
DEFAULT_5118_ICP_API_KEY = "3B7D2B1AEAFC483E98053FE24ED65CCA"
ICP_POLL_INTERVAL_SEC = 2.0
ICP_MAX_WAIT_SEC = 60.0
AUTHORITY_SCORE_CACHE = {}
SEMANTIC_ROLE_LABELS = {
"definition",
"statistical_data",
"comparison",
"procedure",
"background",
"else",
"no-reference",
}
def normalize_field_text(value):
if value is None:
return ""
return re.sub(r"\s+", " ", str(value)).strip()
def get_context(task_id, platform_code):
if not task_id or not platform_code:
return ""
return normalize_field_text(tos_utils.get_string_from_tos(f"geo/{task_id}/{platform_code}/context.txt"))
def get_quote_word_count(quote_context):
return len(re.sub(r"\s+", "", normalize_field_text(quote_context)))
def safe_float(value, default=0.0):
try:
return float(value)
except (TypeError, ValueError):
return default
def safe_int(value, default=0):
try:
return int(float(value))
except (TypeError, ValueError):
return default
def post_llm(payload):
headers = {
"Authorization": LLM_API_KEY,
"Content-Type": "application/json",
}
last_error = None
for attempt in range(LLM_MAX_RETRIES):
try:
response = requests.post(
LLM_API_URL,
headers=headers,
json=payload,
timeout=300,
)
response.raise_for_status()
return (
response.json()
.get("choices", [{}])[0]
.get("message", {})
.get("content", "")
.strip()
)
except Exception as exc:
last_error = str(exc)
if attempt < LLM_MAX_RETRIES - 1:
time.sleep(1 + attempt)
else:
raise RuntimeError(f"大模型调用失败。url={LLM_API_URL}, error={last_error}") from exc
def parse_semantic_role_content(content):
text = normalize_field_text(content)
if text.startswith("```"):
text = re.sub(r"^```(?:json)?", "", text, flags=re.I).strip()
text = re.sub(r"```$", "", text).strip()
candidates = [text]
match = re.search(r"\{.*\}", text, flags=re.S)
if match:
candidates.insert(0, match.group(0))
parsed = None
for candidate in candidates:
try:
parsed = json.loads(candidate.replace(":", ":").replace(",", ","))
break
except json.JSONDecodeError:
continue
if isinstance(parsed, dict):
semantic_role = parsed.get("semantic_role")
if isinstance(semantic_role, list):
raw_labels = [str(item).strip() for item in semantic_role if str(item).strip()]
else:
raw_labels = [item.strip() for item in str(semantic_role).replace(",", ",").split(",") if item.strip()]
else:
lowered = text.lower()
raw_labels = [label for label in SEMANTIC_ROLE_LABELS if label in lowered]
labels = [label for label in raw_labels if label in SEMANTIC_ROLE_LABELS]
return ",".join(labels) if labels else "else"
def get_quote_role(row_number, context, quote_context):
context = normalize_field_text(context)
quote_context = normalize_field_text(quote_context)
if not context and not quote_context:
return "no-reference"
prompt = (
"你是一个语义角色标注器。你的任务是判断回答文本 context 结合引用来源文本 quote_context "
"所共同体现的引用语义角色,并从固定标签集中选出最贴切的标签。\n\n"
"## 标签集(只能使用以下标签)\n"
"- definition:主要在解释概念、给出定义或阐明含义。\n"
"- statistical_data:主要在呈现数字、比例、金额、排名、增长率等统计或量化结果。\n"
"- comparison:主要在对比两个及以上对象的异同、优劣或差异。\n"
"- procedure:主要在说明步骤、流程、操作方法或先后顺序。\n"
"- background:主要在提供背景介绍、常识铺垫或上下文交代,本身不构成核心结论。\n"
"- else:内容构成了有效引用,但不属于上述任何一类。\n"
"- no-reference:内容未形成有效引用语义(如答非所问、空泛、与 quote_context 无实质关联)。\n\n"
"## 判定规则\n"
"1. 先判断是否构成有效引用:若否,直接返回 no-reference(它与其他标签互斥,不可共存)。\n"
"2. 判断「主要用途」而非「是否提及」。例如顺带出现一个数字,但整体在解释概念,应判 definition 而非 statistical_data。\n"
"3. 当多个角色都明显且并重(任一去掉都会丢失核心语义)时,才输出多标签;否则只输出最主要的一个。\n"
"4. 有效引用但确实无法归入前五类时,才使用 else(else 是兜底,不要轻易使用)。\n\n"
"## 输出格式(严格遵守)\n"
"- 只返回 JSON,不要解释、不要 markdown、不要代码块、不要多余文字。\n"
"- 单标签:{\"semantic_role\":\"标签\"}\n"
"- 多标签:{\"semantic_role\":[\"标签1\",\"标签2\"]}\n\n"
f"## 待判定内容\n"
f"【context全文】\n{context}\n\n"
f"【quote_context全文】\n{quote_context}\n\n"
"请输出 JSON:"
)
payload = {
"model": LLM_MODEL,
"messages": [
{"role": "system", "content": "你是一个严格的语义角色分类器,只输出合法 JSON。"},
{"role": "user", "content": prompt},
],
"stream": False,
"thinking": {"type": "disabled"},
"temperature": 0,
}
try:
return parse_semantic_role_content(post_llm(payload))
except Exception as exc:
print(f"[row {row_number}] quote_role 调模型失败,使用默认值: {exc}")
return "else"
def get_context_quote_type(row_number, context, quote_context):
context = normalize_field_text(context)
quote_context = normalize_field_text(quote_context)
if not context or not quote_context:
return "挂名引用"
prompt = (
"# 角色\n"
"你是专业的文本引用关系判定专家。给你两段文本:回答文本context、引用来源文本quote_context,分别是excel表中的两列,\n"
"判断\"回答\"\"来源\"的引用关系属于哪一类,只输出一个标签。\n\n"
"# 三档定义\n"
"1. 深度引用:回答对来源的核心观点/核心信息做了深度整合、拓展延伸或逻辑重构,\n"
" 内容高度依赖来源核心信息支撑,无明显脱离来源核心的无关内容。\n"
"2. 一般引用:回答只引用了来源的零散信息、次要观点,做了简单搬运或浅层拼接,\n"
" 未对来源核心内容做深度加工。\n"
"3. 挂名引用:回答与来源的核心信息、主要观点完全无关,仅标注了来源名称,\n"
" 或仅借用了非常边缘的无关信息,实质未参考来源有效信息。\n\n"
"# 判定要点(降低主观性)\n"
"- 依据是\"回答实际借用了来源多少内容\",而非是否标注出处;仅注明出处、提及来源名,本身不构成深度或一般引用。\n"
"- 相同处若只是该题材谁来写都会有的通用内容(常识、标准框架、大众皆知实体),倾向判\"一般引用\"\"挂名引用\";\n"
" 相同处若是来源特有、本可不同却偏偏一致的内容,才倾向判\"深度引用\"\n\n"
"# 输出(严格遵守)\n"
"仅输出以下三个标签之一,不得输出任何解释、标点或多余文字:\n"
"深度引用 / 一般引用 / 挂名引用\n\n"
"# 输入\n"
f"回答文本:{context}\n"
f"来源文本:{quote_context}\n"
"# 输出\n"
)
payload = {
"model": INFLUENCE_TYPE_MODEL,
"stream": False,
"messages": [
{"role": "system", "content": "你是专业的文本引用关系判定专家,只输出一个合法标签。"},
{"role": "user", "content": prompt},
],
"thinking": {"type": "disabled"},
"temperature": 0,
"top_p": 1,
}
try:
content = post_llm(payload)
for label in ["深度引用", "一般引用", "挂名引用"]:
if label in content:
return label
return "挂名引用"
except Exception as exc:
print(f"[row {row_number}] Context_quote_type 调模型失败,使用默认值: {exc}")
return "挂名引用"
def parse_json_array_response(content):
text = normalize_field_text(content)
if text.startswith("```"):
text = re.sub(r"^```(?:json)?", "", text, flags=re.I).strip()
text = re.sub(r"```$", "", text).strip()
candidates = [text]
match = re.search(r"\[.*\]", text, flags=re.S)
if match:
candidates.insert(0, match.group(0))
for candidate in candidates:
try:
parsed = json.loads(candidate)
except json.JSONDecodeError:
continue
if isinstance(parsed, list):
return [item for item in parsed if isinstance(item, dict)]
if isinstance(parsed, dict):
for key in ("results", "data", "items"):
value = parsed.get(key)
if isinstance(value, list):
return [item for item in value if isinstance(item, dict)]
return []
def normalize_quote_role_value(value):
if isinstance(value, list):
labels = [str(item).strip() for item in value if str(item).strip()]
else:
labels = [item.strip() for item in str(value).replace(",", ",").split(",") if item.strip()]
labels = [label for label in labels if label in SEMANTIC_ROLE_LABELS]
return ",".join(labels) if labels else "else"
def normalize_context_quote_type_value(value):
text = normalize_field_text(value)
for label in ("深度引用", "一般引用", "挂名引用"):
if label in text:
return label
return "挂名引用"
def normalize_context_quote_score_value(value):
score = safe_float(value, default=0.0)
return round(max(0.0, min(1.0, score)), 2)
def default_batch_llm_result():
return {
"quote_role": "else",
"Context_quote_type": "挂名引用",
"Context_quote_score": 0.0,
"Quote_lxfs": "未检测到文本中存在有效联系方式",
}
def get_batch_item_row_number(item):
return safe_int(item.get("_row_number") or item.get("row_number"), default=0)
def compute_quote_batch_fields_with_llm(context, items):
results = {get_batch_item_row_number(item): default_batch_llm_result() for item in items}
context = normalize_field_text(context)
if not context or not items:
return results
quote_blocks = []
for item in items:
row_number = get_batch_item_row_number(item)
quote_blocks.append(
f"row_number={row_number}\n"
f"quote_context:\n{normalize_field_text(item.get('quote_context'))}\n"
)
prompt = (
"# 角色\n"
"你是文本引用关系、语义角色和联系方式抽取专家。给你一段回答文本 context,以及多条引用来源文本 quote_context。\n"
"请分别判断每条 quote_context 与同一个 context 的关系,并分别抽取该 quote_context 内的联系方式。\n\n"
"# 每条引用需要输出的字段\n"
"1. quote_role: definition / statistical_data / comparison / procedure / background / else / no-reference。\n"
"2. Context_quote_type: 只能是 深度引用 / 一般引用 / 挂名引用。\n"
"3. Context_quote_score: 0-1 的实质依赖度分值,保留两位小数。\n"
"4. Quote_lxfs: 只从该条 quote_context 中提取联系方式;没有则填 未检测到文本中存在有效联系方式。\n\n"
"# 引用类型判定\n"
"- 深度引用: 回答对来源核心观点/核心信息做了深度整合、拓展延伸或逻辑重构。\n"
"- 一般引用: 回答只引用来源零散信息、次要观点,简单搬运或浅层拼接。\n"
"- 挂名引用: 回答与来源核心信息无关,或仅标注来源名称。\n\n"
"# 评分原则\n"
"只看回答文本与引用来源文本共享内容是否只能溯源到该来源;通用常识不计入依赖,来源特有内容计入依赖。\n"
"总分 >= 0.65 通常对应深度引用;0.30 <= 总分 < 0.65 通常对应一般引用;总分 < 0.30 通常对应挂名引用。\n\n"
"# 输出格式,严格遵守\n"
"只输出 JSON 数组,不要解释、不要 markdown。数组中每个对象必须包含:\n"
"row_number, quote_role, Context_quote_type, Context_quote_score, Quote_lxfs\n\n"
"# 输入\n"
f"context:\n{context}\n\n"
"quote_context 列表:\n"
+ "\n---\n".join(quote_blocks)
)
payload = {
"model": INFLUENCE_TYPE_MODEL,
"stream": False,
"messages": [
{"role": "system", "content": "你只输出合法 JSON 数组,不要解释。"},
{"role": "user", "content": prompt},
],
"thinking": {"type": "disabled"},
"temperature": 0,
"top_p": 1,
}
try:
rows = parse_json_array_response(post_llm(payload))
for row in rows:
row_number = safe_int(row.get("row_number"), default=0)
if row_number not in results:
continue
quote_lxfs = normalize_field_text(row.get("Quote_lxfs")) or "未检测到文本中存在有效联系方式"
results[row_number] = {
"quote_role": normalize_quote_role_value(row.get("quote_role")),
"Context_quote_type": normalize_context_quote_type_value(row.get("Context_quote_type")),
"Context_quote_score": normalize_context_quote_score_value(row.get("Context_quote_score")),
"Quote_lxfs": quote_lxfs,
}
except Exception as exc:
row_numbers = ",".join(str(get_batch_item_row_number(item)) for item in items)
print(f"[batch rows {row_numbers}] 批量模型失败,使用默认值: {exc}")
return results
def get_5118_config():
return {
"external_link_url": os.getenv("GEO_5118_EXTERNAL_LINK_URL", DEFAULT_5118_EXTERNAL_LINK_URL).strip(),
"weight_url": os.getenv("GEO_5118_WEIGHT_URL", DEFAULT_5118_WEIGHT_URL).strip(),
"icp_url": os.getenv("GEO_5118_ICP_URL", DEFAULT_5118_ICP_URL).strip(),
"external_link_api_key": os.getenv(
"GEO_5118_EXTERNAL_LINK_API_KEY",
os.getenv("GEO_5118_API_KEY", DEFAULT_EXTERNAL_LINK_API_KEY),
).strip(),
"weight_api_key": os.getenv(
"GEO_5118_WEIGHT_API_KEY",
os.getenv("GEO_5118_API_KEY", DEFAULT_5118_WEIGHT_API_KEY),
).strip(),
"icp_api_key": os.getenv(
"GEO_5118_ICP_API_KEY",
os.getenv("GEO_5118_API_KEY", DEFAULT_5118_ICP_API_KEY),
).strip(),
}
def call_5118_json_api(url, params, headers, method="POST"):
if method.upper() == "GET":
response = requests.get(url, params=params, headers=headers, timeout=HTTP_TIMEOUT)
else:
response = requests.post(url, data=params, headers=headers, timeout=HTTP_TIMEOUT)
response.raise_for_status()
data = response.json()
if not isinstance(data, dict):
raise ValueError(f"5118 接口返回不是 JSON 对象: {url}")
return data
def get_nested_data_field(payload, field):
data = payload.get("data")
if isinstance(data, dict):
return data.get(field)
return None
def parse_5118_weight_value(value):
text = normalize_field_text(value)
if not text:
return 0.0
return safe_float(text.replace("+", "").replace("-", ""), default=0.0)
def extract_external_link_count(external_link_data):
result_payload = external_link_data.get("Result")
if isinstance(result_payload, dict) and result_payload.get("SiteCount") is not None:
return safe_float(result_payload.get("SiteCount"), default=0.0)
for field in ["total", "external_link_count"]:
if external_link_data.get(field) is not None:
return safe_float(external_link_data.get(field), default=0.0)
nested_value = get_nested_data_field(external_link_data, field)
if nested_value is not None:
return safe_float(nested_value, default=0.0)
return 0.0
def extract_baidu_pc_weight(weight_data):
data = weight_data.get("data")
result_list = data.get("result") if isinstance(data, dict) else None
if isinstance(result_list, list) and result_list and isinstance(result_list[0], dict):
return parse_5118_weight_value(result_list[0].get("weight"))
direct_value = (
weight_data.get("weight")
or weight_data.get("baidu_pc_weight")
or get_nested_data_field(weight_data, "weight")
or get_nested_data_field(weight_data, "baidu_weight")
)
return parse_5118_weight_value(direct_value)
def has_5118_icp_subject(data):
payload = data.get("data")
subject = payload.get("subject") if isinstance(payload, dict) else None
return isinstance(subject, dict) and (
normalize_field_text(subject.get("company_type"))
or normalize_field_text(subject.get("confirm_time"))
)
def extract_5118_icp_task_id(data):
if data.get("taskid"):
return data.get("taskid")
payload = data.get("data")
if isinstance(payload, dict):
return payload.get("taskid")
return None
def call_5118_icp_instant_api(domain, api_key):
headers = {"User-Agent": USER_AGENT}
if api_key:
headers["Authorization"] = api_key
return call_5118_json_api(
DEFAULT_5118_ICP_URL,
{"searchtext": domain},
headers,
)
def poll_5118_icp_result(domain, task_id, api_key):
headers = {"User-Agent": USER_AGENT}
if api_key:
headers["Authorization"] = api_key
deadline = time.time() + ICP_MAX_WAIT_SEC
last_data = {}
while time.time() < deadline:
last_data = call_5118_json_api(
DEFAULT_5118_ICP_URL,
{"searchtext": domain, "taskid": str(task_id)},
headers,
)
if has_5118_icp_subject(last_data):
return last_data
# time.sleep(ICP_POLL_INTERVAL_SEC)
return last_data
def fetch_5118_icp_detail(domain, api_key):
initial = call_5118_icp_instant_api(domain, api_key)
if has_5118_icp_subject(initial):
return initial
task_id = extract_5118_icp_task_id(initial)
if not task_id:
return initial
return poll_5118_icp_result(domain, task_id, api_key)
def parse_icp_year(value):
match = re.search(r"(19|20)\d{2}", normalize_field_text(value))
return int(match.group()) if match else None
def score_icp_year(year):
if year is None:
return 0.0
if year <= 2015:
return 1.0
if year <= 2020:
return 0.8
return 0.6
def score_company_type(company_type):
text = normalize_field_text(company_type).lower()
if not text:
return 0.0
if "企业" in text or "company" in text or "corp" in text:
return 1.0
if "个人" in text or "person" in text or "individual" in text:
return 0.5
return 0.0
def compute_quote_qwx_score(metric):
external_link_count = safe_float(metric.get("external_link_count"), default=0.0)
external_link_threshold = safe_float(os.getenv("GEO_5118_EXTERNAL_LINK_THRESHOLD", "0"), default=0.0)
external_link_score = 1.0 if external_link_count > external_link_threshold else 0.0
baidu_score = max(0.0, min(safe_float(metric.get("baidu_pc_weight"), default=0.0) / 10.0, 1.0))
icp_year_score = score_icp_year(metric.get("icp_year"))
company_type_score = score_company_type(metric.get("icp_company_type"))
score = (
0.35 * external_link_score
+ 0.30 * baidu_score
+ 0.20 * icp_year_score
+ 0.15 * company_type_score
)
return round(max(0.0, min(score, 1.0)), 6)
def fetch_quote_qwx_metric(domain):
domain = get_main_domain(domain)
if not domain:
return {}
config = get_5118_config()
if not config["external_link_api_key"] or not config["weight_api_key"] or not config["icp_api_key"]:
raise ValueError("缺少 5118 API 配置,请设置 GEO_5118_EXTERNAL_LINK_API_KEY / GEO_5118_WEIGHT_API_KEY / GEO_5118_ICP_API_KEY")
external_domain = domain if domain.startswith("www.") else f"www.{domain}"
external_link_data = call_5118_json_api(
config["external_link_url"],
{"domain": external_domain, "APIKey": config["external_link_api_key"], "ChinazVer": "1.0"},
{"User-Agent": USER_AGENT},
method="GET",
)
weight_data = call_5118_json_api(
config["weight_url"],
{"url": domain},
{"Authorization": config["weight_api_key"], "User-Agent": USER_AGENT},
)
icp_data = fetch_5118_icp_detail(domain, config["icp_api_key"])
payload = icp_data.get("data")
icp_subject = payload.get("subject") if isinstance(payload, dict) else {}
if not isinstance(icp_subject, dict):
icp_subject = {}
return {
"external_link_count": extract_external_link_count(external_link_data),
"baidu_pc_weight": extract_baidu_pc_weight(weight_data),
"icp_approved_time": normalize_field_text(icp_subject.get("confirm_time")),
"icp_company_type": normalize_field_text(icp_subject.get("company_type")),
"icp_year": parse_icp_year(icp_subject.get("confirm_time")),
}
def get_quote_qwx_score(domain):
domain = get_main_domain(domain)
if not domain:
return 0.0
if domain in AUTHORITY_SCORE_CACHE:
return AUTHORITY_SCORE_CACHE[domain]
try:
metric = fetch_quote_qwx_metric(domain)
score = compute_quote_qwx_score(metric)
except Exception as exc:
print(f"[Quote_qwx_score] domain={domain} 调 5118 API 失败,使用默认值 0.0: {exc}")
score = 0.0
AUTHORITY_SCORE_CACHE[domain] = score
return score
def get_context_quote_score(row_number, context, quote_context):
context = normalize_field_text(context)
quote_context = normalize_field_text(quote_context)
if not context or not quote_context:
return 0.0
prompt = (
"# 角色\n"
"你是文本溯源与引用关系判定专家。给你两段文本:回答文本(context)、引用来源文本( quote_context)。\n"
"判断 回答文本 对 引用来源文本 的引用关系类型,并给出 0–1 的实质依赖度分值。\n\n"
"# 核心原则\n"
"只看\"回答文本 与 引用来源文本 共享的内容,是不是只能溯源到 引用来源文本\":\n"
"- 共享的是任何写此题材都会有的通用内容(常识、标准框架、大众皆知实体)→ 不计入依赖\n"
"- 共享的是 引用来源文本 特有、本可不同却偏偏相同的内容 → 计入依赖\n"
"注明出处、挂来源名,本身不提升依赖度;只看内容实际借了多少。\n\n"
"# 评分维度(每维先按重合程度打 0–1,再乘权重,加权求和得总分 0–1)\n"
"1. 精确数据/事实重合(权重 0.30):相同的精确数字、检测值、报告号、专利号、日期等非显然事实。\n"
"2. 特异措辞/生造词/口号重合(权重 0.25):相同的自创术语、品牌话术、独特句式(非通用行业词)。\n"
"3. 属性—对象绑定重合(权重 0.20):对同一对象用同一套理由/评价绑定(如对同一品牌给出同样优缺点逻辑)。\n"
"4. 罕见选择/排序重合(权重 0.15):冷门条目的选取、推荐顺序、排名、主推对象高度一致。\n"
"5. 论证结构/结论路径重合(权重 0.10):不仅结论相同,推理链(\"凭什么得出\")也相同。\n"
"说明:通用题材词、标准框架(如价位分档)、大众皆知实体一律视为题材强制内容,该部分记 0 分。\n\n"
"# 分值与类型(边界不重叠)\n"
"- 总分 ≥ 0.65 →「深度引用」:实质改写/吸收了 引用来源文本 的独占性内容\n"
"- 0.30 ≤ 总分 < 0.65 →「一般引用」:仅共享题材通用内容,引用来源文本 只是众多可能来源之一\n"
"- 总分 < 0.30 →「挂名引用」:仅来源名/标注相连,内容实质独立,借用近乎为零\n\n"
"# 输出(严格遵守)\n"
"填充到每一行对应的列中,列名“Context_quote_score”,分值保留两位小数,不得有其他文字。\n\n\n"
"# 输入\n"
f"A(回答文本):{context}\n"
f"B(来源文本):{quote_context}\n"
"# 输出\n"
)
payload = {
"model": INFLUENCE_TYPE_MODEL,
"stream": False,
"messages": [
{"role": "system", "content": "你是文本溯源与引用关系判定专家,只输出一个两位小数分值。"},
{"role": "user", "content": prompt},
],
"thinking": {"type": "disabled"},
"temperature": 0,
"top_p": 1,
}
try:
content = post_llm(payload)
score_match = re.search(r"\b(?:0(?:\.\d+)?|1(?:\.0+)?)\b", content)
score = float(score_match.group(0)) if score_match else 0.0
return round(max(0.0, min(1.0, score)), 2)
except Exception as exc:
print(f"[row {row_number}] Context_quote_score 调模型失败,使用默认值: {exc}")
return 0.0
def extract_contact_info(row_number, text, field_name):
text = normalize_field_text(text)
if not text:
return "未检测到文本中存在有效联系方式"
prompt = f"""# 角色
你是专业的信息提取专员。你需要从一段文本中完整、准确地提取所有联系方式。
# 提取范围(包括但不限于)
官方网站链接、微博账号/主页链接、微信号/公众号名称、手机号码、固定电话、
邮箱地址、QQ 账号、抖音账号、小红书账号,以及其他任何可用于联系对方的信息。
# 执行规则
1. 完整提取:逐项扫描全文,不得遗漏任何符合范围的联系方式。
2. 忠于原文:提取内容必须与原文逐字一致,不得擅自修改、增删、补全或推断;
仅在网址缺少协议头时可附上可访问链接,其余一律保持原文写法。
3. 去重:同一联系方式重复出现时只保留一条。
4. 空结果:文本不存在任何有效联系方式时,仅输出"未检测到文本中存在有效联系方式"。
# 输出格式(严格遵守)
只输出 {field_name} 这一列要写入的字符串,不要任何前后说明、不要 Markdown 代码块包裹。
有联系方式时,每条联系方式用「; 」分隔,格式为"类型:内容"。
无联系方式时,仅输出:未检测到文本中存在有效联系方式
# 输入
字段名:{field_name}
文本内容:{text}
# 输出 """
payload = {
"model": LLM_MODEL,
"stream": False,
"messages": [
{"role": "system", "content": "你是专业的信息提取专员,只输出提取结果,不要解释。"},
{"role": "user", "content": prompt},
],
"thinking": {"type": "disabled"},
"temperature": 0,
"top_p": 1,
}
try:
return post_llm(payload) or "未检测到文本中存在有效联系方式"
except Exception as exc:
print(f"[row {row_number}] {field_name} 调模型失败,使用默认值: {exc}")
return "未检测到文本中存在有效联系方式"
def get_context_lxfs(row_number, context):
return extract_contact_info(row_number, context, "Context_lxfs")
def get_quote_lxfs(row_number, quote_context):
return extract_contact_info(row_number, quote_context, "Quote_lxfs")
def write_list_dict_to_excel(data_list, output_path):
......@@ -93,16 +794,19 @@ def excel_to_prompt_map(excel_path: str):
try:
prompt_id_idx = headers.index("prompt_id")
prompt_idx = headers.index("prompt")
layer_subcat_idx = headers.index("layer-subcat")
except ValueError as e:
raise ValueError(f"Excel 表头缺少必要字段:{e},当前表头:{headers}")
layer_idx = headers.index("layer") if "layer" in headers else None
subcat_idx = headers.index("subcat") if "subcat" in headers else None
layer_subcat_idx = headers.index("layer-subcat") if "layer-subcat" in headers else None
if layer_subcat_idx is None and (layer_idx is None or subcat_idx is None):
raise ValueError(f"Excel 表头缺少 `layer/subcat` 或 `layer-subcat`,当前表头:{headers}")
result_map = {}
for row in ws.iter_rows(min_row=2, values_only=True):
prompt_id = row[prompt_id_idx]
prompt = row[prompt_idx]
layer_subcat = row[layer_subcat_idx]
# 跳过空 prompt
if prompt is None or str(prompt).strip() == "":
......@@ -110,8 +814,13 @@ def excel_to_prompt_map(excel_path: str):
prompt = str(prompt).strip()
layer = ""
subcat = ""
layer = str(row[layer_idx]).strip() if layer_idx is not None and row[layer_idx] is not None else ""
subcat = str(row[subcat_idx]).strip() if subcat_idx is not None and row[subcat_idx] is not None else ""
if (not layer and not subcat) and layer_subcat_idx is not None:
layer_subcat = row[layer_subcat_idx]
else:
layer_subcat = None
if layer_subcat is not None and str(layer_subcat).strip() != "":
layer_subcat = str(layer_subcat).strip()
......@@ -189,7 +898,7 @@ def get_aidso_data(phone,begin,end,brand_name,platform=None):
else:
query_sql = f"select * from geo_commit_task where reqId in ({req_id_sql}) limit 1"
query_list = bh_utils.query_data(query_sql)
logger.log(f"req_id {req_id} success")
result = []
if query_list:
for q in query_list:
......@@ -264,24 +973,32 @@ def get_aidso_data_v2(phone,begin,end,brand_name,platform=None):
query_list = bh_utils.query_data(query_sql)
result = []
context_lxfs_cache = {}
rows_by_context = defaultdict(list)
if query_list:
for q in query_list:
print(q.get('taskId'))
quote_str = tos_utils.get_string_from_tos(f"geo/{q.get('taskId')}/{q.get('platform')}/quote.txt")
if quote_str:
quto_list = json.loads(tos_utils.get_string_from_tos(f"geo/{q.get('taskId')}/{q.get('platform')}/quote.txt"))
quto_list = json.loads(quote_str)
platform_code= q.get('platform')
prompt= q.get('prompt')
context = get_context(q.get('taskId'), platform_code)
for quto in quto_list:
quote_url = quto.get('url')
quote_title = quto.get('title')
quote_site_name = quto.get('site_name')
quote_index = quto.get('index')
quote_index = quto.get('index','') or quto.get('source_seq','') or None
quote_published_at = quto.get('published_at')
quote_snippet = quto.get('snippet')
quote_snippet = normalize_field_text(quto.get('snippet'))
quote_context = ""
domain = get_main_domain(quote_url)
row_number = len(result) + 2
if context not in context_lxfs_cache:
context_lxfs_cache[context] = get_context_lxfs(row_number, context)
r = {
"_row_number": row_number,
"prompt": prompt,
"platform_code": platform_code,
"quote_url":quote_url,
......@@ -290,6 +1007,16 @@ def get_aidso_data_v2(phone,begin,end,brand_name,platform=None):
"quote_index": quote_index,
"published_at": quote_published_at,
"domain": domain,
"quote_word_count": get_quote_word_count(quote_context),
"quote_role": "else",
"Context_quote_type": "挂名引用",
"Quote_qwx_score": get_quote_qwx_score(domain),
# "Quote_qwx_score": 0,
"context": context,
"quote_context": quote_context,
"Context_quote_score": 0.0,
"Context_lxfs": context_lxfs_cache[context],
"Quote_lxfs": "未检测到文本中存在有效联系方式",
"snippet": quote_snippet
# "content": content,
# "quote": quto_list,
......@@ -298,18 +1025,53 @@ def get_aidso_data_v2(phone,begin,end,brand_name,platform=None):
# "brand_name": req_brand_map[q.get('reqId')],
}
result.append(r)
if context:
rows_by_context[context].append(r)
for context, context_rows in rows_by_context.items():
quote_batches = [
context_rows[start:start + BATCH_QUOTE_SIZE]
for start in range(0, len(context_rows), BATCH_QUOTE_SIZE)
]
for job_start in range(0, len(quote_batches), ROW_MAX_WORKERS):
job_window = quote_batches[job_start:job_start + ROW_MAX_WORKERS]
print(
f"[batch] context_row={context_rows[0]['_row_number']} "
f"进度={job_start + 1}-{job_start + len(job_window)}/{len(quote_batches)}"
)
with ThreadPoolExecutor(max_workers=ROW_MAX_WORKERS) as executor:
future_to_batch = {
executor.submit(compute_quote_batch_fields_with_llm, context, quote_batch): quote_batch
for quote_batch in job_window
}
for future in as_completed(future_to_batch):
quote_batch = future_to_batch[future]
try:
batch_results = future.result()
except Exception as exc:
row_numbers = ",".join(str(item["_row_number"]) for item in quote_batch)
print(f"[batch rows {row_numbers}] 批量模型失败,使用默认值: {exc}")
batch_results = {item["_row_number"]: default_batch_llm_result() for item in quote_batch}
for item in quote_batch:
fields = batch_results.get(item["_row_number"], default_batch_llm_result())
item["quote_role"] = fields.get("quote_role") or "else"
item["Context_quote_type"] = fields.get("Context_quote_type") or "挂名引用"
item["Context_quote_score"] = normalize_context_quote_score_value(fields.get("Context_quote_score"))
item["Quote_lxfs"] = fields.get("Quote_lxfs") or "未检测到文本中存在有效联系方式"
for item in result:
item.pop("_row_number", None)
return result
if __name__ == "__main__":
excel_map = excel_to_prompt_map(
excel_path="/Users/yaowentong/Desktop/shuju.xlsx"
# excel_path="/Users/yaowentong/Desktop/shuju.xlsx"
excel_path="/Users/mac/Downloads/表3_合并完成.xlsx"
)
phone = 13810898434
begin = "2026-06-04"
end = "2026-06-05"
brand_name = ['国内AI平台研究【1】','国内AI平台研究【2】']
begin = "2026-06-17"
end = "2026-06-17"
brand_name = ['国内AI平台研究【3】']
# print(excel_map)
aidso_result =get_aidso_data_v2(phone,begin,end,brand_name)
......@@ -321,10 +1083,10 @@ if __name__ == "__main__":
aidso["subcat"] = aidso_excel_data.get('subcat','')
write_list_dict_to_excel(
data_list=aidso_result,
output_path="/Users/yaowentong/Desktop/aidso_result_v2.xlsx"
output_path="/Users/mac/Downloads/表3_合并完成.xlsx"
)
#
all_file = f"/Users/yaowentong/Desktop/aidso_result_v2.txt"
with open(all_file, "w", encoding="utf-8") as f:
for item in aidso_result:
f.write(json.dumps(item, ensure_ascii=False) + "\n\n\n")
# all_file = f"/Users/mac/Desktop/aidso_result_40.txt"
# with open(all_file, "w", encoding="utf-8") as f:
# for item in aidso_result:
# f.write(json.dumps(item, ensure_ascii=False) + "\n\n\n")
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