171 lines
12 KiB
Python
171 lines
12 KiB
Python
"""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)
|