watch.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///
"""
HF Hub leaderboard watcher — daily email digest.

Run directly via the shebang (`./watch.py`) or `uv run --script watch.py`.
uv will provision an isolated Python interpreter automatically; no
`pip install` and no virtualenv to manage.

Polls official benchmark leaderboards on the Hugging Face Hub, diffs against
local state, and emails a digest of new model entries since the last run.

Designed to be run once per day from cron.

Docs: https://huggingface.co/docs/hub/leaderboard-data-guide

Configuration: edit `config.json` (next to this script) or set the SMTP
credentials via env vars: SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASSWORD,
EMAIL_FROM, EMAIL_TO.
"""

from __future__ import annotations

import json
import os
import smtplib
import ssl
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from email.message import EmailMessage
from pathlib import Path

ROOT = Path(__file__).resolve().parent
CONFIG_PATH = ROOT / "config.json"
STATE_PATH = ROOT / "state.json"
LOG_PATH = ROOT / "watcher.log"

DEFAULT_CONFIG = {
    # If empty, auto-discover all official benchmarks. Otherwise pin a list.
    "benchmarks": [],

    # SMTP settings — override with env vars (SMTP_HOST etc.) if you prefer.
    # For Gmail: host=smtp.gmail.com, port=465, user=you@gmail.com,
    #            password=<App Password from https://myaccount.google.com/apppasswords>.
    "smtp_host": "smtp.gmail.com",
    "smtp_port": 465,
    "smtp_user": "",
    "smtp_password": "",
    "email_from": "",
    "email_to": "",

    # Don't send an email if there are zero new entries.
    "skip_email_when_empty": True,

    # Path to your HF token (used for gated benchmarks). Optional.
    "hf_token_path": "~/.cache/huggingface/token",

    # First run only seeds state silently — no flood email.
    "silent_first_run": True,

    "timeout": 30,
    "request_delay_seconds": 0.3,
}


# ---------- utilities ----------

def log(msg: str) -> None:
    line = f"[{datetime.now(timezone.utc).isoformat(timespec='seconds')}] {msg}"
    print(line, flush=True)
    try:
        with LOG_PATH.open("a") as f:
            f.write(line + "\n")
    except OSError:
        pass


def load_json(path: Path, default):
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text())
    except json.JSONDecodeError:
        log(f"WARN: corrupt JSON at {path}, ignoring")
        return default


def save_json(path: Path, data) -> None:
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, indent=2, sort_keys=True))
    tmp.replace(path)


def load_config() -> dict:
    if not CONFIG_PATH.exists():
        save_json(CONFIG_PATH, DEFAULT_CONFIG)
    cfg = dict(DEFAULT_CONFIG)
    cfg.update(load_json(CONFIG_PATH, {}))
    # Env var overrides for secrets / portability.
    env_map = {
        "smtp_host": "SMTP_HOST",
        "smtp_port": "SMTP_PORT",
        "smtp_user": "SMTP_USER",
        "smtp_password": "SMTP_PASSWORD",
        "email_from": "EMAIL_FROM",
        "email_to": "EMAIL_TO",
    }
    for key, env in env_map.items():
        if os.environ.get(env):
            cfg[key] = os.environ[env]
    cfg["smtp_port"] = int(cfg["smtp_port"])
    return cfg


def read_token(cfg: dict) -> str | None:
    p = cfg.get("hf_token_path", "")
    if p:
        path = Path(os.path.expanduser(p))
        if path.exists():
            tok = path.read_text().strip()
            if tok:
                return tok
    return os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")


def http_get_json(url: str, token: str | None, timeout: int):
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read().decode("utf-8"))


# ---------- HF API ----------

def discover_official_benchmarks(token, timeout):
    url = "https://huggingface.co/api/datasets?filter=benchmark:official&limit=1000"
    data = http_get_json(url, token, timeout)
    return [d["id"] for d in data if isinstance(d, dict) and "id" in d]


def get_leaderboard(dataset_id, token, timeout):
    url = f"https://huggingface.co/api/datasets/{dataset_id}/leaderboard"
    data = http_get_json(url, token, timeout)
    if isinstance(data, dict) and "entries" in data:
        data = data["entries"]
    return data if isinstance(data, list) else []


# ---------- email ----------

