kaizoku/provider_updates.py

#!/usr/bin/env python3

"""Check bundled provider JavaScript against the upstream extensions repo."""

import hashlib
import json
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

from app_support import AGENT, PROJECT_ROOT


UPSTREAM_REPO_URL = "https://github.com/TheYogMehta/extensions"
UPSTREAM_BRANCH = "main"
UPSTREAM_TREE_URL = (
    f"https://api.github.com/repos/TheYogMehta/extensions/git/trees/{UPSTREAM_BRANCH}?recursive=1"
)
UPSTREAM_RAW_BASE = f"https://raw.githubusercontent.com/TheYogMehta/extensions/{UPSTREAM_BRANCH}/"
LOCAL_PROVIDER_ROOT = PROJECT_ROOT / "providers" / "extensions"
UPSTREAM_PROVIDER_PREFIX = "extensions/"


def now_iso():
    return datetime.now(timezone.utc).isoformat()


def default_status(status="idle", message="Provider update check has not run yet."):
    return {
        "status": status,
        "message": message,
        "checked_at": "",
        "upstream_url": UPSTREAM_REPO_URL,
        "upstream_branch": UPSTREAM_BRANCH,
        "upstream_tree_sha": "",
        "checked_files": 0,
        "updated_files": [],
        "missing_upstream_files": [],
    }


def local_provider_js_files(root=LOCAL_PROVIDER_ROOT):
    root = Path(root)
    if not root.exists():
        return {}
    files = {}
    for path in sorted(root.rglob("*.js")):
        if path.is_file():
            files[path.relative_to(root).as_posix()] = path
    return files


def sha256_bytes(content):
    return hashlib.sha256(content).hexdigest()


def fetch_json(url, timeout=10):
    request = urllib.request.Request(
        url,
        headers={
            "Accept": "application/vnd.github+json",
            "User-Agent": AGENT,
        },
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return json.loads(response.read().decode("utf-8"))


def fetch_bytes(url, timeout=10):
    request = urllib.request.Request(
        url,
        headers={
            "Accept": "application/octet-stream",
            "User-Agent": AGENT,
        },
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return response.read()


def upstream_provider_js_paths(timeout=10):
    payload = fetch_json(UPSTREAM_TREE_URL, timeout=timeout)
    tree = payload.get("tree") if isinstance(payload, dict) else []
    paths = set()
    for item in tree or []:
        path = str((item or {}).get("path") or "")
        if (item or {}).get("type") == "blob" and path.startswith(UPSTREAM_PROVIDER_PREFIX) and path.endswith(".js"):
            paths.add(path[len(UPSTREAM_PROVIDER_PREFIX) :])
    return paths, str(payload.get("sha") or "") if isinstance(payload, dict) else ""


def check_provider_updates(local_root=LOCAL_PROVIDER_ROOT, timeout=10):
    checked_at = now_iso()
    local_files = local_provider_js_files(local_root)
    if not local_files:
        status = default_status("error", "No local provider JavaScript files were found.")
        status["checked_at"] = checked_at
        return status

    upstream_paths, upstream_tree_sha = upstream_provider_js_paths(timeout=timeout)
    updated_files = []
    missing_upstream_files = []
    checked_files = 0

    for relative_path, local_path in local_files.items():
        if relative_path not in upstream_paths:
            missing_upstream_files.append(relative_path)
            continue
        checked_files += 1
        local_content = local_path.read_bytes()
        remote_url = f"{UPSTREAM_RAW_BASE}{UPSTREAM_PROVIDER_PREFIX}{relative_path}"
        remote_content = fetch_bytes(remote_url, timeout=timeout)
        local_hash = sha256_bytes(local_content)
        remote_hash = sha256_bytes(remote_content)
        if local_hash != remote_hash:
            updated_files.append(
                {
                    "path": relative_path,
                    "local_sha256": local_hash,
                    "upstream_sha256": remote_hash,
                    "upstream_url": remote_url,
                }
            )

    has_updates = bool(updated_files or missing_upstream_files)
    if has_updates:
        message = "Provider updates are available from TheYogMehta/extensions."
        status_name = "updates_available"
    else:
        message = "Bundled provider JavaScript is up to date."
        status_name = "up_to_date"

    return {
        "status": status_name,
        "message": message,
        "checked_at": checked_at,
        "upstream_url": UPSTREAM_REPO_URL,
        "upstream_branch": UPSTREAM_BRANCH,
        "upstream_tree_sha": upstream_tree_sha,
        "checked_files": checked_files,
        "updated_files": updated_files,
        "missing_upstream_files": missing_upstream_files,
    }


def error_status(exc):
    message = str(exc)
    if isinstance(exc, urllib.error.URLError) and getattr(exc, "reason", None):
        message = str(exc.reason)
    status = default_status("error", f"Provider update check failed: {message}")
    status["checked_at"] = now_iso()
    return status