climperor/_gitea_dispatch_notify.py

"""Push climperor and dispatch site-traffic-notify via Gitea API.

Env: KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN_GITEA_1 (or GITEA_TOKEN),
optional KEYZOO_ASSET_META_URL.
Never prints the token.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import urllib.error
import urllib.request

OWNER = "refining"
REPO = "climperor"
WORKFLOW_FILE = "site-traffic-notify.yml"


def token() -> str:
    for name in (
        "KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN_GITEA_1",
        "KEYZOO_ASSET_SECRET_PERSONAL_ACCESS_TOKEN__GITEA_1",
        "KEYZOO_ASSET_TOKEN",
        "GITEA_TOKEN",
    ):
        v = os.environ.get(name)
        if v:
            return v
    raise SystemExit("missing Gitea token")


def base() -> str:
    return (
        os.environ.get("GITEA_URL")
        or os.environ.get("KEYZOO_ASSET_META_URL")
        or "https://gitea.refining.dev"
    ).rstrip("/")


def api(method: str, path: str, body: dict | None = None) -> tuple[int, dict | list | str]:
    url = base() + path
    data = None if body is None else json.dumps(body).encode()
    req = urllib.request.Request(
        url,
        data=data,
        method=method,
        headers={
            "Authorization": f"token {token()}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as r:
            raw = r.read().decode()
            if not raw:
                return r.status, {}
            try:
                return r.status, json.loads(raw)
            except json.JSONDecodeError:
                return r.status, raw
    except urllib.error.HTTPError as e:
        raw = e.read().decode(errors="replace")
        try:
            return e.code, json.loads(raw)
        except json.JSONDecodeError:
            return e.code, raw


def push() -> None:
    # Prefer HTTPS with token in URL for non-interactive push.
    tok = token()
    remote = f"https://oauth2:{tok}@gitea.refining.dev/{OWNER}/{REPO}.git"
    env = os.environ.copy()
    # Avoid leaking via git trace
    env.pop("GIT_TRACE", None)
    r = subprocess.run(
        ["git", "push", remote, "HEAD:main"],
        cwd=os.path.dirname(os.path.abspath(__file__)) or ".",
        capture_output=True,
        text=True,
        env=env,
        timeout=180,
    )
    # Redact token if it somehow appears
    out = (r.stdout or "") + (r.stderr or "")
    out = out.replace(tok, "***")
    print(out)
    if r.returncode != 0:
        raise SystemExit(f"git push failed: {r.returncode}")


def dispatch(day: str = "") -> None:
    # Gitea: POST /repos/{owner}/{repo}/actions/workflows/{workflow_id}/dispatches
    # workflow_id can be filename
    body: dict = {"ref": "main"}
    if day:
        body["inputs"] = {"day": day}
    code, resp = api(
        "POST",
        f"/api/v1/repos/{OWNER}/{REPO}/actions/workflows/{WORKFLOW_FILE}/dispatches",
        body,
    )
    print(f"dispatch HTTP {code}: {resp}")
    if code not in (200, 201, 204):
        # Fallback: list workflows to find id
        c2, workflows = api("GET", f"/api/v1/repos/{OWNER}/{REPO}/actions/workflows")
        print(f"workflows HTTP {c2}: {workflows}")
        raise SystemExit(1)


def latest_runs() -> None:
    code, resp = api("GET", f"/api/v1/repos/{OWNER}/{REPO}/actions/runs?limit=5")
    print(f"runs HTTP {code}")
    if isinstance(resp, dict):
        rows = resp.get("workflow_runs") or resp.get("runs") or []
    elif isinstance(resp, list):
        rows = resp
    else:
        print(resp)
        return
    for r in rows[:5]:
        if isinstance(r, dict):
            print(
                f"  - id={r.get('id')} name={r.get('name') or r.get('workflow_id')} "
                f"status={r.get('status')} conclusion={r.get('conclusion')} "
                f"event={r.get('event')} created={r.get('created_at')}"
            )


def main() -> int:
    cmd = sys.argv[1] if len(sys.argv) > 1 else "all"
    if cmd == "push":
        push()
    elif cmd == "dispatch":
        day = sys.argv[2] if len(sys.argv) > 2 else ""
        dispatch(day)
    elif cmd == "runs":
        latest_runs()
    elif cmd == "all":
        push()
        dispatch("2026-07-27")
        latest_runs()
    else:
        raise SystemExit("usage: push | dispatch [day] | runs | all")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())