GitLab组仓库多文件存在性检测Python脚本异常排查

问题描述 #

使用Python脚本通过GitLab API按组ID检测指定组下各仓库中serverless.yaml、serverless.yml文件的存在性时，出现两个异常：

1. 仓库中实际存在的文件被判定为不存在
2. 输出的JSON文件中files字段为空，且JSON格式存在语法错误

原脚本代码：

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

# Set your GitLab API URL, private token, and group ID as environment variables
gitlab_api_url = os.getenv("GITLAB_API_URL", "https://gitlab.com/api/v4")
private_token = os.getenv("GITLAB_PRIVATE_TOKEN", "xxx")
group_id = os.getenv("GITLAB_GROUP_ID", "xxx")

def get_project_ids(api_url, private_token, group_id):
    endpoint = f"{api_url}/groups/{group_id}/projects"
    headers = {"PRIVATE-TOKEN": private_token}
    
    response = requests.get(endpoint, headers=headers)

    if response.status_code == 200:
        projects = response.json()
        return [(project['id'], project['name']) for project in projects]
    else:
        print(f"Failed to retrieve projects for group {group_id}. Status code: {response.status_code}")
        return []

def check_files_in_project(api_url, private_token, group_id, project_id, project_name, filenames, output_filename):
    # Specify the private token for authentication
    headers = {"PRIVATE-TOKEN": private_token}

    # Create a dictionary to store the output information
    output_data = {
        "group_id": group_id,
        "project_id": project_id,
        "project_name": project_name,
        "files": []
    }

    for filename in filenames:
        # Encode special characters in the filename and construct the URL
        encoded_filename = quote_plus(filename)
        endpoint = f"{api_url}/projects/{project_id}/repository/tree?recursive=1&path={encoded_filename}&ref=dev"

        print(f"\nChecking for file: {filename} in project {project_name}")
        print(f"Constructed URL: {endpoint}")

        # Make the API request to get the repository tree
        response = requests.get(endpoint, headers=headers)

        # Check if the request was successful
        if response.status_code == 200:
            # Parse the JSON response
            repository_tree = response.json()

            # Check if the specified file exists in any folder
            file_found = any(filename == item.get("name", "") for item in repository_tree)
            print(f"File {filename} found in project {project_name}: {file_found}")

            # Append file information to output_data
            output_data["files"].append({"filename": filename, "file_found": file_found})
        else:
            print(f"Failed to retrieve repository tree for project {project_name}. Status code: {response.status_code}")

        # Append output data to a single JSON file after checking each file
        with open(output_filename, 'a') as json_file:
            json.dump(output_data, json_file, indent=2)
            json_file.write('\n')  # Add newline to separate entries

# Get the list of project IDs and names in the group
projects_info = get_project_ids(gitlab_api_url, private_token, group_id)

# Specify the list of filenames you want to check
filenames_to_check = ["serverless.yaml", "serverless.yml"]

# Single output file for all projects
output_filename = "output_all_projects.json"

# Open the file with an initial '[' to start a JSON array
with open(output_filename, 'w') as json_file:
    json_file.write('[')

# Iterate over each project ID and check for the files in any folder
for project_id, project_name in projects_info:
    check_files_in_project(gitlab_api_url, private_token, group_id, project_id, project_name, filenames_to_check, output_filename)

# Close the file with a ']' to close the JSON array
with open(output_filename, 'a') as json_file:
    json_file.write(']')

异常输出示例（格式错误）：

[
{
  "group_id": "gid",
  "project_id": xxx,
  "project_name": "name",
  "files": []
}
{
  "group_id": "gid",
  "project_id": xxx,
  "project_name": "name",
  "files": []
}
]

问题根源分析 #

1. 文件检测逻辑错误 #

• API接口使用不当：repository/tree接口中，path参数用于指定目录或文件路径，传入文件名时只会返回该路径下的内容（文件在根目录则返回对象，否则返回空数组），结合recursive=1无法实现全局搜索文件的效果。
• 分支硬编码：脚本固定检测dev分支，若文件存在于其他分支（如main/master），会导致检测失败。
• 项目ID未编码：如果项目ID是路径格式（如group/subgroup/project），未编码会导致API请求失败。

