GitLab REST APIで複数リポジトリのnodeパッケージバージョンを一括チェックする

# uv のインストール
curl -LsSf https://astral.sh/uv/install.sh | sh

# プロジェクト作成と依存パッケージの追加
uv init scan-node-versions
cd scan-node-versions
uv add httpx python-dotenv

# uv のインストール
curl -LsSf https://astral.sh/uv/install.sh | sh

# プロジェクト作成と依存パッケージの追加
uv init scan-node-versions
cd scan-node-versions
uv add httpx python-dotenv

"""GitLab上のリポジトリを横断してnodeパッケージのバージョンを調べるスクリプト。"""

import argparse
import base64
import json
import os
import sys
from urllib.parse import quote

import httpx
from dotenv import load_dotenv

# 対象とするロックファイル名
LOCKFILES = ("package-lock.json", "yarn.lock", "pnpm-lock.yaml")


def gitlab_client(base_url: str, token: str) -> httpx.Client:
    """GitLab API用のhttpxクライアントを生成する。"""
    return httpx.Client(
        base_url=base_url.rstrip("/") + "/api/v4",
        headers={"PRIVATE-TOKEN": token},
        timeout=30.0,
    )


def iter_paginated(client: httpx.Client, path: str, params: dict) -> list:
    """ページネーション付きエンドポイントを最後までたどる。"""
    results = []
    page = 1
    while True:
        resp = client.get(path, params={**params, "page": page, "per_page": 100})
        resp.raise_for_status()
        items = resp.json()
        if not items:
            break
        results.extend(items)
        # 次ページの有無はレスポンスヘッダーで判定する
        next_page = resp.headers.get("x-next-page")
        if not next_page:
            break
        page = int(next_page)
    return results


def list_projects(client: httpx.Client, group: str | None) -> list[dict]:
    """ネームスペースまたはグループ配下のプロジェクト一覧を取得する。"""
    if group:
        # グループ配下のプロジェクトをサブグループも含めて取得する
        path = f"/groups/{quote(group, safe='')}/projects"
        params = {"include_subgroups": "true", "archived": "false"}
    else:
        # 自分が所属しているプロジェクトのみを対象にする
        path = "/projects"
        params = {"membership": "true", "archived": "false"}
    return iter_paginated(client, path, params)


def list_lockfiles(client: httpx.Client, project_id: int, ref: str) -> list[dict]:
    """リポジトリツリーを再帰的に走査してロックファイルだけを抽出する。"""
    path = f"/projects/{project_id}/repository/tree"
    params = {"recursive": "true", "ref": ref}
    try:
        items = iter_paginated(client, path, params)
    except httpx.HTTPStatusError as e:
        # 空リポジトリなどでは404が返るのでスキップする
        if e.response.status_code in (403, 404):
            return []
        raise
    return [item for item in items if item["name"] in LOCKFILES]


def fetch_file(client: httpx.Client, project_id: int, file_path: str, ref: str) -> str:
    """指定ファイルの中身を文字列で取得する。"""
    encoded = quote(file_path, safe="")
    resp = client.get(f"/projects/{project_id}/repository/files/{encoded}",
                      params={"ref": ref})
    resp.raise_for_status()
    data = resp.json()
    # contentはbase64エンコードされている
    return base64.b64decode(data["content"]).decode("utf-8", errors="replace")


def extract_versions(content: str, filename: str, package: str) -> list[str]:
    """ロックファイルの中身から対象パッケージのバージョンを抜き出す。"""
    versions: list[str] = []
    if filename == "package-lock.json":
        try:
            doc = json.loads(content)
        except json.JSONDecodeError:
            return versions
        # lockfileVersion 2/3はpackagesキー配下にフラットに格納される
        for key, meta in (doc.get("packages") or {}).items():
            # キーは "node_modules/<name>" または "node_modules/<scope>/<name>"
            if key.endswith(f"node_modules/{package}") and "version" in meta:
                versions.append(meta["version"])
        # 古いlockfileVersion 1向けのフォールバック
        if not versions:
            deps = doc.get("dependencies") or {}
            if package in deps and "version" in deps[package]:
                versions.append(deps[package]["version"])
    elif filename == "yarn.lock":
        # yarn.lockはテキスト形式なので簡易パースで対応する
        for block in content.split("\n\n"):
            header = block.splitlines()[0] if block else ""
            if f'"{package}@' in header or header.startswith(f"{package}@"):
                for line in block.splitlines():
                    line = line.strip()
                    if line.startswith("version "):
                        versions.append(line.split()[1].strip('"'))
                        break
    elif filename == "pnpm-lock.yaml":
        # pnpmは "/<name>@<version>:" の形式でパッケージが現れる
        prefix = f"/{package}@"
        for line in content.splitlines():
            line = line.strip()
            if line.startswith(prefix) and line.endswith(":"):
                versions.append(line[len(prefix):-1])
    return sorted(set(versions))


