README.md

cd ~/Documents/auto-approve

Preview (approves nothing):

python3 auto_approve.py --once --dry-run

Live, background, every 60s:

nohup python3 auto_approve.py > ~/.auto-approve/run.log 2>&1 &

Stop:

pkill -f auto_approve.py

auto-approve.sh

#!/usr/bin/env bash
#
# auto-approve.sh — Background service that auto-approves GitHub PRs
# where you are the requested reviewer, then sends a desktop notification
# so you can open and check each one yourself.
#
# Approvals are posted via your authenticated `gh` account, so they are
# genuinely *you* approving (not a bot or a separate identity).
#
# Usage:
#   ./auto-approve.sh                 # poll every 60s across all repos
#   POLL_INTERVAL=30 ./auto-approve.sh
#   REPO=opswat-eng/mdd-drive-smc ./auto-approve.sh   # limit to one repo
#   APPROVE_BODY="" ./auto-approve.sh                 # plain approval, no comment
#   DRY_RUN=1 ./auto-approve.sh       # show what it WOULD do, approve nothing
#
# Run in background:
#   nohup ./auto-approve.sh > ~/.auto-approve/run.log 2>&1 &
# Stop:
#   pkill -f auto-approve.sh

set -euo pipefail

# ---- Config (override via env) ----------------------------------------------
POLL_INTERVAL="${POLL_INTERVAL:-60}"          # seconds between polls
REPO="${REPO:-}"                              # optional: owner/repo to restrict to
APPROVE_BODY="${APPROVE_BODY:-}"              # review comment; empty = plain approve
DRY_RUN="${DRY_RUN:-0}"                       # 1 = don't actually approve
STATE_DIR="${STATE_DIR:-$HOME/.auto-approve}"
STATE_FILE="$STATE_DIR/approved.txt"          # PR URLs we've already approved
# -----------------------------------------------------------------------------

mkdir -p "$STATE_DIR"
touch "$STATE_FILE"

log() { printf '%s %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*"; }

notify() {
  local title="$1" body="$2"
  if command -v notify-send >/dev/null 2>&1; then
    # -t 0 keeps it until dismissed; drop if you prefer auto-expire
    DISPLAY="${DISPLAY:-:0}" notify-send -u normal -a "PR Auto-Approve" "$title" "$body" || true
  fi
}

# Preflight: gh must be installed and authenticated
if ! command -v gh >/dev/null 2>&1; then
  log "ERROR: gh CLI not found on PATH"; exit 1
fi
# Note: `gh auth status` can exit non-zero on keyring-backed setups even when
# authenticated, so we probe with an actual API call instead.
if ! ME="$(gh api user --jq .login 2>/dev/null)" || [[ -z "$ME" ]]; then
  log "ERROR: gh is not authenticated. Run: gh auth login"; exit 1
fi
log "Started. Approving as '$ME'. Interval=${POLL_INTERVAL}s Repo=${REPO:-<all>} DryRun=${DRY_RUN}"

already_approved() { grep -Fxq "$1" "$STATE_FILE"; }

approve_pr() {
  local url="$1" title="$2" author="$3" repo="$4"
  # Diff stats aren't available from `gh search`; fetch them per new PR.
  local adds="?" dels="?" files="?"
  local stats
  if stats="$(gh pr view "$url" --json additions,deletions,changedFiles \
              --jq '[.additions,.deletions,.changedFiles]|@tsv' 2>/dev/null)"; then
    IFS=$'\t' read -r adds dels files <<<"$stats"
  fi
  if [[ "$DRY_RUN" == "1" ]]; then
    log "[dry-run] would approve: $repo — $title ($url)"
    return   # never persist or notify during a dry run
  fi
  if gh pr review "$url" --approve ${APPROVE_BODY:+--body "$APPROVE_BODY"} >/dev/null 2>&1; then
    log "APPROVED: $repo — $title ($url)"
  else
    log "FAILED to approve (maybe already reviewed / no permission): $url"
    # still record so we don't retry-spam every poll
  fi
  echo "$url" >> "$STATE_FILE"
  notify "✅ PR approved — $repo" \
         "$title
by $author  ·  +$adds −$dels in $files file(s)
$url"
}

poll_once() {
  local search_args=(--review-requested=@me --state=open --limit 50
                     --json title,url,author,repository)
  [[ -n "$REPO" ]] && search_args+=(--repo "$REPO")

  local json
  if ! json="$(gh search prs "${search_args[@]}" 2>/dev/null)"; then
    log "WARN: gh search failed this cycle; will retry"
    return 0
  fi

  # Iterate PRs. Tab-separated to survive spaces in titles.
  while IFS=$'\t' read -r url title author repo; do
    [[ -z "$url" ]] && continue
    already_approved "$url" && continue
    approve_pr "$url" "$title" "$author" "$repo"
  done < <(printf '%s' "$json" | gh_jq)
}

# jq helper kept separate so the field list lives in one place
gh_jq() {
  jq -r '.[] | [.url, .title, .author.login, .repository.nameWithOwner] | @tsv'
}

trap 'log "Stopping."; exit 0' INT TERM

while true; do
  poll_once
  sleep "$POLL_INTERVAL"
done

auto_approve.py

#!/usr/bin/env python3
"""
auto_approve.py — Portable background service that auto-approves GitHub PRs
where you are the requested reviewer, then shows a desktop notification so you
can open and check each one yourself.

Approvals are posted under YOUR GitHub account (via your token), so they are
genuinely you approving — not a separate bot identity.

Dependencies: Python 3.7+ only (standard library). No pip install needed.

Auth: uses $GITHUB_TOKEN if set, otherwise falls back to `gh auth token`.
      Token needs the `repo` scope (or fine-grained pull_requests: write).

Examples:
    python3 auto_approve.py                       # poll all repos every 60s
    python3 auto_approve.py --interval 30
    python3 auto_approve.py --repo opswat-eng/mdd-drive-smc
    python3 auto_approve.py --once --dry-run       # one pass, approve nothing
    python3 auto_approve.py --seed                 # record current PRs WITHOUT
                                                   # approving (ignore backlog,
                                                   # only act on future PRs)

Run in background (Linux/macOS):
    nohup python3 auto_approve.py > ~/.auto-approve/run.log 2>&1 &
Stop:
    pkill -f auto_approve.py
"""
from __future__ import annotations

import argparse
import base64
import json
import os
import platform
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Protocol

API = "https://api.github.com"
STATE_DIR = Path(os.environ.get("STATE_DIR", Path.home() / ".auto-approve"))
STATE_FILE = STATE_DIR / "approved.txt"
DEFAULT_INTERVAL_SECONDS = 60
HTTP_TIMEOUT_SECONDS = 30
SUBPROCESS_TIMEOUT_SECONDS = 10
USER_AGENT = "auto-approve-script"

JsonDict = dict[str, Any]


@dataclass(frozen=True)
class PullRequest:
    owner: str
    repo: str
    number: int
    title: str
    author: str
    url: str

    @property
    def repo_full(self) -> str:
        return f"{self.owner}/{self.repo}"


def log(msg: str) -> None:
    print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}", flush=True)