def render_digest(new_by_ds: dict[str, list[dict]]) -> tuple[str, str]:
    """Return (plain_text, html) bodies."""
    today = datetime.now().strftime("%Y-%m-%d")
    total = sum(len(v) for v in new_by_ds.values())

    # Plain text
    lines = [f"HF Leaderboard digest — {today}", f"{total} new model entr{'y' if total == 1 else 'ies'} across {len(new_by_ds)} benchmark(s).", ""]
    for ds, entries in sorted(new_by_ds.items()):
        lines.append(f"### {ds}  ({len(entries)} new)")
        lines.append(f"https://huggingface.co/datasets/{ds}")
        for e in entries:
            mid = e.get("modelId") or e.get("model_id") or "?"
            rank = e.get("rank")
            val = e.get("value")
            verified = e.get("verified")
            bits = []
            if rank is not None:
                bits.append(f"#{rank}")
            bits.append(mid)
            if val is not None:
                bits.append(f"score={val}")
            if verified:
                bits.append("✓verified")
            lines.append("  - " + "  ".join(bits))
        lines.append("")
    text_body = "\n".join(lines)

    # HTML
    rows = []
    for ds, entries in sorted(new_by_ds.items()):
        ds_url = f"https://huggingface.co/datasets/{ds}"
        rows.append(f'<h3 style="margin:18px 0 6px"><a href="{ds_url}">{ds}</a> '
                    f'<span style="color:#666;font-weight:normal">({len(entries)} new)</span></h3>')
        rows.append('<table style="border-collapse:collapse;width:100%;font-size:14px">')
        rows.append('<thead><tr style="background:#f4f4f5;text-align:left">'
                    '<th style="padding:6px 8px">Rank</th>'
                    '<th style="padding:6px 8px">Model</th>'
                    '<th style="padding:6px 8px">Score</th>'
                    '<th style="padding:6px 8px">Verified</th></tr></thead><tbody>')
        for e in entries:
            mid = e.get("modelId") or e.get("model_id") or "?"
            mid_url = f"https://huggingface.co/{mid}"
            rank = e.get("rank", "")
            val = e.get("value", "")
            verified = "✓" if e.get("verified") else ""
            rows.append(f'<tr style="border-top:1px solid #eee">'
                        f'<td style="padding:6px 8px">#{rank}</td>'
                        f'<td style="padding:6px 8px"><a href="{mid_url}">{mid}</a></td>'
                        f'<td style="padding:6px 8px">{val}</td>'
                        f'<td style="padding:6px 8px">{verified}</td></tr>')
        rows.append("</tbody></table>")

    html_body = (
        f'<div style="font-family:-apple-system,BlinkMacSystemFont,sans-serif;max-width:760px">'
        f'<h2 style="margin-bottom:4px">HF Leaderboard digest — {today}</h2>'
        f'<p style="color:#555;margin-top:0">'
        f'{total} new model entr{"y" if total == 1 else "ies"} across {len(new_by_ds)} benchmark(s).</p>'
        + "\n".join(rows) +
        '<p style="color:#888;font-size:12px;margin-top:24px">'
        'Source: <a href="https://huggingface.co/docs/hub/leaderboard-data-guide">HF leaderboard API</a>. '
        'Sent by ~/leaderboard-watcher/watch.py.</p></div>'
    )
    return text_body, html_body


def send_email(cfg: dict, subject: str, text_body: str, html_body: str) -> None:
    user = cfg["smtp_user"]
    pwd = cfg["smtp_password"]
    sender = cfg["email_from"] or user
    recipient = cfg["email_to"]
    if not (user and pwd and recipient):
        log("ERROR: SMTP credentials or recipient missing — set them in config.json or env vars (SMTP_USER, SMTP_PASSWORD, EMAIL_TO).")
        return

    msg = EmailMessage()
    msg["From"] = sender
    msg["To"] = recipient
    msg["Subject"] = subject
    msg.set_content(text_body)
    msg.add_alternative(html_body, subtype="html")

    host = cfg["smtp_host"]
    port = cfg["smtp_port"]
    ctx = ssl.create_default_context()
    try:
        if port == 465:
            with smtplib.SMTP_SSL(host, port, context=ctx, timeout=30) as s:
                s.login(user, pwd)
                s.send_message(msg)
        else:
            with smtplib.SMTP(host, port, timeout=30) as s:
                s.starttls(context=ctx)
                s.login(user, pwd)
                s.send_message(msg)
        log(f"Email sent to {recipient} via {host}:{port}.")
    except Exception as e:
        log(f"ERROR sending email: {e}")


# ---------- state helpers ----------

def migrate_seen(seen_data) -> dict:
    """Convert legacy list-of-ids to {model_id: first_seen_iso | None} dict."""
    if isinstance(seen_data, list):
        return {m: None for m in seen_data}
    return seen_data


