Python编程：高效检查GitLab群组中多个仓库文件存在性

pip install requests

import requests
from urllib.parse import quote_plus
import json
import os

# 从环境变量获取配置信息
gitlab_api_url = os.getenv("GITLAB_API_URL", "https://gitlab.com/api/v4")
private_token = os.getenv("GITLAB_PRIVATE_TOKEN", "your_private_token_here") # 建议通过环境变量设置
group_id = os.getenv("GITLAB_GROUP_ID", "your_group_id_here") # 建议通过环境变量设置

if private_token == "your_private_token_here" or group_id == "your_group_id_here":
    print("警告：请设置 GITLAB_PRIVATE_TOKEN 和 GITLAB_GROUP_ID 环境变量。")
    print("例如：export GITLAB_PRIVATE_TOKEN='your_token' && export GITLAB_GROUP_ID='your_group_id'")
    exit(1)

def get_all_paginated_results(endpoint, headers):
    """
    通用函数，处理 GitLab API 的分页请求。
    """
    all_results = []
    page = 1
    while True:
        paginated_endpoint = f"{endpoint}?per_page=100&page={page}" # 提高 per_page 到最大值
        response = requests.get(paginated_endpoint, headers=headers)
        if response.status_code == 200:
            current_page_results = response.json()
            if not current_page_results:
                break # 没有更多数据
            all_results.extend(current_page_results)
            if len(current_page_results) < 100: # 如果当前页结果少于 per_page，说明是最后一页
                break
            page += 1
        else:
            print(f"请求失败: {endpoint}. 状态码: {response.status_code}, 错误信息: {response.text}")
            return []
    return all_results

def get_project_ids(api_url, private_token, group_id):
    """
    获取指定群组下的所有项目ID和名称。
    """
    endpoint = f"{api_url}/groups/{group_id}/projects"
    headers = {"PRIVATE-TOKEN": private_token}

    projects = get_all_paginated_results(endpoint, headers)

    if projects:
        return [(project['id'], project['name']) for project in projects]
    else:
        print(f"未能为群组 {group_id} 检索到项目。")
        return []

def check_files_in_project(api_url, private_token, project_id, project_name, filenames, ref="main"):
    """
    在指定项目中检查多个文件的存在性。
    通过获取完整的递归文件树，然后在内存中进行过滤。
    """
    headers = {"PRIVATE-TOKEN": private_token}
    output_data = {
        "project_id": project_id,
        "project_name": project_name,
        "files": []
    }

    print(f"\n正在检查项目: {project_name} (ID: {project_id})")

    # 构建获取整个递归文件树的API端点，不包含 path 参数
    tree_endpoint = f"{api_url}/projects/{project_id}/repository/tree?recursive=1&ref={ref}"

    # 获取完整的递归文件树，处理分页
    repository_tree = get_all_paginated_results(tree_endpoint, headers)

    if not repository_tree:
        print(f"未能检索到项目 {project_name} 的仓库树。")
        for filename in filenames:
            output_data["files"].append({"filename": filename, "file_found": False})
        return output_data

    # 将文件树转换为集合，以便快速查找
    tree_names = {item.get("name") for item in repository_tree if item.get("type") == "blob"}
    tree_paths = {item.get("path") for item in repository_tree if item.get("type") == "blob"}

    for filename in filenames:
        # 检查文件是否存在于文件树中（通过文件名或完整路径）
        # GitLab API返回的`name`是文件名，`path`是完整路径
        file_found = False
        for item in repository_tree:
            if item.get("type") == "blob": # 确保是文件而不是目录
                if item.get("name") == filename:
                    file_found = True
                    break
                # 如果文件名可能包含路径，也可以检查完整路径
                # if item.get("path") == filename: # 仅当 filename 是完整路径时
                #     file_found = True
                #     break

        print(f"  文件 '{filename}' 在项目 '{project_name}' 中存在: {file_found}")
        output_data["files"].append({"filename": filename, "file_found": file_found})

    return output_data

def main():
    # 获取群组中的所有项目信息
    projects_info = get_project_ids(gitlab_api_url, private_token, group_id)

    # 指定要检查的文件名列表
    filenames_to_check = ["serverless.yaml", "serverless.yml", "package.json", "README.md"] # 示例文件列表

    # 存储所有项目的检查结果
    all_projects_results = []

    # 遍历每个项目并检查文件
    for project_id, project_name in projects_info:
        project_result = check_files_in_project(
            gitlab_api_url, private_token, project_id, project_name, filenames_to_check, ref="dev" # 指定检查 dev 分支
        )
        project_result["group_id"] = group_id # 添加群组ID到每个项目结果中
        all_projects_results.append(project_result)

    # 将所有结果写入单个JSON文件
    output_filename = "gitlab_file_check_results.json"
    try:
        with open(output_filename, 'w', encoding='utf-8') as json_file:
            json.dump(all_projects_results, json_file, indent=2, ensure_ascii=False)
        print(f"\n所有项目的检查结果已保存到 '{output_filename}'")
    except IOError as e:
        print(f"写入文件失败: {e}")

if __name__ == "__main__":
    main()