def get_token() -> str:
    tok = os.environ.get("GITHUB_TOKEN")
    if tok:
        return tok.strip()
    if shutil.which("gh"):
        try:
            out = subprocess.run(
                ["gh", "auth", "token"],
                capture_output=True,
                text=True,
                check=True,
                timeout=SUBPROCESS_TIMEOUT_SECONDS,
            )
            if out.stdout.strip():
                return out.stdout.strip()
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
            pass
    log("ERROR: no token. Set $GITHUB_TOKEN or run `gh auth login`.")
    sys.exit(1)


class GitHubClient:
    def __init__(self, token: str):
        self.token = token

    def _api(self, method: str, path: str, body: JsonDict | None = None) -> Any:
        # GitHub returns an object for most endpoints but an array for some
        # (e.g. /pulls/{n}/reviews), so the response type is intentionally Any.
        url = path if path.startswith("http") else API + path
        data = json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Authorization", f"Bearer {self.token}")
        req.add_header("Accept", "application/vnd.github+json")
        req.add_header("X-GitHub-Api-Version", "2022-11-28")
        req.add_header("User-Agent", USER_AGENT)
        with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_SECONDS) as resp:
            raw = resp.read()
            return json.loads(raw) if raw else {}

    def get_login(self) -> str:
        return self._api("GET", "/user")["login"]

    def find_review_requests(self, login: str, repo: str | None) -> list[PullRequest]:
        q = f"is:pr is:open review-requested:{login}"
        if repo:
            q += f" repo:{repo}"
        params = urllib.parse.urlencode({"q": q, "per_page": 50})
        result = self._api("GET", f"/search/issues?{params}")
        items = result.get("items", [])

        prs: list[PullRequest] = []
        for it in items:
            owner, repo_name = it["repository_url"].split("/repos/")[1].split("/", 1)
            prs.append(
                PullRequest(
                    owner=owner,
                    repo=repo_name,
                    number=it["number"],
                    title=it["title"],
                    author=it.get("user", {}).get("login", "?"),
                    url=it["html_url"],
                )
            )
        return prs

    def pr_stats(self, pr: PullRequest) -> str:
        try:
            data = self._api("GET", f"/repos/{pr.owner}/{pr.repo}/pulls/{pr.number}")
            return (
                f"+{data.get('additions', '?')} -{data.get('deletions', '?')} "
                f"in {data.get('changed_files', '?')} file(s)"
            )
        except (urllib.error.URLError, OSError):
            return ""

    def already_approved(self, pr: PullRequest, login: str) -> bool:
        try:
            reviews = self._api(
                "GET", f"/repos/{pr.owner}/{pr.repo}/pulls/{pr.number}/reviews"
            )
        except (urllib.error.URLError, OSError):
            return False

        return any(
            r.get("state") == "APPROVED"
            and r.get("user", {}).get("login") == login
            for r in reviews
        )

    def approve(self, pr: PullRequest, body: str) -> bool:
        payload: JsonDict = {"event": "APPROVE"}
        if body:
            payload["body"] = body
        try:
            self._api(
                "POST", f"/repos/{pr.owner}/{pr.repo}/pulls/{pr.number}/reviews", payload
            )
            return True
        except urllib.error.HTTPError as e:
            log(f"  approve failed ({e.code}): {e.read().decode()[:200]}")
            return False


