"""Single-note handoff from immutable discovery selections to learned reading.""" from __future__ import annotations import json import re 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 .discovery import DiscoveryService, note_reference, origin from .learning_checks import output_assertion_checks from .learning_contracts import ClosedInput from .worker import safe_error_codes class NoteBinding(ClosedInput): note_id: str = Field(pattern=r"^[a-fA-F0-9]{24}$") class ReadingInput(ClosedInput): detail_candidate_id: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_-]+$") note_id: str = Field(pattern=r"^[a-fA-F0-9]{24}$") @model_validator(mode="after") def canonical_id(self): self.note_id = self.note_id.lower() return self def detail_contract(manifest: dict) -> bool: """Business fields and action modes are fixed; observed CSS stays versioned.""" site = origin(manifest.get("entry_url", "")) supported = (manifest.get("platform") == "xiaohongshu" and site == ("https", "www.xiaohongshu.com", 443)) or (manifest.get("platform") == "demo" and site and site[1] in {"127.0.0.1", "localhost", "::1"}) steps = manifest.get("steps") or [] if not supported or len(steps) not in {7, 8}: return False if len(steps) == 8: seek, click = steps[3:5] target = seek.get("until_target") or {} if (not isinstance(target, dict) or seek.get("kind") != "scroll" or seek.get("selector") in {"body", "html"} or seek.get("until_selector") is not None or ("scroll_scope" in seek and seek["scroll_scope"] != "viewport") or target.get("selector") != click.get("selector") or {key: value for key, value in target.items() if key != "selector"} != click.get("target") or target.get("field") != "text" or target.get("parameter") != "note_title" or type(seek.get("timeout_ms")) is not int or not 1 <= seek["timeout_ms"] <= 15000 or type(seek.get("pixels")) is not int or not 1 <= seek["pixels"] <= 1600 or type(seek.get("max_scrolls")) is not int or not 1 <= seek["max_scrolls"] <= 10): return False steps = [*steps[:3], *steps[4:]] if [step.get("kind") for step in steps] != ["fill", "press", "wait", "click", "wait", "extract", "extract"]: return False if set(manifest.get("parameters", {})) != {"keyword", "note_title"} or steps[0].get("parameter") != "keyword" or steps[1].get("key") != "Enter" or steps[0].get("selector") != steps[1].get("selector"): return False if steps[2].get("state") != "visible" or type(steps[2].get("min_count")) is not int or steps[2]["min_count"] < 1 or steps[4].get("state", "visible") != "visible": return False if steps[3].get("target") != {"field": "text", "parameter": "note_title"}: return False detail, comments = steps[5:] fields = {item["name"]: item for item in detail.get("fields", [])} comment_fields = {item["name"]: item for item in comments.get("fields", [])} return bool(detail.get("output_name") == "note_detail" and detail.get("limit") == 1 and set(fields) == {"title", "body", "author"} and all(field.get("source", "text") == "text" and field.get("required") is True for field in fields.values()) and not detail.get("target") and not detail.get("batch") and detail.get("allow_empty") is not True and comments.get("output_name") == "comments" and set(comment_fields) == {"comment_id", "author", "body"} and comment_fields["author"].get("source", "text") == "text" and comment_fields["author"].get("required") is True and comment_fields["body"].get("source", "text") in {"text", "content"} and comment_fields["body"].get("required") is True and comment_fields["comment_id"].get("source") == "attribute" and comment_fields["comment_id"].get("attribute") == "id" and comment_fields["comment_id"].get("required") is True and comments.get("row_selector") and comments.get("complete_only") is True and (comments.get("batch") or {}).get("dedupe_field") == "comment_id" and comments.get("allow_empty") is not True) def audit_note_binding(manifest: dict, binding: dict, proof, outputs: dict, parameters: dict) -> tuple[dict | None, bool]: """Audit the execution proof and actual output sources without trusting titles alone.""" if not isinstance(proof, dict) or set(proof) != {"note_id", "href_checked", "url_checked", "source_url", "title"}: return None, False expected = binding["note_id"] reference = note_reference(proof.get("source_url"), manifest["entry_url"]) raw_url = urlsplit(proof["source_url"]) if reference else None safe_source = proof["source_url"] if reference and raw_url and not raw_url.query and not raw_url.fragment else None title = proof.get("title") title = title if isinstance(title, str) and len(title) <= 2000 else None clean = {"note_id": proof.get("note_id") if isinstance(proof.get("note_id"), str) and re.fullmatch(r"[a-f0-9]{24}", proof["note_id"]) else None, "href_checked": proof.get("href_checked") is True, "url_checked": proof.get("url_checked") is True, "source_url": safe_source, "title": title} detail = outputs.get("note_detail") if isinstance(outputs, dict) else None items = detail.get("items") if isinstance(detail, dict) else None actual_title = items[0].get("title") if isinstance(items, list) and len(items) == 1 and isinstance(items[0], dict) else None passed = (clean["note_id"] == expected and clean["href_checked"] and clean["url_checked"] and reference and reference[0] == expected and safe_source and isinstance(actual_title, str) and bool(actual_title.strip()) and isinstance(parameters.get("note_title"), str) and actual_title.strip() == parameters["note_title"].strip() == (title or "").strip()) for name in ("note_detail", "comments"): value = outputs.get(name) if isinstance(outputs, dict) else None if isinstance(value, dict) and "source_url" in value: output_reference = note_reference(value["source_url"], manifest["entry_url"]) location = urlsplit(value["source_url"]) if output_reference else None passed = passed and bool(output_reference and output_reference[0] == expected and location and not location.query and not location.fragment) elif name == "note_detail": passed = False return clean, bool(passed) class ReadingService: def __init__(self, learning, discovery): self.learning, self.discovery, self.store = learning, discovery, learning.store def get(self, reading_id: str, owner_id: str) -> dict: row = self.store.one("SELECT * FROM reading_runs WHERE id=$1 AND owner_id=$2", (reading_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, run_id: str, owner_id: str) -> list[dict]: self.discovery.get(run_id, owner_id) return [self.get(row["id"], owner_id) for row in self.store.rows(f"SELECT id FROM reading_runs WHERE discovery_run_id=$1 AND owner_id=$2 ORDER BY {self.store.order_column()} DESC", (run_id, owner_id))] async def create(self, run_id: str, owner_id: str, values: dict) -> dict: run = self.discovery.get(run_id, owner_id) selected = next((item for item in run.get("selected", []) if item.get("note_id") == values["note_id"]), None) if run.get("status") != "completed" or not selected: raise HTTPException(422, "只能阅读已完成初筛记录中实际保留的单篇笔记") candidate = self.learning.candidate(values["detail_candidate_id"], owner_id) search = self.learning.candidate(run["search_candidate_id"], owner_id) reference = note_reference(selected.get("note_url"), search["manifest"]["entry_url"]) content = {"status": "failed", "model_connected": False, "discovery_run_id": run_id, "note_id": values["note_id"], "note_title": selected["note_title"], "note_url": reference[1] if reference else None, "keyword": run["keyword"], "detail_candidate_id": candidate["id"], "detail_task_id": candidate["task_id"], "candidate_digest": candidate["digest"], "validation_id": None, "detail": None, "comments": None, "note_binding_result": None, "evidence": [], "errors": [], "error_codes": []} if run.get("business_brief_id"): content.update(business_brief_id=run["business_brief_id"], business_brief_digest=run["business_brief_digest"]) if any(step.get("until_target") for step in candidate["manifest"].get("steps", [])): content["seek_results"] = [] compatible = (detail_contract(candidate["manifest"]) and DiscoveryService.supported_candidate(search) and search["digest"] == run["candidate_digest"] and candidate["manifest"].get("platform") == search["manifest"].get("platform") and origin(candidate["manifest"]["entry_url"]) == origin(search["manifest"]["entry_url"]) and reference and reference[0] == values["note_id"] and isinstance(selected.get("note_title"), str) and 0 < len(selected["note_title"].strip()) <= 2000) if not compatible: content.update(errors=["详情候选或已保留来源不符合当前单篇阅读合同"], error_codes=["READING_CONTRACT_INVALID"]) else: try: validation = await self.learning.verify(candidate["id"], owner_id, {"mode": "exploration", "parameters": {"keyword": run["keyword"], "note_title": selected["note_title"]}, "expected_outputs": {"note_detail": {"state": "nonempty"}, "comments": {"state": "nonempty"}}, "note_binding": {"note_id": values["note_id"]}, "sample_name": "初筛保留笔记：正文与首层评论"}) content.update(validation_id=validation["id"], evidence=validation.get("evidence") or [], note_binding_result=validation.get("note_binding_result")) for key in ("execution_profile", "execution_contract"): if key in validation.get("environment", {}): content[key] = deepcopy(validation["environment"][key]) if "seek_results" in validation: content["seek_results"] = validation["seek_results"] for timing_key in ("step_timings", "verify_timing", "failure_diagnostics"): if timing_key in validation: content[timing_key] = validation[timing_key] bound = any(check.get("name") == "note_source_binding" and check.get("passed") is True for check in validation.get("assertion_checks", [])) outputs = validation.get("actual_outputs") or {} contracts = {step["output_name"]: step for step in candidate["manifest"]["steps"] if step["kind"] == "extract"} if bound: for name, field in (("note_detail", "detail"), ("comments", "comments")): checks = output_assertion_checks(outputs, {name: {"state": "nonempty"}}, contracts) if checks[0]["passed"]: content[field] = outputs[name] detail_ok = bool(content["detail"] and content["detail"]["count"] == 1) comments_ok = bool(content["comments"] and content["comments"]["count"] > 0) if detail_ok: content["status"] = "completed" if validation["status"] == "passed" and comments_ok else "partial" if content["status"] != "completed": content.update(errors=["阅读未完整通过，请核对关联验证和页面证据；已有正文不代表评论或空态已完成"], error_codes=safe_error_codes(validation.get("error_codes")) or ["READING_VALIDATION_FAILED"]) 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 ["READING_NOT_READY"]) reading_id, timestamp = identifier("reading"), now() self.store.execute("INSERT INTO reading_runs(id,owner_id,discovery_run_id,candidate_id,validation_id,data_json,created_at) VALUES($1,$2,$3,$4,$5,$6,$7)", (reading_id, owner_id, run_id, candidate["id"], content["validation_id"], encoded(content), timestamp)) return self.get(reading_id, owner_id)