Coverage for src/ensae_teaching_cs/automation_students/git_helper.py: 0%

181 statements  

« prev     ^ index     » next       coverage.py v7.1.0, created at 2023-01-27 05:44 +0100

1""" 

2@file 

3@brief Some automation helpers to grab mails from students about projects. 

4""" 

5 

6import os 

7import re 

8from pyquickhelper.loghelper import noLOG, run_cmd 

9from pyquickhelper.texthelper import remove_diacritics 

10 

11 

12def git_clone(local_folder, url_https, user=None, password=None, timeout=60, 

13 init=True, fLOG=noLOG): 

14 """ 

15 Clones a project from a git repository in a non empty local folder, 

16 it requires `GIT <http://git-scm.com/>`_ to be installed 

17 and uses the command line. 

18 

19 @param local_folder local folder of the project 

20 @param url_https url, example ``https://gitlab.server/folder/project_name`` 

21 @param user part 1 of the credentials 

22 @param password part 2 of the credentials 

23 @param timeout timeout for the command line 

24 @param init see below (True, use fetch, False, use clone) 

25 @param fLOG logging function 

26 @return local_folder 

27 

28 If the reposity has already been cloned, it does not do it again. 

29 We assume that git can be run without giving its full location. 

30 

31 The function executes the following commands (if init is True):: 

32 

33 cd [folder] 

34 git init 

35 git remote add origin [https://user.password@server/project.git] 

36 git fetch 

37 

38 Otherwise, it does:: 

39 

40 cd [folder] 

41 git clone origin [https://user.password@server/project.git] 

42 git fetch 

43 

44 A folder will be created. 

45 

46 .. exref:: 

47 :tag: Automation 

48 :title: Clone many folders in one row 

49 

50 :: 

51 

52 eleves = "project1;project2;..." 

53 root = r"destination" 

54 

55 for el in eleves.split(";"): 

56 cl = el.lower().replace(".","-") 

57 fold = os.path.join(root, el) 

58 if not os.path.exists(fold): 

59 print("clone", el) 

60 url = "https://<gitlab>/<group>/{0}.git".format(cl) 

61 git_clone( fold, url,user=user,password=password, init=False,fLOG=print) 

62 

63 """ 

64 url_user = git_url_user_password(url_https, user, password) 

65 timeout = 60 

66 local_folder = os.path.normpath(os.path.abspath(local_folder)) 

67 

68 if init: 

69 if not os.path.exists(local_folder): 

70 fLOG("creating folder", local_folder) 

71 os.mkdir(local_folder) 

72 

73 hg = os.path.join(local_folder, ".git") 

74 if os.path.exists(hg): 

75 raise Exception(f"folder {local_folder} should not exist") 

76 

77 if not os.path.exists(hg): 

78 cmds = """ 

79 cd {0} 

80 git init 

81 git remote add origin {1} 

82 git fetch 

83 """.format(local_folder, url_user).replace(" ", "").strip(" \n\r\t") 

84 cmd = cmds.replace("\n", "&") 

85 sin = "" # "{0}\n".format(password) 

86 out, err = run_cmd( 

87 cmd, sin=sin, wait=True, timeout=timeout, fLOG=fLOG) 

88 git_check_error(out, err, fLOG) 

89 

90 return local_folder 

91 else: 

92 if not os.path.exists(local_folder): 

93 fLOG("creating folder", local_folder) 

94 os.mkdir(local_folder) 

95 

96 hg = os.path.join(local_folder, ".git") 

97 if os.path.exists(hg): 

98 raise Exception(f"folder {local_folder} should not exist") 

99 

100 final = os.path.split(url_user)[-1].replace(".git", "") 

101 locf = os.path.join(local_folder, final) 

102 if os.path.exists(locf): 

103 raise Exception( 

104 f"folder {locf} should not exists before cloning") 

105 

106 cmds = f""" 

107 cd {local_folder} 

108 git clone {url_user} . 

109 """.replace(" ", "").strip(" \n\r\t") 

110 cmd = cmds.replace("\n", "&") 

111 sin = "" # "{0}\n".format(password) 

112 out, err = run_cmd(cmd, sin=sin, wait=True, timeout=timeout, fLOG=fLOG) 

113 git_check_error(out, err, fLOG) 

114 

115 return locf 

116 

117 

118def git_change_remote_origin(local_folder, url_https, user=None, password=None, 

119 add_fetch=False, timeout=10, fLOG=noLOG): 

120 """ 

121 Changes the origin of the repository. The url and the password 

122 refer to the new repository. 

123 

124 @param local_folder local folder 

125 @param url_https url, example ``https://gitlab.server/folder/project_name`` 

126 @param user part 1 of the credentials 

127 @param password part 2 of the credentials 

128 @param timeout timeout for the command line 

129 @param add_fetch add instruction ``fetch`` 

130 @param fLOG logging function 

131 @return something 

132 

133 The function runs the instruction:: 

134 

135 git remote remove origin 

136 git remote add origin url 

137 

138 """ 

