210 lines
13 KiB
Python
210 lines
13 KiB
Python
"""Small renovation-only rules pilot over a frozen, evidence-backed search run."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import unicodedata
|
|
from collections import Counter
|
|
from copy import deepcopy
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi import HTTPException
|
|
from pydantic import Field, model_validator
|
|
|
|
from .db import encoded, identifier, now
|
|
from .learning_contracts import ClosedInput
|
|
from .worker import safe_error_codes
|
|
|
|
|
|
class DiscoveryInput(ClosedInput):
|
|
search_candidate_id: str = Field(min_length=1, max_length=120)
|
|
keyword: str = Field(min_length=1, max_length=2000)
|
|
service: str = Field(default="装修服务", min_length=1, max_length=120)
|
|
target_customer: str = Field(default="有装修需求的业主", min_length=1, max_length=300)
|
|
region: str = Field(default="", max_length=120)
|
|
max_selected: int = Field(default=2, ge=1, le=3, strict=True)
|
|
|
|
@model_validator(mode="after")
|
|
def ordinary_conditions(self):
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]+", self.search_candidate_id):
|
|
raise ValueError("invalid candidate reference")
|
|
for key in ("keyword", "service", "target_customer", "region"):
|
|
setattr(self, key, getattr(self, key).strip())
|
|
if not self.keyword or not self.service or not self.target_customer:
|
|
raise ValueError("business conditions cannot be blank")
|
|
if self.service != "装修服务" or self.target_customer != "有装修需求的业主":
|
|
raise ValueError("only the current renovation owner discovery scope is supported")
|
|
return self
|
|
|
|
|
|
class BusinessDiscoveryInput(ClosedInput):
|
|
search_candidate_id: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_-]+$")
|
|
max_selected: int = Field(default=2, ge=1, le=3, strict=True)
|
|
|
|
|
|
RENOVATION = re.compile(r"装修|装潢|家装|旧房|新房|毛坯|硬装|软装|全屋|改造|验房|水电|翻新")
|
|
HELP = re.compile(r"求助|求推荐|有没有.{0,12}推荐|准备.{0,5}装修|打算.{0,5}装修|想.{0,5}装修|要装修|装修小白|装修新手|找.{0,10}(?:工长|师傅|装修公司|设计师)|求.{0,10}(?:公司|工长|报价)|我家.{0,8}(?:装|改造)")
|
|
CASE = re.compile(r"案例|设计师|装修公司|工长|实景|作品|工地|接单")
|
|
FINISHED = re.compile(r"(?<![没未])(?:完工|竣工|交付|完装|装完|入住)")
|
|
UNCERTAINTIES = ["作者是否业主及真实装修需求尚未核实", "地区与预算未核实，未作地域过滤", "仅装修样例规则初筛，未调用模型"]
|
|
|
|
|
|
def origin(value: str) -> tuple | None:
|
|
try:
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
|
|
return None
|
|
return parsed.scheme, parsed.hostname.lower(), parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def note_reference(value, entry_url: str) -> tuple[str, str] | None:
|
|
if not isinstance(value, str) or origin(value) != origin(entry_url):
|
|
return None
|
|
parsed = urlsplit(value)
|
|
match = re.fullmatch(r"/(?:explore|search_result)/([a-fA-F0-9]{24})", parsed.path)
|
|
if not match:
|
|
return None
|
|
entry = urlsplit(entry_url)
|
|
note_id = match[1].lower()
|
|
return note_id, f"{entry.scheme}://{entry.netloc}/explore/{note_id}"
|
|
|
|
|
|
def title_key(value: str) -> str:
|
|
return " ".join(unicodedata.normalize("NFKC", value).split()).casefold()
|
|
|
|
|
|
def select_results(items: list[dict], entry_url: str, max_selected: int) -> tuple[list[dict], list[dict]]:
|
|
title_counts = Counter(title_key(item["title"]) for item in items if isinstance(item.get("title"), str) and item["title"].strip())
|
|
results, by_id, scores = [], {}, {}
|
|
for index, item in enumerate(items, 1):
|
|
title = item.get("title") if isinstance(item.get("title"), str) else None
|
|
author = item.get("author") if isinstance(item.get("author"), str) else None
|
|
reference = note_reference(item.get("url"), entry_url)
|
|
row = {"index": index, "title": title, "author": author, "url": reference[1] if reference else None, "canonical_url": reference[1] if reference else None,
|
|
"note_id": reference[0] if reference else None, "classification": "invalid", "selected": False, "reasons": [], "uncertainties": list(UNCERTAINTIES)}
|
|
if reference and reference[0] in by_id:
|
|
first = by_id[reference[0]]
|
|
first["reasons"].append("同一笔记 ID 的重复结果已合并")
|
|
# Different card texts for one object are not a sound click target.
|
|
if title_key(title or "") != title_key(first["title"] or ""):
|
|
first["classification"] = "ambiguous"
|
|
first["reasons"].append("同一笔记出现不同标题，需要人工核对")
|
|
continue
|
|
results.append(row)
|
|
if not reference:
|
|
row["reasons"].append("来源不是候选同站的真实笔记路径，未选择")
|
|
continue
|
|
by_id[reference[0]] = row
|
|
if not title or not title.strip():
|
|
row["classification"] = "ambiguous"
|
|
row["reasons"].append("当前未读到标题，不能安全按标题选择")
|
|
elif "[已隐藏]" in title or "[手机号已隐藏]" in title:
|
|
row["classification"] = "ambiguous"
|
|
row["reasons"].append("标题含脱敏替代文字，不能作为精确点击依据")
|
|
elif title_counts[title_key(title)] > 1:
|
|
row["classification"] = "ambiguous"
|
|
row["reasons"].append("标题重复，无法唯一定位笔记，未自动选择")
|
|
elif not RENOVATION.search(title):
|
|
row["classification"] = "unrelated"
|
|
row["reasons"].append("标题未出现装修相关信号，不推定采购需求")
|
|
elif FINISHED.search(title):
|
|
row["classification"] = "peer_case"
|
|
row["reasons"].append("存在已完工或入住分享信号，保留评论来源，不判定为待装修需求")
|
|
scores[index] = 60
|
|
elif HELP.search(title):
|
|
row["classification"] = "consumer_help"
|
|
row["reasons"].append("标题出现装修求助或准备装修表述，优先进一步阅读；不确认真实业主身份")
|
|
scores[index] = 100
|
|
elif CASE.search(title):
|
|
row["classification"] = "peer_case"
|
|
row["reasons"].append("存在案例或服务展示信号，保留作为评论线索来源；不确认作者为商家")
|
|
scores[index] = 60
|
|
else:
|
|
row["classification"] = "related_content"
|
|
row["reasons"].append("标题与装修相关，可进一步查看；相关或好看不等于需求")
|
|
scores[index] = 40
|
|
ranked = sorted((item for item in results if item["classification"] in {"consumer_help", "peer_case", "related_content"}), key=lambda item: (-scores[item["index"]], item["index"]))
|
|
selected = []
|
|
for row in ranked[:max_selected]:
|
|
row["selected"] = True
|
|
selected.append({"note_title": row["title"], "note_url": row["canonical_url"], "note_id": row["note_id"], "reason": row["reasons"][0]})
|
|
for row in ranked[max_selected:]:
|
|
row["reasons"].append("达到本次保留数量上限，本轮未保留用于后续详情读取")
|
|
return results, selected
|
|
|
|
|
|
class DiscoveryService:
|
|
def __init__(self, learning):
|
|
self.learning, self.store = learning, learning.store
|
|
|
|
def get(self, run_id: str, owner_id: str) -> dict:
|
|
row = self.store.one("SELECT * FROM discovery_runs WHERE id=$1 AND owner_id=$2", (run_id, owner_id))
|
|
if not row:
|
|
raise HTTPException(404, "发现记录不存在")
|
|
return {"id": row["id"], "created_at": row["created_at"], **json.loads(row["data_json"])}
|
|
|
|
def list(self, owner_id: str) -> list[dict]:
|
|
return [self.get(row["id"], owner_id) for row in self.store.rows(f"SELECT id FROM discovery_runs WHERE owner_id=$1 ORDER BY {self.store.order_column()} DESC", (owner_id,))]
|
|
|
|
@staticmethod
|
|
def supported_candidate(candidate: dict) -> bool:
|
|
manifest = candidate["manifest"]
|
|
entry_origin = origin(manifest.get("entry_url", ""))
|
|
platform = manifest.get("platform")
|
|
supported_origin = (platform == "xiaohongshu" and entry_origin == ("https", "www.xiaohongshu.com", 443)) or (platform == "demo" and entry_origin and entry_origin[1] in {"127.0.0.1", "localhost", "::1"})
|
|
steps = manifest.get("steps") or []
|
|
if not supported_origin or len(steps) != 4 or [step.get("kind") for step in steps] != ["fill", "press", "wait", "extract"]:
|
|
return False
|
|
if "keyword" not in manifest.get("parameters", {}) or steps[0].get("parameter") != "keyword" or steps[1].get("key") != "Enter" or steps[2].get("state") != "visible" or type(steps[2].get("min_count")) is not int or steps[2]["min_count"] < 1:
|
|
return False
|
|
output = steps[3]
|
|
fields = {item["name"]: item for item in output.get("fields", [])}
|
|
return (output.get("output_name") == "search_results" and set(fields) == {"title", "author", "url"}
|
|
and all(fields[name].get("source", "text") == "text" and not fields[name].get("required") for name in ("title", "author"))
|
|
and fields["url"].get("source") == "href" and fields["url"].get("required") is True)
|
|
|
|
async def create_for_brief(self, owner_id: str, brief: dict, values: dict) -> dict:
|
|
"""Use one explicitly confirmed snapshot; never accept condition overrides."""
|
|
conditions = brief["conditions"]
|
|
inputs = DiscoveryInput(search_candidate_id=values["search_candidate_id"],
|
|
max_selected=values["max_selected"], keyword=conditions["search_terms"][0],
|
|
service=conditions["service"], target_customer=conditions["target_customer"],
|
|
region=conditions["region"] or "")
|
|
return await self.create(owner_id, inputs.model_dump(), business_brief=brief)
|
|
|
|
async def create(self, owner_id: str, values: dict, *, business_brief: dict | None = None) -> dict:
|
|
candidate = self.learning.candidate(values["search_candidate_id"], owner_id)
|
|
content = {"status": "failed", "analysis_mode": "rules", "model_connected": False, "rule_scope": "renovation_only",
|
|
"search_candidate_id": candidate["id"], "search_task_id": candidate["task_id"], "candidate_digest": candidate["digest"], "validation_id": None,
|
|
"keyword": values["keyword"], "conditions": {key: values[key] for key in ("service", "target_customer", "region")}, "max_selected": values["max_selected"],
|
|
"results": [], "selected": [], "evidence": [], "errors": [], "error_codes": []}
|
|
if business_brief is not None:
|
|
content.update(business_brief_id=business_brief["id"], business_brief_digest=business_brief["digest"],
|
|
business_search_term_index=0)
|
|
if not self.supported_candidate(candidate):
|
|
content.update(errors=["当前候选不符合已支持的装修搜索合同"], error_codes=["DISCOVERY_CONTRACT_INVALID"])
|
|
else:
|
|
try:
|
|
validation = await self.learning.verify(candidate["id"], owner_id, {"mode": "exploration", "parameters": {"keyword": values["keyword"]}, "expected_outputs": {"search_results": {"state": "nonempty"}}, "sample_name": "装修规则初筛：结构化搜索"})
|
|
content.update(validation_id=validation["id"], evidence=validation.get("evidence") or [])
|
|
for key in ("execution_profile", "execution_contract"):
|
|
if key in validation.get("environment", {}):
|
|
content[key] = deepcopy(validation["environment"][key])
|
|
if validation["status"] != "passed":
|
|
content.update(status="search_failed", errors=["结构化搜索未通过验证，请查看关联验证和页面证据"], error_codes=safe_error_codes(validation.get("error_codes")) or ["SEARCH_VALIDATION_FAILED"])
|
|
else:
|
|
output = validation["actual_outputs"]["search_results"]
|
|
if origin(output["source_url"]) != origin(candidate["manifest"]["entry_url"]):
|
|
content.update(status="search_failed", errors=["搜索结果来源与当前候选站点不一致"], error_codes=["DISCOVERY_SOURCE_MISMATCH"])
|
|
else:
|
|
content["results"], content["selected"] = select_results(output["items"], candidate["manifest"]["entry_url"], values["max_selected"])
|
|
content["status"] = "completed"
|
|
except HTTPException as error:
|
|
detail = error.detail if isinstance(error.detail, dict) else {}
|
|
content.update(errors=["搜索未执行完成，请核对学习会话和候选静态校验"], error_codes=safe_error_codes([detail.get("code")]) or ["SEARCH_NOT_READY"])
|
|
run_id, timestamp = identifier("discovery"), now()
|
|
self.store.execute("INSERT INTO discovery_runs(id,owner_id,candidate_id,validation_id,data_json,created_at) VALUES($1,$2,$3,$4,$5,$6)", (run_id, owner_id, candidate["id"], content["validation_id"], encoded(content), timestamp))
|
|
return self.get(run_id, owner_id)
|