github_notes.py

"""Import GitHub PRs and issues into rich notes (thread + file patches)."""

from __future__ import annotations

import json
import os
import shutil
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import notes_store
import workspace as workspace_mod

TOOL_DIR = Path(__file__).resolve().parent


def _notes_path() -> Path:
    return workspace_mod.get().notes_path


def _notes_db_path() -> Path:
    return workspace_mod.get().notes_db_path


def _utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def _gh_executable() -> str | None:
    """Resolve gh binary — PATH first, then common Windows install dirs."""
    found = shutil.which("gh")
    if found:
        return found
    env = (os.environ.get("GH_PATH") or os.environ.get("GITHUB_CLI_PATH") or "").strip().strip('"')
    if env:
        p = Path(env)
        if p.is_file():
            return str(p)
        if (p / "gh.exe").is_file():
            return str(p / "gh.exe")
    for base in (
        os.environ.get("ProgramFiles"),
        os.environ.get("ProgramFiles(x86)"),
        os.environ.get("LOCALAPPDATA"),
    ):
        if not base:
            continue
        for rel in (
            Path("GitHub CLI") / "gh.exe",
            Path("Programs") / "GitHub CLI" / "gh.exe",
        ):
            cand = Path(base) / rel
            if cand.is_file():
                return str(cand)
    return None


def _gh_available() -> bool:
    return _gh_executable() is not None


def _run_gh(args: list[str], *, cwd: Path | None = None) -> tuple[int, str, str]:
    gh = _gh_executable()
    if not gh:
        return 1, "", "gh CLI not available"
    cmd = [gh, *args]
    try:
        proc = subprocess.run(
            cmd,
            cwd=str(cwd) if cwd else None,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=120,
        )
        return proc.returncode, proc.stdout or "", proc.stderr or ""
    except (OSError, subprocess.TimeoutExpired) as exc:
        return 1, "", str(exc)


def status(repo: Path) -> dict[str, Any]:
    ok = _gh_available()
    remote = ""
    if ok:
        code, out, err = _run_gh(["repo", "view", "--json", "nameWithOwner,url"], cwd=repo)
        if code == 0:
            try:
                data = json.loads(out)
                remote = data.get("nameWithOwner") or data.get("url") or ""
            except json.JSONDecodeError:
                remote = out.strip()[:200]
        else:
            return {"ok": False, "gh": True, "error": err.strip() or "gh repo view failed", "repo": str(repo)}
    return {
        "ok": ok,
        "gh": ok,
        "repo": str(repo),
        "remote": remote,
        "hint": None if ok else "Install GitHub CLI (gh) and authenticate (gh auth login)",
    }


def _map_issue(item: dict[str, Any], *, kind: str) -> dict[str, Any]:
    number = item.get("number")
    title = item.get("title") or f"{kind.upper()} #{number}"
    body = item.get("body") or ""
    url = item.get("url") or ""
    state = (item.get("state") or "").lower()
    author = ""
    if isinstance(item.get("author"), dict):
        author = item["author"].get("login") or ""
    elif isinstance(item.get("user"), dict):
        author = item["user"].get("login") or ""
    labels = []
    for lab in item.get("labels") or []:
        if isinstance(lab, dict):
            labels.append(lab.get("name") or "")
        elif isinstance(lab, str):
            labels.append(lab)
    comments_raw = item.get("comments") or []
    thread = []
    if body:
        thread.append({"author": author or "author", "at": item.get("createdAt") or item.get("created_at"), "body": body})
    if isinstance(comments_raw, list):
        for c in comments_raw:
            if not isinstance(c, dict):
                continue
            c_author = ""
            if isinstance(c.get("author"), dict):
                c_author = c["author"].get("login") or ""
            elif isinstance(c.get("user"), dict):
                c_author = c["user"].get("login") or ""
            thread.append(
                {
                    "author": c_author,
                    "at": c.get("createdAt") or c.get("created_at"),
                    "body": c.get("body") or "",
                }
            )
    files = []
    for f in item.get("files") or []:
        if not isinstance(f, dict):
            continue
        files.append(
            {
                "path": f.get("path") or f.get("filename") or "",
                "status": f.get("status") or "",
                "additions": f.get("additions"),
                "deletions": f.get("deletions"),
                "patch": f.get("patch") or "",
            }
        )
    base = ""
    head = ""
    if kind == "pr":
        base_ref = item.get("baseRefName") or (item.get("base") or {}).get("ref")
        head_ref = item.get("headRefName") or (item.get("head") or {}).get("ref")
        base = base_ref or ""
        head = head_ref or ""
    summary = f"# {title}\n\n{body}\n" if body else f"# {title}\n"
    return {
        "title": f"{'PR' if kind == 'pr' else 'Issue'} #{number}: {title}",
        "body": summary,
        "kind": f"github-{kind}",
        "github": {
            "number": number,
            "state": state,
            "url": url,
            "base": base,
            "head": head,
            "author": author,
            "labels": [x for x in labels if x],
            "thread": thread,
            "files": files,
            "synced_at": _utc_now(),
        },
        "meta": {"source": "github-sync", "at": _utc_now()},
    }


