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-04-28 06:23 +0200
« prev ^ index » next coverage.py v7.1.0, created at 2023-04-28 06:23 +0200
1"""
2@file
3@brief Some automation helpers to grab mails from students about projects.
4"""
6import os
7import re
8from pyquickhelper.loghelper import noLOG, run_cmd
9from pyquickhelper.texthelper import remove_diacritics
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.
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
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.
31 The function executes the following commands (if init is True)::
33 cd [folder]
34 git init
35 git remote add origin [https://user.password@server/project.git]
36 git fetch
38 Otherwise, it does::
40 cd [folder]
41 git clone origin [https://user.password@server/project.git]
42 git fetch
44 A folder will be created.
46 .. exref::
47 :tag: Automation
48 :title: Clone many folders in one row
50 ::
52 eleves = "project1;project2;..."
53 root = r"destination"
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)
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))
68 if init:
69 if not os.path.exists(local_folder):
70 fLOG("creating folder", local_folder)
71 os.mkdir(local_folder)
73 hg = os.path.join(local_folder, ".git")
74 if os.path.exists(hg):
75 raise RuntimeError(f"folder {local_folder} should not exist")
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)
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)
96 hg = os.path.join(local_folder, ".git")
97 if os.path.exists(hg):
98 raise RuntimeError(f"folder {local_folder} should not exist")
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 RuntimeError(
104 f"folder {locf} should not exists before cloning")
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)
115 return locf
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.
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
133 The function runs the instruction::
135 git remote remove origin
136 git remote add origin url
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)
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.
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
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.
172 The function executes the following commands::
174 cd [folder]
175 git add -A
176 git commit -m "[message]"
177 git push -u origin master
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)
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)
209 with open(filename, "r", encoding="utf8") as f:
210 content = f.read()
212 _gitlab_regex = re.compile(".+1.*")
213 gitlab = _gitlab_regex.findall(content)
214 if len(gitlab) == 0:
215 raise RuntimeError(
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 RuntimeError(
223 "more than one gitlab repo is mentioned {0} in {1}".format(
224 _gitlab_regex.pattern,
225 filename))
226 gitlab = gitlab[0]
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
242 return commit
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.
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
261 The function *email_function* has the following signature::
263 def email_function(first_name, last_name):
264 # ....
265 """
267 def split_name(name):
268 name = remove_diacritics(name).split(" ")
269 first = name[-1]
270 last = " ".join(name[:-1])
271 return first, last
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
284 folds = []
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 RuntimeError(
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()
297 title = ", ".join(eleves)
298 content = [title]
299 content.append("=" * len(title))
300 content.append("")
302 content.append("* subject: " + title)
303 content.append("* G: %d" % int(name))
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)
310 content.append("")
311 content.append("")
313 last = ".".join(ul(a[-1]) for a in sorted(names))
315 folder = os.path.join(root, last)
316 filename = os.path.join(folder, report)
318 if not os.path.exists(folder):
319 os.mkdir(folder)
321 if overwrite or not os.path.exists(filename):
322 with open(filename, "w", encoding="utf8") as f:
323 f.write("\n".join(content))
325 folds.append(folder)
326 return folds
329def get_sections(path, suivi="suivi.rst"):
330 """
331 Extracts sections from a filename used to follow a group of students.
333 @param path where to find filename
334 @param suivi file, RST format, section are followed by ``+++++``
335 @return dictionary { section : content }
337 Example of a file::
339 rapport
340 +++++++
342 * bla 1
344 extrait
345 +++++++
347 ::
349 paragraphe 1
351 paragraphe 2
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)
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
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)
405 return sections
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.
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
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 RuntimeError(f"OUT:\n{out}\nERR:\n{err}")
432 raise RuntimeError(err)