gss.py

#!/usr/bin/env python3
"""
Fastest approach: GitLab Group Search API (scope=blobs) with extension filters.
Finds matches across ALL projects under a group without scanning each repo tree.

Prereqs:
  pip install -r requirements.txt

Usage:
  # Option A: .env file in the project root
  # GITLAB_URL="https://gitlab.com"
  # GITLAB_TOKEN="glpat-xxxx"
  #
  # Option B: exported environment variables
  export GITLAB_URL="https://gitlab.com"
  export GITLAB_TOKEN="glpat-xxxx"

  python gss.py \
    --group-path platform \
    --search "kaniko-project/executor:debug" \
    --out kaniko_old_image.xlsx

Optional:
  --search-type basic   # override default "advanced"
  --no-fallback         # skip per-project fallback (faster, but may miss matches)
  --max-workers 6       # worker threads for fallback (only used if fallback runs)

If group blob search isn't available on your GitLab instance, the script
automatically falls back to per-project blob search (slower but works on basic search),
unless --no-fallback is set.
"""

import argparse
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Dict, List, Any
import requests
import pandas as pd

VERSION = "0.1.0"


def load_env_file(path: str = ".env") -> None:
    if not os.path.isfile(path):
        return
    try:
        with open(path, "r", encoding="utf-8") as f:
            for raw in f:
                line = raw.strip()
                if not line or line.startswith("#"):
                    continue
                if line.startswith("export "):
                    line = line[len("export "):].lstrip()
                if "=" not in line:
                    continue
                key, val = line.split("=", 1)
                key = key.strip()
                val = val.strip()
                if not key:
                    continue
                if (val.startswith('"') and val.endswith('"')) or (val.startswith("'") and val.endswith("'")):
                    val = val[1:-1]
                if key not in os.environ:
                    os.environ[key] = val
    except OSError as e:
        print(f"Warning: failed to read {path}: {e}")


def gl_headers(token: str) -> dict:
    return {"PRIVATE-TOKEN": token}


def request_with_retries(session, method, url, headers, params=None, timeout=30, retries=5):
    backoff = 0.7
    last_exc = None
    for attempt in range(retries + 1):
        try:
            r = session.request(method, url, headers=headers, params=params, timeout=timeout)
            if r.status_code in (429, 500, 502, 503, 504):
                sleep_s = backoff * (2 ** attempt)
                ra = r.headers.get("Retry-After")
                if ra and ra.isdigit():
                    sleep_s = max(sleep_s, int(ra))
                time.sleep(sleep_s)
                continue
            return r
        except requests.RequestException as e:
            last_exc = e
            time.sleep(backoff * (2 ** attempt))
    raise last_exc or RuntimeError("request failed after retries")


def paginate(session, url, headers, params=None):
    page = 1
    per_page = 100
    while True:
        p = dict(params or {})
        p.update({"page": page, "per_page": per_page})
        r = request_with_retries(session, "GET", url, headers, params=p)
        r.raise_for_status()
        items = r.json()
        if not isinstance(items, list):
            yield items
            return
        for it in items:
            yield it
        next_page = r.headers.get("X-Next-Page")
        if not next_page:
            return
        page = int(next_page)


def get_group(session, base_url, token, group_path: str) -> dict:
    enc = requests.utils.quote(group_path, safe="")
    url = f"{base_url}/api/v4/groups/{enc}"
    r = request_with_retries(session, "GET", url, gl_headers(token))
    r.raise_for_status()
    return r.json()


def list_group_projects_map(session, base_url, token, group_id: int, include_subgroups=True) -> Dict[int, dict]:
    url = f"{base_url}/api/v4/groups/{group_id}/projects"
    params = {
        "include_subgroups": "true" if include_subgroups else "false",
        "archived": "false",
        "simple": "true",
    }
    m = {}
    for p in paginate(session, url, gl_headers(token), params=params):
        m[p["id"]] = {
            "project_path": p.get("path_with_namespace"),
            "project_url": p.get("web_url"),
            "last_activity_at": p.get("last_activity_at"),
        }
    return m