class Notifier(Protocol):
    def notify(self, title: str, message: str) -> None: ...


class LinuxNotifier:
    def notify(self, title: str, message: str) -> None:
        env = dict(os.environ)
        env.setdefault("DISPLAY", ":0")
        subprocess.run(
            ["notify-send", "-a", "PR Auto-Approve", title, message],
            env=env,
            check=False,
            timeout=SUBPROCESS_TIMEOUT_SECONDS,
        )


class MacNotifier:
    def notify(self, title: str, message: str) -> None:
        script = (
            f'display notification {json.dumps(message)} '
            f'with title {json.dumps(title)}'
        )
        subprocess.run(
            ["osascript", "-e", script],
            check=False,
            timeout=SUBPROCESS_TIMEOUT_SECONDS,
        )


# AppUserModelID of the built-in Windows PowerShell shortcut. Toasts are
# silently dropped unless their notifier uses an AUMID that is registered in
# the Start Menu, and this one ships with Windows — so the toast actually
# renders without us having to install our own shortcut.
WINDOWS_TOAST_APP_ID = (
    "{1AC14E77-02E7-4E5D-B744-2EB1AE5198B7}\\WindowsPowerShell\\v1.0\\powershell.exe"
)

# Title/message are read from the environment (set in WindowsNotifier.notify)
# rather than interpolated into the script, and CreateTextNode handles XML
# escaping — so no value ever needs manual quoting/escaping.
WINDOWS_TOAST_POWERSHELL = (
    "$ErrorActionPreference='Stop'; "
    "$title=$env:AUTO_APPROVE_TOAST_TITLE; "
    "$msg=$env:AUTO_APPROVE_TOAST_MSG; "
    "[void][Windows.UI.Notifications.ToastNotificationManager, "
    "Windows.UI.Notifications, ContentType=WindowsRuntime]; "
    "$tt=[Windows.UI.Notifications.ToastTemplateType]::ToastText02; "
    "$doc=[Windows.UI.Notifications.ToastNotificationManager]::GetTemplateContent($tt); "
    "$texts=$doc.GetElementsByTagName('text'); "
    "[void]$texts.Item(0).AppendChild($doc.CreateTextNode($title)); "
    "[void]$texts.Item(1).AppendChild($doc.CreateTextNode($msg)); "
    "$toast=[Windows.UI.Notifications.ToastNotification]::new($doc); "
    f"$notifier=[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier('{WINDOWS_TOAST_APP_ID}'); "
    "$notifier.Show($toast)"
)


class WindowsNotifier:
    def notify(self, title: str, message: str) -> None:
        encoded = base64.b64encode(
            WINDOWS_TOAST_POWERSHELL.encode("utf-16-le")
        ).decode("ascii")
        env = dict(os.environ)
        env["AUTO_APPROVE_TOAST_TITLE"] = title
        env["AUTO_APPROVE_TOAST_MSG"] = message
        out = subprocess.run(
            [
                "powershell",
                "-NoProfile",
                "-NonInteractive",
                "-EncodedCommand",
                encoded,
            ],
            env=env,
            check=False,
            capture_output=True,
            text=True,
            timeout=SUBPROCESS_TIMEOUT_SECONDS,
        )
        if out.returncode != 0:
            err = (out.stderr or "").strip()
            log(f"[notify] {title} - {message}" + (f" (toast failed: {err[:200]})" if err else ""))


class LogNotifier:
    def notify(self, title: str, message: str) -> None:
        log(f"[notify] {title} - {message}")