def main() -> int:
    load_dotenv()
    parser = argparse.ArgumentParser(description="GitLabリポジトリを横断してnodeパッケージのバージョンを集計する")
    parser.add_argument("package", help="調べたいパッケージ名（例: axios, next）")
    parser.add_argument("--gitlab-url", default=os.environ.get("GITLAB_URL", "https://gitlab.com"))
    parser.add_argument("--group", default=None, help="対象のグループパス（省略時は自分の所属プロジェクト）")
    args = parser.parse_args()

    token = os.environ.get("GITLAB_TOKEN")
    if not token:
        print("環境変数 GITLAB_TOKEN を設定してください。", file=sys.stderr)
        return 1

    with gitlab_client(args.gitlab_url, token) as client:
        projects = list_projects(client, args.group)
        print(f"対象プロジェクト数: {len(projects)}", file=sys.stderr)

        hits: list[tuple[str, str, str, str]] = []
        for project in projects:
            ref = project.get("default_branch")
            if not ref:
                continue
            lockfiles = list_lockfiles(client, project["id"], ref)
            for lf in lockfiles:
                content = fetch_file(client, project["id"], lf["path"], ref)
                versions = extract_versions(content, lf["name"], args.package)
                for v in versions:
                    hits.append((project["path_with_namespace"], lf["path"], lf["name"], v))

    # 結果をTSVで出力する（spreadsheetに貼り付けやすいため）
    print("project\tlockfile\ttype\tversion")
    for project_path, file_path, filetype, version in hits:
        print(f"{project_path}\t{file_path}\t{filetype}\t{version}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""GitLab上のリポジトリを横断してnodeパッケージのバージョンを調べるスクリプト。"""

import argparse
import base64
import json
import os
import sys
from urllib.parse import quote

import httpx
from dotenv import load_dotenv

# 対象とするロックファイル名
LOCKFILES = ("package-lock.json", "yarn.lock", "pnpm-lock.yaml")


def gitlab_client(base_url: str, token: str) -> httpx.Client:
    """GitLab API用のhttpxクライアントを生成する。"""
    return httpx.Client(
        base_url=base_url.rstrip("/") + "/api/v4",
        headers={"PRIVATE-TOKEN": token},
        timeout=30.0,
    )


def iter_paginated(client: httpx.Client, path: str, params: dict) -> list:
    """ページネーション付きエンドポイントを最後までたどる。"""
    results = []
    page = 1
    while True:
        resp = client.get(path, params={**params, "page": page, "per_page": 100})
        resp.raise_for_status()
        items = resp.json()
        if not items:
            break
        results.extend(items)
        # 次ページの有無はレスポンスヘッダーで判定する
        next_page = resp.headers.get("x-next-page")
        if not next_page:
            break
        page = int(next_page)
    return results


def list_projects(client: httpx.Client, group: str | None) -> list[dict]:
    """ネームスペースまたはグループ配下のプロジェクト一覧を取得する。"""
    if group:
        # グループ配下のプロジェクトをサブグループも含めて取得する
        path = f"/groups/{quote(group, safe='')}/projects"
        params = {"include_subgroups": "true", "archived": "false"}
    else:
        # 自分が所属しているプロジェクトのみを対象にする
        path = "/projects"
        params = {"membership": "true", "archived": "false"}
    return iter_paginated(client, path, params)


def list_lockfiles(client: httpx.Client, project_id: int, ref: str) -> list[dict]:
    """リポジトリツリーを再帰的に走査してロックファイルだけを抽出する。"""
    path = f"/projects/{project_id}/repository/tree"
    params = {"recursive": "true", "ref": ref}
    try:
        items = iter_paginated(client, path, params)
    except httpx.HTTPStatusError as e:
        # 空リポジトリなどでは404が返るのでスキップする
        if e.response.status_code in (403, 404):
            return []
        raise
    return [item for item in items if item["name"] in LOCKFILES]


def fetch_file(client: httpx.Client, project_id: int, file_path: str, ref: str) -> str:
    """指定ファイルの中身を文字列で取得する。"""
    encoded = quote(file_path, safe="")
    resp = client.get(f"/projects/{project_id}/repository/files/{encoded}",
                      params={"ref": ref})
    resp.raise_for_status()
    data = resp.json()
    # contentはbase64エンコードされている
    return base64.b64decode(data["content"]).decode("utf-8", errors="replace")


def extract_versions(content: str, filename: str, package: str) -> list[str]:
    """ロックファイルの中身から対象パッケージのバージョンを抜き出す。"""
    versions: list[str] = []
    if filename == "package-lock.json":
        try:
            doc = json.loads(content)
        except json.JSONDecodeError:
            return versions
        # lockfileVersion 2/3はpackagesキー配下にフラットに格納される
        for key, meta in (doc.get("packages") or {}).items():
            # キーは "node_modules/<name>" または "node_modules/<scope>/<name>"
            if key.endswith(f"node_modules/{package}") and "version" in meta:
                versions.append(meta["version"])
        # 古いlockfileVersion 1向けのフォールバック
        if not versions:
            deps = doc.get("dependencies") or {}
            if package in deps and "version" in deps[package]:
                versions.append(deps[package]["version"])
    elif filename == "yarn.lock":
        # yarn.lockはテキスト形式なので簡易パースで対応する
        for block in content.split("\n\n"):
            header = block.splitlines()[0] if block else ""
            if f'"{package}@' in header or header.startswith(f"{package}@"):
                for line in block.splitlines():
                    line = line.strip()
                    if line.startswith("version "):
                        versions.append(line.split()[1].strip('"'))
                        break
    elif filename == "pnpm-lock.yaml":
        # pnpmは "/<name>@<version>:" の形式でパッケージが現れる
        prefix = f"/{package}@"
        for line in content.splitlines():
            line = line.strip()
            if line.startswith(prefix) and line.endswith(":"):
                versions.append(line[len(prefix):-1])
    return sorted(set(versions))


def main() -> int:
    load_dotenv()
    parser = argparse.ArgumentParser(description="GitLabリポジトリを横断してnodeパッケージのバージョンを集計する")
    parser.add_argument("package", help="調べたいパッケージ名（例: axios, next）")
    parser.add_argument("--gitlab-url", default=os.environ.get("GITLAB_URL", "https://gitlab.com"))
    parser.add_argument("--group", default=None, help="対象のグループパス（省略時は自分の所属プロジェクト）")
    args = parser.parse_args()

    token = os.environ.get("GITLAB_TOKEN")
    if not token:
        print("環境変数 GITLAB_TOKEN を設定してください。", file=sys.stderr)
        return 1

    with gitlab_client(args.gitlab_url, token) as client:
        projects = list_projects(client, args.group)
        print(f"対象プロジェクト数: {len(projects)}", file=sys.stderr)

        hits: list[tuple[str, str, str, str]] = []
        for project in projects:
            ref = project.get("default_branch")
            if not ref:
                continue
            lockfiles = list_lockfiles(client, project["id"], ref)
            for lf in lockfiles:
                content = fetch_file(client, project["id"], lf["path"], ref)
                versions = extract_versions(content, lf["name"], args.package)
                for v in versions:
                    hits.append((project["path_with_namespace"], lf["path"], lf["name"], v))

    # 結果をTSVで出力する（spreadsheetに貼り付けやすいため）
    print("project\tlockfile\ttype\tversion")
    for project_path, file_path, filetype, version in hits:
        print(f"{project_path}\t{file_path}\t{filetype}\t{version}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

# 自分が所属するすべてのプロジェクトを対象にする
uv run python scan_node_versions.py axios

# 特定のグループ配下のみを対象にする
uv run python scan_node_versions.py axios --group my-team/web

# 自分が所属するすべてのプロジェクトを対象にする
uv run python scan_node_versions.py axios

# 特定のグループ配下のみを対象にする
uv run python scan_node_versions.py axios --group my-team/web