自动化Python脚本检查GitLab仓库文件存在性与API实践指南

# 错误用法：将文件名作为path参数
endpoint = f"{api_url}/projects/{project_id}/repository/tree?recursive=1&path={encoded_filename}&ref=dev"

# 正确用法：移除path参数，获取整个仓库树
endpoint = f"{api_url}/projects/{project_id}/repository/tree?recursive=1&ref=dev"

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

# 配置GitLab API信息
# 建议将敏感信息通过环境变量设置，以增强安全性
gitlab_api_url = os.getenv("GITLAB_API_URL", "https://gitlab.com/api/v4")
private_token = os.getenv("GITLAB_PRIVATE_TOKEN", "your_private_token_here") # 请替换为您的实际Token
group_id = os.getenv("GITLAB_GROUP_ID", "your_group_id_here") # 请替换为您的实际Group ID

def get_project_ids(api_url: str, private_token: str, group_id: str) -> list:
    """
    获取指定群组下的所有项目ID和名称。
    处理分页以确保获取所有项目。
    """
    projects_info = []
    page = 1
    while True:
        endpoint = f"{api_url}/groups/{group_id}/projects?per_page=100&page={page}" # per_page可适当增大
        headers = {"PRIVATE-TOKEN": private_token}

        print(f"Fetching projects from group {group_id}, page {page}...")
        response = requests.get(endpoint, headers=headers)

        if response.status_code == 200:
            projects = response.json()
            if not projects: # 如果当前页没有数据，说明已到最后一页
                break
            projects_info.extend([(project['id'], project['name']) for project in projects])
            page += 1
        else:
            print(f"Failed to retrieve projects for group {group_id}. Status code: {response.status_code}, Response: {response.text}")
            break # 发生错误时退出循环
    return projects_info

def check_files_in_project(api_url: str, private_token: str, project_id: int, project_name: str, filenames: list) -> dict:
    """
    检查指定项目中的文件是否存在，并处理仓库树的分页。
    """
    headers = {"PRIVATE-TOKEN": private_token}
    output_data = {
        "project_id": project_id,
        "project_name": project_name,
        "files": []
    }

    all_repo_items = []
    page = 1
    while True:
        # 正确的API端点：不包含path参数，使用recursive=1获取完整树
        endpoint = f"{api_url}/projects/{project_id}/repository/tree?recursive=1&ref=dev&per_page=100&page={page}"

        print(f"  Checking repository tree for project '{project_name}' (ID: {project_id}), page {page}...")
        response = requests.get(endpoint, headers=headers)

        if response.status_code == 200:
            repository_tree = response.json()
            if not repository_tree: # 如果当前页没有数据，说明已到最后一页
                break
            all_repo_items.extend(repository_tree)
            page += 1
        else:
            print(f"  Failed to retrieve repository tree for project '{project_name}'. Status code: {response.status_code}, Response: {response.text}")
            break # 发生错误时退出循环

    # 在获取所有仓库条目后，统一检查文件是否存在
    for filename in filenames:
        file_found = any(filename == item.get("name", "") and item.get("type") == "blob" for item in all_repo_items)
        print(f"  File '{filename}' found in project '{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)
    if not projects_info:
        print("No projects found or failed to retrieve projects. Exiting.")
        return

    # 指定要检查的文件名列表
    filenames_to_check = ["serverless.yaml", "serverless.yml", "README.md"] # 可以添加更多文件

    # 用于存储所有项目检查结果的列表
    all_results = []

    print(f"\nStarting file check across {len(projects_info)} projects in group {group_id}...")
    # 遍历每个项目并检查文件
    for project_id, project_name in projects_info:
        print(f"\nProcessing project: '{project_name}' (ID: {project_id})")
        project_result = check_files_in_project(gitlab_api_url, private_token, project_id, project_name, filenames_to_check)
        all_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_results, json_file, indent=2, ensure_ascii=False)
        print(f"\nAll results saved to '{output_filename}'")
    except IOError as e:
        print(f"Error writing to output file: {e}")

if __name__ == "__main__":
    main()