Gh Deps Intel

interface:
  display_name: "Dependency Intel"
  short_description: "Plan dependency upgrades with release and refactor context"
  default_prompt: "Use $gh-deps-intel to inventory upgrades, research release notes, map refactors, and either plan or execute the upgrade workflow with a stable Markdown and JSON contract."
policy:
  allow_implicit_invocation: false
dependencies:
  tools:
    - type: "mcp"
      value: "context7"
      description: "Current library documentation"
    - type: "mcp"
      value: "exa"
      description: "Current migration and release-note research"

#!/usr/bin/env python3
"""Dependency extraction from package.json and pyproject.toml files."""

from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any


def _read_package_json(path: Path) -> dict[str, Any]:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _extract_npm_version_hint(spec: str | None) -> str | None:
    if not spec:
        return None
    s = spec.strip()
    if s.startswith("workspace:"):
        s = s.split(":", 1)[1].strip()
    if s in {"*", "latest", "next"}:
        return None
    m = re.search(r"(\d+\.\d+(?:\.\d+)?)", s)
    return m.group(1) if m else None


def collect_js_dependencies(package_json_files: list[str], repo_root: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    sections = ["dependencies", "devDependencies", "peerDependencies", "optionalDependencies"]
    for file in package_json_files:
        path = Path(file)
        data = _read_package_json(path)
        for section in sections:
            bucket = data.get(section)
            if not isinstance(bucket, dict):
                continue
            for name, spec in bucket.items():
                if not isinstance(name, str) or not isinstance(spec, str):
                    continue
                rows.append(
                    {
                        "ecosystem": "npm",
                        "name": name,
                        "spec": spec,
                        "current_version_hint": _extract_npm_version_hint(spec),
                        "source_file": str(path),
                        "source_rel": str(path.relative_to(repo_root)),
                        "dependency_type": section,
                    }
                )
    return rows


def _parse_pep508_name(req: str) -> str | None:
    req = req.strip()
    if not req or req.startswith("#"):
        return None
    m = re.match(r"^([A-Za-z0-9_.-]+)", req)
    if not m:
        return None
    return m.group(1).lower().replace("_", "-")


def _parse_pep508_spec(req: str) -> str:
    req = req.strip()
    marker_split = req.split(";", 1)[0].strip()
    name_match = re.match(r"^([A-Za-z0-9_.-]+)", marker_split)
    if not name_match:
        return ""
    rest = marker_split[name_match.end() :].strip()
    return rest


def collect_python_dependencies(pyproject_files: list[str], repo_root: Path) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    try:
        import tomllib
    except Exception:
        return rows

    for file in pyproject_files:
        path = Path(file)
        try:
            data = tomllib.loads(path.read_text(encoding="utf-8"))
        except Exception:
            continue
        project = data.get("project") if isinstance(data, dict) else None
        if not isinstance(project, dict):
            continue

        deps = project.get("dependencies")
        if isinstance(deps, list):
            for dep in deps:
                if not isinstance(dep, str):
                    continue
                name = _parse_pep508_name(dep)
                if not name:
                    continue
                rows.append(
                    {
                        "ecosystem": "pypi",
                        "name": name,
                        "spec": _parse_pep508_spec(dep),
                        "current_version_hint": None,
                        "source_file": str(path),
                        "source_rel": str(path.relative_to(repo_root)),
                        "dependency_type": "dependencies",
                    }
                )

        optional_deps = project.get("optional-dependencies")
        if isinstance(optional_deps, dict):
            for group, reqs in optional_deps.items():
                if not isinstance(group, str) or not isinstance(reqs, list):
                    continue
                for dep in reqs:
                    if not isinstance(dep, str):
                        continue
                    name = _parse_pep508_name(dep)
                    if not name:
                        continue
                    rows.append(
                        {
                            "ecosystem": "pypi",
                            "name": name,
                            "spec": _parse_pep508_spec(dep),
                            "current_version_hint": None,
                            "source_file": str(path),
                            "source_rel": str(path.relative_to(repo_root)),
                            "dependency_type": f"optional:{group}",
                        }
                    )

    return rows


def collect_dependencies(repo_context: dict[str, Any]) -> list[dict[str, Any]]:
    repo_root = Path(repo_context["repo_root"])
    deps: list[dict[str, Any]] = []
    deps.extend(collect_js_dependencies(repo_context.get("package_json_files", []), repo_root))
    deps.extend(collect_python_dependencies(repo_context.get("pyproject_files", []), repo_root))
    return deps


def aggregate_dependencies(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    grouped: dict[tuple[str, str], dict[str, Any]] = {}
    for row in rows:
        key = (row["ecosystem"], row["name"])
        target = grouped.setdefault(
            key,
            {
                "ecosystem": row["ecosystem"],
                "name": row["name"],
                "specs": set(),
                "contexts": [],
                "current_version_hint": row.get("current_version_hint"),
            },
        )
        if row.get("spec"):
            target["specs"].add(row["spec"])
        target["contexts"].append(
            {
                "source_file": row.get("source_file"),
                "source_rel": row.get("source_rel"),
                "dependency_type": row.get("dependency_type"),
                "spec": row.get("spec"),
            }
        )
        if not target.get("current_version_hint") and row.get("current_version_hint"):
            target["current_version_hint"] = row["current_version_hint"]

    out: list[dict[str, Any]] = []
    for value in grouped.values():
        value["specs"] = sorted(value["specs"])
        out.append(value)
    out.sort(key=lambda x: (x["ecosystem"], x["name"]))
    return out


def main() -> None:
    import argparse
    from detect_repo import detect_repo_context

    parser = argparse.ArgumentParser(description="Collect dependencies from repo manifests.")
    parser.add_argument("repo", nargs="?", default=".")
    args = parser.parse_args()

    ctx = detect_repo_context(Path(args.repo))
    rows = collect_dependencies(ctx)
    print(json.dumps(aggregate_dependencies(rows), indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Repository and runtime detection for gh-deps-intel."""

from __future__ import annotations

import glob
import json
import os
from pathlib import Path
from typing import Any

from utils import run_cmd
from utils import extract_node_major

IGNORED_DIRS = {
    ".git",
    ".hg",
    ".svn",
    "node_modules",
    ".next",
    ".turbo",
    ".venv",
    "venv",
    "dist",
    "build",
    "coverage",
    "tmp",
    "out",
    ".cache",
}


def _read_json(path: Path) -> dict[str, Any]:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _read_text(path: Path) -> str | None:
    if not path.exists():
        return None
    txt = path.read_text(encoding="utf-8", errors="ignore").strip()
    return txt or None


def _workspace_globs_from_package_json(pkg: dict[str, Any]) -> list[str]:
    ws = pkg.get("workspaces")
    if isinstance(ws, list):
        return [x for x in ws if isinstance(x, str)]
    if isinstance(ws, dict):
        packages = ws.get("packages")
        if isinstance(packages, list):
            return [x for x in packages if isinstance(x, str)]
    return []


def _workspace_globs_from_pnpm(path: Path) -> list[str]:
    file = path / "pnpm-workspace.yaml"
    if not file.exists():
        return []
    globs: list[str] = []
    in_packages = False
    for line in file.read_text(encoding="utf-8", errors="ignore").splitlines():
        s = line.strip()
        if not s or s.startswith("#"):
            continue
        if s.startswith("packages:"):
            in_packages = True
            continue
        if in_packages and s.startswith("-"):
            item = s[1:].strip().strip("'\"")
            if item:
                globs.append(item)
        elif in_packages and not s.startswith("-"):
            # Exit package list when indentation/shape changes.
            in_packages = False
    return globs


def _expand_workspace_globs(root: Path, globs_in: list[str]) -> list[Path]:
    ignored = _load_git_ignored_entries(root)
    paths: set[Path] = set()
    for pattern in globs_in:
        pat = pattern.rstrip("/") + "/package.json"
        for match in glob.glob(str(root / pat), recursive=True):
            p = Path(match).resolve()
            if p.is_file() and not _is_git_ignored(root, p, ignored):
                paths.add(p)
    return sorted(paths)


def _recursive_package_scan(root: Path) -> list[Path]:
    ignored = _load_git_ignored_entries(root)
    found: list[Path] = []
    for dirpath, dirnames, filenames in os.walk(root):
        current = Path(dirpath)
        if _is_git_ignored(root, current, ignored):
            dirnames[:] = []
            continue
        keep_dirs = []
        for d in dirnames:
            if d in IGNORED_DIRS or d.startswith("."):
                continue
            candidate = current / d
            if _is_git_ignored(root, candidate, ignored):
                continue
            keep_dirs.append(d)
        dirnames[:] = keep_dirs
        if "package.json" in filenames:
            pkg = Path(dirpath) / "package.json"
            if not _is_git_ignored(root, pkg, ignored):
                found.append(pkg)
    return sorted(p.resolve() for p in found)


def _recursive_pyproject_scan(root: Path) -> list[Path]:
    ignored = _load_git_ignored_entries(root)
    found: list[Path] = []
    for dirpath, dirnames, filenames in os.walk(root):
        current = Path(dirpath)
        if _is_git_ignored(root, current, ignored):
            dirnames[:] = []
            continue
        keep_dirs = []
        for d in dirnames:
            if d in IGNORED_DIRS or d.startswith("."):
                continue
            candidate = current / d
            if _is_git_ignored(root, candidate, ignored):
                continue
            keep_dirs.append(d)
        dirnames[:] = keep_dirs
        if "pyproject.toml" in filenames:
            pyproject = Path(dirpath) / "pyproject.toml"
            if not _is_git_ignored(root, pyproject, ignored):
                found.append(pyproject)
    return sorted(p.resolve() for p in found)


def _is_git_repo(root: Path) -> bool:
    proc = run_cmd(["git", "-C", str(root), "rev-parse", "--is-inside-work-tree"], check=False)
    return proc.returncode == 0


def _load_git_ignored_entries(root: Path) -> list[str]:
    """Return ignored paths from .gitignore as normalized relative entries."""
    if not _is_git_repo(root):
        return []
    proc = run_cmd(
        ["git", "-C", str(root), "ls-files", "--ignored", "--exclude-standard", "--others", "--directory"],
        check=False,
    )
    if proc.returncode != 0:
        return []
    entries: list[str] = []
    for raw in proc.stdout.splitlines():
        line = raw.strip()
        if not line:
            continue
        normalized = line.replace("\\", "/").lstrip("./")
        entries.append(normalized)
    return entries


def _is_git_ignored(root: Path, path: Path, ignored_entries: list[str]) -> bool:
    if not ignored_entries:
        return False
    try:
        rel = path.resolve().relative_to(root.resolve()).as_posix().lstrip("./")
    except Exception:
        return False
    for entry in ignored_entries:
        if entry.endswith("/"):
            prefix = entry[:-1]
            if rel == prefix or rel.startswith(entry):
                return True
        else:
            if rel == entry:
                return True
    return False


def _detect_node_manager(root: Path, root_pkg: dict[str, Any]) -> str:
    package_manager = str(root_pkg.get("packageManager") or "")
    if package_manager.startswith("bun@") or (root / "bun.lock").exists() or (root / "bun.lockb").exists():
        return "bun"
    if package_manager.startswith("pnpm@") or (root / "pnpm-lock.yaml").exists():
        return "pnpm"
    if package_manager.startswith("yarn@") or (root / "yarn.lock").exists():
        return "yarn"
    if package_manager.startswith("npm@") or (root / "package-lock.json").exists():
        return "npm"
    return "npm"


def _detect_python_manager(root: Path) -> str:
    if (root / "uv.lock").exists() or (root / "uv.toml").exists():
        return "uv"
    if (root / "poetry.lock").exists():
        return "poetry"
    if (root / "requirements.txt").exists():
        return "pip"
    return "uv"


def _detect_node_runtime(root: Path, root_pkg: dict[str, Any]) -> dict[str, Any]:
    hints: list[dict[str, str]] = []
    engines = root_pkg.get("engines") if isinstance(root_pkg, dict) else None
    if isinstance(engines, dict) and isinstance(engines.get("node"), str):
        hints.append({"source": "package.json#engines.node", "value": engines["node"]})

    for name in [".nvmrc", ".node-version"]:
        txt = _read_text(root / name)
        if txt:
            hints.append({"source": name, "value": txt})

    tool_versions = _read_text(root / ".tool-versions")
    if tool_versions:
        for line in tool_versions.splitlines():
            s = line.strip()
            if s.startswith("nodejs "):
                hints.append({"source": ".tool-versions", "value": s.split(" ", 1)[1].strip()})

    volta = root_pkg.get("volta") if isinstance(root_pkg, dict) else None
    if isinstance(volta, dict) and isinstance(volta.get("node"), str):
        hints.append({"source": "package.json#volta.node", "value": volta["node"]})

    selected = hints[0]["value"] if hints else None
    major = extract_node_major(selected)
    return {"detected": selected, "major": major, "hints": hints}


def _detect_python_runtime(root: Path, pyproject_files: list[Path]) -> dict[str, Any]:
    hints: list[dict[str, str]] = []

    pyver = _read_text(root / ".python-version")
    if pyver:
        hints.append({"source": ".python-version", "value": pyver})

    for pp in pyproject_files:
        try:
            import tomllib

            data = tomllib.loads(pp.read_text(encoding="utf-8"))
            project = data.get("project") if isinstance(data, dict) else None
            if isinstance(project, dict) and isinstance(project.get("requires-python"), str):
                rel = str(pp.relative_to(root))
                hints.append({"source": f"{rel}#project.requires-python", "value": project["requires-python"]})
        except Exception:
            continue

    selected = hints[0]["value"] if hints else None
    return {"detected": selected, "hints": hints}


def detect_repo_context(repo_root: Path) -> dict[str, Any]:
    repo_root = repo_root.resolve()
    root_pkg_path = repo_root / "package.json"
    root_pkg = _read_json(root_pkg_path) if root_pkg_path.exists() else {}

    declared_globs = _workspace_globs_from_package_json(root_pkg) + _workspace_globs_from_pnpm(repo_root)
    workspace_pkgs = _expand_workspace_globs(repo_root, declared_globs)

    recursive_pkgs = _recursive_package_scan(repo_root)
    package_json_files: list[Path] = []
    seen: set[Path] = set()
    for p in [root_pkg_path] + workspace_pkgs + recursive_pkgs:
        if p and p.exists() and p not in seen:
            seen.add(p)
            package_json_files.append(p.resolve())

    pyproject_files = _recursive_pyproject_scan(repo_root)

    has_node = len(package_json_files) > 0
    has_python = len(pyproject_files) > 0

    ctx = {
        "repo_root": str(repo_root),
        "has_node": has_node,
        "has_python": has_python,
        "node_manager": _detect_node_manager(repo_root, root_pkg) if has_node else None,
        "python_manager": _detect_python_manager(repo_root) if has_python else None,
        "package_json_files": [str(p) for p in package_json_files],
        "pyproject_files": [str(p) for p in pyproject_files],
        "is_monorepo": len(package_json_files) > 1 or len(pyproject_files) > 1,
        "workspace_globs": declared_globs,
        "node_runtime": _detect_node_runtime(repo_root, root_pkg) if has_node else {"detected": None, "major": None, "hints": []},
        "python_runtime": _detect_python_runtime(repo_root, pyproject_files) if has_python else {"detected": None, "hints": []},
    }
    return ctx


def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(description="Detect repository context.")
    parser.add_argument("repo", nargs="?", default=".")
    args = parser.parse_args()

    ctx = detect_repo_context(Path(args.repo))
    print(json.dumps(ctx, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Fallback metadata links when GitHub data is missing or limited."""

from __future__ import annotations

from typing import Any


def collect_fallback_links(dep: dict[str, Any], resolved: dict[str, Any]) -> list[dict[str, str]]:
    ecosystem = dep.get("ecosystem")
    name = dep.get("name")
    links: list[dict[str, str]] = []

    if ecosystem == "npm":
        links.append({"label": "npm package", "url": f"https://www.npmjs.com/package/{name}"})
    elif ecosystem == "pypi":
        links.append({"label": "PyPI package", "url": f"https://pypi.org/project/{name}/"})

    meta_links = resolved.get("links") if isinstance(resolved, dict) else None
    if isinstance(meta_links, dict):
        for key, value in meta_links.items():
            if not isinstance(value, str) or not value.startswith("http"):
                continue
            label = f"{key}"
            links.append({"label": label, "url": value})

    project_urls = (resolved.get("metadata") or {}).get("project_urls")
    if isinstance(project_urls, dict):
        for k, v in project_urls.items():
            if isinstance(v, str) and v.startswith("http"):
                links.append({"label": f"project_url:{k}", "url": v})

    # Deduplicate by URL while preserving order.
    seen: set[str] = set()
    deduped: list[dict[str, str]] = []
    for item in links:
        url = item.get("url")
        if not url or url in seen:
            continue
        seen.add(url)
        deduped.append(item)
    return deduped

#!/usr/bin/env python3
"""Summarize compare commits between two tags/refs."""

from __future__ import annotations

import argparse
import json

from gh_release_fetch import GitHubClient


def main() -> None:
    parser = argparse.ArgumentParser(description="Summarize GitHub compare output.")
    parser.add_argument("repo", help="owner/repo")
    parser.add_argument("base")
    parser.add_argument("head")
    args = parser.parse_args()

    if "/" not in args.repo:
        raise SystemExit("repo must be owner/repo")
    owner, repo = args.repo.split("/", 1)

    client = GitHubClient(mode="safe")
    data = client.get_compare(owner, repo, args.base, args.head)
    client.flush_cache()
    if not data:
        print(json.dumps({"error": "compare data unavailable"}, indent=2))
        return

    commits = data.get("commits") if isinstance(data, dict) else []
    summary: list[str] = []
    if isinstance(commits, list):
        for c in commits[:40]:
            if not isinstance(c, dict):
                continue
            sha = str(c.get("sha") or "")[:7]
            msg = ""
            commit = c.get("commit")
            if isinstance(commit, dict):
                m = commit.get("message")
                if isinstance(m, str):
                    msg = m.splitlines()[0]
            if msg:
                summary.append(f"{sha} {msg}")

    print(json.dumps({"repo": args.repo, "base": args.base, "head": args.head, "summary": summary}, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Dependency upgrade intelligence orchestrator.

This script analyzes a repository, gathers outdated dependency signals,
retrieves GitHub releases/changelogs, and produces a refactor-focused report.
"""

from __future__ import annotations

import argparse
import json
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any

from collect_deps import aggregate_dependencies, collect_dependencies
from detect_repo import detect_repo_context
from fallback_registry_fetch import collect_fallback_links
from gh_release_fetch import GitHubApiError, GitHubClient, filter_releases_between
from impact_analyzer import analyze_dependency_changes
from outdated_probe import probe_outdated
from render_report import write_reports
from repo_resolver import flush_cache as flush_registry_cache
from repo_resolver import resolve_dependency
from runtime_policy import choose_target_version
from utils import ensure_dir, now_iso, run_cmd, write_json


def _build_outdated_lookup(scan: dict[str, Any], ecosystem: str) -> dict[str, dict[str, str]]:
    if ecosystem == "npm":
        return scan.get("outdated", {}).get("js", {})
    if ecosystem == "pypi":
        return scan.get("outdated", {}).get("python", {})
    return {}


def _normalize(name: str) -> str:
    return name.strip().lower()


def _dep_matches_selector(dep_name: str, selector: str) -> bool:
    dep = _normalize(dep_name)
    sel = _normalize(selector)
    if not sel:
        return False
    if dep == sel:
        return True
    if dep.endswith("/" + sel):
        return True
    return sel in dep


def _filter_scan_dependencies(scan: dict[str, Any], selectors: list[str]) -> tuple[dict[str, Any], list[str]]:
    if not selectors:
        return scan, []
    deps = scan.get("dependencies", [])
    if not isinstance(deps, list):
        return scan, []

    warnings: list[str] = []
    selected: list[dict[str, Any]] = []
    for dep in deps:
        name = dep.get("name")
        if not isinstance(name, str):
            continue
        if any(_dep_matches_selector(name, s) for s in selectors):
            selected.append(dep)

    if not selected:
        warnings.append(f"No dependencies matched selectors: {', '.join(selectors)}")

    out = dict(scan)
    out["dependencies"] = selected
    out["targeted_dependencies"] = selectors
    return out, warnings


def _repo_usage_map_for_dependency(repo_root: Path, dep: dict[str, Any], limit: int = 120) -> dict[str, Any]:
    name = str(dep.get("name") or "")
    ecosystem = str(dep.get("ecosystem") or "")
    if not name:
        return {"summary": "No dependency name available.", "hits": [], "files": []}

    patterns: list[str] = []
    if ecosystem == "npm":
        escaped = re.escape(name)
        patterns.append(rf"from\\s+['\\\"]{escaped}['\\\"]")
        patterns.append(rf"require\\(\\s*['\\\"]{escaped}['\\\"]\\s*\\)")
        patterns.append(rf"import\\(\\s*['\\\"]{escaped}['\\\"]\\s*\\)")
        patterns.append(rf"['\\\"]{escaped}['\\\"]")
    elif ecosystem == "pypi":
        mod = name.replace("-", "_")
        patterns.append(rf"^\\s*import\\s+{re.escape(mod)}\\b")
        patterns.append(rf"^\\s*from\\s+{re.escape(mod)}\\b")
        patterns.append(rf"['\\\"]{re.escape(name)}['\\\"]")
    else:
        patterns.append(re.escape(name))

    glob_args = [
        "--glob",
        "!.git/**",
        "--glob",
        "!node_modules/**",
        "--glob",
        "!.venv/**",
        "--glob",
        "!venv/**",
        "--glob",
        "!dist/**",
        "--glob",
        "!build/**",
        "--glob",
        "!.next/**",
        "--glob",
        "!.turbo/**",
    ]

    hits: list[dict[str, Any]] = []
    seen = set()
    for pat in patterns:
        cmd = ["rg", "-n", "--no-heading", "--color", "never", "--hidden", *glob_args, pat, str(repo_root)]
        proc = run_cmd(cmd, check=False)
        if proc.returncode not in (0, 1):
            continue
        for line in (proc.stdout or "").splitlines():
            # path:line:text (text may contain colons)
            parts = line.split(":", 2)
            if len(parts) < 3:
                continue
            path, line_no, text = parts[0], parts[1], parts[2]
            key = (path, line_no, text)
            if key in seen:
                continue
            seen.add(key)
            hits.append({"path": path, "line": int(line_no) if line_no.isdigit() else None, "text": text.strip()})
            if len(hits) >= limit:
                break
        if len(hits) >= limit:
            break

    files = sorted({h["path"] for h in hits if isinstance(h.get("path"), str)})
    summary = f"Found {len(hits)} reference hits across {len(files)} files."
    if not hits:
        summary = "No direct usage references found with default patterns; validate dynamic/runtime usage manually."
    return {"summary": summary, "hits": hits, "files": files}


def run_scan(repo_root: Path) -> dict[str, Any]:
    ctx = detect_repo_context(repo_root)
    dep_rows = collect_dependencies(ctx)
    deps = aggregate_dependencies(dep_rows)
    outdated = probe_outdated(ctx)

    return {
        "generated_at": now_iso(),
        "repo_root": str(repo_root.resolve()),
        "repo_context": ctx,
        "dependencies": deps,
        "outdated": outdated,
    }


def _extract_compare_summary(compare_obj: dict[str, Any] | None) -> str:
    if not compare_obj:
        return ""
    commits = compare_obj.get("commits")
    if not isinstance(commits, list):
        return ""
    lines: list[str] = []
    for c in commits[:25]:
        if not isinstance(c, dict):
            continue
        sha = str(c.get("sha") or "")[:7]
        msg = ""
        commit = c.get("commit")
        if isinstance(commit, dict):
            m = commit.get("message")
            if isinstance(m, str):
                msg = m.splitlines()[0]
        if msg:
            lines.append(f"- {sha}: {msg}")
    return "\n".join(lines)


def _find_tag_for_version(tags: list[dict[str, Any]], version: str | None) -> str | None:
    if not version:
        return None
    candidates = {version, f"v{version}"}
    for tag in tags:
        name = tag.get("name")
        if isinstance(name, str) and name in candidates:
            return name
    for tag in tags:
        name = tag.get("name")
        if isinstance(name, str) and version in name:
            return name
    return None


def _enrich_one_dependency(
    dep: dict[str, Any],
    repo_context: dict[str, Any],
    outdated_lookup: dict[str, dict[str, str]],
    mode: str,
    compatibility_policy: str,
) -> tuple[dict[str, Any], list[str]]:
    warnings: list[str] = []

    resolved = resolve_dependency(dep)
    target = choose_target_version(
        dep,
        outdated_lookup,
        resolved,
        repo_context,
        compatibility_policy=compatibility_policy,
    )

    row: dict[str, Any] = {
        "ecosystem": dep.get("ecosystem"),
        "name": dep.get("name"),
        "specs": dep.get("specs") or [],
        "contexts": dep.get("contexts") or [],
        "current_version": target.get("current"),
        "latest_available": target.get("latest_available"),
        "target_version": target.get("target"),
        "target_reason": target.get("reason"),
        "outdated_source": target.get("outdated_source"),
        "resolved": {
            "source_repo": resolved.get("source_repo"),
            "source_repo_url": resolved.get("source_repo_url"),
            "links": resolved.get("links") or {},
        },
        "release_notes": [],
        "changelog_text": "",
        "fallback_links": [],
        "source_links": [],
    }

    source_repo = resolved.get("source_repo")
    if not isinstance(source_repo, str) or "/" not in source_repo:
        row["fallback_links"] = collect_fallback_links(dep, resolved)
        row["source_links"] = row["fallback_links"]
        return row, warnings

    owner, repo = source_repo.split("/", 1)
    gh = GitHubClient(mode=mode)

    try:
        releases = gh.get_releases(owner, repo)
        selected = filter_releases_between(releases, row.get("current_version"), row.get("target_version"), max_items=25)
        row["release_notes"] = selected

        changelog = gh.get_changelog(owner, repo)
        if changelog:
            row["changelog_text"] = changelog.get("text") or ""
            if changelog.get("html_url"):
                row["source_links"].append({"label": "changelog", "url": changelog["html_url"]})

        # If no releases were found in range, attempt tag compare notes.
        if not row["release_notes"]:
            tags = gh.get_tags(owner, repo)
            base = _find_tag_for_version(tags, row.get("current_version"))
            head = _find_tag_for_version(tags, row.get("target_version"))
            if base and head and base != head:
                compare_obj = gh.get_compare(owner, repo, base, head)
                compare_text = _extract_compare_summary(compare_obj)
                if compare_text:
                    row["release_notes"] = [
                        {
                            "name": f"Compare {base}...{head}",
                            "tag_name": f"{base}...{head}",
                            "version": row.get("target_version"),
                            "published_at": None,
                            "html_url": f"https://github.com/{owner}/{repo}/compare/{base}...{head}",
                            "body": compare_text,
                            "draft": False,
                            "prerelease": False,
                        }
                    ]

        if not row["release_notes"]:
            warnings.append(f"No GitHub releases/compare notes found for {source_repo} in selected range.")

    except GitHubApiError as exc:
        warnings.append(f"GitHub API error for {source_repo}: {exc}")

    fallback = collect_fallback_links(dep, resolved)
    row["fallback_links"] = fallback

    links: list[dict[str, str]] = []
    links.append({"label": "repository", "url": f"https://github.com/{owner}/{repo}"})
    for rel in row.get("release_notes") or []:
        url = rel.get("html_url")
        if isinstance(url, str) and url.startswith("http"):
            links.append({"label": f"release:{rel.get('tag_name') or rel.get('name')}", "url": url})
    links.extend(fallback)

    deduped: list[dict[str, str]] = []
    seen: set[str] = set()
    for item in links:
        url = item.get("url")
        if not url or url in seen:
            continue
        seen.add(url)
        deduped.append(item)
    row["source_links"] = deduped

    gh.flush_cache()
    return row, warnings


def run_enrich(
    scan: dict[str, Any],
    mode: str = "safe",
    max_concurrency: int = 3,
    compatibility_policy: str = "runtime-pinned",
    deep_repo_map: bool = False,
) -> dict[str, Any]:
    repo_context = scan["repo_context"]
    deps = scan["dependencies"]

    warnings: list[str] = list(scan.get("outdated", {}).get("warnings", []))
    command_traces = list(scan.get("outdated", {}).get("command_traces", []))

    enriched: list[dict[str, Any]] = []
    rate_limited: list[dict[str, Any]] = []

    def enrich_task(dep: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
        lookup = _build_outdated_lookup(scan, dep.get("ecosystem"))
        return _enrich_one_dependency(dep, repo_context, lookup, mode=mode, compatibility_policy=compatibility_policy)

    if mode == "fast" and max_concurrency > 1:
        with ThreadPoolExecutor(max_workers=max_concurrency) as pool:
            future_map = {pool.submit(enrich_task, dep): dep for dep in deps}
            for fut in as_completed(future_map):
                dep = future_map[fut]
                try:
                    row, w = fut.result()
                    enriched.append(row)
                    warnings.extend(w)
                    if any("rate limit" in x.lower() for x in w):
                        rate_limited.append(dep)
                except Exception as exc:
                    warnings.append(f"Fast mode error for {dep.get('name')}: {exc}")
                    rate_limited.append(dep)

        # Auto-fallback for dependencies that hit limits/errors in fast mode.
        if rate_limited:
            warnings.append(
                f"Fast mode fallback: re-running {len(rate_limited)} dependencies serially with safe mode due to rate-limit/errors."
            )
            # Remove any partial rows for those deps first.
            retry_keys = {(d.get("ecosystem"), d.get("name")) for d in rate_limited}
            enriched = [r for r in enriched if (r.get("ecosystem"), r.get("name")) not in retry_keys]
            for dep in rate_limited:
                lookup = _build_outdated_lookup(scan, dep.get("ecosystem"))
                row, w = _enrich_one_dependency(
                    dep,
                    repo_context,
                    lookup,
                    mode="safe",
                    compatibility_policy=compatibility_policy,
                )
                enriched.append(row)
                warnings.extend(w)
    else:
        for dep in deps:
            lookup = _build_outdated_lookup(scan, dep.get("ecosystem"))
            row, w = _enrich_one_dependency(
                dep,
                repo_context,
                lookup,
                mode="safe",
                compatibility_policy=compatibility_policy,
            )
            enriched.append(row)
            warnings.extend(w)

    enriched.sort(key=lambda x: (x.get("ecosystem") or "", x.get("name") or ""))

    return {
        "generated_at": now_iso(),
        "repo_root": scan["repo_root"],
        "repo_context": repo_context,
        "mode": mode,
        "targeted_dependencies": scan.get("targeted_dependencies", []),
        "deep_repo_map": deep_repo_map,
        "dependencies": enriched,
        "warnings": warnings,
        "command_traces": command_traces,
    }


def run_analyze(enriched: dict[str, Any]) -> dict[str, Any]:
    repo_root = Path(enriched["repo_root"])
    deep_repo_map = bool(enriched.get("deep_repo_map"))
    analyzed: list[dict[str, Any]] = []
    for dep in enriched["dependencies"]:
        impact = analyze_dependency_changes(dep)
        merged = dict(dep)
        merged.update(impact)
        if deep_repo_map:
            usage = _repo_usage_map_for_dependency(repo_root, merged)
            merged["repo_usage"] = usage
            if usage.get("files"):
                merged.setdefault("refactor_actions", [])
                merged["refactor_actions"].append(
                    f"Update all usage points in {len(usage.get('files') or [])} files identified by repo impact scan."
                )
                merged["refactor_actions"].append(
                    "Refactor imports/usages first, then run tests for modules listed in repo impact map."
                )
        analyzed.append(merged)

    analyzed.sort(key=lambda x: (x.get("ecosystem") or "", x.get("name") or ""))

    return {
        "generated_at": now_iso(),
        "repo_root": enriched["repo_root"],
        "repo_context": enriched["repo_context"],
        "mode": enriched.get("mode", "safe"),
        "targeted_dependencies": enriched.get("targeted_dependencies", []),
        "deep_repo_map": deep_repo_map,
        "dependencies": analyzed,
        "warnings": enriched.get("warnings", []),
        "command_traces": enriched.get("command_traces", []),
    }


def run_report(
    analyzed: dict[str, Any],
    out_dir: Path,
    compatibility_policy: str = "runtime-pinned",
) -> dict[str, Any]:
    paths = write_reports(
        out_dir=out_dir,
        repo_root=analyzed["repo_root"],
        repo_context=analyzed["repo_context"],
        dependencies=analyzed["dependencies"],
        mode=analyzed.get("mode", "safe"),
        compatibility_policy=compatibility_policy,
        command_traces=analyzed.get("command_traces", []),
        warnings=analyzed.get("warnings", []),
        targeted_dependencies=analyzed.get("targeted_dependencies", []),
        deep_repo_map=bool(analyzed.get("deep_repo_map")),
    )
    return {
        "generated_at": now_iso(),
        "report_paths": paths,
        "warnings": analyzed.get("warnings", []),
    }


def run_rate_limit_diag() -> dict[str, Any]:
    gh = GitHubClient(mode="safe")
    data = gh.get_rate_limit()
    gh.flush_cache()
    return data


def save_stage_json(out_dir: Path, name: str, payload: dict[str, Any]) -> str:
    ensure_dir(out_dir)
    path = out_dir / f"{name}.json"
    write_json(path, payload)
    return str(path)


def main() -> None:
    parser = argparse.ArgumentParser(description="GitHub dependency intelligence orchestrator")
    sub = parser.add_subparsers(dest="command", required=True)

    def add_common(p: argparse.ArgumentParser) -> None:
        p.add_argument("--repo", default=".", help="Target repository root (default: current directory)")
        p.add_argument("--out", default="reports", help="Output directory relative to target repo")
        p.add_argument("--mode", choices=["safe", "fast"], default="safe", help="Execution mode")
        p.add_argument("--max-concurrency", type=int, default=3, help="Fast mode worker cap")
        p.add_argument(
            "--dependency",
            action="append",
            default=[],
            help="Dependency selector (repeatable). Supports exact or partial name match.",
        )
        p.add_argument(
            "--deep-repo-map",
            action="store_true",
            help="Run repo-wide usage mapping with rg and include impacted files/usages in report.",
        )
        p.add_argument(
            "--compatibility-policy",
            default="runtime-pinned",
            choices=["runtime-pinned", "semver-only", "always-latest"],
            help="Target version selection policy",
        )

    add_common(sub.add_parser("scan", help="Detect repo and collect dependencies/outdated data"))
    add_common(sub.add_parser("enrich", help="Scan + enrich dependencies with registry and GitHub release metadata"))
    add_common(sub.add_parser("analyze", help="Scan + enrich + impact analysis"))

    p_report = sub.add_parser("report", help="Scan + enrich + analyze + report outputs")
    add_common(p_report)

    p_full = sub.add_parser("full", help="Same as report")
    add_common(p_full)
    p_package = sub.add_parser("package", help="Single-dependency comprehensive upgrade spec")
    add_common(p_package)

    sub.add_parser("rate-limit", help="Show current GitHub API rate-limit status")

    args = parser.parse_args()

    if args.command == "rate-limit":
        print(json.dumps(run_rate_limit_diag(), indent=2))
        return

    repo_root = Path(args.repo).resolve()
    out_dir = repo_root / args.out

    scan = run_scan(repo_root)
    selectors = list(args.dependency or [])
    if selectors:
        scan, filter_warnings = _filter_scan_dependencies(scan, selectors)
        scan.setdefault("outdated", {}).setdefault("warnings", []).extend(filter_warnings)
    if args.command == "package":
        if not selectors:
            raise SystemExit("`package` requires at least one --dependency selector")
        if not scan.get("dependencies"):
            raise SystemExit(f"No dependencies matched selector(s): {', '.join(selectors)}")

    if args.command == "scan":
        stage_path = save_stage_json(out_dir, "gh-deps-intel-scan", scan)
        print(json.dumps({"scan": stage_path, "summary": {"dependencies": len(scan['dependencies'])}}, indent=2))
        flush_registry_cache()
        return

    enriched = run_enrich(
        scan,
        mode=args.mode,
        max_concurrency=max(1, int(args.max_concurrency)),
        compatibility_policy=args.compatibility_policy,
        deep_repo_map=bool(args.deep_repo_map or args.command == "package"),
    )
    if args.command == "enrich":
        stage_path = save_stage_json(out_dir, "gh-deps-intel-enrich", enriched)
        print(json.dumps({"enrich": stage_path, "summary": {"dependencies": len(enriched['dependencies'])}}, indent=2))
        flush_registry_cache()
        return

    analyzed = run_analyze(enriched)
    if args.command == "analyze":
        stage_path = save_stage_json(out_dir, "gh-deps-intel-analyze", analyzed)
        print(json.dumps({"analyze": stage_path, "summary": {"dependencies": len(analyzed['dependencies'])}}, indent=2))
        flush_registry_cache()
        return

    if args.command in {"report", "full", "package"}:
        report = run_report(analyzed, out_dir, compatibility_policy=args.compatibility_policy)
        print(json.dumps(report, indent=2))
        flush_registry_cache()
        return


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Quick GitHub rate-limit diagnostic."""

from __future__ import annotations

import json

from gh_release_fetch import GitHubClient


def main() -> None:
    client = GitHubClient(mode="safe")
    data = client.get_rate_limit()
    client.flush_cache()
    print(json.dumps(data, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Fetch release notes between two versions for a GitHub repository."""

from __future__ import annotations

import argparse
import json

from gh_release_fetch import GitHubClient, filter_releases_between


def main() -> None:
    parser = argparse.ArgumentParser(description="Get releases between two versions.")
    parser.add_argument("repo", help="owner/repo")
    parser.add_argument("--current", default=None, help="Current version")
    parser.add_argument("--target", default=None, help="Target version")
    parser.add_argument("--mode", choices=["safe", "fast"], default="safe")
    args = parser.parse_args()

    if "/" not in args.repo:
        raise SystemExit("repo must be owner/repo")
    owner, repo = args.repo.split("/", 1)

    client = GitHubClient(mode=args.mode)
    releases = client.get_releases(owner, repo)
    selected = filter_releases_between(releases, args.current, args.target, max_items=30)
    client.flush_cache()
    print(json.dumps(selected, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""GitHub API access for releases/changelogs with rate-limit aware behavior."""

from __future__ import annotations

import base64
import json
import time
from pathlib import Path
from typing import Any

from utils import cmp_version, ensure_dir, read_json, run_cmd, sleep_with_jitter, write_json

CACHE_PATH = Path.home() / ".cache" / "gh-deps-intel" / "github-api-cache.json"
DEFAULT_TTL_SECONDS = 60 * 60 * 6


class GitHubApiError(RuntimeError):
    pass


class GitHubClient:
    def __init__(
        self,
        mode: str = "safe",
        max_retries: int = 4,
        min_interval_seconds: float = 0.2,
        cache_ttl_seconds: int = DEFAULT_TTL_SECONDS,
    ) -> None:
        self.mode = mode
        self.max_retries = max_retries
        self.min_interval_seconds = min_interval_seconds
        self.cache_ttl_seconds = cache_ttl_seconds
        self.cache = read_json(CACHE_PATH, default={}) or {}
        self.last_request_at = 0.0

    def flush_cache(self) -> None:
        ensure_dir(CACHE_PATH.parent)
        write_json(CACHE_PATH, self.cache)

    def _cache_get(self, key: str) -> Any:
        item = self.cache.get(key)
        if not isinstance(item, dict):
            return None
        ts = item.get("fetched_at")
        if not isinstance(ts, (int, float)):
            return None
        if time.time() - float(ts) > self.cache_ttl_seconds:
            return None
        return item.get("data")

    def _cache_set(self, key: str, data: Any) -> None:
        self.cache[key] = {"fetched_at": time.time(), "data": data}

    def _throttle(self) -> None:
        if self.mode != "safe":
            return
        elapsed = time.time() - self.last_request_at
        wait_for = self.min_interval_seconds - elapsed
        if wait_for > 0:
            sleep_with_jitter(wait_for)

    def _exec_api(self, path: str) -> Any:
        cmd = [
            "gh",
            "api",
            path,
            "--header",
            "Accept: application/vnd.github+json",
            "--header",
            "X-GitHub-Api-Version: 2022-11-28",
        ]

        for attempt in range(self.max_retries + 1):
            self._throttle()
            proc = run_cmd(cmd, check=False)
            self.last_request_at = time.time()
            if proc.returncode == 0:
                try:
                    return json.loads(proc.stdout)
                except json.JSONDecodeError as exc:
                    raise GitHubApiError(f"Invalid JSON from gh api {path}: {exc}") from exc

            stderr = (proc.stderr or "").lower()
            stdout = (proc.stdout or "").lower()
            is_secondary = "secondary rate limit" in stderr or "secondary rate limit" in stdout
            is_rate = "rate limit" in stderr or "rate limit" in stdout
            if attempt < self.max_retries and (is_secondary or is_rate or "http 429" in stderr or "http 403" in stderr):
                delay = min(120, 5 * (2**attempt))
                sleep_with_jitter(delay)
                continue

            raise GitHubApiError(
                f"gh api failed for {path} (attempt {attempt + 1}): rc={proc.returncode}\n"
                f"stderr={proc.stderr[-2000:]}"
            )

        raise GitHubApiError(f"gh api failed for {path}: retries exhausted")

    def _exec_graphql(self, query: str, fields: dict[str, str]) -> Any:
        cmd = [
            "gh",
            "api",
            "graphql",
            "-f",
            f"query={query}",
            "--header",
            "Accept: application/vnd.github+json",
            "--header",
            "X-GitHub-Api-Version: 2022-11-28",
        ]
        for key, value in fields.items():
            cmd.extend(["-F", f"{key}={value}"])

        for attempt in range(self.max_retries + 1):
            self._throttle()
            proc = run_cmd(cmd, check=False)
            self.last_request_at = time.time()
            if proc.returncode == 0:
                try:
                    return json.loads(proc.stdout)
                except json.JSONDecodeError as exc:
                    raise GitHubApiError(f"Invalid JSON from gh graphql: {exc}") from exc

            stderr = (proc.stderr or "").lower()
            if attempt < self.max_retries and ("rate limit" in stderr or "http 429" in stderr or "http 403" in stderr):
                delay = min(120, 5 * (2**attempt))
                sleep_with_jitter(delay)
                continue
            raise GitHubApiError(f"gh graphql failed: rc={proc.returncode} stderr={proc.stderr[-2000:]}")

        raise GitHubApiError("gh graphql failed: retries exhausted")

    def get_json(self, path: str, cache_key: str | None = None) -> Any:
        key = cache_key or f"gh:{path}"
        cached = self._cache_get(key)
        if cached is not None:
            return cached
        data = self._exec_api(path)
        self._cache_set(key, data)
        return data

    def get_paginated_list(self, path: str, max_pages: int = 10) -> list[dict[str, Any]]:
        all_rows: list[dict[str, Any]] = []
        for page in range(1, max_pages + 1):
            sep = "&" if "?" in path else "?"
            page_path = f"{path}{sep}per_page=100&page={page}"
            key = f"gh:{page_path}"
            rows = self.get_json(page_path, cache_key=key)
            if not isinstance(rows, list):
                break
            items = [r for r in rows if isinstance(r, dict)]
            all_rows.extend(items)
            if len(items) < 100:
                break
        return all_rows

    def get_releases(self, owner: str, repo: str) -> list[dict[str, Any]]:
        try:
            rows = self.get_paginated_list(f"repos/{owner}/{repo}/releases")
            if rows:
                return rows
        except GitHubApiError:
            rows = []

        # GraphQL fallback when REST data is unavailable/empty.
        gql = """
query($owner:String!, $repo:String!, $first:Int!) {
  repository(owner:$owner, name:$repo) {
    releases(first:$first, orderBy:{field:CREATED_AT, direction:DESC}) {
      nodes {
        name
        tagName
        publishedAt
        isDraft
        isPrerelease
        url
        description
      }
    }
  }
}
"""
        try:
            data = self._exec_graphql(gql, {"owner": owner, "repo": repo, "first": "100"})
        except GitHubApiError:
            return rows
        nodes = (
            (((data or {}).get("data") or {}).get("repository") or {}).get("releases") or {}
        ).get("nodes")
        if not isinstance(nodes, list):
            return rows
        mapped: list[dict[str, Any]] = []
        for n in nodes:
            if not isinstance(n, dict):
                continue
            mapped.append(
                {
                    "name": n.get("name"),
                    "tag_name": n.get("tagName"),
                    "published_at": n.get("publishedAt"),
                    "draft": bool(n.get("isDraft")),
                    "prerelease": bool(n.get("isPrerelease")),
                    "html_url": n.get("url"),
                    "body": n.get("description") or "",
                }
            )
        return mapped

    def get_tags(self, owner: str, repo: str) -> list[dict[str, Any]]:
        return self.get_paginated_list(f"repos/{owner}/{repo}/tags")

    def get_rate_limit(self) -> dict[str, Any]:
        data = self.get_json("rate_limit")
        return data if isinstance(data, dict) else {}

    def get_compare(self, owner: str, repo: str, base: str, head: str) -> dict[str, Any] | None:
        try:
            data = self.get_json(f"repos/{owner}/{repo}/compare/{base}...{head}")
            return data if isinstance(data, dict) else None
        except Exception:
            return None

    def get_changelog(self, owner: str, repo: str) -> dict[str, Any] | None:
        candidate_paths = [
            "CHANGELOG.md",
            "changelog.md",
            "CHANGES.md",
            "changes.md",
            "docs/CHANGELOG.md",
        ]
        for path in candidate_paths:
            api_path = f"repos/{owner}/{repo}/contents/{path}"
            try:
                data = self.get_json(api_path, cache_key=f"gh:contents:{owner}/{repo}:{path}")
            except Exception:
                continue
            if not isinstance(data, dict):
                continue
            content = data.get("content")
            encoding = data.get("encoding")
            if isinstance(content, str) and encoding == "base64":
                try:
                    decoded = base64.b64decode(content).decode("utf-8", errors="ignore")
                except Exception:
                    continue
                return {
                    "path": path,
                    "text": decoded,
                    "html_url": data.get("html_url"),
                }
        return None


def _coerce_version(tag: str | None) -> str | None:
    if not tag:
        return None
    t = str(tag).strip()
    if not t:
        return None
    t = t.split("/")[-1]
    if t.lower().startswith("release-"):
        t = t[len("release-") :]
    if t.lower().startswith("v"):
        t = t[1:]
    return t


def filter_releases_between(
    releases: list[dict[str, Any]],
    current_version: str | None,
    target_version: str | None,
    max_items: int = 25,
) -> list[dict[str, Any]]:
    current = _coerce_version(current_version)
    target = _coerce_version(target_version)

    normalized: list[dict[str, Any]] = []
    for rel in releases:
        tag = rel.get("tag_name") or rel.get("name")
        ver = _coerce_version(tag)
        normalized.append(
            {
                "name": rel.get("name") or tag,
                "tag_name": rel.get("tag_name"),
                "version": ver,
                "published_at": rel.get("published_at"),
                "html_url": rel.get("html_url"),
                "body": rel.get("body") or "",
                "draft": bool(rel.get("draft")),
                "prerelease": bool(rel.get("prerelease")),
            }
        )

    normalized = [x for x in normalized if not x["draft"]]

    selected: list[dict[str, Any]] = []
    for rel in normalized:
        ver = rel.get("version")
        if not ver:
            continue
        if current and cmp_version(ver, current) <= 0:
            continue
        if target and cmp_version(ver, target) > 0:
            continue
        selected.append(rel)

    if not selected:
        # If we have an explicit version window but cannot confidently map releases
        # into that window, return none instead of injecting potentially unrelated notes.
        if current or target:
            return []
        selected = normalized[:max_items]

    # Stable latest-first ordering.
    selected.sort(key=lambda x: x.get("published_at") or "", reverse=True)
    return selected[:max_items]