254 lines
16 KiB
Python
254 lines
16 KiB
Python
"""Resolve and replay a small learned read route without business classification.
|
|
|
|
The route itself is an exploration validation, not a published capability. Each
|
|
stage still executes the immutable candidate through LearningService.verify.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections import Counter
|
|
from copy import deepcopy
|
|
from typing import Literal
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi import HTTPException
|
|
from pydantic import Field, model_validator
|
|
|
|
from .discovery import DiscoveryService, note_reference, origin, title_key
|
|
from .learning_checks import output_assertion_checks
|
|
from .learning_contracts import ClosedInput
|
|
from .reading import audit_note_binding, detail_contract
|
|
from .worker import safe_error_codes
|
|
|
|
|
|
class RouteResolveInput(ClosedInput):
|
|
platform: Literal["demo", "xiaohongshu"]
|
|
task_type: Literal["search_note_comments"]
|
|
search_candidate_id: str | None = Field(default=None, min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_-]+$")
|
|
detail_candidate_id: str | None = Field(default=None, min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_-]+$")
|
|
|
|
|
|
class RouteVerifyInput(RouteResolveInput):
|
|
search_candidate_id: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_-]+$")
|
|
detail_candidate_id: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_-]+$")
|
|
search_digest: str = Field(pattern=r"^[a-f0-9]{64}$")
|
|
detail_digest: str = Field(pattern=r"^[a-f0-9]{64}$")
|
|
keyword: str = Field(min_length=1, max_length=2000)
|
|
note_id: str | None = Field(default=None, pattern=r"^[a-fA-F0-9]{24}$")
|
|
|
|
@model_validator(mode="after")
|
|
def ordinary_parameters(self):
|
|
self.keyword = self.keyword.strip()
|
|
if not self.keyword:
|
|
raise ValueError("keyword cannot be blank")
|
|
if self.note_id is not None:
|
|
self.note_id = self.note_id.lower()
|
|
return self
|
|
|
|
|
|
def route_error(code: str, message: str) -> HTTPException:
|
|
return HTTPException(409, {"code": code, "message": message})
|
|
|
|
|
|
def select_note(items: list[dict], entry_url: str, requested_id: str | None) -> dict | None:
|
|
"""Choose in actual output order; titles and IDs must each be unique.
|
|
|
|
No industry keywords, author-role assumptions or demand scores participate.
|
|
Invalid and repeated DOM cards are deliberately not merged into click targets.
|
|
"""
|
|
titles = Counter(title_key(item["title"]) for item in items
|
|
if isinstance(item, dict) and isinstance(item.get("title"), str) and item["title"].strip())
|
|
references = [note_reference(item.get("url"), entry_url) if isinstance(item, dict) else None for item in items]
|
|
identities = Counter(reference[0] for reference in references if reference)
|
|
for index, (item, reference) in enumerate(zip(items, references), 1):
|
|
title = item.get("title") if isinstance(item, dict) else None
|
|
if (not reference or identities[reference[0]] != 1
|
|
or not isinstance(title, str) or not 0 < len(title.strip()) <= 2000
|
|
or "[已隐藏]" in title or "[手机号已隐藏]" in title
|
|
or titles[title_key(title)] != 1 or (requested_id is not None and reference[0] != requested_id)):
|
|
continue
|
|
return {"index": index, "note_id": reference[0], "note_title": title.strip(), "note_url": reference[1]}
|
|
return None
|
|
|
|
|
|
class RouteService:
|
|
def __init__(self, learning):
|
|
self.learning, self.store = learning, learning.store
|
|
self.running = learning.route_tasks
|
|
|
|
def catalogue(self, owner_id: str) -> list[dict]:
|
|
items = []
|
|
for candidate in self.learning.candidates(owner_id):
|
|
manifest = candidate["manifest"]
|
|
if DiscoveryService.supported_candidate(candidate):
|
|
ability = "search_notes"
|
|
elif detail_contract(manifest):
|
|
ability = "read_note_comments"
|
|
else:
|
|
continue
|
|
rows = self.learning.validations(owner_id, candidate_id=candidate["id"])
|
|
static = next((row for row in rows if row["kind"] == "static"), None)
|
|
trial = next((row for row in rows if row["kind"] == "browser"), None)
|
|
parsed = urlsplit(manifest["entry_url"])
|
|
items.append({"candidate_id": candidate["id"], "task_id": candidate["task_id"], "name": candidate["name"],
|
|
"version": candidate["version"], "digest": candidate["digest"], "platform": manifest["platform"],
|
|
"execution_profile": manifest.get("execution_profile"), "ability": ability,
|
|
"source_origin": f"{parsed.scheme}://{parsed.netloc}",
|
|
"static_status": static["status"] if static else "not_run", "integrity_rechecked": False,
|
|
"static_errors": deepcopy(static.get("errors") or []) if static else [], "static_validation_id": static["id"] if static else None,
|
|
"trial_status": trial["status"] if trial else "not_run", "trial_validation_id": trial["id"] if trial else None,
|
|
"maintenance_status": candidate["quality"]["state"], "published": False})
|
|
return items
|
|
|
|
@staticmethod
|
|
def compatible(search: dict, detail: dict) -> bool:
|
|
left, right = search["manifest"], detail["manifest"]
|
|
return (search["task_id"] == detail["task_id"] and left.get("platform") == right.get("platform")
|
|
and left.get("execution_profile") == right.get("execution_profile")
|
|
and left.get("execution_contract") == right.get("execution_contract")
|
|
and origin(left["entry_url"]) == origin(right["entry_url"]))
|
|
|
|
def resolve(self, owner_id: str, values: dict) -> dict:
|
|
# Resolve explicit references first, preserving the normal owner-scoped
|
|
# 404 contract instead of revealing whether somebody else's ID exists.
|
|
explicit = {}
|
|
for key in ("search_candidate_id", "detail_candidate_id"):
|
|
if values.get(key):
|
|
explicit[key] = self.learning.candidate(values[key], owner_id)
|
|
# Explicit replay must not hash every historical candidate's screenshots.
|
|
# For automatic resolution, structural/session compatibility is cheap and
|
|
# narrows the set before checking the actual evidence integrity.
|
|
candidates = list(explicit.values()) if len(explicit) == 2 else self.learning.candidates(owner_id)
|
|
searches = [item for item in candidates if item["manifest"].get("platform") == values["platform"]
|
|
and DiscoveryService.supported_candidate(item)
|
|
and (not values.get("search_candidate_id") or item["id"] == values["search_candidate_id"])]
|
|
details = [item for item in candidates if item["manifest"].get("platform") == values["platform"]
|
|
and detail_contract(item["manifest"])
|
|
and (not values.get("detail_candidate_id") or item["id"] == values["detail_candidate_id"])]
|
|
structural_pairs = [(search, detail) for search in searches for detail in details if self.compatible(search, detail)]
|
|
eligible = {}
|
|
for candidate in {candidate["id"]: candidate for pair in structural_pairs for candidate in pair}.values():
|
|
static = self.store.one(f"SELECT status FROM validations WHERE candidate_id=$1 AND kind='static' ORDER BY {self.store.order_column()} DESC LIMIT 1", (candidate["id"],))
|
|
eligible[candidate["id"]] = bool(static and static["status"] == "passed" and not self.learning.static_errors(candidate))
|
|
pairs = []
|
|
for search, detail in structural_pairs:
|
|
if eligible[search["id"]] and eligible[detail["id"]]:
|
|
pairs.append((search, detail))
|
|
if not pairs:
|
|
explicit = values.get("search_candidate_id") and values.get("detail_candidate_id")
|
|
raise route_error("ROUTE_PAIR_INCOMPATIBLE" if explicit else "ROUTE_NOT_AVAILABLE",
|
|
"没有同任务、同站点、同执行模式且当前固定校验通过的搜索与详情链路")
|
|
if len(pairs) != 1:
|
|
raise route_error("ROUTE_AMBIGUOUS", "有多组可用链路，请显式选择搜索和详情候选版本")
|
|
search, detail = pairs[0]
|
|
def reference(candidate):
|
|
return {"candidate_id": candidate["id"], "task_id": candidate["task_id"], "digest": candidate["digest"],
|
|
"name": candidate["name"], "version": candidate["version"]}
|
|
return {"route_version": "learned-read-route-v1", "platform": values["platform"], "task_type": values["task_type"],
|
|
"mode": "exploration", "task_id": search["task_id"], "execution_profile": search["manifest"].get("execution_profile"),
|
|
"search": reference(search), "detail": reference(detail), "model_calls": 0, "published": False}
|
|
|
|
async def verify(self, owner_id: str, values: dict) -> dict:
|
|
route = self.resolve(owner_id, values)
|
|
if route["search"]["digest"] != values["search_digest"] or route["detail"]["digest"] != values["detail_digest"]:
|
|
raise route_error("ROUTE_DIGEST_MISMATCH", "候选摘要与已选择链路不同，请重新召回确认版本")
|
|
task_id = route["task_id"]
|
|
if task_id in self.running:
|
|
raise route_error("ROUTE_BUSY", "当前学习任务正在验证链路，请等待本次结束")
|
|
self.running.add(task_id)
|
|
token = self.learning.route_task.set(task_id)
|
|
try:
|
|
return await self._verify(owner_id, values, route)
|
|
finally:
|
|
self.learning.route_task.reset(token)
|
|
self.running.discard(task_id)
|
|
|
|
async def _verify(self, owner_id: str, values: dict, route: dict) -> dict:
|
|
started = time.monotonic()
|
|
search = self.learning.candidate(route["search"]["candidate_id"], owner_id)
|
|
detail = self.learning.candidate(route["detail"]["candidate_id"], owner_id)
|
|
content = {"route": route, "route_status": "failed", "keyword": values["keyword"], "selected": None,
|
|
"search_validation_id": None, "detail_validation_id": None, "actual_outputs": {}, "usable_outputs": {},
|
|
"error_codes": [], "evidence": [], "model_calls": 0, "model_usage_status": "not_used", "published": False,
|
|
"mode": "exploration", "sample_name": "通用只读链路召回与变参验证", "assertion_checks": []}
|
|
errors: list[str] = []
|
|
|
|
def fail(message: str, codes: list[str]):
|
|
errors.append(message)
|
|
content["error_codes"] = list(dict.fromkeys([*content["error_codes"], *safe_error_codes(codes)]))
|
|
|
|
def preserve(stage: str, validation: dict):
|
|
content[f"{stage}_validation_id"] = validation["id"]
|
|
content["actual_outputs"].update(deepcopy(validation.get("actual_outputs") or {}))
|
|
content["evidence"].extend(deepcopy(validation.get("evidence") or []))
|
|
|
|
try:
|
|
validation = await self.learning.verify(search["id"], owner_id, {"mode": "exploration",
|
|
"parameters": {"keyword": values["keyword"]}, "expected_outputs": {"search_results": {"state": "nonempty"}},
|
|
"sample_name": "通用只读链路：搜索"})
|
|
except HTTPException:
|
|
fail("搜索未执行完成，请核对已确认会话和候选校验", ["SEARCH_NOT_READY"])
|
|
validation = None
|
|
if validation:
|
|
preserve("search", validation)
|
|
contracts = {step["output_name"]: step for step in search["manifest"]["steps"] if step.get("kind") == "extract"}
|
|
checks = output_assertion_checks(content["actual_outputs"], {"search_results": {"state": "nonempty"}}, contracts)
|
|
content["assertion_checks"].extend(checks)
|
|
output = content["actual_outputs"].get("search_results")
|
|
source_ok = isinstance(output, dict) and origin(output.get("source_url")) == origin(search["manifest"]["entry_url"])
|
|
content["assertion_checks"].append({"name": "search_source_binding", "passed": bool(source_ok), "message": "核对搜索输出同站来源"})
|
|
if validation["status"] != "passed" or not all(check["passed"] for check in checks):
|
|
fail("结构化搜索未通过，请查看搜索阶段验证", safe_error_codes(validation.get("error_codes")) or ["SEARCH_VALIDATION_FAILED"])
|
|
elif not source_ok:
|
|
fail("搜索输出来源不属于当前链路站点", ["ROUTE_SOURCE_MISMATCH"])
|
|
else:
|
|
content["usable_outputs"]["search_results"] = deepcopy(output)
|
|
content["selected"] = select_note(output["items"], search["manifest"]["entry_url"], values.get("note_id"))
|
|
if content["selected"] is None:
|
|
fail("本次搜索没有标题和笔记编号均唯一的安全定位项", ["ROUTE_NOTE_NOT_SELECTABLE"])
|
|
|
|
if content["selected"] is not None and not errors:
|
|
selected = content["selected"]
|
|
parameters = {"keyword": values["keyword"], "note_title": selected["note_title"]}
|
|
binding = {"note_id": selected["note_id"]}
|
|
try:
|
|
validation = await self.learning.verify(detail["id"], owner_id, {"mode": "exploration", "parameters": parameters,
|
|
"expected_outputs": {"note_detail": {"state": "nonempty"}, "comments": {"state": "nonempty"}},
|
|
"note_binding": binding, "sample_name": "通用只读链路：详情与首层评论"})
|
|
except HTTPException:
|
|
fail("详情未执行完成，请核对已确认会话和候选校验", ["DETAIL_NOT_READY"])
|
|
validation = None
|
|
if validation:
|
|
preserve("detail", validation)
|
|
outputs = validation.get("actual_outputs") or {}
|
|
proof, bound = audit_note_binding(detail["manifest"], binding, validation.get("note_binding_result"), outputs, parameters)
|
|
bound = bound and any(check.get("name") == "note_source_binding" and check.get("passed") is True
|
|
for check in validation.get("assertion_checks", []))
|
|
content["note_binding_result"] = proof
|
|
content["assertion_checks"].append({"name": "note_source_binding", "passed": bool(bound), "message": "核对实际笔记编号、标题与输出来源"})
|
|
contracts = {step["output_name"]: step for step in detail["manifest"]["steps"] if step.get("kind") == "extract"}
|
|
note_checks = output_assertion_checks(outputs, {"note_detail": {"state": "nonempty"}}, contracts)
|
|
comment_checks = output_assertion_checks(outputs, {"comments": {"state": "nonempty"}}, contracts)
|
|
content["assertion_checks"].extend([*note_checks, *comment_checks])
|
|
detail_ok = bound and all(check["passed"] for check in note_checks) and outputs["note_detail"]["count"] == 1
|
|
comment_value = outputs.get("comments")
|
|
comments_ok = (all(check["passed"] for check in comment_checks) and isinstance(comment_value, dict)
|
|
and bool(note_reference(comment_value.get("source_url"), detail["manifest"]["entry_url"]))
|
|
and note_reference(comment_value["source_url"], detail["manifest"]["entry_url"])[0] == binding["note_id"])
|
|
if detail_ok:
|
|
content["usable_outputs"]["note_detail"] = deepcopy(outputs["note_detail"])
|
|
if bound and comments_ok:
|
|
content["usable_outputs"]["comments"] = deepcopy(comment_value)
|
|
if detail_ok:
|
|
content["route_status"] = "completed" if validation["status"] == "passed" and comments_ok else "partial"
|
|
if content["route_status"] != "completed":
|
|
fail("详情与评论未完整通过，已得输出及阶段失败记录保留", safe_error_codes(validation.get("error_codes")) or ["DETAIL_VALIDATION_FAILED"])
|
|
content["duration_ms"] = round((time.monotonic() - started) * 1000)
|
|
content["parameters"] = {"keyword": values["keyword"]}
|
|
# A route summary is stored under its search candidate, but never counts
|
|
# as a browser run, maintenance quality or release qualification.
|
|
record = self.learning.add_validation(search, "route", errors, **content)
|
|
record["quality_observations"] = []
|
|
record["quality"] = self.learning.validation_quality(record)
|
|
return record
|