238 lines
13 KiB
Python
238 lines
13 KiB
Python
"""Small, versioned input boundary shared by recording and frozen checks."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
FIELD_NAME = re.compile(r"[A-Za-z][A-Za-z0-9_]{0,63}")
|
|
PRIVATE_NAME = re.compile(r"password|passwd|credential|token|secret|session|cookie|authorization|captcha|verification|phone|mobile|sms|code|value", re.I)
|
|
|
|
EXECUTION_PROFILE = "playwright-xhs-native-ui-v1"
|
|
EXECUTION_CONTRACT = {"engine": "playwright", "action_policy": "xhs-bounded-readonly-v1", "network_policy": "xhs-page-native-v1"}
|
|
NAVIGATION_PROFILE = "playwright-xhs-navigation-ui-v1"
|
|
NAVIGATION_CONTRACT = {"engine": "playwright", "action_policy": "xhs-observed-navigation-readonly-v1", "network_policy": "xhs-page-navigation-v1"}
|
|
EXECUTION_CONTRACTS = {EXECUTION_PROFILE: EXECUTION_CONTRACT, NAVIGATION_PROFILE: NAVIGATION_CONTRACT}
|
|
|
|
|
|
def execution_metadata(profile) -> dict:
|
|
"""Resolve a fixed contract; absence means legacy without rewriting facts."""
|
|
if profile is None:
|
|
return {}
|
|
if not isinstance(profile, str) or profile not in EXECUTION_CONTRACTS:
|
|
raise ValueError("unsupported execution profile")
|
|
return {"execution_profile": profile, "execution_contract": dict(EXECUTION_CONTRACTS[profile])}
|
|
|
|
|
|
def execution_metadata_matches(value, profile) -> bool:
|
|
if not isinstance(value, dict):
|
|
return False
|
|
try:
|
|
expected = execution_metadata(profile)
|
|
except ValueError:
|
|
return False
|
|
actual = {key: value[key] for key in ("execution_profile", "execution_contract") if key in value}
|
|
return actual == expected
|
|
|
|
|
|
def valid_selector(value: str, *, relative: bool = False) -> bool:
|
|
forbidden = r">>|/|[\x00-\x1f]|(?:^|\s)(?:xpath|text|css|id|role|nth|javascript)\s*=|javascript:|:has-text\s*\(|:text(?:-is|-matches)?\s*\("
|
|
if not value.strip() or len(value) > 1000 or re.search(forbidden, value, re.I):
|
|
return False
|
|
return not relative or not re.search(r"(?:^|[\s,>+~])(?:html|body)(?:$|[\s,.#:[>+~])|:root", value, re.I)
|
|
|
|
|
|
def valid_attribute(value: str | None) -> bool:
|
|
return bool(value and (value in {"id", "title", "role", "datetime", "aria-label"} or (re.fullmatch(r"data-[a-z][a-z0-9_-]{0,63}", value) and not PRIVATE_NAME.search(value))))
|
|
|
|
|
|
class ClosedInput(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
class ExtractField(ClosedInput):
|
|
name: str = Field(min_length=1, max_length=64)
|
|
selector: str | None = Field(default=None, min_length=1, max_length=1000)
|
|
source: Literal["text", "href", "attribute", "content"] = "text"
|
|
attribute: str | None = Field(default=None, max_length=100)
|
|
required: bool = Field(default=False, strict=True)
|
|
text_selector: str | None = Field(default=None, min_length=1, max_length=1000)
|
|
media_selector: str | None = Field(default=None, min_length=1, max_length=1000)
|
|
max_media: int | None = Field(default=None, ge=1, le=4, strict=True)
|
|
|
|
@model_validator(mode="after")
|
|
def safe_field(self):
|
|
if not FIELD_NAME.fullmatch(self.name) or self.name in {"constructor", "prototype", "__proto__"}:
|
|
raise ValueError("invalid business field name")
|
|
if self.selector is not None and not valid_selector(self.selector, relative=True):
|
|
raise ValueError("only relative CSS selectors are supported")
|
|
if self.source == "attribute" and not valid_attribute(self.attribute):
|
|
raise ValueError("unsupported attribute")
|
|
if self.source != "attribute" and self.attribute is not None:
|
|
raise ValueError("attribute requires attribute source")
|
|
content_fields = {"text_selector", "media_selector", "max_media"}
|
|
if self.source == "content":
|
|
if not self.required or not self.text_selector or not self.media_selector or "attribute" in self.model_fields_set:
|
|
raise ValueError("content requires explicit text/media selectors and a required content root")
|
|
if not all(valid_selector(value, relative=True) for value in (self.text_selector, self.media_selector)):
|
|
raise ValueError("content selectors must stay within the ordinary content root")
|
|
if "max_media" in self.model_fields_set and self.max_media is None:
|
|
raise ValueError("max_media must be omitted or explicit")
|
|
if self.max_media is None:
|
|
self.max_media = 4
|
|
elif content_fields & self.model_fields_set:
|
|
raise ValueError("content configuration requires an explicit content source")
|
|
return self
|
|
|
|
|
|
class DynamicTarget(ClosedInput):
|
|
field: Literal["text", "href", "attribute"]
|
|
attribute: str | None = Field(default=None, max_length=100)
|
|
value: str | None = Field(default=None, min_length=1, max_length=2000)
|
|
parameter: str | None = Field(default=None, min_length=1, max_length=64)
|
|
|
|
@model_validator(mode="after")
|
|
def safe_target(self):
|
|
if (self.value is None) == (self.parameter is None):
|
|
raise ValueError("target requires exactly one value or parameter")
|
|
if self.value is not None and not self.value.strip():
|
|
raise ValueError("target value must be nonempty")
|
|
if self.field == "attribute" and not valid_attribute(self.attribute):
|
|
raise ValueError("unsupported target attribute")
|
|
if self.field != "attribute" and self.attribute is not None:
|
|
raise ValueError("attribute requires attribute field")
|
|
return self
|
|
|
|
|
|
class ScrollTarget(DynamicTarget):
|
|
selector: str = Field(min_length=1, max_length=1000)
|
|
|
|
@model_validator(mode="after")
|
|
def relative_target(self):
|
|
if not valid_selector(self.selector, relative=True):
|
|
raise ValueError("scroll target requires a safe descendant CSS selector")
|
|
return self
|
|
|
|
|
|
class ExtractBatch(ClosedInput):
|
|
scroll_selector: str = Field(min_length=1, max_length=1000)
|
|
pixels: int = Field(ge=1, le=1600, strict=True)
|
|
max_batches: int = Field(ge=1, le=5, strict=True)
|
|
dedupe_field: str = Field(min_length=1, max_length=64)
|
|
|
|
@model_validator(mode="after")
|
|
def bounded_batch(self):
|
|
if not valid_selector(self.scroll_selector) or not FIELD_NAME.fullmatch(self.dedupe_field):
|
|
raise ValueError("batch requires observed CSS and a declared identity field")
|
|
return self
|
|
|
|
|
|
class EmptyState(ClosedInput):
|
|
selector: str = Field(min_length=1, max_length=1000)
|
|
text: str = Field(min_length=1, max_length=2000)
|
|
|
|
@model_validator(mode="after")
|
|
def bounded_empty_state(self):
|
|
self.text = self.text.strip()
|
|
if not valid_selector(self.selector, relative=True) or not self.text.strip():
|
|
raise ValueError("empty state requires relative CSS and explicit visible text")
|
|
return self
|
|
|
|
|
|
class EmptyExpectation(ClosedInput):
|
|
state: Literal["empty"]
|
|
|
|
|
|
class NonemptyExpectation(ClosedInput):
|
|
state: Literal["nonempty"]
|
|
|
|
|
|
class Action(ClosedInput):
|
|
kind: Literal["fill", "click", "press", "read", "scroll", "wait", "extract"]
|
|
selector: str = Field(min_length=1, max_length=1000)
|
|
key: Literal["Enter"] | None = None
|
|
value: str | None = Field(default=None, max_length=2000)
|
|
parameter: str | None = Field(default=None, min_length=1, max_length=64)
|
|
output_name: str | None = Field(default=None, max_length=64)
|
|
target: DynamicTarget | None = None
|
|
direction: Literal["down", "up"] | None = None
|
|
pixels: int | None = Field(default=None, ge=1, le=1600, strict=True)
|
|
max_scrolls: int | None = Field(default=None, ge=1, le=10, strict=True)
|
|
until_selector: str | None = Field(default=None, min_length=1, max_length=1000)
|
|
until_target: ScrollTarget | None = None
|
|
scroll_scope: Literal["viewport"] | None = None
|
|
state: Literal["visible", "hidden", "text"] | None = None
|
|
wait_scope: str | None = Field(default=None, min_length=1, max_length=1000)
|
|
text: str | None = Field(default=None, min_length=1, max_length=2000)
|
|
timeout_ms: int | None = Field(default=None, ge=1, le=15000, strict=True)
|
|
min_count: int | None = Field(default=None, ge=1, le=100, strict=True)
|
|
fields: list[ExtractField] | None = Field(default=None, min_length=1, max_length=12)
|
|
limit: int | None = Field(default=None, ge=1, le=100, strict=True)
|
|
complete_only: bool | None = Field(default=None, strict=True)
|
|
row_selector: str | None = Field(default=None, min_length=1, max_length=1000)
|
|
batch: ExtractBatch | None = None
|
|
allow_empty: bool | None = Field(default=None, strict=True)
|
|
empty_state: EmptyState | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def bounded_action(self):
|
|
if not valid_selector(self.selector):
|
|
raise ValueError("only bounded CSS selectors are supported")
|
|
if any(name in self.model_fields_set and getattr(self, name) is None for name in ("until_selector", "until_target", "scroll_scope", "wait_scope", "complete_only", "row_selector", "batch", "allow_empty", "empty_state", "min_count")):
|
|
raise ValueError("optional completion fields must be omitted or have an explicit valid value")
|
|
fields = self.model_dump(exclude_none=True)
|
|
if self.target is not None and self.kind not in {"click", "read", "extract"}:
|
|
raise ValueError("dynamic target only applies to click, read and extract")
|
|
common = {"kind", "selector", "target"}
|
|
allowed = {
|
|
"fill": {"value", "parameter"}, "click": set(), "press": {"key"}, "read": {"output_name"},
|
|
"scroll": {"direction", "pixels", "max_scrolls", "until_selector", "until_target", "scroll_scope", "timeout_ms"},
|
|
"wait": {"state", "text", "timeout_ms", "min_count", "wait_scope"},
|
|
"extract": {"output_name", "fields", "limit", "complete_only", "row_selector", "batch", "allow_empty", "empty_state"},
|
|
}[self.kind] | common
|
|
if set(fields) - allowed:
|
|
raise ValueError("fields do not apply to this action")
|
|
if self.kind == "fill" and (self.value is None) == (self.parameter is None):
|
|
raise ValueError("fill requires exactly one value or parameter")
|
|
if self.until_selector is not None and not valid_selector(self.until_selector, relative=True):
|
|
raise ValueError("until_selector requires a bounded CSS selector within the scroll root")
|
|
if self.until_target is not None:
|
|
if self.until_selector is not None or self.timeout_ms is None:
|
|
raise ValueError("until_target requires an explicit time budget and excludes until_selector")
|
|
if self.selector in {"body", "html"}:
|
|
raise ValueError("dynamic target requires an observed ordinary root")
|
|
elif self.kind == "scroll" and self.timeout_ms is not None:
|
|
raise ValueError("scroll time budget requires an explicit dynamic target")
|
|
if self.scroll_scope is not None and (self.kind != "scroll" or self.until_target is None):
|
|
raise ValueError("viewport scope only applies to bounded dynamic target scrolling")
|
|
if self.wait_scope is not None:
|
|
if self.kind != "wait" or self.state != "visible" or self.min_count is None:
|
|
raise ValueError("scoped wait requires explicit visible state and collection count")
|
|
if not valid_selector(self.wait_scope, relative=True) or not valid_selector(self.selector, relative=True):
|
|
raise ValueError("scoped wait requires ordinary root and descendant CSS selectors")
|
|
if self.kind == "press" and self.key != "Enter":
|
|
raise ValueError("press supports only an explicit search Enter")
|
|
if self.kind in {"read", "extract"} and not FIELD_NAME.fullmatch(self.output_name or ""):
|
|
raise ValueError("read and extract require an output name")
|
|
if self.kind == "extract":
|
|
if not self.fields or len({item.name for item in self.fields}) != len(self.fields):
|
|
raise ValueError("extract fields are required and must be unique")
|
|
if self.row_selector is not None and (not valid_selector(self.row_selector, relative=True) or self.target is not None):
|
|
raise ValueError("scoped rows require relative CSS and cannot use a dynamic target")
|
|
if self.batch is not None:
|
|
identity = next((item for item in self.fields if item.name == self.batch.dedupe_field), None)
|
|
if not self.row_selector or self.complete_only is not True or self.allow_empty is True or not identity or identity.source != "attribute" or not identity.required or not (identity.attribute == "id" or (identity.attribute or "").startswith("data-")):
|
|
raise ValueError("batch requires complete scoped rows and a required safe identity attribute")
|
|
if self.allow_empty is True and (not self.row_selector or not self.empty_state):
|
|
raise ValueError("empty extraction requires scoped rows and explicit empty-state evidence")
|
|
if self.empty_state is not None and self.allow_empty is not True:
|
|
raise ValueError("empty-state evidence requires allow_empty")
|
|
if self.kind == "wait" and (self.state == "text") != (self.text is not None):
|
|
raise ValueError("text wait requires exactly one text expectation")
|
|
if self.min_count is not None and self.state not in {None, "visible"}:
|
|
raise ValueError("min_count applies only to visible waits")
|
|
if self.text is not None and not self.text.strip():
|
|
raise ValueError("text expectation must be nonempty")
|
|
return self
|