将 GitLab 指定群组下的所有项目设置为内部可见

GitLab
1 
#!/usr/bin/env python3
"""
将 GitLab 指定群组下的所有项目设置为内部可见。

Usage:
    python3 set-internal.py <group>
    python3 set-internal.py <group> --yes   # 跳过确认

Environment variables:
    GITLAB_HOST     GitLab 地址 (required)
    GITLAB_TOKEN    GitLab Personal Access Token (required, scope: api)
"""

import json
import os
import sys

from urllib.parse import urlencode
from urllib.request import Request, urlopen


def main() -> None:
    host = os.environ.get("GITLAB_HOST")
    if not host:
        sys.exit("错误: 请设置 GITLAB_HOST 环境变量")

    token = os.environ.get("GITLAB_TOKEN")
    if not token:
        sys.exit("错误: 请设置 GITLAB_TOKEN 环境变量")

    if len(sys.argv) < 2:
        sys.exit(f"用法: {sys.argv[0]} <group> [--yes]")

    group = sys.argv[1]
    auto_yes = "--yes" in sys.argv

    api_base = f"https://{host}/api/v4"

    # 1. 查找群组
    group_id = _find_group(api_base, token, group)
    if not group_id:
        sys.exit(f"错误: 未找到群组 '{group}'")

    print(f"==> 群组: {group} (ID: {group_id})\n")

    # 2. 获取所有项目
    projects = _list_projects(api_base, token, group_id)
    if not projects:
        print(f"群组 '{group}' 下没有项目。")
        return

    # 3. 列出需要修改的项目
    to_change = [p for p in projects if p["visibility"] != "internal"]
    skipped = [p for p in projects if p["visibility"] == "internal"]

    print("==> 以下项目将被设置为 internal:\n")
    for p in to_change:
        print(f"  {p['id']}\t{p['path_with_namespace']}\t{p['visibility']}\t-> internal")

    if skipped:
        print(f"\n  (另外 {len(skipped)} 个项目已是 internal, 将跳过)")

    if not to_change:
        print("\n所有项目已是 internal 可见性，无需修改。")
        return

    print(f"\n共 {len(to_change)} 个项目待修改。")

    # 4. 确认
    if not auto_yes:
        answer = input("\n确认继续? (y/N) ")
        if answer.lower() != "y":
            print("已取消。")
            return

    # 5. 执行修改
    print("\n==> 开始修改...")
    success = 0
    failed = 0

    for p in to_change:
        name = p["path_with_namespace"]
        print(f"  {name}: ", end="", flush=True)
        try:
            _api_put(api_base, token, f"/projects/{p['id']}", {"visibility": "internal"})
            print("OK")
            success += 1
        except ApiError as e:
            print(f"失败 (HTTP {e.status})")
            failed += 1

    print(f"\n==> 完成: 成功 {success}, 失败 {failed}")


class ApiError(Exception):
    def __init__(self, status: int, body: str) -> None:
        super().__init__(f"HTTP {status}")
        self.status = status
        self.body = body


def _api_get(api_base: str, token: str, path: str, params: dict | None = None) -> list | dict:
    url = f"{api_base}{path}"
    if params:
        url += "?" + urlencode(params)
    req = Request(url, headers={"PRIVATE-TOKEN": token})
    with urlopen(req) as resp:
        return json.loads(resp.read())


def _api_put(api_base: str, token: str, path: str, data: dict) -> None:
    url = f"{api_base}{path}"
    body = urlencode(data).encode()
    req = Request(url, data=body, method="PUT", headers={"PRIVATE-TOKEN": token})
    with urlopen(req) as resp:
        if resp.status != 200:
            raise ApiError(resp.status, resp.read().decode(errors="replace"))


def _find_group(api_base: str, token: str, group: str) -> int | None:
    """按 path 或 full_path 查找群组 ID."""
    groups = _api_get(api_base, token, "/groups", {"search": group, "per_page": 100})
    for g in groups:
        if g.get("full_path") == group or g.get("path") == group:
            return g["id"]
    return None


def _list_projects(api_base: str, token: str, group_id: int) -> list[dict]:
    """分页获取群组下所有项目 (含子群组)."""
    projects: list[dict] = []
    page = 1

    while True:
        batch = _api_get(
            api_base, token, f"/groups/{group_id}/projects",
            {"per_page": 100, "page": page, "include_subgroups": True},
        )
        if not batch:
            break
        projects.extend(batch)
        page += 1

    return projects


if __name__ == "__main__":
    main()