下载gitlab群组或子群组全部项目_gitlab整个群组导出

脚本来自https://www.jianshu.com/p/67d827fbb4e8
添加一些修改解决群组包含项目过多时下载不全的问题

from urllib.request import urlopen
import json
import subprocess, shlex
import time
import os


gitlabToken = 'token'
gitlabAddr  = 'gitlab.com'              # gitlab地址
targets     = ['intel']             # 所属群组或子群组,也可填其id （ps: 为空克隆所有群组，慎用）
withShared  = 'false'                   # 是否包含共享的项目，默认false （ps：为true会拉取到归属其他群组的项目）
max_projects = '100'				#脚本只能读取一页，设置最大项目数量，超过的部分不会下载

#-------------------------

counter = 0
procs = []

def get_next(group_id):
    global counter
    global procs

    print('get_next group_id:', group_id)
    url = gen_next_url(group_id)
    allProjects     = urlopen(url)
    allProjectsDict = json.loads(allProjects.read().decode())
    if len(allProjectsDict) == 0:
        return
    for thisProject in allProjectsDict: 
        try:
            thisProjectURL  = thisProject['http_url_to_repo']
            thisProjectPath = thisProject['path_with_namespace']
            if os.path.exists(thisProjectPath):
                command     = shlex.split('git -C "%s" pull' % (thisProjectPath))               
            else:
                print("=========== 开始克隆 %s %s ===========" % (group_id, thisProject['name']))
                print('执行：git clone %s %s' % (thisProjectURL, thisProjectPath))                        
                command     = shlex.split('git clone %s %s' % (thisProjectURL, thisProjectPath))
                
            proc  = subprocess.Popen(command)
            procs.append(proc)
            time.sleep(1)
            counter += 1

        except Exception as e:
            print("Error on %s: %s" % (thisProjectURL, e.strerror))

    print("=========== 等待子线程执行结束 ===========")
    for p in procs:
        p.wait()
        p.kill()
    print("=========== 子线程执行结束 ===========")
    return

def have_next_projects(group_id):
    url = gen_next_url(group_id)
    allProjects     = urlopen(url)
    allProjectsDict = json.loads(allProjects.read().decode())
    if len(allProjectsDict) == 0:
        return False
    return True


def get_sub_groups(parent_id):
    url = gen_subgroups_url(parent_id)
    allProjects     = urlopen(url)
    allProjectsDict = json.loads(allProjects.read().decode())
    sub_ids = []
    if len(allProjectsDict) == 0:
        return sub_ids
    for thisProject in allProjectsDict: 
        try:
            id = thisProject['id']
            sub_ids.append(id)
        except Exception as e:
            print("Error on %s: %s" % (id, e.strerror))
    return sub_ids

def cal_next_sub_groupids(parent_id):
    parent = ''
    parent = parent_id
    
    is_start = 1
    parent_list = []
    sub_ids = get_sub_groups(parent_id)
    print('cal_next_sub_groupids sub_ids and parent_id:',sub_ids, parent_id)
    ok = have_next_projects(parent_id) 
    print('have_next_projects result:', ok)
    if len(sub_ids)!=0 and ok == False:
        for i in range(len(sub_ids)):
            print('cal_next_sub_groupids sub_ids[i]:', sub_ids[i])
            parent = sub_ids[i]
            a = cal_next_sub_groupids(sub_ids[i])
            return a
    if len(sub_ids) !=0 and ok == True:
        for i in range(len(sub_ids)):
            print('cal_next_sub_groupids parent:', parent)
            parent = sub_ids[i]
            parent_list.append(sub_ids[i])
            a = cal_next_sub_groupids(sub_ids[i])
            parent_list.extend(a)
    if len(sub_ids) == 0 and ok == True:
        print('cal_next_sub_groupids is_start:',is_start)
        parent_list.append(parent)
        return parent_list
    if len(sub_ids) ==0 and ok == False:
        return parent_list
    return parent_list

def download_code(parent_id):
    data =cal_next_sub_groupids(parent_id)
    print('download_code result: ',data)
    for group_id in data:
        get_next(group_id)
    return
 
def gen_next_url(target_id):
    return "https://%s/api/v4/groups/%s/projects?per_page=%s&private_token=%s&with_shared=%s&order_by=updated_at" % (max_projects, gitlabAddr, target_id, gitlabToken, withShared)

def gen_subgroups_url(target_id):
    return "https://%s/api/v4/groups/%s/subgroups?private_token=%s" % (gitlabAddr, target_id, gitlabToken)

def gen_global_url():
    return "http://%s/api/v4/projects?private_token=%s" % (gitlabAddr, gitlabToken)

def download_global_code():
    global counter
    global procs

    url = gen_global_url()
    allProjects     = urlopen(url)
    allProjectsDict = json.loads(allProjects.read().decode())
    if len(allProjectsDict) == 0:
        return
    for thisProject in allProjectsDict: 
        try:
            thisProjectURL  = thisProject['http_url_to_repo']
            thisProjectPath = thisProject['path_with_namespace']
            print(thisProjectURL + ' ' + thisProjectPath)
            
            if os.path.exists(thisProjectPath):       
                command     = shlex.split('git -C "%s" pull' % (thisProjectPath))
            else:
                print("=========== 开始克隆 %s %s ===========" % (group_id, thisProject['name']))
                print('执行：git clone %s %s' % (thisProjectURL, thisProjectPath))    
                command     = shlex.split('git clone %s %s' % (thisProjectURL, thisProjectPath))

            proc  = subprocess.Popen(command)
            procs.append(proc)
            time.sleep(1)
            counter += 1
            
        except Exception as e:
            print("Error on %s: %s" % (thisProjectURL, e.strerror))

    print("=========== 等待子线程执行结束 ===========")
    for p in procs:
        p.wait()
        p.kill()
    print("=========== 子线程执行结束 ===========")
    return

def download_targets_code():
    for target in targets:
        url     = "https://%s/api/v4/groups?private_token=%s&search=%s" % (gitlabAddr, gitlabToken, target)
        allProjects     = urlopen(url)
        allProjectsDict = json.loads(allProjects.read().decode())
        if len(allProjectsDict) == 0:
            return
        target_id = '' 
        for thisProject in allProjectsDict: 
            try:
                this_name = thisProject['name']
                if target == this_name:
                    target_id = thisProject['id']
                    break
            except Exception as e:
                print("Error on %s: %s" % (this_name, e.strerror))    
        download_code(target_id)
    return

def main():
    if len(targets) == 0:
        download_global_code()
    else:
        download_targets_code()
    
    print("=========== 执行结束，克隆项目数： %s ===========" % (counter))
    return


main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192

1