139 url_user = git_url_user_password(url_https, user, password) 

140 cmds = f""" 

141 cd {local_folder} 

142 git remote remove origin 

143 git remote add origin {url_user} 

144 """.replace(" ", "").strip(" \n\r\t") 

145 if add_fetch: 

146 cmds += "\ngit fetch" 

147 cmd = cmds.replace("\n", "&") 

148 sin = "" # "{0}\n".format(password) 

149 out, err = run_cmd(cmd, sin=sin, wait=True, timeout=timeout, fLOG=fLOG) 

150 git_check_error(out, err, fLOG) 

151 

152 

153def git_commit_all(local_folder, url_https, message, user=None, 

154 password=None, timeout=300, fLOG=noLOG): 

155 """ 

156 From a git repository, 

157 it requires `GIT <http://git-scm.com/>`_ to be installed 

158 and uses the command line. 

159 

160 @param local_folder local folder of the project 

161 @param url_https url, example ``https://gitlab.server/folder/project_name`` 

162 @param message message for the commit 

163 @param user part 1 of the credentials 

164 @param password part 2 of the credentials 

165 @param timeout timeout for the command line 

166 @param fLOG logging function 

167 @return None 

168 

169 If the reposity has already been cloned, it does not do it again. 

170 We assume that git can be run without giving its full location. 

171 

172 The function executes the following commands:: 

173 

174 cd [folder] 

175 git add -A 

176 git commit -m "[message]" 

177 git push -u origin master 

178 

179 """ 

180 cmds = f""" 

181 cd {local_folder} 

182 git add -A 

183 git commit -m "{message}" 

184 git push -u origin master 

185 """.replace(" ", "").strip(" \n\r\t") 

186 cmd = cmds.replace("\n", "&") 

187 sin = "" # "{0}\n".format(password) 

188 out, err = run_cmd(cmd, sin=sin, wait=True, timeout=timeout, fLOG=fLOG) 

189 git_check_error(out, err, fLOG) 

190 

191 

192def git_first_commit_all_projects(local_folder, user=None, password=None, 

193 timeout=300, suivi="suivi.rst", fLOG=noLOG): 

194 """ 

195 @param local_folder folder 

196 @param user part 1 of the credentials 

197 @param password part 2 of the credentials 

198 @param timeout timeout for the command line 

199 @param suivi file to open to get the gitlab account 

200 @param fLOG logging function 

201 @return None or ( local_folder, gitlab ) 

202 """ 

203 if not os.path.exists(local_folder): 

204 raise FileNotFoundError(local_folder) 

205 filename = os.path.join(local_folder, suivi) 

206 if not os.path.exists(filename): 

207 raise FileNotFoundError(filename) 

208 

209 with open(filename, "r", encoding="utf8") as f: 

210 content = f.read() 

211 

212 _gitlab_regex = re.compile(".+1.*") 

213 gitlab = _gitlab_regex.findall(content) 

214 if len(gitlab) == 0: 

215 raise Exception( 

216 "unable to find the regular expression {0} in {1}".format( 

217 _gitlab_regex.pattern, 

218 filename)) 

219 if not isinstance(gitlab, list): 

220 raise TypeError("we expect a list for: " + str(gitlab)) 

221 if len(gitlab) != 1: 

222 raise Exception( 

223 "more than one gitlab repo is mentioned {0} in {1}".format( 

224 _gitlab_regex.pattern, 

225 filename)) 

226 gitlab = gitlab[0] 

227 

228 fLOG("* gitlab", gitlab) 

229 g = os.path.join(local_folder, ".git") 

230 commit = None 

231 if not os.path.exists(g): 

232 fLOG("* initialize", local_folder) 

233 git_clone(local_folder, gitlab, 

234 user=user, password=password, fLOG=fLOG) 

235 sub = os.path.split(local_folder)[-1] 

236 fLOG("* first commit ", gitlab) 

237 git_commit_all(local_folder, gitlab, 

238 "first commit to " + sub, 

239 user=user, password=password, fLOG=print) 

240 commit = local_folder, gitlab 

241 

242 return commit 

243 

244 

245def create_folders_from_dataframe(df, root, report="suivi.rst", col_student="Eleves", 

246 col_group="Groupe", col_subject="Sujet", 

247 overwrite=False, email_function=None): 

248 """ 

249 Creates a series of folders for groups of students. 

250 

251 @param root where to create the folders 

252 @param col_student column which contains the student name (firt name + last name) 

253 @param col_group index of the grou 

254 @param col_subject column which contains the subject 

255 @param df DataFrame 

256 @param email_function function which infers email from first and last names, see below 

257 @param report report file 

258 @param overwrite if False, skip if the report already exists 

259 @return list of creates folders 

260 

261 The function *email_function* has the following signature:: 

262 

263 def email_function(first_name, last_name): 

264 # .... 

265 """ 

266 

