Scrape a website hourly into MySQL

from urllib.parse import urlsplit

import requests
from pydantic import BaseModel, Field
from pydantic.v1 import BaseSettings

NOTTE_REQUIRED_SECRETS = ["MYSQL_INGEST_URL", "MYSQL_INGEST_TOKEN"]


class ScrapeSettings(BaseSettings):
    notte_api_key: str
    notte_api_url: str = "https://api.notte.cc"


class IngestSettings(BaseSettings):
    mysql_ingest_url: str
    mysql_ingest_token: str


class PageData(BaseModel):
    title: str = Field(min_length=1)
    description: str = Field(min_length=1)


def scrape_page(url: str) -> PageData:
    settings = ScrapeSettings()  # type: ignore[call-arg]  # BaseSettings reads required fields from the environment.
    response = requests.post(
        settings.notte_api_url.rstrip("/") + "/scrape",
        headers={"Authorization": f"Bearer {settings.notte_api_key}"},
        json={
            "url": url,
            "instructions": "Extract the page title and a short description of the page.",
            "response_format": PageData.model_json_schema(),
            "only_main_content": True,
        },
        timeout=120,
    )
    response.raise_for_status()
    structured = response.json().get("structured") or {}
    if not structured.get("success"):
        raise RuntimeError("The scrape did not produce structured data")
    return PageData.model_validate(structured["data"])


def run(url: str) -> dict:
    settings = IngestSettings()  # type: ignore[call-arg]  # BaseSettings reads required fields from the environment.
    receiver = settings.mysql_ingest_url
    if urlsplit(receiver).scheme != "https":
        raise ValueError("MYSQL_INGEST_URL must use HTTPS")
    page = scrape_page(url)
    response = requests.post(
        receiver,
        headers={"Authorization": f"Bearer {settings.mysql_ingest_token}"},
        json={"url": url, "data": page.model_dump()},
        timeout=30,
        allow_redirects=False,
    )
    response.raise_for_status()
    if response.status_code != 200 or response.json().get("saved") is not True:
        raise RuntimeError("The receiver did not confirm the database write")
    return {"url": url, "saved": True}

import os
import ssl

import pymysql
from scrape_function import PageData, scrape_page


def save_page(url: str, page: PageData) -> None:
    connection = pymysql.connect(
        host=os.environ["MYSQL_HOST"],
        port=int(os.environ.get("MYSQL_PORT", "3306")),
        user=os.environ["MYSQL_USER"],
        password=os.environ["MYSQL_PASSWORD"],
        database=os.environ["MYSQL_DATABASE"],
        ssl=ssl.create_default_context(cafile=os.environ.get("MYSQL_SSL_CA")),
        charset="utf8mb4",
        connect_timeout=10,
        read_timeout=30,
        write_timeout=30,
        autocommit=False,
    )
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                "INSERT INTO scraped_pages (url, title, description, scraped_at) "
                "VALUES (%s, %s, %s, UTC_TIMESTAMP()) "
                "ON DUPLICATE KEY UPDATE title=%s, description=%s, scraped_at=UTC_TIMESTAMP()",
                (url, page.title, page.description, page.title, page.description),
            )
        connection.commit()
    finally:
        connection.close()


def run(url: str) -> dict:
    page = scrape_page(url)
    save_page(url, page)
    return {"url": url, "saved": True}


if __name__ == "__main__":
    print(run("https://example.com"))

import os
import secrets

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field
from scrape_function import PageData
from scrape_to_mysql import save_page

app = FastAPI()


class ScrapeResult(BaseModel):
    url: str = Field(min_length=1, max_length=512, pattern=r"^https?://")
    data: PageData


@app.post("/ingest")
def ingest(payload: ScrapeResult, authorization: str = Header(default="")) -> dict:
    expected = f"Bearer {os.environ['MYSQL_INGEST_TOKEN']}"
    if not secrets.compare_digest(authorization.encode(), expected.encode()):
        raise HTTPException(status_code=401, detail="Invalid ingest token")
    save_page(str(payload.url), payload.data)
    return {"saved": True}

from notte_sdk import NotteClient

client = NotteClient()
function = client.Function(
    path="scrape_function.py",
    name="Hourly scrape to MySQL",
)
print(function.function_id)
print(function.run(url="https://example.com", stream=False).result)