"""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