RugMunchMedia/pryscraper
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(x402): extract constants/enums to pry_x402 sub-package #12

Closed
cryptorugmunch wants to merge 16 commits from refactor/x402-split into main
pull from: refactor/x402-split
merge into: RugMunchMedia:main
Conversation 0 Commits 16 Files changed 64 +538 -587
64 changed files with 538 additions and 587 deletions
Download patch file Download diff file Expand all files Collapse all files

156
.forgejo/workflows/ci.yml
View file

@ -10,147 +10,135 @@ on:
pull_request:
branches: [main]
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true
env:
UV_BIN: /root/.local/bin/uv
PYTHON_VERSION: "3.12"
jobs:
lint:
runs-on: docker-x64
container:
image: python:3.11-slim
image: python:3.12-slim
steps:
- name: Install Node.js for actions/checkout
run: |
apt-get update
apt-get install -y --no-install-recommends curl ca-certificates
apt-get install -y --no-install-recommends curl ca-certificates git
curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
apt-get install -y --no-install-recommends nodejs
rm -rf /var/lib/apt/lists/*
- uses: actions/checkout@v4
- name: Install system deps
run: |
apt-get update
apt-get install -y --no-install-recommends gcc libpq-dev curl git ca-certificates
rm -rf /var/lib/apt/lists/*
- name: Install uv
run: |
curl -LsSf https://astral.sh/uv/install.sh | sh
ln -sf /root/.local/bin/uv /usr/local/bin/uv
ln -sf /root/.local/bin/uvx /usr/local/bin/uvx
- name: Setup Python
run: |
uv python install 3.11
uv venv --python 3.11 .venv
- name: Install dependencies
run: |
uv pip install --python .venv/bin/python -r requirements.txt
uv pip install --python .venv/bin/python -e ".[dev]"
- name: Lint (ruff)
run: source .venv/bin/activate && ruff check . --exit-non-zero-on-fix
- name: Format check (ruff)
run: source .venv/bin/activate && ruff format --check .
- name: Install ruff
run: pip install ruff
- name: Lint
run: ruff check .
- name: Format check
run: ruff format --check .
typecheck:
runs-on: docker-x64
container:
image: python:3.11-slim
image: python:3.12-slim
steps:
- name: Install Node.js for actions/checkout
run: |
apt-get update && apt-get install -y --no-install-recommends curl ca-certificates
apt-get update
apt-get install -y --no-install-recommends curl ca-certificates git
curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
apt-get install -y --no-install-recommends nodejs
rm -rf /var/lib/apt/lists/*
- uses: actions/checkout@v4
- name: Install uv + setup
- name: Install dependencies
run: |
apt-get update && apt-get install -y --no-install-recommends curl ca-certificates && rm -rf /var/lib/apt/lists/*
curl -LsSf https://astral.sh/uv/install.sh | sh
ln -sf /root/.local/bin/uv /usr/local/bin/uv
ln -sf /root/.local/bin/uvx /usr/local/bin/uvx
uv python install 3.11
uv venv --python 3.11 .venv
uv pip install --python .venv/bin/python -e ".[dev]"
- name: Type check (mypy)
run: source .venv/bin/activate && mypy .
apt-get update
apt-get install -y --no-install-recommends gcc libpq-dev
rm -rf /var/lib/apt/lists/*
pip install -e ".[dev]"
- name: Type check
run: mypy --python-version 3.12 --strict --ignore-missing-imports --exclude 'build/|dist/|.git/|__pycache__/' *.py
test:
runs-on: docker-x64
container:
image: python:3.11-slim
image: python:3.12-slim
steps:
- name: Install Node.js for actions/checkout
run: |
apt-get update && apt-get install -y --no-install-recommends curl ca-certificates
apt-get update
apt-get install -y --no-install-recommends curl ca-certificates git
curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
apt-get install -y --no-install-recommends nodejs
rm -rf /var/lib/apt/lists/*
- uses: actions/checkout@v4
- name: Install uv + setup
- name: Install dependencies
run: |
apt-get update && apt-get install -y --no-install-recommends gcc libpq-dev curl ca-certificates && rm -rf /var/lib/apt/lists/*
curl -LsSf https://astral.sh/uv/install.sh | sh
ln -sf /root/.local/bin/uv /usr/local/bin/uv
ln -sf /root/.local/bin/uvx /usr/local/bin/uvx
uv python install 3.11
uv venv --python 3.11 .venv
uv pip install --python .venv/bin/python -e ".[dev]"
- name: Test (pytest)
run: source .venv/bin/activate && pytest tests/ -x -q --tb=short -m "not integration"
apt-get update
apt-get install -y --no-install-recommends gcc libpq-dev
rm -rf /var/lib/apt/lists/*
pip install -e ".[dev]"
- name: Test
run: pytest tests/ -v --cov=. --cov-report=term-missing --cov-fail-under=40
secrets:
name: Secret scan (gitleaks)
security:
runs-on: docker-x64
container:
image: python:3.11-slim
image: python:3.12-slim
steps:
- name: Install Node.js for actions/checkout
run: |
apt-get update && apt-get install -y --no-install-recommends curl ca-certificates
apt-get update
apt-get install -y --no-install-recommends curl ca-certificates git
curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
apt-get install -y --no-install-recommends nodejs
rm -rf /var/lib/apt/lists/*
- uses: actions/checkout@v4
- name: Install bandit
run: pip install bandit
- name: Bandit
run: bandit -r . -x tests/,.venv,__pycache__
gitleaks:
runs-on: docker-x64
container:
image: python:3.12-slim
steps:
- name: Install Node.js for actions/checkout
run: |
apt-get update
apt-get install -y --no-install-recommends curl ca-certificates git
curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
apt-get install -y --no-install-recommends nodejs
rm -rf /var/lib/apt/lists/*
- uses: actions/checkout@v4
with:
fetch-depth: 0
- name: Run gitleaks
- name: Install gitleaks
run: |
apt-get update && apt-get install -y --no-install-recommends wget git ca-certificates && rm -rf /var/lib/apt/lists/*
apt-get update
apt-get install -y --no-install-recommends wget git ca-certificates
rm -rf /var/lib/apt/lists/*
wget -qO- https://github.com/gitleaks/gitleaks/releases/download/v8.24.0/gitleaks_8.24.0_linux_x64.tar.gz | tar -xz -C /usr/local/bin gitleaks
gitleaks detect --source . --redact --no-banner --no-git
- name: Run gitleaks
run: gitleaks detect --source . --redact --no-banner --no-git
security:
name: Security audit (bandit)
commitlint:
if: github.event_name == 'pull_request'
runs-on: docker-x64
container:
image: python:3.11-slim
image: node:22-slim
steps:
- name: Install Node.js for actions/checkout
- name: Install git and ca-certificates
run: |
apt-get update && apt-get install -y --no-install-recommends curl ca-certificates
curl -fsSL https://deb.nodesource.com/setup_22.x | bash -
apt-get install -y --no-install-recommends nodejs
apt-get update
apt-get install -y --no-install-recommends git ca-certificates
rm -rf /var/lib/apt/lists/*
- uses: actions/checkout@v4
- name: Install uv + setup
run: |
apt-get update && apt-get install -y --no-install-recommends curl ca-certificates && rm -rf /var/lib/apt/lists/*
curl -LsSf https://astral.sh/uv/install.sh | sh
ln -sf /root/.local/bin/uv /usr/local/bin/uv
ln -sf /root/.local/bin/uvx /usr/local/bin/uvx
uv python install 3.11
uv venv --python 3.11 /tmp/bandit-venv
uv pip install --python /tmp/bandit-venv/bin/python bandit
- name: Bandit (high+medium severity)
run: source /tmp/bandit-venv/bin/activate && bandit -r . -x tests/,.venv,__pycache__ -ll
with:
fetch-depth: 0
- name: Install commitlint
run: npm install -g @commitlint/cli @commitlint/config-conventional
- name: Run commitlint
run: commitlint --from ${{ github.event.pull_request.base.sha }} --to ${{ github.event.pull_request.head.sha }} --verbose

2
.github/workflows/ci.yml vendored
View file

@ -2,6 +2,8 @@
# Copyright (c) 2026 Rug Munch Media LLC
# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
# Licensed under MIT. See LICENSE.
#
# Force workflow re-parse: bump cache key 2026-07-03-a
name: CI
on:

333
api.py
View file

@ -388,51 +388,21 @@ class PryHttpMiddleware:
app.add_middleware(PryHttpMiddleware)
# ── Stats ──
# ── Costing ──
# ── Freshness ──
# ── Parse (Documents) ──
# ── Automate (Browser) ──
# ── Vision — analyze an image via free OpenRouter vision models ───────────────
# Free models default to google/gemma-4-31b-it:free. Accepts:
# - image (base64 data URI or raw base64)
@ -445,113 +415,39 @@ app.add_middleware(PryHttpMiddleware)
# Auto-fallback: if primary model 429s, tries other free models in order.
# ── Sessions ──────────────────────────────────────────────────────────────────
# ── Email Inbox Scraping ──
# ── Integrations ──
# ── Alerts ──
# ── Jobs ──
# ── Config ──
# ── Win 3: WebSocket streaming ──
# ── Win 5: Batch from file + template ──
# ── Win 6: Browser recorder ──
# ── Win 8: Multi-format transform ──
# ── Win 1: Pryfile execution ──
# ── Win 10: Health dashboard HTML ──
@ -561,120 +457,42 @@ app.add_middleware(PryHttpMiddleware)
# ── Win 2: Circuit Breaker ──
# ── Win 3: Stable Extraction ──
# ── Pipeline hooks ──
# ── Win 4: Data Pipeline ──
# ── Win 5: Shareable Results ──
# ── Compliance ──
# ── GDPR Compliance Portal ──
# ── AI Training Data Pipeline ──
# ── Monitoring (cron-based content change detection with AI judging) ──
# ── Structure Monitor ──
# ── Intelligence (Competitive Intelligence Engine) ──
# ── Quality ──
# ── Reconciliation ──
# ── Auth ──
# (split into routers/auth.py on the api-router-split refactor)
@ -733,201 +551,50 @@ app.include_router(webhooks_router)
app.include_router(x402_router)
# ── Review ──
# ── Commerce Sync ──
# ── CRM Sync ──
# ── Pipeline Builder Endpoints ──
# ── Agency ──
# ── SEO Content Change Monitor ──
# ── Reports ──
# ── Enrichment ──
# ── AI Agent Integration ──
# ── Proxy Provider & Affiliate Signup Flow ──
# ── Advanced Scraping (TLS, GraphQL, Schema.org, WebSocket) ──
# ── Advanced: Cookie Warming, PDF, OCR, Dedup, Behavior ──
# ── x402 payment processing ──
# ── Actor marketplace ──
# ── Webhook delivery ──
if __name__ == "__main__":
uvicorn.run(app, host=settings.host, port=settings.port)

1
pry_mcp/fallback_server.py
View file

@ -31,6 +31,7 @@ from pry_mcp.notifications import (
logger = logging.getLogger(__name__)
def make_fallback_server(
base_url: str = DEFAULT_BASE_URL,
tool_executor: Callable[[str, dict[str, Any]], Coroutine[Any, Any, dict[str, Any]]]

1
pry_mcp/sdk_server.py
View file

@ -59,6 +59,7 @@ try:
except ImportError:
_has_mcp_sdk = False
def create_server() -> Any:
"""Create the MCP server (uses official SDK if available, fallback otherwise)."""
if _has_mcp_sdk:

44
pry_x402/__init__.py Normal file
View file

@ -0,0 +1,44 @@
"""Pry — x402 payment protocol sub-package.
Split from x402.py for maintainability.
"""
# 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.
from __future__ import annotations
from pry_x402.constants import (
ASSET_DECIMALS,
EIP7702_RPC,
X402_DEFAULT_ASSET,
X402_DEFAULT_NETWORK,
X402_DIR,
X402_FACILITATOR_URL,
X402_OFFLINE_MODE,
X402_PRICING,
X402_TIMEOUT_SECONDS,
X402_WALLET,
X402Asset,
X402Network,
X402Scheme,
)
__all__ = [
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
"X402Asset",
"X402Network",
"X402Scheme",
]

141
pry_x402/constants.py Normal file
View file

@ -0,0 +1,141 @@
"""Pry — x402 constants and enums.
Split from x402.py for maintainability and to avoid import cycles.
"""
# 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.
from __future__ import annotations
import os
from enum import StrEnum
from typing import Any
from paths import PRY_DATA_DIR
X402_DIR = PRY_DATA_DIR / "x402"
X402_DIR.mkdir(parents=True, exist_ok=True)
# x402 pricing per operation (USDC atomic units, 6 decimals)
X402_PRICING: dict[str, dict[str, Any]] = {
"scrape": {"price_usd": 0.001, "description": "Single URL scrape"},
"crawl": {"price_usd": 0.01, "description": "Crawl up to 10 pages"},
"extract": {"price_usd": 0.005, "description": "Structured extraction"},
"monitor": {"price_usd": 0.02, "description": "Create scheduled monitor"},
"llm_call": {"price_usd": 0.01, "description": "LLM extraction call"},
"template_execute": {"price_usd": 0.002, "description": "Execute scraper template"},
"template_batch_execute": {"price_usd": 0.01, "description": "Execute up to 20 template items"},
"browser_automation": {"price_usd": 0.05, "description": "Browser automation"},
"bulk_crawl": {"price_usd": 0.10, "description": "Crawl up to 1000 pages"},
"pdf_extract": {"price_usd": 0.01, "description": "PDF table extraction"},
"ocr_extract": {"price_usd": 0.005, "description": "Image OCR"},
"graphql_query": {"price_usd": 0.003, "description": "GraphQL query execution"},
"schema_extract": {"price_usd": 0.002, "description": "Schema.org/JSON-LD extraction"},
}
class X402Scheme(StrEnum):
"""Payment schemes supported by x402."""
EXACT = "exact"
UPTO = "upto"
class X402Network(StrEnum):
"""Blockchain networks supported by x402."""
BASE = "base"
SOLANA = "solana"
ETHEREUM = "ethereum"
POLYGON = "polygon"
ARBITRUM = "arbitrum"
OPTIMISM = "optimism"
BNB = "bnb"
AVALANCHE = "avalanche"
BASE_SEPOLIA = "base-sepolia"
TRON = "tron"
BITCOIN = "bitcoin"
LITECOIN = "litecoin"
DOGECOIN = "dogecoin"
class X402Asset(StrEnum):
"""Token assets supported by x402."""
USDC = "USDC"
USDT = "USDT"
DAI = "DAI"
ETH = "ETH"
SOL = "SOL"
MATIC = "MATIC"
TRX = "TRX"
BTC = "BTC"
LTC = "LTC"
DOGE = "DOGE"
# Prefer gopass for the x402 wallet and facilitator URL. Env vars override only
# if the backend returns an empty string. This lets you:
# gopass insert -m pry/x402_wallet # persistent across restarts
# PRY_X402_WALLET=0x... # one-off override (CI, dev)
try:
from secrets_backend import get_secret as _gs
X402_WALLET = _gs("x402_wallet") or os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = _gs("x402_facilitator") or os.getenv(
"PRY_X402_FACILITATOR", "https://x402.org/facilitator"
)
except ImportError:
X402_WALLET = os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = os.getenv("PRY_X402_FACILITATOR", "https://x402.org/facilitator")
X402_DEFAULT_NETWORK = os.getenv("PRY_X402_NETWORK", X402Network.BASE.value)
X402_DEFAULT_ASSET = os.getenv("PRY_X402_ASSET", X402Asset.USDC.value)
X402_OFFLINE_MODE = os.getenv("PRY_X402_OFFLINE", "false").lower() in ("1", "true", "yes")
X402_TIMEOUT_SECONDS = float(os.getenv("PRY_X402_TIMEOUT", "15"))
# EIP-7702 self-verify RPC endpoints (public defaults, override via env)
EIP7702_RPC: dict[str, str] = {
"base": os.getenv("PRY_X402_BASE_RPC", "https://mainnet.base.org"),
"ethereum": os.getenv("PRY_X402_ETH_RPC", "https://eth.llamarpc.com"),
"polygon": os.getenv("PRY_X402_POLYGON_RPC", "https://polygon.drpc.org"),
"arbitrum": os.getenv("PRY_X402_ARBITRUM_RPC", "https://arb1.arbitrum.io/rpc"),
"optimism": os.getenv("PRY_X402_OPTIMISM_RPC", "https://mainnet.optimism.io"),
"bnb": os.getenv("PRY_X402_BNB_RPC", "https://bsc-dataseed.binance.org"),
"avalanche": os.getenv("PRY_X402_AVALANCHE_RPC", "https://api.avax.network/ext/bc/C/rpc"),
"base-sepolia": os.getenv("PRY_X402_BASE_SEPOLIA_RPC", "https://sepolia.base.org"),
}
# Asset decimals for atomic conversion
ASSET_DECIMALS: dict[str, int] = {
X402Asset.USDC.value: 6,
X402Asset.USDT.value: 6,
X402Asset.DAI.value: 18,
X402Asset.ETH.value: 18,
X402Asset.SOL.value: 9,
X402Asset.MATIC.value: 18,
X402Asset.TRX.value: 6,
X402Asset.BTC.value: 8,
X402Asset.LTC.value: 8,
X402Asset.DOGE.value: 8,
}
__all__ = [
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
"X402Asset",
"X402Network",
"X402Scheme",
]

1
routers/advanced.py
View file

@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Advanced"])
@router.post(
"/v1/tls/impersonate",
tags=["Advanced"],

9
routers/agency.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Agency"])
@router.post("/v1/agency/create", tags=["Agency"], summary="Create a white-label agency profile")
async def create_agency_endpoint(
name: str = Body(...),
@ -67,7 +68,9 @@ async def update_branding(
return {"success": result["success"], "data": result}
@router.post("/v1/agency/{agency_id}/clients", tags=["Agency"], summary="Create a client sub-account")
@router.post(
"/v1/agency/{agency_id}/clients", tags=["Agency"], summary="Create a client sub-account"
)
async def create_client_endpoint(
agency_id: str,
client_name: str = Body(...),
@ -93,7 +96,9 @@ async def list_clients_endpoint(agency_id: str) -> dict[str, Any]:
return {"success": True, "data": {"clients": clients, "total": len(clients)}}
@router.get("/v1/agency/{agency_id}/analytics", tags=["Agency"], summary="Get agency usage analytics")
@router.get(
"/v1/agency/{agency_id}/analytics", tags=["Agency"], summary="Get agency usage analytics"
)
async def get_analytics(agency_id: str) -> dict[str, Any]:
"""Get aggregate usage analytics for an agency."""
from agency import get_agency_analytics

1
routers/ai.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["AI"])
@router.get("/v1/ai/gpt-manifest", tags=["AI"], summary="Get GPT Action manifest for ChatGPT")
async def get_gpt_manifest() -> JSONResponse:
"""Get the GPT Action manifest for ChatGPT integration.

1
routers/alerts.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Alerts"])
@router.post("/v1/alert/send", tags=["Alerts"], summary="Send an alert to any channel")
async def send_alert_endpoint(
channel: str = Body(...),

1
routers/analysis.py
View file

@ -27,6 +27,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Analysis"])
@router.post("/v1/compare", tags=["Analysis"], summary="Compare content of two URLs")
async def compare(url1: str = Body(...), url2: str = Body(...)) -> dict[str, Any]:
"""Scrape two URLs and compare their content. Shows additions, deletions, changes."""

1
routers/automation.py
View file

@ -22,6 +22,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Automation"])
@router.post("/v1/automate", tags=["Automation"], summary="Execute browser automation steps")
async def automate(request: AutomateRequest) -> dict[str, Any]:
"""Execute browser automation steps. Sessions persist for login flows."""

5
routers/batch.py
View file

@ -20,7 +20,10 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Batch"])
@router.post("/v1/batch-file", tags=["Batch"], summary="Batch scrape URLs from a file with templates")
@router.post(
"/v1/batch-file", tags=["Batch"], summary="Batch scrape URLs from a file with templates"
)
async def batch_from_file(request: BatchFileRequest) -> dict[str, Any]:
bp = BatchProcessor()
results = await bp.from_file(

1
routers/circuit_breaker.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Circuit Breaker"])
@router.post("/v1/breaker/status", tags=["Circuit Breaker"], summary="Get circuit breaker status")
async def breaker_status() -> dict[str, Any]:
return {

9
routers/commerce.py
View file

@ -19,7 +19,10 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Commerce"])
@router.post("/v1/commerce/sync", tags=["Commerce"], summary="Sync products to WooCommerce or Shopify")
@router.post(
"/v1/commerce/sync", tags=["Commerce"], summary="Sync products to WooCommerce or Shopify"
)
async def sync_commerce(
platform: str = Body(...),
products: list[dict[str, Any]] = Body(...),
@ -58,7 +61,9 @@ async def sync_commerce(
return {"success": result["success"], "data": result}
@router.get("/v1/commerce/platforms", tags=["Commerce"], summary="List supported commerce platforms")
@router.get(
"/v1/commerce/platforms", tags=["Commerce"], summary="List supported commerce platforms"
)
async def list_commerce_platforms() -> dict[str, Any]:
"""List supported e-commerce platforms and their credential requirements."""
return {

9
routers/compliance.py
View file

@ -17,7 +17,10 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Compliance"])
@router.post("/v1/compliance/check", tags=["Compliance"], summary="Run full compliance check on a URL")
@router.post(
"/v1/compliance/check", tags=["Compliance"], summary="Run full compliance check on a URL"
)
async def compliance_check(url: str = Body(...)) -> dict[str, Any]:
"""Run a full legal compliance check on a target URL.
@ -35,7 +38,9 @@ async def compliance_check(url: str = Body(...)) -> dict[str, Any]:
return {"success": True, "data": result}
@router.get("/v1/compliance/check", tags=["Compliance"], summary="Get compliance check documentation")
@router.get(
"/v1/compliance/check", tags=["Compliance"], summary="Get compliance check documentation"
)
async def compliance_docs() -> dict[str, Any]:
"""Get information about the compliance check endpoint."""
return {

4
routers/config.py
View file

@ -34,7 +34,9 @@ async def update_config(updates: dict[str, Any] = Body(...)) -> dict[str, Any]:
return {"success": True, "data": result}
@router.post("/v1/config/profile/tor", tags=["Config"], summary="Enable Tor routing for all requests")
@router.post(
"/v1/config/profile/tor", tags=["Config"], summary="Enable Tor routing for all requests"
)
async def enable_tor() -> dict[str, Any]:
"""Enable Tor routing for all requests."""
result = settings.enable_tor()

1
routers/costing.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Costing"])
@router.get("/v1/costing/dashboard", tags=["Costing"], summary="Get cost analytics dashboard")
async def cost_dashboard() -> dict[str, Any]:
"""Get the full cost analytics dashboard.

1
routers/crm.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["CRM"])
@router.post("/v1/crm/sync", tags=["CRM"], summary="Sync scraped data to CRM")
async def sync_crm(
platform: str = Body(...),

1
routers/dashboard.py
View file

@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Dashboard"])
@router.get("/dashboard", tags=["Dashboard"], summary="Pry health and performance dashboard")
async def dashboard() -> HTMLResponse:
cache_stats = cache.stats()

9
routers/email.py
View file

@ -17,7 +17,10 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Email"])
@router.post("/v1/email/scrape", tags=["Email"], summary="Extract data from an email (subject + body)")
@router.post(
"/v1/email/scrape", tags=["Email"], summary="Extract data from an email (subject + body)"
)
async def scrape_email(
subject: str = Body(""),
body: str = Body(""),
@ -56,7 +59,9 @@ async def fetch_gmail(
return {"success": result.get("success", False), "data": result}
@router.post("/v1/email/outlook", tags=["Email"], summary="Fetch and extract data from Outlook inbox")
@router.post(
"/v1/email/outlook", tags=["Email"], summary="Fetch and extract data from Outlook inbox"
)
async def fetch_outlook(
access_token: str = Body(...),
max_results: int = Body(20),

1
routers/enrichment.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Enrichment"])
@router.post(
"/v1/enrich",
tags=["Enrichment"],

1
routers/execute.py
View file

@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Execute"])
@router.post("/v1/run", tags=["Execute"], summary="Execute a Pryfile")
async def run_pryfile(path: str = Body("pry.yml")) -> dict[str, Any]:
from pryfile import Pryfile

1
routers/export.py
View file

@ -22,6 +22,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Export"])
@router.post("/v1/export", tags=["Export"], summary="Export scraped content in multiple formats")
async def export_data(url: str = Body(...), format: str = "json") -> dict[str, Any]:
"""Export scraped content in multiple formats: json, csv, txt, rss.

9
routers/extraction.py
View file

@ -26,6 +26,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Extraction"])
@router.post(
"/v1/extract-table", tags=["Extraction"], summary="Extract HTML tables as structured data"
)
@ -122,7 +123,9 @@ async def analyze_seo(url: str = Body(...)) -> dict[str, Any]:
}
@router.post("/v1/schema", tags=["Extraction"], summary="Extract Schema.org/JSON-LD structured data")
@router.post(
"/v1/schema", tags=["Extraction"], summary="Extract Schema.org/JSON-LD structured data"
)
async def extract_schema(url: str = Body(...)) -> dict[str, Any]:
"""Extract Schema.org/JSON-LD structured data from a page."""
client = await get_client()
@ -211,7 +214,9 @@ async def extract_css(
}
@router.post("/v1/extract/llm", tags=["Extraction"], summary="Extract with LLM + chunking strategies")
@router.post(
"/v1/extract/llm", tags=["Extraction"], summary="Extract with LLM + chunking strategies"
)
async def extract_llm(
url: str = Body(...),
instruction: str = Body("Extract all key information from this content."),

5
routers/freshness.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Freshness"])
@router.post(
"/v1/freshness/check",
tags=["Freshness"],
@ -63,7 +64,9 @@ async def freshness_frequency(
return {"success": True, "data": result}
@router.get("/v1/freshness/dashboard", tags=["Freshness"], summary="Get content staleness dashboard")
@router.get(
"/v1/freshness/dashboard", tags=["Freshness"], summary="Get content staleness dashboard"
)
async def freshness_dashboard() -> dict[str, Any]:
"""Get the content staleness dashboard.

1
routers/gdpr.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["GDPR"])
@router.post("/v1/gdpr/consent", tags=["GDPR"], summary="Record user consent for data processing")
async def record_consent_endpoint(
user_id: str = Body(...),

1
routers/integrations.py
View file

@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Integrations"])
@router.get(
"/v1/destinations",
tags=["Integrations"],

5
routers/intelligence.py
View file

@ -17,7 +17,10 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Intelligence"])
@router.post("/v1/intel/snapshot", tags=["Intelligence"], summary="Record a competitor data snapshot")
@router.post(
"/v1/intel/snapshot", tags=["Intelligence"], summary="Record a competitor data snapshot"
)
async def record_intel_snapshot(
competitor_id: str = Body(...),
competitor_name: str = Body(...),

1
routers/jobs.py
View file

@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Jobs"])
@router.get("/v1/job/{job_id}", tags=["Jobs"], summary="Get async job status and result")
async def get_job(job_id: str) -> dict[str, Any]:
"""Get async job status and result."""

1
routers/marketplace.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Marketplace"])
@router.post(
"/v1/actors/create",
tags=["Marketplace"],

1
routers/monitoring.py
View file

@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Monitoring"])
@router.post("/v1/monitor", tags=["Monitoring"], summary="Create a scheduled content monitor")
async def create_monitor_endpoint(
name: str = Body(...),

5
routers/parsing.py
View file

@ -23,7 +23,10 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Parsing"])
@router.post("/v1/parse", tags=["Parsing"], summary="Parse a document (PDF, DOCX, image, CSV, JSON)")
@router.post(
"/v1/parse", tags=["Parsing"], summary="Parse a document (PDF, DOCX, image, CSV, JSON)"
)
async def parse_document(request: ParseRequest) -> dict[str, Any]:
"""Parse a document (PDF, DOCX, image, CSV, JSON) to text."""
try:

1
routers/pipeline.py
View file

@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Pipeline"])
@router.post("/v1/pipeline/hook", tags=["Pipeline"], summary="Register a hook function")
async def register_hook(
hook_point: str = Body(...),

1
routers/pipelines.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Pipelines"])
@router.get(
"/v1/pipelines/steps", tags=["Pipelines"], summary="List all available pipeline step types"
)

13
routers/proxy.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Proxy"])
@router.get("/v1/proxy/providers", tags=["Proxy"], summary="List all available proxy providers")
async def list_proxy_providers() -> dict[str, Any]:
"""List free + premium proxy providers with affiliate details."""
@ -25,7 +26,9 @@ async def list_proxy_providers() -> dict[str, Any]:
return {"success": True, "data": ProxyManager().list_providers()}
@router.post("/v1/proxy/signup", tags=["Proxy"], summary="Open affiliate signup link for a provider")
@router.post(
"/v1/proxy/signup", tags=["Proxy"], summary="Open affiliate signup link for a provider"
)
async def proxy_signup(provider: str = Body(...)) -> dict[str, Any]:
"""Get the affiliate signup URL for a proxy provider.
@ -39,7 +42,9 @@ async def proxy_signup(provider: str = Body(...)) -> dict[str, Any]:
return {"success": True, "data": {"signup_url": url, "provider": provider}}
@router.get("/v1/proxy/referrals", tags=["Proxy"], summary="List proxy provider affiliate referrals")
@router.get(
"/v1/proxy/referrals", tags=["Proxy"], summary="List proxy provider affiliate referrals"
)
async def list_proxy_referrals() -> dict[str, Any]:
"""Return the curated proxy provider affiliate catalog (proxy_referrals.py).
@ -126,7 +131,9 @@ async def proxy_status() -> dict[str, Any]:
}
@router.post("/v1/proxy/recommend", tags=["Proxy"], summary="Get proxy recommendation after a block")
@router.post(
"/v1/proxy/recommend", tags=["Proxy"], summary="Get proxy recommendation after a block"
)
async def proxy_recommend(last_error: str = Body("")) -> dict[str, Any]:
"""After a scrape fails with anti-bot detection, get a recommendation
for which premium proxy provider to sign up with."""

1
routers/quality.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Quality"])
@router.post(
"/v1/quality/check", tags=["Quality"], summary="Run data quality check on extraction results"
)

1
routers/reconciliation.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Reconciliation"])
@router.post(
"/v1/reconcile",
tags=["Reconciliation"],

1
routers/recorder.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Recorder"])
@router.post("/v1/record/start", tags=["Recorder"], summary="Start recording browser actions")
async def start_recording(session_id: str = Body(...)) -> dict[str, Any]:
recorder.start(session_id)

1
routers/referrals.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Referrals"])
@router.get(
"/v1/referrals/catalog",
tags=["Referrals"],

1
routers/reports.py
View file

@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Reports"])
@router.post(
"/v1/reports/generate",
tags=["Reports"],

1
routers/review.py
View file

@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Review"])
@router.post("/v1/review/submit", tags=["Review"], summary="Submit extracted data for human review")
async def review_submit(
data: dict[str, Any] = Body(...),

4
routers/scraping.py
View file

@ -264,7 +264,9 @@ async def crawl(request: CrawlRequest) -> dict[str, Any]:
{
"max_pages": request.maxPages,
"max_depth": request.maxDepth,
"timeout": request.scrapeOptions.get("timeout", 60) if request.scrapeOptions else 60,
"timeout": request.scrapeOptions.get("timeout", 60)
if request.scrapeOptions
else 60,
},
)
return {"success": True, "data": {"id": "sync", "url": request.url, "pages": pages}}

1
routers/scraping_api.py
View file

@ -24,6 +24,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Scraping"])
@router.post(
"/v1/ultimate-scrape", tags=["Scraping"], summary="Scrape with 10-tier anti-bot fallback system"
)

1
routers/seo.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["SEO"])
@router.post("/v1/seo/analyze", tags=["SEO"], summary="Analyze SEO elements from a URL")
async def seo_analyze(url: str = Body(...)) -> dict[str, Any]:
"""Analyze all SEO elements from a URL.

1
routers/sessions.py
View file

@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Sessions"])
@router.post("/v1/session/create", tags=["Sessions"], summary="Create a persistent browser session")
async def create_session(url: str = Body(...), persist: bool = True) -> dict[str, Any]:
"""Create a persistent browser session."""

1
routers/share.py
View file

@ -23,6 +23,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Share"])
@router.post("/v1/share", tags=["Share"], summary="Share scraped content via link")
async def share_scrape(data: dict[str, Any] = Body(...)) -> dict[str, Any]:
url = data.get("url", "")

1
routers/stats.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Stats"])
@router.get("/v0/stats", tags=["Stats"], summary="Get cache, rate limiter, and session stats")
async def stats() -> dict[str, Any]:
return {

1
routers/structure.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Structure"])
@router.post(
"/v1/structure/check",
tags=["Structure"],

1
routers/system.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["System"])
@router.get("/metrics", tags=["System"], summary="Prometheus metrics", include_in_schema=False)
async def metrics() -> Response:
"""Expose Prometheus metrics for scraping."""

5
routers/training.py
View file

@ -19,6 +19,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Training"])
@router.post(
"/v1/training/classify-license",
tags=["Training"],
@ -36,7 +37,9 @@ async def classify_license_endpoint(text: str = Body(...)) -> dict[str, Any]:
return {"success": True, "data": result}
@router.post("/v1/training/clean", tags=["Training"], summary="Strip PII and copyright from content")
@router.post(
"/v1/training/clean", tags=["Training"], summary="Strip PII and copyright from content"
)
async def clean_content(
text: str = Body(...),
strip_names: bool = Body(False),

1
routers/transform.py
View file

@ -22,6 +22,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Transform"])
@router.post("/v1/transform", tags=["Transform"], summary="Transform data to multiple formats")
async def transform_data(
data: list[dict[str, Any]] = Body(...), format: str = "csv"

1
routers/untagged.py
View file

@ -18,6 +18,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Untagged"])
@router.websocket("/v1/stream/{job_id}")
async def stream_endpoint(websocket: WebSocket, job_id: str = "live") -> None:
await websocket.accept()

1
routers/vision.py
View file

@ -26,6 +26,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Vision"])
@router.post("/v1/vision", tags=["Vision"], summary="Analyze an image with a free vision model")
async def vision(
question: str = Body("Describe what is visible in this image.", embed=True),

1
routers/webhooks.py
View file

@ -17,6 +17,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["Webhooks"])
@router.post(
"/v1/webhooks/test",
tags=["Webhooks"],

1
routers/x402.py
View file

@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=["x402"])
@router.get("/v1/x402/pricing", tags=["x402"], summary="Get x402 pricing for all paid operations")
async def x402_pricing() -> dict[str, Any]:
"""Get the price list for pay-per-scrape operations."""

4
settings.py
View file

@ -236,7 +236,9 @@ class PrySettings(BaseSettings):
proxies.append(f"socks5://{self.tor_socks5_host}:{self.tor_socks5_port}")
if self.proxy_enabled and self.proxy_url:
if self.proxy_username:
netloc = self.proxy_url.split("://")[1] if "://" in self.proxy_url else self.proxy_url
netloc = (
self.proxy_url.split("://")[1] if "://" in self.proxy_url else self.proxy_url
)
proxies.append(
f"{self.proxy_type}://{self.proxy_username}:{self.proxy_password}@{netloc}"
)

18
tests/test_fallback_integration.py
View file

@ -17,8 +17,8 @@ from retry import Jitter, RetryConfig, async_retry
def _valid_html(text: str = "content") -> str:
body = f"<p>{text}</p>" * 40
return (
"<!DOCTYPE html><html lang=\"en\"><head><title>T</title>"
"<meta charset=\"utf-8\"></head>"
'<!DOCTYPE html><html lang="en"><head><title>T</title>'
'<meta charset="utf-8"></head>'
f"<body>{body}</body></html>"
)
@ -152,7 +152,9 @@ class TestCircuitBreakerIntegration:
monkeypatch.setattr(resilience.time, "time", lambda: future)
fn = AsyncMock(return_value="ok")
cfg = RetryConfig(max_retries=1, circuit_breaker="ollama_int", base_delay=0.01, jitter=Jitter.NONE)
cfg = RetryConfig(
max_retries=1, circuit_breaker="ollama_int", base_delay=0.01, jitter=Jitter.NONE
)
result = await async_retry(fn, config=cfg)
assert result == "ok"
@ -202,7 +204,9 @@ class TestDomainTierMemoryWithRetry:
async def test_domain_memory_used_on_retry(self, fresh_registry, scraper_with_mocks):
"""After domain learns best tier, prefers it."""
s, _mocks = scraper_with_mocks
fresh_registry.record_domain_tier_result("example.com", "archive", success=True, latency=0.1)
fresh_registry.record_domain_tier_result(
"example.com", "archive", success=True, latency=0.1
)
html = _valid_html("archive content")
_mocks["_tier_archive"].return_value = html
_mocks["_tier_direct"].return_value = _valid_html("direct")
@ -242,7 +246,9 @@ class TestGracefulDegradation:
raise RuntimeError(result.get("error", "scrape failed"))
return result
cfg = RetryConfig(max_retries=1, circuit_breaker=name, base_delay=0.01, jitter=Jitter.NONE)
cfg = RetryConfig(
max_retries=1, circuit_breaker=name, base_delay=0.01, jitter=Jitter.NONE
)
with pytest.raises(RuntimeError):
await async_retry(scrape_and_raise, "https://example.com/page", config=cfg)
@ -255,11 +261,13 @@ class TestRetrySettings:
def test_retry_config_from_settings(self) -> None:
from settings import PrySettings
settings = PrySettings()
assert settings.retry_backoff == "exponential"
def test_api_dict_includes_backoff(self) -> None:
from settings import PrySettings
settings = PrySettings()
api_dict = settings.to_api_dict()
assert "retry" in api_dict

17
tests/test_resilience.py
View file

@ -16,8 +16,8 @@ from resilience import CircuitBreaker, CircuitState, DomainTierMemory, Resilienc
def _valid_html(text: str) -> str:
body = f"<p>{text}</p>" * 40
return (
"<!DOCTYPE html><html lang=\"en\"><head><title>T</title>"
"<meta charset=\"utf-8\"></head>"
'<!DOCTYPE html><html lang="en"><head><title>T</title>'
'<meta charset="utf-8"></head>'
f"<body>{body}</body></html>"
)
@ -159,10 +159,7 @@ class TestUltimateScraperResilience:
fresh_registry.record_service_result("flaresolverr", False)
html = _valid_html("archive content")
mocks = {
name: AsyncMock(return_value=None)
for name in _TIER_NAMES
}
mocks = {name: AsyncMock(return_value=None) for name in _TIER_NAMES}
mocks["_tier_archive"].return_value = html
with ExitStack() as stack:
for name in _TIER_NAMES:
@ -201,9 +198,7 @@ class TestUltimateScraperResilience:
for name in _TIER_NAMES:
stack.enter_context(patch.object(scraper, name, new=mocks[name]))
await scraper.scrape("https://example.com/page")
assert fresh_registry.best_tier_for_domain(
"example.com", ["direct", "archive"]
) == "direct"
assert fresh_registry.best_tier_for_domain("example.com", ["direct", "archive"]) == "direct"
@pytest.mark.anyio
async def test_records_tier_failure_no_success_memory(self, fresh_registry):
@ -214,6 +209,4 @@ class TestUltimateScraperResilience:
stack.enter_context(patch.object(scraper, name, new=mocks[name]))
result = await scraper.scrape("https://example.com/page")
assert result["status"] == "error"
assert fresh_registry.best_tier_for_domain(
"example.com", ["direct", "archive"]
) is None
assert fresh_registry.best_tier_for_domain("example.com", ["direct", "archive"]) is None

41
tests/test_ultimate_scraper.py
View file

@ -30,12 +30,21 @@ def _clear_registry() -> None:
_TIER_ALL = [
"_tier_direct", "_tier_tls_fingerprint", "_tier_cloudscraper",
"_tier_flaresolverr", "_tier_undetected", "_tier_camoufox",
"_tier_playwright", "_tier_googlebot", "_tier_premium_proxy",
"_tier_archive", "_tier_google_cache", "_tier_tor",
"_tier_direct",
"_tier_tls_fingerprint",
"_tier_cloudscraper",
"_tier_flaresolverr",
"_tier_undetected",
"_tier_camoufox",
"_tier_playwright",
"_tier_googlebot",
"_tier_premium_proxy",
"_tier_archive",
"_tier_google_cache",
"_tier_tor",
]
def test_ultimate_init() -> None:
s = UltimateScraper()
assert s is not None
@ -132,17 +141,19 @@ async def test_tier_camoufox_failure() -> None:
async def test_scrape_falls_through_all_tiers() -> None:
_clear_registry()
s = UltimateScraper()
with patch.object(s, "_tier_direct", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_tls_fingerprint", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_cloudscraper", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_flaresolverr", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_undetected", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_camoufox", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_playwright", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_googlebot", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_archive", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_google_cache", new=AsyncMock(return_value=None)), \
patch.object(s, "_tier_tor", new=AsyncMock(return_value=None)):
with (
patch.object(s, "_tier_direct", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_tls_fingerprint", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_cloudscraper", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_flaresolverr", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_undetected", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_camoufox", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_playwright", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_googlebot", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_archive", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_google_cache", new=AsyncMock(return_value=None)),
patch.object(s, "_tier_tor", new=AsyncMock(return_value=None)),
):
result = await s.scrape("http://example.com")
assert result["status"] == "error"

98
tests/test_x402_surface_snapshot.py Normal file
View file

@ -0,0 +1,98 @@
"""Snapshot tests for the x402 public API surface.
These tests guard against accidental breakage when refactoring
x402.py into the pry_x402/ sub-package.
"""
# 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.
from __future__ import annotations
class TestX402PublicSurface:
"""The x402 module must expose the same public API as before."""
def test_required_symbols_exist(self) -> None:
import x402
required = {
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402Asset",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402Network",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402Scheme",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
"FacilitatorConfig",
"FacilitatorRouter",
"X402Handler",
"build_payment_required",
"create_batch_payment",
"create_payment_request",
"get_facilitator_router",
"is_batch_paid",
"require_payment_legacy",
"set_facilitator_router",
"verify_batch_payment",
}
missing = required - set(dir(x402))
assert not missing, f"Missing x402 public symbols: {missing}"
def test_pry_x402_re_exports_constants(self) -> None:
import pry_x402
required = {
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402Asset",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402Network",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402Scheme",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
}
missing = required - set(dir(pry_x402))
assert not missing, f"Missing pry_x402 public symbols: {missing}"
def test_enum_values_unchanged(self) -> None:
from x402 import X402Asset, X402Network, X402Scheme
assert X402Scheme.EXACT == "exact"
assert X402Scheme.UPTO == "upto"
assert X402Network.BASE == "base"
assert X402Asset.USDC == "USDC"
def test_pricing_keys_unchanged(self) -> None:
import x402
expected = {
"scrape",
"crawl",
"extract",
"monitor",
"llm_call",
"template_execute",
"template_batch_execute",
"browser_automation",
"bulk_crawl",
"pdf_extract",
"ocr_extract",
"graphql_query",
"schema_extract",
}
assert set(x402.X402_PRICING.keys()) == expected

18
url_guard.py
View file

@ -25,15 +25,15 @@ logger = logging.getLogger(__name__)
# Private/reserved IP ranges (RFC 1918, RFC 6598, RFC 4193, link-local, loopback)
_PRIVATE_NETS: list[str] = [
"127.0.0.0/8", # loopback
"10.0.0.0/8", # RFC 1918
"172.16.0.0/12", # RFC 1918
"192.168.0.0/16", # RFC 1918
"100.64.0.0/10", # RFC 6598 (CGNAT)
"169.254.0.0/16", # link-local
"fc00::/7", # RFC 4193 (ULA)
"fe80::/10", # link-local IPv6
"::1/128", # IPv6 loopback
"127.0.0.0/8", # loopback
"10.0.0.0/8", # RFC 1918
"172.16.0.0/12", # RFC 1918
"192.168.0.0/16", # RFC 1918
"100.64.0.0/10", # RFC 6598 (CGNAT)
"169.254.0.0/16", # link-local
"fc00::/7", # RFC 4193 (ULA)
"fe80::/10", # link-local IPv6
"::1/128", # IPv6 loopback
]
# Common internal hostnames that suggest SSRF targeting

123
x402.py
View file

@ -28,121 +28,28 @@ import threading
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any, cast
import httpx
from paths import PRY_DATA_DIR
from pry_x402.constants import (
ASSET_DECIMALS,
EIP7702_RPC,
X402_DEFAULT_ASSET,
X402_DEFAULT_NETWORK,
X402_DIR,
X402_FACILITATOR_URL,
X402_OFFLINE_MODE,
X402_PRICING,
X402_TIMEOUT_SECONDS,
X402_WALLET,
X402Asset,
X402Network,
X402Scheme,
)
logger = logging.getLogger(__name__)
X402_DIR = PRY_DATA_DIR / "x402"
X402_DIR.mkdir(parents=True, exist_ok=True)
# x402 pricing per operation (USDC atomic units, 6 decimals)
X402_PRICING: dict[str, dict[str, Any]] = {
"scrape": {"price_usd": 0.001, "description": "Single URL scrape"},
"crawl": {"price_usd": 0.01, "description": "Crawl up to 10 pages"},
"extract": {"price_usd": 0.005, "description": "Structured extraction"},
"monitor": {"price_usd": 0.02, "description": "Create scheduled monitor"},
"llm_call": {"price_usd": 0.01, "description": "LLM extraction call"},
"template_execute": {"price_usd": 0.002, "description": "Execute scraper template"},
"template_batch_execute": {"price_usd": 0.01, "description": "Execute up to 20 template items"},
"browser_automation": {"price_usd": 0.05, "description": "Browser automation"},
"bulk_crawl": {"price_usd": 0.10, "description": "Crawl up to 1000 pages"},
"pdf_extract": {"price_usd": 0.01, "description": "PDF table extraction"},
"ocr_extract": {"price_usd": 0.005, "description": "Image OCR"},
"graphql_query": {"price_usd": 0.003, "description": "GraphQL query execution"},
"schema_extract": {"price_usd": 0.002, "description": "Schema.org/JSON-LD extraction"},
}
class X402Scheme(StrEnum):
"""Payment schemes supported by x402."""
EXACT = "exact"
UPTO = "upto"
class X402Network(StrEnum):
"""Blockchain networks supported by x402."""
BASE = "base"
SOLANA = "solana"
ETHEREUM = "ethereum"
POLYGON = "polygon"
ARBITRUM = "arbitrum"
OPTIMISM = "optimism"
BNB = "bnb"
AVALANCHE = "avalanche"
BASE_SEPOLIA = "base-sepolia"
TRON = "tron"
BITCOIN = "bitcoin"
LITECOIN = "litecoin"
DOGECOIN = "dogecoin"
class X402Asset(StrEnum):
"""Token assets supported by x402."""
USDC = "USDC"
USDT = "USDT"
DAI = "DAI"
ETH = "ETH"
SOL = "SOL"
MATIC = "MATIC"
TRX = "TRX"
BTC = "BTC"
LTC = "LTC"
DOGE = "DOGE"
# Prefer gopass for the x402 wallet and facilitator URL. Env vars override only
# if the backend returns an empty string. This lets you:
# gopass insert -m pry/x402_wallet # persistent across restarts
# PRY_X402_WALLET=0x... # one-off override (CI, dev)
try:
from secrets_backend import get_secret as _gs
X402_WALLET = _gs("x402_wallet") or os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = _gs("x402_facilitator") or os.getenv(
"PRY_X402_FACILITATOR", "https://x402.org/facilitator"
)
except ImportError:
X402_WALLET = os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = os.getenv("PRY_X402_FACILITATOR", "https://x402.org/facilitator")
X402_DEFAULT_NETWORK = os.getenv("PRY_X402_NETWORK", X402Network.BASE.value)
X402_DEFAULT_ASSET = os.getenv("PRY_X402_ASSET", X402Asset.USDC.value)
X402_OFFLINE_MODE = os.getenv("PRY_X402_OFFLINE", "false").lower() in ("1", "true", "yes")
X402_TIMEOUT_SECONDS = float(os.getenv("PRY_X402_TIMEOUT", "15"))
# EIP-7702 self-verify RPC endpoints (public defaults, override via env)
EIP7702_RPC: dict[str, str] = {
"base": os.getenv("PRY_X402_BASE_RPC", "https://mainnet.base.org"),
"ethereum": os.getenv("PRY_X402_ETH_RPC", "https://eth.llamarpc.com"),
"polygon": os.getenv("PRY_X402_POLYGON_RPC", "https://polygon.drpc.org"),
"arbitrum": os.getenv("PRY_X402_ARBITRUM_RPC", "https://arb1.arbitrum.io/rpc"),
"optimism": os.getenv("PRY_X402_OPTIMISM_RPC", "https://mainnet.optimism.io"),
"bnb": os.getenv("PRY_X402_BNB_RPC", "https://bsc-dataseed.binance.org"),
"avalanche": os.getenv("PRY_X402_AVALANCHE_RPC", "https://api.avax.network/ext/bc/C/rpc"),
"base-sepolia": os.getenv("PRY_X402_BASE_SEPOLIA_RPC", "https://sepolia.base.org"),
}
# Asset decimals for atomic conversion
ASSET_DECIMALS: dict[str, int] = {
X402Asset.USDC.value: 6,
X402Asset.USDT.value: 6,
X402Asset.DAI.value: 18,
X402Asset.ETH.value: 18,
X402Asset.SOL.value: 9,
X402Asset.MATIC.value: 18,
X402Asset.TRX.value: 6,
X402Asset.BTC.value: 8,
X402Asset.LTC.value: 8,
X402Asset.DOGE.value: 8,
}
def _to_atomic(amount_usd: float, asset: str = X402_DEFAULT_ASSET) -> int:
"""Convert USD amount to atomic units for the given asset."""