def diff_models(seen: dict, current_ids: list, now: datetime) -> tuple[list, dict]:
    """Return (new_ids, updated_seen_dict) given current leaderboard snapshot.

    A model is only reported as "new" if it has appeared in at least 2 consecutive
    runs. First-time appearances are recorded in state with a ``pending`` sentinel
    and promoted on the next sighting. This guards against flaky external
    leaderboards that intermittently drop and restore model entries.
    """
    now_iso = now.isoformat(timespec="seconds")
    pending_marker = "pending"
    current_set = set(current_ids)
    new_ids: list[str] = []
    updated = dict(seen)

    # Handle pending models from the previous run that are no longer in
    # the API. They stay in state so they won't be re-reported later,
    # but we demote the pending marker to None so they don't persist.
    for m, first_seen in list(updated.items()):
        if first_seen == pending_marker and m not in current_set:
            updated[m] = None  # treat as seeded, not new

    for m in current_ids:
        if m not in updated:
            # First ever sighting — mark as pending, don't report yet.
            updated[m] = pending_marker
        elif updated[m] == pending_marker:
            # Seen in the previous run too — promote to confirmed new.
            updated[m] = now_iso
            new_ids.append(m)
        # else: already confirmed (has a real first_seen date) — no change.
    return new_ids, updated


# ---------- core ----------

def run_once() -> int:
    cfg = load_config()
    token = read_token(cfg)
    first_run_overall = not STATE_PATH.exists()
    state = load_json(STATE_PATH, {})

    benchmarks = cfg.get("benchmarks") or []
    if not benchmarks:
        log("Discovering official benchmark datasets...")
        try:
            benchmarks = discover_official_benchmarks(token, cfg["timeout"])
        except Exception as e:
            log(f"ERROR discovering benchmarks: {e}")
            return 1
    log(f"Checking {len(benchmarks)} benchmark(s).")

    new_by_ds: dict[str, list[dict]] = {}

    for ds in benchmarks:
        try:
            entries = get_leaderboard(ds, token, cfg["timeout"])
        except urllib.error.HTTPError as e:
            log(f"  {ds}: HTTP {e.code} ({e.reason}) — skipping")
            continue
        except Exception as e:
            log(f"  {ds}: error {e} — skipping")
            continue

        current_ids = []
        details: dict[str, dict] = {}
        for entry in entries:
            mid = entry.get("modelId") or entry.get("model_id") or entry.get("model")
            if not mid:
                continue
            current_ids.append(mid)
            details[mid] = entry

        ds_state = state.get(ds, {})
        seen = migrate_seen(ds_state.get("seen", {}))
        first_time_for_ds = ds not in state
        now = datetime.now(timezone.utc)
        new_ids, updated_seen = diff_models(seen, current_ids, now)

        state[ds] = {
            "seen": updated_seen,
            "last_checked": now.isoformat(timespec="seconds"),
            "count": len(current_ids),
        }

        if not new_ids:
            log(f"  {ds}: no new models ({len(current_ids)} total)")
            continue

        log(f"  {ds}: {len(new_ids)} new (of {len(current_ids)} total)")
        for m in new_ids[:10]:
            e = details.get(m, {})
            rank = e.get("rank")
            val = e.get("value")
            first_seen = updated_seen.get(m)
            log(f"      + #{rank} {m} score={val}  first_seen={first_seen}")
        if len(new_ids) > 10:
            log(f"      + … and {len(new_ids) - 10} more")
        silent = (first_run_overall or first_time_for_ds) and cfg["silent_first_run"]
        if not silent:
            new_by_ds[ds] = [details[m] for m in new_ids]

        time.sleep(cfg["request_delay_seconds"])

    save_json(STATE_PATH, state)

    total_new = sum(len(v) for v in new_by_ds.values())
    if total_new == 0:
        log("No new models to report.")
        if cfg["skip_email_when_empty"]:
            return 0
        text_body = f"No new models on any of the {len(benchmarks)} watched benchmarks today."
        html_body = f"<p>{text_body}</p>"
        send_email(cfg, "HF Leaderboard digest — no new models", text_body, html_body)
        return 0

    subject = f"HF Leaderboard digest — {total_new} new model entr{'y' if total_new == 1 else 'ies'}"
    text_body, html_body = render_digest(new_by_ds)
    log(f"Sending digest: {total_new} new entries across {len(new_by_ds)} benchmark(s).")
    send_email(cfg, subject, text_body, html_body)
    return 0


def main(argv):
    if "--list" in argv:
        cfg = load_config()
        for b in discover_official_benchmarks(read_token(cfg), cfg["timeout"]):
            print(b)
        return 0
    if "--reset" in argv:
        if STATE_PATH.exists():
            STATE_PATH.unlink()
            print(f"Removed {STATE_PATH}")
        return 0
    if "--test-email" in argv:
        cfg = load_config()
        send_email(cfg, "HF Leaderboard watcher — test email",
                   "If you can read this, SMTP works. 🎉",
                   "<p>If you can read this, SMTP works. 🎉</p>")
        return 0
    return run_once()


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))