Build a refreshable dataset from websites

[
  { "id": "supplier-a:mug-blue", "url": "https://shop.example.com/products/blue-mug" }
]

import json
import math
import os
import sqlite3
import sys
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

SCHEMA = {
    "type": "object",
    "properties": {
        "name": {"type": ["string", "null"], "description": "Stated product name; null if absent."},
        "variant": {"type": ["string", "null"], "description": "Selected size, color, or variant; null if unclear."},
        "price_amount": {"type": ["number", "null"], "description": "Current stated price for this variant; null if unclear."},
        "currency": {"type": ["string", "null"], "description": "Explicit three-letter currency code; null if unclear."},
        "unit": {"type": ["string", "null"], "description": "Price basis, such as per item or per kilogram; null if absent."},
    },
    "required": ["name", "variant", "price_amount", "currency", "unit"],
    "additionalProperties": False,
}

def validate_record(data):
    # Local validation for the small schema above; reject booleans as prices.
    if not isinstance(data, dict) or set(data) != set(SCHEMA["required"]):
        raise ValueError("Unexpected record shape")
    for key in ("name", "variant", "currency", "unit"):
        if data[key] is not None and not isinstance(data[key], str):
            raise ValueError(f"Invalid {key}")
    price = data["price_amount"]
    if price is not None and (type(price) not in (int, float) or not math.isfinite(price) or price < 0):
        raise ValueError("Invalid price")
    if not data["name"] or not data["name"].strip():
        raise ValueError("No usable product identity")
    normalized = dict(data)
    if normalized["currency"] is not None:
        normalized["currency"] = normalized["currency"].strip().upper()
        if len(normalized["currency"]) != 3 or not normalized["currency"].isascii() or not normalized["currency"].isalpha():
            raise ValueError("Currency needs review")
    return normalized

def open_catalog(path):
    db = sqlite3.connect(path)
    db.row_factory = sqlite3.Row
    db.executescript("""
      CREATE TABLE IF NOT EXISTS sources (
        id TEXT PRIMARY KEY, url TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending',
        attempts INTEGER NOT NULL DEFAULT 0, next_attempt_at REAL NOT NULL DEFAULT 0,
        error TEXT, last_success_at TEXT
      );
      CREATE TABLE IF NOT EXISTS records (
        id TEXT PRIMARY KEY, source_url TEXT NOT NULL, observed_at TEXT NOT NULL,
        data_json TEXT NOT NULL, analyzed_urls_json TEXT NOT NULL
      );
      CREATE TABLE IF NOT EXISTS observations (
        id TEXT NOT NULL, observed_at TEXT NOT NULL, source_url TEXT NOT NULL,
        raw_data_json TEXT NOT NULL, analyzed_urls_json TEXT NOT NULL,
        PRIMARY KEY (id, observed_at)
      );
    """)
    return db

def enqueue(db, items, refresh=False):
    with db:
        for item in items:
            url = urlsplit(item["url"])
            if url.scheme != "https" or not url.hostname or url.username or url.password:
                raise ValueError("Use reviewed HTTPS source URLs")
            db.execute("""
              INSERT INTO sources (id, url) VALUES (?, ?)
              ON CONFLICT(id) DO UPDATE SET
                status=CASE WHEN url <> excluded.url THEN 'pending' ELSE status END,
                attempts=CASE WHEN url <> excluded.url THEN 0 ELSE attempts END,
                next_attempt_at=CASE WHEN url <> excluded.url THEN 0 ELSE next_attempt_at END,
                url=excluded.url
            """, (item["id"], item["url"]))
            if refresh:
                db.execute("UPDATE sources SET status='pending', attempts=0, next_attempt_at=0, error=NULL WHERE id=?", (item["id"],))

def extract_page(url):
    request = Request(
        "https://api.context.dev/v1/web/extract",
        data=json.dumps({
            "url": url, "schema": SCHEMA, "factCheck": True,
            "maxPages": 1, "maxDepth": 0, "maxAgeMs": 0,
        }).encode(),
        headers={"Authorization": f"Bearer {os.environ['CONTEXT_DEV_API_KEY']}",
                 "Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(request, timeout=120) as response:
        return json.load(response)

def retry_delay(headers):
    value = headers.get("Retry-After")
    try:
        return max(0, float(value))
    except (TypeError, ValueError):
        try:
            return max(0, parsedate_to_datetime(value).timestamp() - time.time())
        except (TypeError, ValueError, AttributeError):
            return 30

def run_pending(db):
    jobs = db.execute("""
      SELECT * FROM sources WHERE status IN ('pending', 'retryable')
      AND attempts < 3 AND next_attempt_at <= ?
    """, (time.time(),)).fetchall()
    for job in jobs:
        with db:
            db.execute("UPDATE sources SET attempts=attempts+1 WHERE id=?", (job["id"],))
        try:
            result = extract_page(job["url"])
            data = validate_record(result["data"])
            urls = result["urls_analyzed"]
            if not isinstance(urls, list) or not urls or not all(isinstance(url, str) for url in urls):
                raise ValueError("No source provenance")
            observed = datetime.now(timezone.utc).isoformat()
            with db:
                db.execute("INSERT INTO observations VALUES (?, ?, ?, ?, ?)",
                    (job["id"], observed, job["url"], json.dumps(result["data"]), json.dumps(urls)))
                db.execute("""
                  INSERT INTO records VALUES (?, ?, ?, ?, ?)
                  ON CONFLICT(id) DO UPDATE SET source_url=excluded.source_url,
                    observed_at=excluded.observed_at, data_json=excluded.data_json,
                    analyzed_urls_json=excluded.analyzed_urls_json
                """, (job["id"], job["url"], observed, json.dumps(data), json.dumps(urls)))
                db.execute("UPDATE sources SET status='ready', error=NULL, last_success_at=? WHERE id=?", (observed, job["id"]))
        except HTTPError as error:
            retryable = error.code in (408, 429) or error.code >= 500
            delay = retry_delay(error.headers)
            error.close()
            with db:
                db.execute("UPDATE sources SET status=?, error=?, next_attempt_at=? WHERE id=?",
                    ("retryable" if retryable else "error", f"HTTP {error.code}",
                     time.time() + delay, job["id"]))
            if error.code in (401, 403, 429):
                break
        except (URLError, TimeoutError):
            with db:
                db.execute("UPDATE sources SET status='retryable', error='Network failure', next_attempt_at=? WHERE id=?", (time.time() + 30, job["id"]))
        except (ValueError, KeyError, TypeError) as error:
            with db:
                db.execute("UPDATE sources SET status='review', error=? WHERE id=?", (str(error), job["id"]))

if __name__ == "__main__":
    with open(sys.argv[1]) as source_file:
        manifest = json.load(source_file)
    db = open_catalog("catalog.sqlite")
    enqueue(db, manifest, refresh="--refresh" in sys.argv[2:])
    run_pending(db)
    for row in db.execute("SELECT status, count(*) AS count FROM sources GROUP BY status"):
        print(dict(row))
    db.close()

python3 catalog.py sources.json
python3 catalog.py sources.json --refresh