267 def split_name(name): 

268 name = remove_diacritics(name).split(" ") 

269 first = name[-1] 

270 last = " ".join(name[:-1]) 

271 return first, last 

272 

273 def ul(last): 

274 res = "" 

275 for i, c in enumerate(last): 

276 if c == " ": 

277 res += "_" 

278 elif i == 0 or last[i - 1] in [" ", "-", "_"]: 

279 res += c.upper() 

280 else: 

281 res += c.lower() 

282 return res 

283 

284 folds = [] 

285 

286 gr = df.groupby(col_group) 

287 for name, group in gr: 

288 s = list(set(group[col_subject].copy())) 

289 if len(s) > 1: 

290 raise Exception( 

291 "more than one subject for group: " + str(name) + "\n" + str(s)) 

292 # subject = s[0] 

293 eleves = list(group[col_student]) 

294 names = [(_,) + split_name(_) for _ in eleves] 

295 eleves.sort() 

296 

297 title = ", ".join(eleves) 

298 content = [title] 

299 content.append("=" * len(title)) 

300 content.append("") 

301 

302 content.append("* subject: " + title) 

303 content.append("* G: %d" % int(name)) 

304 

305 if email_function is not None: 

306 mails = [email_function(a[1], a[2]) for a in names] 

307 jmail = "; ".join(mails) 

308 content.append("* mails: " + jmail) 

309 

310 content.append("") 

311 content.append("") 

312 

313 last = ".".join(ul(a[-1]) for a in sorted(names)) 

314 

315 folder = os.path.join(root, last) 

316 filename = os.path.join(folder, report) 

317 

318 if not os.path.exists(folder): 

319 os.mkdir(folder) 

320 

321 if overwrite or not os.path.exists(filename): 

322 with open(filename, "w", encoding="utf8") as f: 

323 f.write("\n".join(content)) 

324 

325 folds.append(folder) 

326 return folds 

327 

328 

329def get_sections(path, suivi="suivi.rst"): 

330 """ 

331 Extracts sections from a filename used to follow a group of students. 

332 

333 @param path where to find filename 

334 @param suivi file, RST format, section are followed by ``+++++`` 

335 @return dictionary { section : content } 

336 

337 Example of a file:: 

338 

339 rapport 

340 +++++++ 

341 

342 * bla 1 

343 

344 extrait 

345 +++++++ 

346 

347 :: 

348 

349 paragraphe 1 

350 

351 paragraphe 2 

352 

353 """ 

354 if not os.path.exists(path): 

355 raise FileNotFoundError(path) 

356 filename = os.path.join(path, suivi) 

357 if not os.path.exists(filename): 

358 raise FileNotFoundError(filename) 

359 

360 try: 

361 with open(filename, "r", encoding="utf8") as f: 

362 content = f.read() 

363 except UnicodeDecodeError as e: 

364 raise ValueError( 

365 f'unable to parse file:\n File "{filename}", line 1') from e 

366 

367 lines = [_.strip("\r").rstrip() for _ in content.split("\n")] 

368 added_in = [] 

369 sections = {"": []} 

370 title = "" 

371 for i, line in enumerate(lines): 

372 if len(line) == 0: 

373 sections[title].append(line) 

374 added_in.append(title) 

375 else: 

376 f = line[0] 

377 if f == " ": 

378 if title is not None: 

379 sections[title].append(line) 

380 added_in.append(title) 

381 else: 

382 sections[""].append(line) 

383 added_in.append("") 

384 elif f in "=+-": 

385 if line == f * len(line): 

386 title = lines[i - 1] 

387 if len(added_in) > 0: 

388 t = added_in[-1] 

389 sections[t] = sections[t][:-1] 

390 added_in[-1] = title 

391 if f == "=": 

392 sections["title"] = [title] 

393 added_in.append("title") 

394 title = "title" 

395 else: 

396 sections[title] = [] 

397 added_in.append(title) 

398 else: 

399 sections[title].append(line) 

400 added_in.append(title) 

401 else: 

402 sections[title].append(line) 

403 added_in.append(title) 

404 

405 return sections 

406 

407 

408def git_url_user_password(url_https, user, password): 

409 """ 

410 Builds a url (starting with https) and add the user and the password 

411 to skip the authentification. 

412 

413 :param url_https: example ``https://gitlab.server/folder/project_name`` 

414 :param user: part 1 of the credentials 

415 :param password: part 2 of the credentials 

416 :return: url 

417 """ 

418 url_user = url_https.replace( 

419 "https://", f"https://{user}:{password}@") 

420 return url_user 

421 

422 

423def git_check_error(out, err, fLOG): 

424 """ 

425 Private function, analyse the output. 

426 """ 

427 if len(out) > 0: 

428 fLOG("OUT:\n" + out) 

429 if len(err) > 0: 

430 if "error" in err.lower(): 

431 raise Exception(f"OUT:\n{out}\nERR:\n{err}") 

432 raise Exception(err)