zzhyyyyy
244 lines
8.3 KiB
Python
244 lines
8.3 KiB
Python
"""把无障碍树扁平化 + 候选簇喂给 LLM,一次出 商品标题、到手价、归属簇。"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Optional
|
|
|
|
from app.llm_client import MOCK_LLM, chat
|
|
from app.schemas import ClusterDto, NodeDto
|
|
|
|
logger = logging.getLogger("shagua.extractor")
|
|
|
|
PKG_TO_BRAND: dict[str, str] = {
|
|
"com.taobao.taobao": "淘宝",
|
|
"com.jingdong.app.mall": "京东",
|
|
"com.xunmeng.pinduoduo": "拼多多",
|
|
"com.ss.android.ugc.aweme": "抖音",
|
|
"com.sankuai.meituan": "美团",
|
|
"com.sankuai.meituan.takeoutnew": "美团",
|
|
"me.ele": "饿了么",
|
|
}
|
|
|
|
MAX_FLAT_CHARS = 12000
|
|
|
|
SYSTEM_PROMPT = """你是一个商品页结构化提取、商品归簇与市场参考价估算助手。
|
|
|
|
# 任务
|
|
|
|
用户会发给你两部分输入:
|
|
1. 一段从 Android 无障碍树取出的某个购物 / 外卖 / 团购 App 页面的可见文本与控件信息
|
|
2. 用户已有的"商品簇"列表(每个簇用一个代表标题描述)
|
|
|
|
请你完成三件事:
|
|
A. 从页面信息中识别当前商品的「标题」与「到手价」(单位:元,数字)
|
|
B. 判断当前商品是否归属于已有簇中的某一个,若是返回该簇 id,若否返回 null(由客户端新建簇)
|
|
C. 估算该商品的「市场常见价」(typical_price):主流电商平台常见售价区间的中位值,不含双 11 / 618 等特殊促销价
|
|
|
|
# 簇匹配规则
|
|
|
|
两件商品视为「同一簇」,核心商品名一致即可,无视规格、颜色、容量、装数、性别、码数等差异。
|
|
|
|
例:
|
|
- "iPhone 15 Pro 256GB" ↔ "iPhone 15 Pro 1TB 暮光紫" → 同簇 ✓
|
|
- "iPhone 15 Pro" ↔ "iPhone 14 Pro" → 不同簇 ✗ (型号不同)
|
|
- "海尔保温杯 500ml" ↔ "九阳保温杯 500ml" → 不同簇 ✗ (品牌不同)
|
|
- "PaulFrank 卫衣 男款 L" ↔ "PaulFrank 卫衣 女款 M" → 同簇 ✓ (性别码数算规格)
|
|
|
|
# 市场常见价估算
|
|
|
|
- 取主流电商(淘宝/京东/拼多多/抖音电商)常见售价区间的中位值
|
|
- 不含双 11、618、年货节、品牌大促等特殊促销价
|
|
- 单位:元,可以有小数,必须为正数
|
|
- **必须给出一个数字,不允许 null**。即使你对该商品不熟悉,也要根据**品类**(如矿泉水、卫衣、手机、咖啡等)、**品牌**(如有)、**规格**(数量/重量/容量/型号)的常识给出合理估算
|
|
- 估算思路:
|
|
* 知名品牌 → 该品牌该品类的官方建议零售价或主流电商常见价
|
|
* 冷门品牌 → 同品类(品类无关品牌)的市场常见价区间中位值
|
|
* 完全陌生 → 参考给定到手价 ±20% 的合理区间
|
|
|
|
# 输出格式
|
|
|
|
严格 JSON,无任何额外文字、不要 markdown 代码块、不要解释:
|
|
|
|
{"title": "完整商品标题", "price": 99.9, "cluster_id": 3, "typical_price": 120.0}
|
|
|
|
字段说明:
|
|
- title:商品标题字符串(最显眼最完整那段,不截断、不拼规格)
|
|
- price:到手价数字(优先级:实付到手价 > 优惠后价 > 标价)
|
|
- cluster_id:命中已有簇返回其 id (整数);未命中返回 null
|
|
- typical_price:市场常见价数字(正数),**必须返回数字,不允许 null**
|
|
|
|
# 失败处理
|
|
|
|
页面不是商品/团购/外卖详情页,或无法提取标题/价格时:
|
|
|
|
{"title": null, "price": null, "cluster_id": null, "typical_price": null}
|
|
|
|
# 价格细则
|
|
|
|
到手价选择优先级:实付 > 优惠后 > 单品标价。
|
|
"""
|
|
|
|
|
|
def flatten_tree(node: Optional[NodeDto]) -> str:
|
|
if node is None:
|
|
return ""
|
|
lines: list[str] = []
|
|
_walk(node, depth=0, out=lines)
|
|
text = "\n".join(lines)
|
|
if len(text) > MAX_FLAT_CHARS:
|
|
text = text[:MAX_FLAT_CHARS] + "\n...(truncated)"
|
|
return text
|
|
|
|
|
|
def _walk(node: NodeDto, depth: int, out: list[str]) -> None:
|
|
parts: list[str] = []
|
|
if node.view_id:
|
|
parts.append(f"id={node.view_id}")
|
|
if node.text:
|
|
parts.append(f'text="{node.text}"')
|
|
if node.desc:
|
|
parts.append(f'desc="{node.desc}"')
|
|
if parts:
|
|
out.append(" " * depth + " | ".join(parts))
|
|
for child in node.children or []:
|
|
_walk(child, depth + 1, out)
|
|
|
|
|
|
def _format_clusters(clusters: list[ClusterDto]) -> str:
|
|
if not clusters:
|
|
return "(无,这是用户记录的第一件商品)"
|
|
return "\n".join(f"- id={c.id}: {c.title}" for c in clusters)
|
|
|
|
|
|
def _mock_raw(flat: str) -> str:
|
|
"""占坑期 mock(DUOBIBI_MOCK_LLM):按页面文本哈希从预置商品池挑一件,
|
|
造 LLM 同格式 JSON(不归簇,由客户端按 cluster 规则新建)。"""
|
|
from app.mock_extractor import MOCK_PRODUCTS
|
|
|
|
h = hashlib.sha256(flat.encode("utf-8")).digest()
|
|
p = MOCK_PRODUCTS[h[0] % len(MOCK_PRODUCTS)]
|
|
return json.dumps(
|
|
{"title": p["title"], "price": p["price"], "cluster_id": None,
|
|
"typical_price": p["typical_price"]},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
|
|
def extract(
|
|
package_name: str,
|
|
tree: Optional[NodeDto],
|
|
clusters: list[ClusterDto],
|
|
) -> dict:
|
|
"""返回 {success, title?, price?, source_app, cluster_id?, reason?}。"""
|
|
brand = PKG_TO_BRAND.get(package_name, "未知")
|
|
if tree is None:
|
|
return {"success": False, "reason": "no_tree", "source_app": brand}
|
|
|
|
flat = flatten_tree(tree)
|
|
if not flat.strip():
|
|
return {"success": False, "reason": "empty_tree", "source_app": brand}
|
|
|
|
user_msg = (
|
|
f"App: {brand}\n\n"
|
|
f"页面信息:\n{flat}\n\n"
|
|
f"已有商品簇:\n{_format_clusters(clusters)}"
|
|
)
|
|
|
|
messages = [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": user_msg},
|
|
]
|
|
if MOCK_LLM:
|
|
raw = _mock_raw(flat)
|
|
else:
|
|
try:
|
|
raw = chat(messages)
|
|
logger.debug("LLM raw output: %s", raw[:500])
|
|
except Exception as e:
|
|
# LLM 不可用(限流/网络/超时):不 500,返回提取失败让客户端提示重试。
|
|
logger.warning("extract LLM call failed: %s", e)
|
|
raw = ""
|
|
|
|
title, price, cluster_id, typical_price = _parse_llm_output(raw)
|
|
if title is None or price is None:
|
|
return {
|
|
"success": False,
|
|
"reason": "llm_no_extract",
|
|
"source_app": brand,
|
|
"raw": raw[:200],
|
|
}
|
|
|
|
# 校验 cluster_id 是不是用户提供过的(防 LLM 编造)
|
|
valid_ids = {c.id for c in clusters}
|
|
if cluster_id is not None and cluster_id not in valid_ids:
|
|
logger.warning("LLM returned unknown cluster_id=%s, treating as new cluster", cluster_id)
|
|
cluster_id = None
|
|
|
|
# typical_price 兜底: prompt 已要求 LLM 必须给数字,但仍可能不听话或给负数。
|
|
# 这种极少数情况下用当前价兜底(客户端会显示"持平"),保证字段总有值。
|
|
if typical_price is None or typical_price <= 0:
|
|
typical_price = price
|
|
|
|
return {
|
|
"success": True,
|
|
"title": title,
|
|
"price": price,
|
|
"source_app": brand,
|
|
"cluster_id": cluster_id,
|
|
"typical_price": typical_price,
|
|
}
|
|
|
|
|
|
def _parse_number(raw) -> Optional[float]:
|
|
if isinstance(raw, bool):
|
|
return None
|
|
if isinstance(raw, (int, float)):
|
|
return float(raw)
|
|
if isinstance(raw, str):
|
|
try:
|
|
return float(raw.strip())
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _parse_llm_output(s: str) -> tuple[Optional[str], Optional[float], Optional[int], Optional[float]]:
|
|
s = s.strip()
|
|
s = re.sub(r"^```(?:json)?\s*", "", s)
|
|
s = re.sub(r"\s*```$", "", s)
|
|
|
|
data: Optional[dict] = None
|
|
try:
|
|
data = json.loads(s)
|
|
except json.JSONDecodeError:
|
|
m = re.search(r'\{[^{}]*"title"\s*:\s*[^{}]+\}', s)
|
|
if m:
|
|
try:
|
|
data = json.loads(m.group(0))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
if not isinstance(data, dict):
|
|
return None, None, None, None
|
|
|
|
title = data.get("title")
|
|
title = title.strip() if isinstance(title, str) and title.strip() else None
|
|
|
|
price = _parse_number(data.get("price"))
|
|
typical_price = _parse_number(data.get("typical_price"))
|
|
|
|
cid_raw = data.get("cluster_id")
|
|
if isinstance(cid_raw, bool):
|
|
cluster_id = None
|
|
elif isinstance(cid_raw, int):
|
|
cluster_id = cid_raw
|
|
elif isinstance(cid_raw, float) and cid_raw.is_integer():
|
|
cluster_id = int(cid_raw)
|
|
else:
|
|
cluster_id = None
|
|
|
|
return title, price, cluster_id, typical_price
|