def group_search_blobs(session, base_url, token, group_id_or_path: str, query: str, search_type: str | None):
    url = f"{base_url}/api/v4/groups/{group_id_or_path}/search"
    params = {"scope": "blobs", "search": query}
    if search_type:
        params["search_type"] = search_type  # basic|advanced
    return list(paginate(session, url, gl_headers(token), params=params))


def project_search_blobs(session, base_url, token, project_id: int, query: str, search_type: str | None = None):
    url = f"{base_url}/api/v4/projects/{project_id}/search"
    params = {"scope": "blobs", "search": query}
    if search_type:
        params["search_type"] = search_type  # basic|advanced (if supported)
    return list(paginate(session, url, gl_headers(token), params=params))


def normalize_extensions(raw: str | None) -> list[str]:
    if not raw:
        return []
    raw = raw.strip()
    if raw in ("*", "all"):
        return []
    parts = []
    for p in raw.split(","):
        p = p.strip().lower()
        if not p:
            continue
        if p.startswith("."):
            p = p[1:]
        if p:
            parts.append(p)
    out: list[str] = []
    seen = set()
    for p in parts:
        if p in seen:
            continue
        seen.add(p)
        out.append(p)
    return out


def is_allowed_path(path: str | None, exts: list[str]) -> bool:
    if not path:
        return False
    if not exts:
        return True
    lower = path.lower()
    return any(lower.endswith(f".{e}") for e in exts)


def extract_project_team(project_path: str | None, group_path: str) -> str:
    if not project_path:
        return ""
    path_parts = [p for p in project_path.split("/") if p]
    group_parts = [p for p in (group_path or "").split("/") if p]
    if group_parts and path_parts[: len(group_parts)] == group_parts:
        remainder = path_parts[len(group_parts):]
        return remainder[0] if remainder else ""
    if len(path_parts) >= 2:
        return path_parts[1]
    return path_parts[0] if path_parts else ""


def status_code(err: requests.HTTPError) -> int | None:
    resp = getattr(err, "response", None)
    return getattr(resp, "status_code", None)