def sync(
    repo: Path,
    *,
    types: list[str] | None = None,
    state: str = "all",
    limit: int = 30,
) -> dict[str, Any]:
    if not _gh_available():
        return {"ok": False, "error": "gh CLI not available"}
    types = types or ["pr", "issue"]
    state = state if state in ("all", "open", "closed") else "all"
    notes = notes_store.load_notes_file(_notes_path())
    imported = 0
    errors: list[str] = []

    if "pr" in types:
        fields = (
            "number,title,body,url,state,createdAt,author,labels,baseRefName,headRefName,"
            "comments,files"
        )
        args = ["pr", "list", "--state", state, "--limit", str(limit), "--json", fields]
        code, out, err = _run_gh(args, cwd=repo)
        if code != 0:
            errors.append(err.strip() or "pr list failed")
        else:
            try:
                items = json.loads(out)
            except json.JSONDecodeError:
                items = []
                errors.append("invalid pr JSON")
            for item in items if isinstance(items, list) else []:
                # Enrich files with patches via REST if missing
                number = item.get("number")
                if number and not any((f or {}).get("patch") for f in (item.get("files") or [])):
                    c2, o2, _ = _run_gh(
                        ["api", f"repos/{{owner}}/{{repo}}/pulls/{number}/files", "--paginate"],
                        cwd=repo,
                    )
                    if c2 == 0:
                        try:
                            item["files"] = json.loads(o2)
                        except json.JSONDecodeError:
                            pass
                # comments as list: gh may return count; fetch if needed
                if isinstance(item.get("comments"), int) or not isinstance(item.get("comments"), list):
                    c3, o3, _ = _run_gh(
                        [
                            "api",
                            f"repos/{{owner}}/{{repo}}/issues/{number}/comments",
                            "--paginate",
                        ],
                        cwd=repo,
                    )
                    if c3 == 0:
                        try:
                            item["comments"] = json.loads(o3)
                        except json.JSONDecodeError:
                            item["comments"] = []
                    else:
                        item["comments"] = []
                note = _map_issue(item, kind="pr")
                key = f"gh:pr:{number}"
                notes[key] = note
                imported += 1

    if "issue" in types:
        fields = "number,title,body,url,state,createdAt,author,labels,comments"
        args = ["issue", "list", "--state", state, "--limit", str(limit), "--json", fields]
        code, out, err = _run_gh(args, cwd=repo)
        if code != 0:
            errors.append(err.strip() or "issue list failed")
        else:
            try:
                items = json.loads(out)
            except json.JSONDecodeError:
                items = []
                errors.append("invalid issue JSON")
            for item in items if isinstance(items, list) else []:
                number = item.get("number")
                if isinstance(item.get("comments"), int) or not isinstance(item.get("comments"), list):
                    c3, o3, _ = _run_gh(
                        [
                            "api",
                            f"repos/{{owner}}/{{repo}}/issues/{number}/comments",
                            "--paginate",
                        ],
                        cwd=repo,
                    )
                    if c3 == 0:
                        try:
                            item["comments"] = json.loads(o3)
                        except json.JSONDecodeError:
                            item["comments"] = []
                    else:
                        item["comments"] = []
                note = _map_issue(item, kind="issue")
                key = f"gh:issue:{number}"
                notes[key] = note
                imported += 1

    if imported:
        notes_store.save_notes_with_history(_notes_path(), notes, db_path=_notes_db_path())

    return {
        "ok": len(errors) == 0,
        "imported": imported,
        "errors": errors,
        "notes": notes,
        "at": _utc_now(),
    }


def list_branches(repo: Path) -> dict[str, Any]:
    try:
        proc = subprocess.run(
            ["git", "branch", "-a", "--format=%(refname:short)"],
            cwd=str(repo),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=30,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return {"ok": False, "error": str(exc), "branches": []}
    if proc.returncode != 0:
        return {"ok": False, "error": proc.stderr.strip(), "branches": []}
    branches = [b.strip() for b in (proc.stdout or "").splitlines() if b.strip()]
    return {"ok": True, "branches": branches}


def git_status(repo: Path) -> dict[str, Any]:
    import git_ops

    return git_ops.status(repo)


def git_switch(repo: Path, branch: str, *, stash_if_dirty: bool = False) -> dict[str, Any]:
    import git_ops

    return git_ops.switch(repo, branch, stash_if_dirty=stash_if_dirty)


def diff_branch(repo: Path, ref: str, base: str = "HEAD") -> dict[str, Any]:
    """File list + patches for ref vs base (not a full tree swap)."""
    try:
        name_proc = subprocess.run(
            ["git", "diff", "--name-status", f"{base}...{ref}"],
            cwd=str(repo),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=60,
        )
        patch_proc = subprocess.run(
            ["git", "diff", f"{base}...{ref}", "--", "."],
            cwd=str(repo),
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=60,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return {"ok": False, "error": str(exc)}
    files = []
    for line in (name_proc.stdout or "").splitlines():
        parts = line.split("\t")
        if len(parts) >= 2:
            files.append({"status": parts[0], "path": parts[-1], "patch": ""})
    return {
        "ok": name_proc.returncode == 0,
        "base": base,
        "head": ref,
        "files": files,
        "patch": (patch_proc.stdout or "")[:500_000],
        "error": name_proc.stderr.strip() if name_proc.returncode else None,
    }