2. JSON输出格式错误 #

• 重复写入项目结果：每检查一个文件就写入一次项目对象，导致同一个项目在输出文件中出现多次条目。
• 数组元素无分隔符：写入JSON数组时，对象之间未添加逗号，导致整个JSON格式非法，解析时可能显示异常（如files字段为空）。
• 项目分页未处理：GitLab API默认每页返回20个项目，原脚本未处理分页，会遗漏大量项目。

修复后的脚本 #

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", "xxx")
group_id = os.getenv("GITLAB_GROUP_ID", "xxx")
target_branch = os.getenv("GITLAB_TARGET_BRANCH", "dev")  # 可配置目标分支

def get_all_projects(api_url, private_token, group_id):
    """分页获取组下所有项目"""
    endpoint = f"{api_url}/groups/{group_id}/projects"
    headers = {"PRIVATE-TOKEN": private_token}
    projects = []
    page = 1
    per_page = 100  # 每页最大项目数
    
    while True:
        params = {"page": page, "per_page": per_page}
        response = requests.get(endpoint, headers=headers, params=params)
        if response.status_code != 200:
            print(f"获取第 {page} 页项目失败，状态码: {response.status_code}")
            break
        page_projects = response.json()
        if not page_projects:
            break
        projects.extend(page_projects)
        page += 1
    return [(project['id'], project['name']) for project in projects]

def check_files_in_project(api_url, private_token, group_id, project_id, project_name, filenames, branch):
    """检测单个项目中指定文件的存在性"""
    headers = {"PRIVATE-TOKEN": private_token}
    output_data = {
        "group_id": group_id,
        "project_id": project_id,
        "project_name": project_name,
        "files": []
    }

    for filename in filenames:
        # 编码项目ID和文件名，避免特殊字符导致请求失败
        encoded_project_id = quote_plus(str(project_id))
        encoded_filename = quote_plus(filename)
        # 使用repository/files接口直接检测文件存在性
        endpoint = f"{api_url}/projects/{encoded_project_id}/repository/files/{encoded_filename}?ref={branch}"
        
        print(f"\n检测项目 {project_name} 中的文件: {filename} (分支: {branch})")
        print(f"请求URL: {endpoint}")
        
        response = requests.get(endpoint, headers=headers)
        file_found = response.status_code == 200
        print(f"文件 {filename} 存在: {file_found}")
        
        output_data["files"].append({
            "filename": filename,
            "file_found": file_found,
            "status_code": response.status_code
        })
    
    return output_data

# 执行检测流程
projects_info = get_all_projects(gitlab_api_url, private_token, group_id)
filenames_to_check = ["serverless.yaml", "serverless.yml"]
output_filename = "output_all_projects.json"

# 收集所有项目的检测结果
output_data_list = []
for project_id, project_name in projects_info:
    project_result = check_files_in_project(gitlab_api_url, private_token, group_id, project_id, project_name, filenames_to_check, target_branch)
    output_data_list.append(project_result)

# 写入合法的JSON文件
with open(output_filename, 'w') as json_file:
    json.dump(output_data_list, json_file, indent=2)

print(f"\n检测完成，结果已保存到 {output_filename}")

修复说明 #

1. 改用repository/files接口：直接请求单个文件的存在性，返回200则文件存在，404则不存在，无需遍历仓库树，效率更高且结果准确。
2. 处理项目分页：循环获取所有分页的项目，避免遗漏组内的项目。
3. 修复JSON格式：先收集所有项目的检测结果到列表，再一次性写入文件，确保JSON数组格式合法。
4. URL编码关键参数：对项目ID和文件名进行URL编码，适配特殊格式的项目ID。
5. 分支可配置：通过环境变量指定目标分支，提升脚本灵活性。

内容的提问来源于stack exchange，提问作者Yuday