github_ops.py

#!/usr/bin/env python3
"""
GitHub Operations Helper for Sales Ops
============================================

Provides GitHub API operations via Python (requests) or CLI (gh/curl).
Loads token from .env file automatically.

Usage:
    python scripts/github_ops.py list-branches
    python scripts/github_ops.py create-branch feature/add-new-leads
    python scripts/github_ops.py create-pr --title "Add new leads" --body "Added 10 new GA leads" --head feature/add-new-leads --base test
    python scripts/github_ops.py delete-branch feature/add-new-leads
    python scripts/github_ops.py list-prs
    python scripts/github_ops.py protect-branch main
    python scripts/github_ops.py setup-environments
"""

import os
import sys
import json
import argparse
import subprocess
from pathlib import Path

try:
    import requests
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)

# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

def load_env():
    """Load .env file from project root."""
    env_path = Path(__file__).resolve().parent.parent / ".env"
    if env_path.exists():
        for line in env_path.read_text().splitlines():
            line = line.strip()
            if line and not line.startswith("#") and "=" in line:
                key, _, value = line.partition("=")
                os.environ.setdefault(key.strip(), value.strip())

load_env()

GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
GITHUB_OWNER = os.environ.get("GITHUB_OWNER", "your-org")
GITHUB_REPO = os.environ.get("GITHUB_REPO", "gcs")
API_BASE = f"https://api.github.com/repos/{GITHUB_OWNER}/{GITHUB_REPO}"

HEADERS = {
    "Authorization": f"Bearer {GITHUB_TOKEN}",
    "Accept": "application/vnd.github+json",
    "X-GitHub-Api-Version": "2022-11-28",
}

# ---------------------------------------------------------------------------
# API Helpers
# ---------------------------------------------------------------------------

def api_get(endpoint):
    """GET request to GitHub API."""
    r = requests.get(f"{API_BASE}{endpoint}", headers=HEADERS)
    r.raise_for_status()
    return r.json()

def api_post(endpoint, data):
    """POST request to GitHub API."""
    r = requests.post(f"{API_BASE}{endpoint}", headers=HEADERS, json=data)
    if not r.ok:
        print(f"Error {r.status_code}: {r.text}")
    r.raise_for_status()
    return r.json()

def api_put(endpoint, data=None):
    """PUT request to GitHub API."""
    r = requests.put(f"{API_BASE}{endpoint}", headers=HEADERS, json=data or {})
    r.raise_for_status()
    return r.json() if r.text else {}

def api_delete(endpoint):
    """DELETE request to GitHub API."""
    r = requests.delete(f"{API_BASE}{endpoint}", headers=HEADERS)
    if not r.ok:
        print(f"Error {r.status_code}: {r.text}")
    r.raise_for_status()

def api_patch(endpoint, data):
    """PATCH request to GitHub API."""
    r = requests.patch(f"{API_BASE}{endpoint}", headers=HEADERS, json=data)
    if not r.ok:
        print(f"Error {r.status_code}: {r.text}")
    r.raise_for_status()
    return r.json()

# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------

def list_branches():
    """List all branches."""
    branches = api_get("/branches")
    print(f"\n{'Branch':<40} {'Protected':<10}")
    print("-" * 50)
    for b in branches:
        prot = "Yes" if b["protected"] else "No"
        print(f"  {b['name']:<38} {prot:<10}")
    print(f"\nTotal: {len(branches)} branches")

def create_branch(name, base="test"):
    """Create a new branch from base."""
    # Get the SHA of the base branch
    base_info = api_get(f"/git/ref/heads/{base}")
    sha = base_info["object"]["sha"]
    result = api_post("/git/refs", {"ref": f"refs/heads/{name}", "sha": sha})
    print(f"Created branch '{name}' from '{base}' at {sha[:8]}")
    return result

def delete_branch(name):
    """Delete a remote branch."""
    api_delete(f"/git/refs/heads/{name}")
    print(f"Deleted branch '{name}'")

def list_prs(state="open"):
    """List pull requests."""
    prs = api_get(f"/pulls?state={state}")
    if not prs:
        print(f"\nNo {state} pull requests.")
        return
    print(f"\n{'#':<6} {'Title':<50} {'Head → Base':<30}")
    print("-" * 86)
    for pr in prs:
        arrow = f"{pr['head']['ref']} → {pr['base']['ref']}"
        print(f"  #{pr['number']:<4} {pr['title'][:48]:<50} {arrow:<30}")
    print(f"\nTotal: {len(prs)} {state} PRs")

def create_pr(title, body, head, base="test"):
    """Create a pull request."""
    result = api_post("/pulls", {
        "title": title,
        "body": body,
        "head": head,
        "base": base,
    })
    print(f"Created PR #{result['number']}: {result['html_url']}")
    return result

