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
RULES = {
"title": "h1",
"instructor": ".course-instructor",
"level": ".course-level",
"price": {"selector": "[itemprop=price]", "output": "@content"},
"currency": {"selector": "[itemprop=priceCurrency]", "output": "@content"},
"topics": {"selector": ".syllabus li", "type": "list"},
}
def validate_record(parsed):
# Every rule name is present: missing items are None and missing lists are [].
if not isinstance(parsed, dict) or set(parsed) != set(RULES):
raise ValueError("Unexpected record shape")
for key in ("title", "instructor", "level", "price", "currency"):
if parsed[key] is not None and not isinstance(parsed[key], str):
raise ValueError(f"Invalid {key}")
if not isinstance(parsed["topics"], list) or not all(isinstance(topic, str) for topic in parsed["topics"]):
raise ValueError("Invalid topics")
if not parsed["title"] or not parsed["title"].strip():
raise ValueError("No usable record identity")
record = {key: parsed[key] for key in ("title", "instructor", "level", "topics")}
record["price_text"] = parsed["price"]
record["price_amount"] = None
if parsed["price"] is not None:
try:
amount = float(parsed["price"])
except ValueError:
raise ValueError("Price needs review") from None
if not math.isfinite(amount) or amount < 0:
raise ValueError("Invalid price")
record["price_amount"] = amount
record["currency"] = None
if parsed["currency"] is not None:
currency = parsed["currency"].strip().upper()
if len(currency) != 3 or not currency.isascii() or not currency.isalpha():
raise ValueError("Currency needs review")
record["currency"] = currency
return record
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, final_url TEXT NOT NULL,
observed_at TEXT NOT NULL, data_json TEXT NOT NULL, request_id TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS observations (
id TEXT NOT NULL, observed_at TEXT NOT NULL, source_url TEXT NOT NULL,
final_url TEXT NOT NULL, request_id TEXT NOT NULL,
parsed_json TEXT NOT NULL, markdown TEXT,
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 scrape_page(url):
request = Request(
"https://api.context.dev/v1/web/scrape",
data=json.dumps({
"url": url,
"formats": {"parse": True, "markdown": True},
"parseParams": {"rules": RULES},
"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 = scrape_page(job["url"])
data = validate_record(result["parsed"]["data"])
final_url, markdown = result["url"], result["markdown"]["data"]
if urlsplit(final_url).hostname != urlsplit(job["url"]).hostname:
raise ValueError("Final URL is on another host")
observed = datetime.now(timezone.utc).isoformat()
with db:
db.execute("INSERT INTO observations VALUES (?, ?, ?, ?, ?, ?, ?)",
(job["id"], observed, job["url"], final_url, result["request_id"],
json.dumps(result["parsed"]["data"]), markdown))
db.execute("""
INSERT INTO records VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET source_url=excluded.source_url,
final_url=excluded.final_url, observed_at=excluded.observed_at,
data_json=excluded.data_json, request_id=excluded.request_id
""", (job["id"], job["url"], final_url, observed, json.dumps(data), result["request_id"]))
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()