def fallback_project_search(
    base_url,
    token,
    proj_map: Dict[int, dict],
    search: str,
    search_type: str | None,
    exts: list[str],
    max_workers: int,
) -> list[dict]:
    hits: list[dict] = []
    total = len(proj_map)

    def search_project(pid: int) -> list[dict]:
        session = requests.Session()
        try:
            results = project_search_blobs(session, base_url, token, pid, search, search_type=search_type)
        except requests.HTTPError as e:
            sc = status_code(e)
            if search_type and sc == 400:
                try:
                    results = project_search_blobs(session, base_url, token, pid, search, search_type=None)
                except requests.HTTPError as e2:
                    sc = status_code(e2)
                    if sc in (403, 404):
                        print(f"Skipping project {pid}: search not allowed (HTTP {sc})")
                        return []
                    raise
            elif sc in (403, 404):
                print(f"Skipping project {pid}: search not allowed (HTTP {sc})")
                return []
            else:
                raise
        filtered: list[dict] = []
        for h in results:
            path = h.get("path") or h.get("filename")
            if not is_allowed_path(path, exts):
                continue
            if "project_id" not in h:
                h["project_id"] = pid
            filtered.append(h)
        return filtered

    if max_workers <= 1:
        for idx, pid in enumerate(proj_map.keys(), 1):
            if idx == 1 or idx % 50 == 0 or idx == total:
                print(f"Project search progress: {idx}/{total}")
            hits.extend(search_project(pid))
        return hits

    print(f"Project search fallback with {max_workers} workers...")
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(search_project, pid): pid for pid in proj_map.keys()}
        done = 0
        for future in as_completed(futures):
            done += 1
            if done == 1 or done % 50 == 0 or done == total:
                print(f"Project search progress: {done}/{total}")
            hits.extend(future.result())
    return hits


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--version", action="version", version=f"%(prog)s {VERSION}")
    ap.add_argument("--group-path", required=True, help="e.g. ultim or ultim/subgroup")
    ap.add_argument("--search", required=True, help="string to search in files")
    ap.add_argument("--out", default="group_yaml_hits.xlsx")
    ap.add_argument("--search-type", default="advanced", help="optional: basic|advanced (default: advanced)")
    ap.add_argument("--include-subgroups", action="store_true", default=True)
    ap.add_argument("--no-fallback", action="store_true", help="do not scan per-project on group search failure")
    ap.add_argument(
        "--extensions",
        default="yml,yaml",
        help="comma-separated list of extensions (default: yml,yaml). Use '*' or 'all' for no filter.",
    )
    ap.add_argument("--max-workers", type=int, default=6, help="workers for per-project fallback (default: 6)")
    args = ap.parse_args()
    if args.max_workers < 1:
        raise SystemExit("--max-workers must be >= 1")

    load_env_file()
    base_url = os.environ.get("GITLAB_URL")
    token = os.environ.get("GITLAB_TOKEN")
    if not base_url or not token:
        raise SystemExit("Missing env vars: GITLAB_URL and/or GITLAB_TOKEN")
    base_url = base_url.rstrip("/")

    s = requests.Session()

    group = get_group(s, base_url, token, args.group_path)
    group_id = group["id"]
    group_enc = requests.utils.quote(args.group_path, safe="")  # usable as :id for group search endpoint
    print(f"Group: {args.group_path} (id={group_id}) {group.get('web_url')}")
    proj_map: Dict[int, dict] | None = None

    exts = normalize_extensions(args.extensions)
    if exts:
        queries = [f"{args.search} extension:{ext}" for ext in exts]
    else:
        queries = [args.search]

    hits = None
    try:
        hits = []
        for q in queries:
            hits.extend(group_search_blobs(s, base_url, token, group_enc, q, args.search_type))
    except requests.HTTPError as e:
        sc = status_code(e)
        if args.search_type and sc == 400:
            print(f"Group search with search_type={args.search_type} failed (HTTP 400). Retrying without search_type...")
            try:
                hits = []
                for q in queries:
                    hits.extend(group_search_blobs(s, base_url, token, group_enc, q, None))
            except requests.HTTPError as e2:
                e = e2
                sc = status_code(e2)
        if hits is None:
            if sc == 400:
                if args.no_fallback:
                    raise SystemExit(
                        "Group blob search isn't available (HTTP 400) and --no-fallback was set."
                    )
                print("Group blob search isn't available (HTTP 400). Falling back to per-project search (slower).")
                proj_map = list_group_projects_map(
                    s,
                    base_url,
                    token,
                    group_id,
                    include_subgroups=args.include_subgroups,
                )
                print(f"Projects in group: {len(proj_map)}")
                hits = fallback_project_search(
                    base_url,
                    token,
                    proj_map,
                    args.search,
                    args.search_type,
                    exts,
                    args.max_workers,
                )
            else:
                raise SystemExit(
                    f"Group blob search failed. This usually means advanced search / exact code search isn't enabled "
                    f"for blobs on this instance. Error: {e}"
                )
    print(f"Raw hits: {len(hits)}")

    if hits and proj_map is None:
        proj_map = list_group_projects_map(s, base_url, token, group_id, include_subgroups=args.include_subgroups)
        print(f"Projects in group: {len(proj_map)}")
    if proj_map is None:
        proj_map = {}

    # Normalize + enrich (unique by project + file)
    rows: List[Dict[str, Any]] = []
    seen: set[tuple[int | None, str | None]] = set()
    for h in hits:
        pid = h.get("project_id")
        path = h.get("path") or h.get("filename")
        if not path:
            continue
        key = (pid, path)
        if key in seen:
            continue
        seen.add(key)

        meta = proj_map.get(pid, {})
        project_path = meta.get("project_path")
        project_url = meta.get("project_url")
        if not project_path and project_url and project_url.startswith(base_url):
            project_path = project_url[len(base_url):].lstrip("/")
        if not project_url and project_path:
            project_url = f"{base_url}/{project_path}"
        rows.append({
            "project_team": extract_project_team(project_path, args.group_path),
            "project_path": project_url or project_path or "",
            "target_file": path or "",
            "status": "",
        })

    with pd.ExcelWriter(args.out, engine="openpyxl") as w:
        df = pd.DataFrame(rows, columns=["project_team", "project_path", "target_file", "status"])
        if df.empty:
            print("No matches found.")
        df.sort_values(by=["project_team", "project_path", "target_file"]).to_excel(
            w,
            index=False,
            sheet_name="results",
        )

    print(f"Done. Wrote: {args.out}")


if __name__ == "__main__":
    main()