-
Notifications
You must be signed in to change notification settings - Fork 1
/
main.py
199 lines (184 loc) · 8.87 KB
/
main.py
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
193
194
195
196
197
198
199
#################################################################
# Author: Ismael Maurice #
# Language: Python #
# Projet: Github Clone #
# Version: V1 #
# File: main.py #
#################################################################
"""
The main package which will load the env variable to clone the gihub repositories.
"""
# Import the modules
import pyfiglet
import os
from helpers import *
import constants
import logging
# Init the date
import datetime
now = datetime.datetime.now()
displayDate = now.strftime("%Y-%m-%d")
# Configure the logging
if not os.path.exists('logs') or not os.path.isdir('logs'):
os.makedirs('logs')
logging.basicConfig(filename=f'logs/{displayDate}.log', format='%(levelname)s:%(message)s', filemode='w', level=logging.DEBUG)
# Display the application logo
ASCII_art = pyfiglet.figlet_format(constants.APP_NAME + f'\n{displayDate}', justify="center")
print(ASCII_art)
message = constants.APP_NAME + f' - {now.strftime("%Y-%m-%d")}'
logging.debug(message)
# Init start time
import time
startTime = time.time()
try:
# Import the env
from dotenv import dotenv_values
config = dotenv_values(".env")
# Verification of config
verificationCnfig(config=config)
# Create the folder for clone
resultPath = config["FOLDER"]
isNewFolder = createFolder(path=resultPath)
# Get the domain API url
DOMAIN_API = getUrl(config=config, urlTYpe=constants.API_URL_TYPE)
TOKEN = config['TOKEN']
# Get the profile of user
import requests
headers = {'Authorization': f'Bearer {TOKEN}'}
response = requests.get(DOMAIN_API + "/user", headers=headers)
if response.status_code != 200:
message = "Request to Github connexion failed !"
logMessage(message=message, logType="error")
logMessage(message=response.text, logType="error", addSeparator=False)
exit(0)
# Convert the response to json
responseData = response.json()
USERNAME = responseData["login"]
message = f"The connexion is successfully we will clone the repositories of {USERNAME}"
logMessage(message=message, logType="info")
# Add the username in config
config["USERNAME"] = USERNAME
# Get the list of repositories and clone each after each
metric = {
"success": 0,
"failed": 0,
"update": 0,
"new": 0
}
# Define the list of repo list which has failed to be cloned
repoListFailed = []
# Define the list of repo list which branches have failed to be cloned
repoListPartial = []
# Define the list of failed update forked
repoForkFailed = []
repoData = getRepositoryData(config=config)
# Display the repository list
message = f"{len(repoData)} repositories found for user {USERNAME}"
logMessage(message=message, logType="info")
for repo in repoData:
message = f"{repo['name']} ({repo['owner']})"
logMessage(message=message, logType="info", addSeparator=False)
for repo in repoData:
try:
# Get the repo name and default branch
repoName = repo["name"]
defaulBranch = repo["defaultBranch"]
# Specify the right user of repository because the owner of repository can be another user not login user
config["USERNAME"] = repo["owner"]
# If the repository is fork repository run the update of fork before any actions
if repo["isFork"] == True:
message = f"Starting synchronization of fork repository {repoName}"
logMessage(message=message, logType="info")
isSync = updateFork(config=config, repoName=repoName, branch=defaulBranch)
if not isSync:
repoForkFailed.append(repoName)
# Specify if the repo is clone or not
isCloneRepo = False
# Build the folder and clone the repository if necessary
RESULT_FOLDER = resultPath + "/" + repoName.replace(" ", '-')
if isNewFolder or not os.path.exists(RESULT_FOLDER):
createFolder(RESULT_FOLDER)
message = f"Starting cloning of repository {repoName} inside {RESULT_FOLDER}"
logMessage(message=message, logType="info")
# Call the function to clone the repository
isCloneRepo = cloneRepository(config=config, repoName=repoName, location=RESULT_FOLDER)
if isCloneRepo:
metric["new"] += 1
else:
metric["failed"] += 1
# Add in failded list
repoListFailed.append(repoName)
else:
# Set true it is already clone
isCloneRepo = True
# Increment the number of new repository which should be updated
metric["update"] += 1
# Checkout and update the branch if repo is clone successfully
if isCloneRepo:
# Get the list of branches
listOfBranchs = getRepositoryBranchesNames(config=config, repoName=repoName)
# Display the repository list
message = f"{len(listOfBranchs)} branch(es) found on repository {repoName}"
logMessage(message=message, logType="info")
for branch in listOfBranchs:
logMessage(message=branch, logType="info", addSeparator=False)
if len(listOfBranchs) > 0:
message = f"Starting updating of repository {repoName} branches inside {RESULT_FOLDER}"
logMessage(message=message, logType="info")
# Clone each branch
isBranchClone = cloneRepoBranches(location=RESULT_FOLDER, listOfBranch=listOfBranchs, defaultBranch=defaulBranch)
if isBranchClone:
# Increment here the number of success
metric["success"] += 1
else:
# Increment the number of failed
metric["failed"] += 1
# Add in branch failed list
repoListPartial.append(repoName)
else:
# Increment the number of failed
metric["failed"] += 1
# Add in branch failed list
repoListPartial.append(repoName)
except Exception as err:
message = f"Unexpected {err}, {type(err)}"
logMessage(message=message, logType="error")
# Increment here the number of failed
metric["failed"] += 1
# Display the result of metric
message = f"The summary of actions are:"
logMessage(message=message, logType="info")
message = f"Number of new repository clones: {metric['new']}"
logMessage(message=message, logType="info", addSeparator=False)
message = f"Number of repository updates: {metric['update']}"
logMessage(message=message, logType="info", addSeparator=False)
message = f"Number of failures: {metric['failed']}"
logMessage(message=message, logType="info", addSeparator=False)
message = f"Number of successes: {metric['success']}"
logMessage(message=message, logType="info", addSeparator=False)
if len(repoForkFailed) > 0:
message = f"The list of {len(repoForkFailed)} fork {'repository' if len(repoForkFailed) < 2 else 'repositories'} which failed to be synchronized:"
logMessage(message=message, logType="info")
for repo in repoForkFailed:
logMessage(message=repo, logType="info", addSeparator=False)
logMessage(message="\n", logType="info", addSeparator=False)
if len(repoListFailed) > 0:
message = f"The list of {len(repoListFailed)} {'repository which' if len(repoListFailed) < 2 else 'repositories which are failed'} failed to be cloned:"
logMessage(message=message, logType="info")
for repo in repoListFailed:
logMessage(message=repo, logType="info", addSeparator=False)
logMessage(message="\n", logType="info", addSeparator=False)
if len(repoListPartial) > 0:
message = f"The list of {len(repoListPartial)} {'repository' if len(repoListPartial) < 2 else 'repositories'} which failed to be updated:"
logMessage(message=message, logType="info")
for repo in repoListPartial:
logMessage(message=repo, logType="info", addSeparator=False)
logMessage(message="\n", logType="info", addSeparator=False)
except Exception as err:
message = f"Unexpected {err}, {type(err)}"
logMessage(message=message, logType="error")
finally:
# Show analyse time
endTime = time.time()
message = f"This programme takes: {getSecondsConvertion(endTime-startTime)}"
logMessage(message=message, logType="info")