背景

在软件开发中，版本控制是非常重要的一个环节。SNAPSHOT版本是一个特殊的版本，通常用于表示尚处于开发阶段、不稳定、可能随时会有变化的代码。在Maven等构建工具中，SNAPSHOT版本以’-SNAPSHOT’后缀标识，例如’1.0-SNAPSHOT’。
SNAPSHOT版本的主要特点有：

1. 不稳定性：由于SNAPSHOT版本是开发过程中的快照，因此它可能包含未完成的功能、未修复的bug以及不稳定的状态。使用SNAPSHOT版本需要谨慎，因为可能会遇到运行时错误或其他问题。

2. 快速迭代：SNAPSHOT版本常常用于快速迭代和测试新功能。开发人员可以在短时间内提交大量代码更改，并在SNAPSHOT版本中进行集成测试。这样可以在早期发现潜在问题，并进行修复，从而提高代码质量和开发效率。

3. 易于回滚：由于SNAPSHOT版本只是一个快照，如果出现问题，可以快速回滚到之前的版本，避免长时间等待新版本的发布。
   使用SNAPSHOT版本的场景包括：

开发阶段：开发人员可以使用SNAPSHOT版本进行本地开发和测试，以便快速迭代和验证新功能。

集成测试：在将代码提交到主分支之前，可以使用SNAPSHOT版本进行集成测试，以确保代码的质量和稳定性。

临时性功能：如果需要临时添加某个功能，而该功能可能不会长期保留在主分支中，可以使用SNAPSHOT版本。
使用SNAPSHOT版本时需要注意以下几点：

风险控制：由于SNAPSHOT版本可能不稳定，因此在使用时应谨慎评估风险。尽量避免在生产环境中使用SNAPSHOT版本，以免造成不可预测的后果。

依赖管理：在使用SNAPSHOT版本时，需要确保依赖的库或框架也是SNAPSHOT版本，否则可能会出现版本不兼容的问题。

避免长期使用：SNAPSHOT版本只应作为临时解决方案使用。一旦代码稳定并准备好发布为正式版本，应及时发布并移除SNAPSHOT版本。

配置仓库：在使用SNAPSHOT版本时，需要配置正确的Maven仓库来获取SNAPSHOT版本的依赖。否则可能会出现找不到SNAPSHOT版本的错误。

备份与恢复：在使用SNAPSHOT版本之前，建议备份当前的工作成果，以防止出现问题时无法恢复。
在实际应用中，需要根据具体情况评估是否使用SNAPSHOT版本。在开发阶段，为了快速迭代和测试新功能，可以使用SNAPSHOT版本；但在生产环境中，应谨慎使用SNAPSHOT版本，以确保系统的稳定性和可靠性。同时，要关注SNAPSHOT版本的更新和维护，及时发布为正式版本并移除不再需要的SNAPSHOT版本。通过合理使用SNAPSHOT版本，可以提高软件开发的效率和代码质量。

公司内的产品快速上线过程中，后端应用中大量使用并保留了SNAPSHOT包，快照包可以随时发布最新版本，一旦作者没有做好向下兼容，就会导致线上应用拉取最新快照版本，引发故障，所以必须对生产预发环境使用snapshot的情况进行治理。

业务流程

整体的业务流程设计如上图，以使用git仓库存储代码为例，需要获取到所有应用的代码拉取到本地，通过执行mvn命令输出每个应用的依赖项来查找是否存在snapshot。如果代码仓库对接了应用管理发布平台的话，可以从应用管理平台对应用进行分组（如是否是java应用，是否还在使用），从应用管理平台进行Git代码地址的获取。如果直接扫描整个git仓库地址消耗的资源会比较多，同时有相当多不活跃的代码，会做较多的无用功，roi比较低.

拉取git代码

def git_clone(repo_url,local_directory):
    try:
        subprocess.run(["git", "clone", repo_url, local_directory], check=True)
        print(f"Repository cloned to {local_directory}")
    except subprocess.CalledProcessError as e:
        print(f"Cloning failed: {e}")

直接调取git clone命令，需要设置一个本机存储地址，建议以应用维度进行目录命名分类

需要注意的点是，可以自行增加拉取Git 分支的选项，执行环境需要配置git运行环境及进行身份认证，如有周期扫描的需求，由于Git clone遇到本地存在同名目录不会再次拉取的原因，需要修改代码或者删除本地代码库（长期存储代码有资源消耗和代码泄露的潜在风险）

如果是周期任务的话也可以对代码是否有变动进行判断来确定是否需要拉取Git代码，通过git pull实现，代码如下：