def make_notifier() -> Notifier:
    system = platform.system()
    if system == "Linux" and shutil.which("notify-send"):
        return LinuxNotifier()
    if system == "Darwin":
        return MacNotifier()
    if system == "Windows":
        return WindowsNotifier()
    return LogNotifier()


class StateStore:
    def __init__(self, path: Path):
        self.path = path
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.path.touch(exist_ok=True)
        self.seen = self._load()

    def _load(self) -> set[str]:
        return {
            ln.strip()
            for ln in self.path.read_text(encoding="utf-8").splitlines()
            if ln.strip()
        }

    def contains(self, url: str) -> bool:
        return url in self.seen

    def add(self, url: str) -> None:
        # Append-only log: O(1) per approval instead of rewriting the whole
        # file. The in-memory `seen` set guards against duplicate lines, so the
        # log never needs deduping or compaction.
        if url in self.seen:
            return
        self.seen.add(url)
        with self.path.open("a", encoding="utf-8") as f:
            f.write(url + "\n")


class AutoApprover:
    def __init__(
        self,
        github: GitHubClient,
        state: StateStore,
        notifier: Notifier,
        login: str,
        repo: str | None,
        body: str,
        dry_run: bool,
        seed: bool,
    ):
        self.github = github
        self.state = state
        self.notifier = notifier
        self.login = login
        self.repo = repo
        self.body = body
        self.dry_run = dry_run
        self.seed = seed

    def poll_once(self) -> None:
        try:
            prs = self.github.find_review_requests(self.login, self.repo)
        except urllib.error.HTTPError as e:
            log(f"WARN: search failed ({e.code}); will retry next cycle")
            return
        except urllib.error.URLError as e:
            log(f"WARN: network error ({e.reason}); will retry next cycle")
            return

        for pr in prs:
            if self.state.contains(pr.url):
                continue

            if self.seed:
                self.state.add(pr.url)
                log(f"SEEDED (ignored, not approved): {pr.repo_full} - {pr.title}")
                continue

            if self.github.already_approved(pr, self.login):
                self.state.add(pr.url)
                continue

            if self.dry_run:
                log(f"[dry-run] would approve: {pr.repo_full} - {pr.title} ({pr.url})")
                continue

            stats = self.github.pr_stats(pr)
            ok = self.github.approve(pr, self.body)
            self.state.add(pr.url)
            if ok:
                log(f"APPROVED: {pr.repo_full} - {pr.title} ({pr.url})")
                self._safe_notify(
                    f"PR approved - {pr.repo_full}",
                    f"{pr.title}\nby {pr.author}  .  {stats}\n{pr.url}",
                )
            else:
                log(f"NOT approved: {pr.repo_full} - {pr.title} ({pr.url})")

    def _safe_notify(self, title: str, message: str) -> None:
        try:
            self.notifier.notify(title, message)
        except Exception as e:
            log(f"[notify failed: {e}] {title} - {message}")


def main() -> None:
    ap = argparse.ArgumentParser(description="Auto-approve PRs awaiting your review.")
    ap.add_argument(
        "--interval",
        type=int,
        default=int(os.environ.get("POLL_INTERVAL", str(DEFAULT_INTERVAL_SECONDS))),
        help="seconds between polls (default 60)",
    )
    ap.add_argument("--repo", default=os.environ.get("REPO") or None,
                    help="restrict to owner/repo")
    ap.add_argument("--body", default=os.environ.get("APPROVE_BODY", ""),
                    help="review comment (default: none / plain approve)")
    ap.add_argument("--once", action="store_true", help="run a single pass and exit")
    ap.add_argument("--dry-run", action="store_true", help="show actions, approve nothing")
    ap.add_argument("--seed", action="store_true",
                    help="record current matching PRs WITHOUT approving (ignore backlog)")
    args = ap.parse_args()

    mode = "SEED" if args.seed else ("DRY-RUN" if args.dry_run else "LIVE")
    log(f"Starting [{mode}]. Resolving GitHub token and identity...")

    token = get_token()
    github = GitHubClient(token)
    state = StateStore(STATE_FILE)
    notifier = make_notifier()
    login = github.get_login()

    approver = AutoApprover(
        github=github,
        state=state,
        notifier=notifier,
        login=login,
        repo=args.repo,
        body=args.body,
        dry_run=args.dry_run,
        seed=args.seed,
    )

    log(f"Started [{mode}]. Approving as '{login}'. "
        f"interval={args.interval}s repo={args.repo or '<all>'}")

    while True:
        approver.poll_once()
        if args.once or args.seed:
            break
        time.sleep(args.interval)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        log("Stopping.")