kaizoku/provider_updates.py

#!/usr/bin/env python3

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

import hashlib
import json
import re
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": [],
        "locally_modified_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 provider_module_version(content):
    match = re.search(rb'version:\s*["\']([^"\']+)["\']', content)
    if not match:
        return ""
    return match.group(1).decode("utf-8", errors="replace")


def compare_versions(left, right):
    left_parts = [int(part) for part in re.findall(r"\d+", left or "")]
    right_parts = [int(part) for part in re.findall(r"\d+", right or "")]
    length = max(len(left_parts), len(right_parts))
    for index in range(length):
        left_value = left_parts[index] if index < len(left_parts) else 0
        right_value = right_parts[index] if index < len(right_parts) else 0
        if left_value != right_value:
            return 1 if left_value > right_value else -1
    return 0


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 = []
    locally_modified_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:
            local_version = provider_module_version(local_content)
            upstream_version = provider_module_version(remote_content)
            item = {
                "path": relative_path,
                "local_sha256": local_hash,
                "upstream_sha256": remote_hash,
                "upstream_url": remote_url,
                "local_version": local_version,
                "upstream_version": upstream_version,
            }
            if local_version and upstream_version and compare_versions(local_version, upstream_version) > 0:
                locally_modified_files.append(item)
            else:
                updated_files.append(item)

    if updated_files or missing_upstream_files:
        message = "Provider updates are available from TheYogMehta/extensions."
        status_name = "updates_available"
    elif locally_modified_files:
        message = "Bundled provider JavaScript includes local changes ahead of upstream."
        status_name = "local_changes"
    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,
        "locally_modified_files": locally_modified_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