接着上篇批量 git 拉取文章，说下更新操作

Python3 多线程批量更新文件夹下所有 gitlab 项目代码

#!/usr/bin/python3
# -*- coding: UTF-8 -*-

import os
import os.path
import sys
import requests
import time
import functools
import threading
import logging
from requests_pkcs12 import Pkcs12Adapter
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED

session = requests.session()
BASE_HOST = 'https://gitlab.co-xxxxxx.com'
HEADERS = {'PRIVATE-TOKEN': 'xxxxxxx'}
session.mount(BASE_HOST, Pkcs12Adapter(pkcs12_filename='cert/client-xxxxxx.p12', pkcs12_password='xxxxxx'))


def time_me(info='耗时'):
    def _time_me(fn):
        @functools.wraps(fn)
        def _wrapper(*args, **kwargs):
            start = time.perf_counter()
            fn(*args, **kwargs)
            print('{} {} {}'.format(fn.__name__, info, int(time.perf_counter() - start)), ' 秒')
        return _wrapper
    return _time_me


def get_dir(path):
    file_list = os.listdir(path)
    try:
        for tmp in file_list:
            path_tmp = os.path.join(path, tmp)
            if os.path.isdir(path_tmp):
                if not path_tmp.endswith('.git'):
                    get_dir(path_tmp)
                else:
                    paths.append(path_tmp)
                    # logging.info(path_tmp)
    except PermissionError:
        pass


def batch_pull(_path: str):
    try:
        # 切换目录
        os.chdir(_path.replace('/.git', ''))
        name = _path.split('/')[-2]

        if name not in skip_dif:
            logging.info('{} 开始拉取 [{}]'.format(threading.currentThread().getName(), name))
            command = 'git pull'
            # 执行pull
            f = os.popen(command)
            # print(f.readline())
            result = f.read()
            if 'Updating' in result:
                has_result.append(name)
                print(name, '有更新内容')
            logging.info('{} 拉取 [{}] 完毕'.format(threading.currentThread().getName(), name))
    except Exception as e:
        logging.error('执行异常，{}'.format(_path), e)


@time_me()
def main():
    # 创建线程池
    all_task = []
    pool = ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix='code-read-exec')

    # 获取目录
    get_dir(base_code_path)
    logging.info('共 {} 个项目'.format(len(paths)))
    for p in paths:
        time.sleep(0.05)
        all_task.append(pool.submit(batch_pull, p))

    # 等待所有完成
    wait(all_task, return_when=ALL_COMPLETED)
    print("\n------所有项目拉取完毕--------")


if __name__ == '__main__':
    # 操作目录
    base_code_path = '/Users/liurenkui/workSpace/Java/all-code'
    os.chdir(base_code_path)

    # 跳过的目录
    skip_dif = ['abc-code']
    paths = []        # 所有目录
    max_workers = 20        # 最大线程数量
    has_result = list()

    # 执行主方法
    main()

    print('\n 更新项目列表：', has_result)

执行结果示例

------所有项目拉取完毕--------
main 耗时 17  秒

更新项目列表： ['xxx-business-mall', 'xxx-product', 'xxx-saas-weapp-eshop', 'xxx-base', 'xxx-config-nacos', 'xxx-salerule']

未经允许请勿转载：程序喵 » Python3 多线程批量更新文件夹下所有 gitlab 项目代码