239 lines
8.2 KiB
Python
239 lines
8.2 KiB
Python
"""Pry — Page Structure Monitor.
|
|
Track CSS selector validity over time. Alert before scrapers break."""
|
|
|
|
# SPDX-License-Identifier: MIT
|
|
# Copyright (c) 2026 Rug Munch Media LLC
|
|
#
|
|
# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
|
|
# Licensed under MIT. See LICENSE.
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from contextlib import suppress
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from lxml import html as lxml_html
|
|
|
|
from client import get_client
|
|
from paths import PRY_DATA_DIR
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
STRUCTURE_DIR = PRY_DATA_DIR / "structure"
|
|
STRUCTURE_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
async def check_selectors(
|
|
url: str,
|
|
selectors: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
"""Check if CSS selectors still match on a page.
|
|
|
|
Args:
|
|
url: The URL to check
|
|
selectors: List of selector dicts with name, selector, and type fields
|
|
|
|
Returns per-selector match status with similarity score.
|
|
"""
|
|
result: dict[str, Any] = {
|
|
"url": url,
|
|
"checked_at": datetime.now(UTC).isoformat(),
|
|
"selectors": [],
|
|
"all_matched": True,
|
|
"matched_count": 0,
|
|
"failed_count": 0,
|
|
}
|
|
|
|
# Fetch the page
|
|
try:
|
|
client = await get_client()
|
|
resp = await client.get(
|
|
url,
|
|
timeout=30,
|
|
follow_redirects=True,
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (StructureMonitor/1.0)"
|
|
},
|
|
)
|
|
if not resp.is_success:
|
|
return {"url": url, "error": f"HTTP {resp.status_code}", "all_matched": False}
|
|
html_content = resp.text
|
|
except (httpx.HTTPError, httpx.RequestError) as e:
|
|
return {"url": url, "error": str(e)[:200], "all_matched": False}
|
|
|
|
try:
|
|
tree = lxml_html.fromstring(html_content)
|
|
except (httpx.HTTPError, httpx.RequestError):
|
|
return {"url": url, "error": "Failed to parse HTML", "all_matched": False}
|
|
|
|
# Check each selector
|
|
for sel in selectors:
|
|
selector_str = sel.get("selector", "")
|
|
name = sel.get("name", selector_str[:30])
|
|
sel_type = sel.get("type", "css")
|
|
|
|
try:
|
|
if sel_type == "css":
|
|
elements = tree.cssselect(selector_str)
|
|
elif sel_type == "xpath":
|
|
elements = tree.xpath(selector_str)
|
|
else:
|
|
elements = []
|
|
|
|
match_count = len(elements)
|
|
is_matched = match_count > 0
|
|
|
|
selector_result: dict[str, Any] = {
|
|
"name": name,
|
|
"selector": selector_str,
|
|
"type": sel_type,
|
|
"matched": is_matched,
|
|
"match_count": match_count,
|
|
"status": "ok" if is_matched else "broken",
|
|
}
|
|
|
|
# If matched, sample content for change detection
|
|
if is_matched and elements:
|
|
sample_text = " ".join(e.text_content().strip()[:100] for e in elements[:3])
|
|
selector_result["sample"] = sample_text[:200]
|
|
|
|
result["selectors"].append(selector_result)
|
|
if is_matched:
|
|
result["matched_count"] += 1
|
|
else:
|
|
result["failed_count"] += 1
|
|
|
|
except (httpx.HTTPError, httpx.RequestError) as e:
|
|
result["selectors"].append(
|
|
{
|
|
"name": name,
|
|
"selector": selector_str,
|
|
"type": sel_type,
|
|
"matched": False,
|
|
"match_count": 0,
|
|
"status": "error",
|
|
"error": str(e)[:100],
|
|
}
|
|
)
|
|
result["failed_count"] += 1
|
|
|
|
result["all_matched"] = result["failed_count"] == 0
|
|
return result
|
|
|
|
|
|
async def monitor_page_structure(
|
|
url: str,
|
|
selectors: list[dict[str, Any]],
|
|
template_id: str = "",
|
|
) -> dict[str, Any]:
|
|
"""Monitor page structure by checking selectors and comparing to history.
|
|
|
|
Returns the current check result plus any detected changes since last check.
|
|
"""
|
|
url_hash = hashlib.sha256(url.encode()).hexdigest()[:16]
|
|
# Use template_id if provided for the filename, else url_hash
|
|
file_key = template_id or url_hash
|
|
history_path = STRUCTURE_DIR / f"structure_{file_key}.json"
|
|
|
|
# Run current check
|
|
current = await check_selectors(url, selectors)
|
|
if "error" in current:
|
|
return current
|
|
|
|
# Load previous state
|
|
previous: dict[str, Any] | None = None
|
|
if history_path.exists():
|
|
with suppress(json.JSONDecodeError, OSError):
|
|
previous = json.loads(history_path.read_text())
|
|
|
|
# Detect changes
|
|
changes = []
|
|
if previous and "selectors" in previous:
|
|
prev_map = {s["name"]: s for s in previous["selectors"]}
|
|
curr_map = {s["name"]: s for s in current["selectors"]}
|
|
|
|
for name, curr_sel in curr_map.items():
|
|
prev_sel = prev_map.get(name)
|
|
if prev_sel:
|
|
# Check status change
|
|
if prev_sel.get("matched") and not curr_sel.get("matched"):
|
|
changes.append(
|
|
{
|
|
"type": "selector_broken",
|
|
"severity": "critical",
|
|
"name": name,
|
|
"selector": curr_sel.get("selector", ""),
|
|
"message": f"CSS selector '{name}' stopped matching (was matching before)",
|
|
}
|
|
)
|
|
elif not prev_sel.get("matched") and curr_sel.get("matched"):
|
|
changes.append(
|
|
{
|
|
"type": "selector_fixed",
|
|
"severity": "info",
|
|
"name": name,
|
|
"message": f"CSS selector '{name}' started matching again",
|
|
}
|
|
)
|
|
# Check count change
|
|
elif curr_sel.get("match_count") != prev_sel.get("match_count"):
|
|
changes.append(
|
|
{
|
|
"type": "count_changed",
|
|
"severity": "medium",
|
|
"name": name,
|
|
"message": f"Selector '{name}' matched {prev_sel.get('match_count')} elements before, now {curr_sel.get('match_count')}",
|
|
"previous_count": prev_sel.get("match_count"),
|
|
"current_count": curr_sel.get("match_count"),
|
|
}
|
|
)
|
|
|
|
# Detect new failures
|
|
if previous and previous.get("all_matched") and not current.get("all_matched"):
|
|
changes.append(
|
|
{
|
|
"type": "structure_broken",
|
|
"severity": "critical",
|
|
"name": "page_structure",
|
|
"message": f"Page structure changed — {current.get('failed_count')} selectors now fail",
|
|
"failed_selectors": [
|
|
s["name"] for s in current.get("selectors", []) if not s.get("matched")
|
|
],
|
|
}
|
|
)
|
|
|
|
# Save current state
|
|
with suppress(OSError):
|
|
history_path.write_text(json.dumps(current, indent=2))
|
|
|
|
current["changes"] = changes
|
|
current["change_count"] = len(changes)
|
|
current["has_changes"] = len(changes) > 0
|
|
current["template_id"] = template_id
|
|
current["url_hash"] = url_hash
|
|
|
|
return current
|
|
|
|
|
|
def get_structure_history(url: str, template_id: str = "") -> dict[str, Any]:
|
|
"""Get structure check history for a URL."""
|
|
url_hash = hashlib.sha256(url.encode()).hexdigest()[:16]
|
|
file_key = template_id or url_hash
|
|
history_path = STRUCTURE_DIR / f"structure_{file_key}.json"
|
|
|
|
if not history_path.exists():
|
|
return {"url": url, "has_history": False}
|
|
|
|
try:
|
|
data = json.loads(history_path.read_text())
|
|
return {
|
|
"url": url,
|
|
"has_history": True,
|
|
"last_checked": data.get("checked_at", ""),
|
|
"last_all_matched": data.get("all_matched", True),
|
|
"last_selectors": data.get("selectors", []),
|
|
}
|
|
except (json.JSONDecodeError, OSError):
|
|
return {"url": url, "has_history": False}
|