def git_clone(repo_url, local_directory):
    try:
        # 克隆仓库
        subprocess.run(["git", "clone", repo_url, local_directory], check=True)
        print(f"Repository cloned to {local_directory}")

        # 检查并拉取最新更改
        with subprocess.Popen(["git", "pull"], cwd=local_directory, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as process:
            out, err = process.communicate()
            if process.returncode == 0:
                print("Updated to the latest commit")
            else:
                print(f"Failed to update: {err.decode('utf-8').strip()}")
    except subprocess.CalledProcessError as e:
        print(f"Cloning failed: {e}")

对接应用平台

如果本身有应用平台的话，可以从应用平台获取需要扫描的Git项目地址，或者就需要自行考虑如何导入扫描的项目地址。对于应用平台一般需要判断应用类型（mvn的扫描显然只支持后端java），应用等级，活跃状况等，对于一些特殊的应用可能不存在Git代码地址（非开源的三方应用），以下为一个代码示例：

def get_application_list():
    url = ""
    data = {"": "", "pageSize": 20}
    new_data_available = True
    page_num = 1
    while new_data_available:
        data["pageNum"] = page_num
        response = requests.post(url, json=data)
        if response.status_code != 200:
            print(f"请求失败，状态码：{response.status_code}")
            break
        result = response.json()
        # 如果结果为空，说明没有更多数据
        if not result.get("data") or len(result.get("data")) == 0:
            new_data_available = False
        else:
            for item in result['data']:
                if item['codeLanguage'] == 'Java' and item['gitlabProjecturl'] is not None:
                    yield result
            page_num += 1

访问应用平台的接口url，设置相关参数来获取其应用信息，然后做一个简单的循环判断，由于data可能返回[]，所以除了判空还要判断词典长度来break循环。

在data数据的判断中可以增加应用类型的判断减少脚本运行的次数提高效率

如果应用平台无法获取Git仓库地址而只提供git项目id的话，这里提供一种写法：

def get_project_info(project_id):
    url = f"https://gitexample.com/api/v4/projects/{project_id}"
    headers = {
        "Accept": "application/json",
        "Private-Token": ""
    }

    response = requests.get(url, headers=headers)

    if response.status_code == 200:
        project_data = response.json()
        path_with_namespace = project_data['http_url_to_repo']
        return path_with_namespace
    else:
        print(f"请求失败，状态码：{response.status_code}")
        return None

如何生成连接git用的private-token此处不再展开

mvn依赖树分析

def execute_mvn_command(directory,appcode):
    command =f"/Library/apache-maven-3.5.4/bin/mvn dependency:list -f {directory}"
    command_append = """| awk -F '[,:]' '{if ($3 == "jar") print $2,$3,$4}'"""
    command = command + command_append
    process = subprocess.Popen(command, stdout=subprocess.PIPE, shell=True)
    output, error = process.communicate()
    if error:
        print(f"Error occurred: {error.decode('utf-8')}")
    else:
        snapshot_found = 'SNAPSHOT'.lower()  in output.decode('utf-8').lower()
        if snapshot_found:
            print("Snapshot found! Writing error to /scan.txt")
            unique_snapshot_output = set(output.decode('utf-8').splitlines())
            file_path = f'/snapshot/{appcode}/scan.txt'
            directory_structure = '/'.join(file_path.split('/')[:-1])
            os.makedirs(directory_structure, exist_ok=True)
            with open(directory + '/mvn-result.txt', 'w') as result_file:
                with open(file_path, 'a') as snapshot_file:
                    for line in unique_snapshot_output:
                        result_file.write(line + '\n')
                        if 'SNAPSHOT'.lower() in line.lower():
                            snapshot_file.write(line + '\n')
        else:
            unique_output = set(output.decode('utf-8').splitlines())
            with open(directory + '/mvn-result.txt', 'w') as file:
                for line in unique_output:
                    file.write(line + '\n')

        print(f"Results saved to mvn-result.txt")

调取mvn dependency命令结合awk对文本进行处理提取依赖包版本，对于所有的项目依赖包的结果进行记录，每个应用的结果保存在拉取代码的目录下，对于存在snapshot的情况，讲该依赖包单独记录

由于后期治理其实除了每个应用单独升级依赖以外，对于所有的snapshot版本可能也需要升级处理，可以汇总去重一个snapshot的版本记录：

import os

# 定义目标目录和输出文件名
directory = '/snapshot'
output_filename = 'merged_unique_lines.txt'

# 初始化一个空集合用于存储唯一行
unique_lines = set()

# 遍历目录及其子目录下的所有txt文件
for root, dirs, files in os.walk(directory):
    for filename in files:
        if filename.endswith('.txt'):
            filepath = os.path.join(root, filename)
            #print(filepath)
            # 打开每个txt文件并读取内容
            with open(filepath, 'r') as file:
                for line in file:
                    # 去除行尾的换行符并添加到集合中
                    unique_lines.add(line.strip())

print(unique_lines)
# 将所有唯一行写入新的txt文件
with open(os.path.join(directory, output_filename), 'w') as output_file:
    for line in unique_lines:
        output_file.write(f'{line}\n')

4A评测 - 免责申明

本站提供的一切软件、教程和内容信息仅限用于学习和研究目的。

不得将上述内容用于商业或者非法用途，否则一切后果请用户自负。

本站信息来自网络，版权争议与本站无关。您必须在下载后的24个小时之内，从您的电脑或手机中彻底删除上述内容。

如果您喜欢该程序，请支持正版，购买注册，得到更好的正版服务。如有侵权请邮件与我们联系处理。敬请谅解！

程序来源网络，不确保不包含木马病毒等危险内容，请在确保安全的情况下或使用虚拟机使用。

侵权违规投诉邮箱：4ablog168#gmail.com（#换成@）