def protect_branch(name, require_pr=True, required_approvals=1):
    """Set branch protection rules."""
    protection = {
        "required_status_checks": None,
        "enforce_admins": True,
        "required_pull_request_reviews": {
            "required_approving_review_count": required_approvals,
            "dismiss_stale_reviews": True,
        } if require_pr else None,
        "restrictions": None,
    }
    try:
        api_put(f"/branches/{name}/protection", protection)
        print(f"Protected branch '{name}' (require PR: {require_pr}, approvals: {required_approvals})")
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 403:
            print(f"  Skipped protection for '{name}' — requires GitHub Pro or public repo")
            print(f"  To fix: make the repo public, or upgrade to GitHub Pro ($4/mo)")
        else:
            raise

def setup_environments():
    """Set up the recommended branch structure: main, staging, test."""
    print("\n Setting up environment branches...\n")

    # Get main branch SHA
    try:
        main_info = api_get("/git/ref/heads/main")
        sha = main_info["object"]["sha"]
    except Exception:
        print("Error: 'main' branch not found.")
        return

    # Create staging and test if they don't exist
    existing = [b["name"] for b in api_get("/branches")]

    for branch in ["staging", "test"]:
        if branch in existing:
            print(f"  '{branch}' already exists — skipping")
        else:
            try:
                api_post("/git/refs", {"ref": f"refs/heads/{branch}", "sha": sha})
                print(f"  Created '{branch}' from main at {sha[:8]}")
            except requests.exceptions.HTTPError as e:
                if e.response.status_code == 422:
                    print(f"  '{branch}' already exists — skipping")
                else:
                    raise

    # Protect branches
    print("\n Setting up branch protection...\n")
    print("  Note: Branch protection requires GitHub Pro or a public repo.")
    print("  Attempting to set protection rules...\n")

    # main: strictest — require PR + 1 approval
    protect_branch("main", require_pr=True, required_approvals=1)

    # staging: require PR, no approval needed
    protection_staging = {
        "required_status_checks": None,
        "enforce_admins": True,
        "required_pull_request_reviews": {
            "required_approving_review_count": 0,
            "dismiss_stale_reviews": True,
        },
        "restrictions": None,
    }
    try:
        api_put("/branches/staging/protection", protection_staging)
        print("  Protected 'staging' (require PR, 0 approvals)")
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 403:
            print("  Skipped protection for 'staging' — requires GitHub Pro or public repo")
        else:
            raise

    # test: lightest protection
    protection_test = {
        "required_status_checks": None,
        "enforce_admins": False,
        "required_pull_request_reviews": None,
        "restrictions": None,
    }
    try:
        api_put("/branches/test/protection", protection_test)
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 403:
            print("  Skipped protection for 'test' — requires GitHub Pro or public repo")
        else:
            raise
    print("  Protected 'test' (lightweight)")

    print("\n Environment setup complete!")
    print("""
    Branch Flow:
    ┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ feature/ │ -> │   test   │ -> │ staging  │ -> │   main   │
    │  fix/    │    │   (QA)   │    │(pre-prod)│    │  (prod)  │
    │  docs/   │    └──────────┘    └──────────┘    └──────────┘
    └──────────┘
    """)

def enable_auto_delete():
    """Enable auto-delete of branches after PR merge (repo setting)."""
    api_patch("", {"delete_branch_on_merge": True})
    print("Enabled auto-delete of branches after PR merge.")

# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(description="GitHub Operations for Sales Ops")
    sub = parser.add_subparsers(dest="command")

    sub.add_parser("list-branches", help="List all branches")
    sub.add_parser("list-prs", help="List open pull requests")

    cb = sub.add_parser("create-branch", help="Create a new branch")
    cb.add_argument("name", help="Branch name (e.g., feature/add-leads)")
    cb.add_argument("--base", default="test", help="Base branch (default: test)")

    db = sub.add_parser("delete-branch", help="Delete a remote branch")
    db.add_argument("name", help="Branch name to delete")

    pr = sub.add_parser("create-pr", help="Create a pull request")
    pr.add_argument("--title", required=True)
    pr.add_argument("--body", default="")
    pr.add_argument("--head", required=True, help="Source branch")
    pr.add_argument("--base", default="test", help="Target branch (default: test)")

    pb = sub.add_parser("protect-branch", help="Protect a branch")
    pb.add_argument("name", help="Branch name")
    pb.add_argument("--approvals", type=int, default=1)

    sub.add_parser("setup-environments", help="Create staging/test branches + protection rules")
    sub.add_parser("auto-delete", help="Enable auto-delete of merged branches")

    args = parser.parse_args()

    if not GITHUB_TOKEN:
        print("Error: GITHUB_TOKEN not found. Set it in .env or environment.")
        sys.exit(1)

    commands = {
        "list-branches": list_branches,
        "list-prs": list_prs,
        "create-branch": lambda: create_branch(args.name, args.base),
        "delete-branch": lambda: delete_branch(args.name),
        "create-pr": lambda: create_pr(args.title, args.body, args.head, args.base),
        "protect-branch": lambda: protect_branch(args.name, required_approvals=args.approvals),
        "setup-environments": setup_environments,
        "auto-delete": enable_auto_delete,
    }

    if args.command in commands:
        commands[args.command]()
    else:
        parser.print_help()

if __name__ == "__main__":
    main()