Reworked bash script to Python script

commands.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+
+
+import re
+
+
+from constants import HOMEWORK, SUFFIX
+
+from git import add_files, commit, push
+from gitlab import post_mr, get_mrs_for_branch, set_assignees, merge_mr
+from utils import run_cmd, get_branch
+
+
+class MergeRequests:
+    @staticmethod
+    def get_files(submission: str) -> None:
+        relative_path = f"/home/kontr/kontr/_tmp_/ib111/{HOMEWORK}/{submission}"
+        files = f"{{master-naostro/LoadTest/{HOMEWORK}.py,teacher_email}}"
+
+        if run_cmd("rsync", "-avzP", f"aisa:{relative_path}/{files}", "./")[0] != 0:
+            exit(1)
+
+    @staticmethod
+    def call_flake() -> None:
+        process = run_cmd("flake8", "--exit-zero", f"{HOMEWORK}.py")[1]
+        with open("flake.log", "w") as f:
+            print(process.stdout, file=f)
+
+    @staticmethod
+    def get_mail() -> None:
+        with open("teacher_email") as file:
+            contents = file.read()
+            match = re.search(r"<pre>((.*\s+)+)<\/pre>", contents)
+
+            return match.group(1) if match else contents
+
+    def __call__(self, login: str, submission: str) -> None:
+        branch = get_branch(login)
+
+        self.get_files(submission)
+        self.call_flake()
+
+        add_files(f"{HOMEWORK}.py", "flake.log")
+        commit(f'"Add sources and flake log ({HOMEWORK}{SUFFIX} {login})"')
+        push("origin", branch)
+
+        post_mr(
+            source_branch=branch,
+            target_branch="master",
+            title=f"[{HOMEWORK}{SUFFIX}] {login}",
+            description=f"```\n{self.get_mail()}\n```",
+            labels=HOMEWORK,
+            remove_source_branch="true",
+            assignee_ids=["1772"],
+        )
+
+
+class UpdateAssignees:
+    def __call__(self, login: str, submission: str) -> None:
+        branch = get_branch(login)
+        iid = get_mrs_for_branch(branch)[0]["iid"]
+
+        print(f"{login} @ {branch} : {iid}")
+        set_assignees(iid, ["39", "1772"])
+
+
+class Merge:
+    def __call__(self, login: str, submission: str):
+        branch = get_branch(login)
+        iid = get_mrs_for_branch(branch)[0]["iid"]
+
+        merge_mr(iid)
+
+class Test:
+    def __call__(self, login: str, submission: str):
+        print(f"{login} - {submission}")

constants.py

@@ -0,0 +1,12 @@
+#!/usr/bin/env python3
+
+
+import os
+
+
+DRY_RUN = False
+PROJECT = "xfocko%2Fib111-reviews"
+HOMEWORK = "hwX"
+SUFFIX = ("", "-opravne")[0]
+]
+TOKEN = os.getenv("GITLAB_FI_TOKEN")

git.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+
+
+from utils import handle_error, run_cmd
+
+
+def checkout_branch(branch: str) -> None:
+    if run_cmd("git", "checkout", branch)[0] != 0:
+        if run_cmd("git", "checkout", "-b", branch)[0] != 0:
+            exit(1)
+
+
+def add_files(*files: str) -> None:
+    run_cmd("git", "add", *files)
+
+
+def commit(msg: str) -> None:
+    run_cmd("git", "commit", "-m", msg)
+
+
+def push(remote: str, branch: str):
+    run_cmd("git", "push", "-u", remote, branch)

gitlab.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+
+
+import requests
+
+
+from constants import PROJECT, TOKEN
+
+
+def post_mr(
+    source_branch,
+    target_branch,
+    title,
+    description,
+    labels,
+    remove_source_branch,
+    assignee_ids,
+):
+    params = {
+        "source_branch": source_branch,
+        "target_branch": target_branch,
+        "title": title,
+        "description": description,
+        "labels": labels,
+        "remove_source_branch": remove_source_branch,
+        "assignee_ids": assignee_ids,
+    }
+    headers = {"Private-Token": TOKEN}
+
+    with requests.post(
+        f"https://gitlab.fi.muni.cz/api/v4/projects/{PROJECT}/merge_requests",
+        params=params,
+        headers=headers,
+    ) as req:
+        print(req.status_code)
+
+
+def get_mrs_for_branch(branch):
+    params = {"source_branch": branch}
+    headers = {"Private-Token": TOKEN}
+
+    with requests.get(
+        f"https://gitlab.fi.muni.cz/api/v4/projects/{PROJECT}/merge_requests",
+        params=params,
+        headers=headers,
+    ) as req:
+        return req.json()
+
+
+def merge_mr(iid):
+    headers = {"Private-Token": TOKEN}
+
+    with requests.put(
+        f"https://gitlab.fi.muni.cz/api/v4/projects/{PROJECT}/merge_requests/{iid}/merge",
+        headers=headers,
+    ) as req:
+        print(req.status_code)
+
+
+def set_assignees(iid, assignee_ids):
+    params = {"assignee_ids": assignee_ids}
+    headers = {"Private-Token": TOKEN}
+
+    with requests.put(
+        f"https://gitlab.fi.muni.cz/api/v4/projects/{PROJECT}/merge_requests/{iid}",
+        params=params,
+        headers=headers,
+    ) as req:
+        print(req.status_code)

reviews.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+
+import os
+import re
+import requests
+from subprocess import run
+import sys
+
+
+from commands import MergeRequests, UpdateAssignees, Merge, Test
+from constants import SUBMISSIONS, HOMEWORK
+from git import checkout_branch
+from utils import get_branch, mkcd, make_pair
+
+
+def iterate_logins(func):
+    for login, submission in map(make_pair, SUBMISSIONS):
+        branch = get_branch(login)
+
+        checkout_branch(branch)
+        mkcd(f"{HOMEWORK}/{login}")
+
+        func(login, submission)
+
+        os.chdir("../..")
+        checkout_branch("master")
+
+
+COMMANDS = {
+    "mrs": MergeRequests(),
+    "update-assignees": UpdateAssignees(),
+    "merge": Merge(),
+    "test": Test(),
+}
+
+
+def main():
+    if sys.argv[1] not in COMMANDS:
+        print("Invalid command")
+        exit(2)
+
+    iterate_logins(COMMANDS[sys.argv[1]])
+
+
+if __name__ == "__main__":
+    main()

utils.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+
+
+import os
+import re
+from subprocess import run, CompletedProcess
+from typing import Tuple
+
+
+from constants import DRY_RUN, HOMEWORK, SUFFIX
+
+
+def handle_error(process: CompletedProcess) -> int:
+    if process.stdout:
+        print(f"stdout: {process.stdout}")
+    if process.stderr:
+        print(f"stderr: {process.stderr}")
+    return process.returncode
+
+
+def run_cmd(*args: str) -> Tuple[int, CompletedProcess]:
+    if DRY_RUN:
+        print(" ".join(args))
+        return (0, None)
+    process = run(args, capture_output=True)
+    return (handle_error(process), process)
+
+
+def make_pair(submission: str) -> Tuple[str, str]:
+    login = re.search("(x[a-z0-9]*)_.*", submission).group(1)
+    return (login, submission)
+
+
+def mkcd(directory: str) -> None:
+    os.makedirs(directory, exist_ok=True)
+    os.chdir(directory)
+
+
+def get_branch(login: str) -> str:
+    return f"{HOMEWORK}{SUFFIX}-{login}"