feat: joj3 failed table / fix: joj3 scoreboard (#29)

joint_teapot/app.py

@@ -197,18 +197,20 @@ def unsubscribe_from_repos(pattern: str = Argument("")) -> None:
 
 @app.command(
     "JOJ3-scoreboard",
-    help="parse JOJ3 scoreboard json file and upload to gitea",
+    help="parse JOJ3 score json file into scoreboard and upload to gitea",
 )
 def JOJ3_scoreboard(
     scorefile_path: str = Argument(
         "", help="path to score json file generated by JOJ3"
     ),
-    student_name: str = Argument("", help="name of student"),
+    submitter: str = Argument(
-    student_id: str = Argument("", help="id of student"),
+        "", help="name of submitter, either student name + id, or group name"
+    ),
     repo_name: str = Argument(
         "",
         help="name of local gitea repo folder, or link to remote gitea repo, to push scoreboard file",
     ),
+    exercise_name: str = Argument("", help="exercise name of this json score file"),
     scoreboard_file_name: str = Argument(
         "", help="name of scoreboard file in the gitea repo"
     ),
@@ -226,8 +228,8 @@ def JOJ3_scoreboard(
     repo.git.reset("--hard", "origin/grading")
     joj3.generate_scoreboard(
         scorefile_path,
-        student_name,
+        submitter,
-        student_id,
+        exercise_name,
         os.path.join(repo_path, scoreboard_file_name),
     )
     now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@@ -236,6 +238,53 @@ def JOJ3_scoreboard(
     )
 
 
+@app.command(
+    "JOJ3-failed-table",
+    help="parse JOJ3 score json file into failed table markdown file and upload to gitea",
+)
+def JOJ3_failed_table(
+    scorefile_path: str = Argument(
+        "", help="path to score json file generated by JOJ3"
+    ),
+    repo_name: str = Argument(
+        "",
+        help="name of local gitea repo folder, or link to remote gitea repo, to push scoreboard file",
+    ),
+    submitter_repo_name: str = Argument(
+        "",
+        help="repository's name of the submitter",
+    ),
+    submitter_repo_link: str = Argument(
+        "",
+        help="repository's url link of the submitter",
+    ),
+    failedtable_file_name: str = Argument(
+        "", help="name of failed table file in the gitea repo"
+    ),
+) -> None:
+    repo_path = tea.pot.git.repo_clean_and_checkout(repo_name, "grading")
+    repo: Repo = tea.pot.git.get_repo(repo_name)
+    if "grading" not in repo.remote().refs:
+        logger.error(
+            '"grading" branch not found in remote, create and push it to origin first.'
+        )
+        return
+    if "grading" not in repo.branches:
+        logger.error('"grading" branch not found in local, create it first.')
+        return
+    repo.git.reset("--hard", "origin/grading")
+    joj3.generate_failed_table(
+        scorefile_path,
+        submitter_repo_name,
+        submitter_repo_link,
+        os.path.join(repo_path, failedtable_file_name),
+    )
+    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    tea.pot.git.add_commit_and_push(
+        repo_name, [failedtable_file_name], f"test: JOJ3-dev testing at {now}"
+    )
+
+
 if __name__ == "__main__":
     try:
         app()

joint_teapot/utils/joj3.py

@@ -1,14 +1,15 @@
+import bisect
 import csv
 import json
 import os
 from datetime import datetime
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from joint_teapot.utils.logger import logger
 
 
 def generate_scoreboard(
-    score_file_path: str, student_name: str, student_id: str, scoreboard_file_path: str
+    score_file_path: str, submitter: str, exercise_name: str, scoreboard_file_path: str
 ) -> None:
     if not scoreboard_file_path.endswith(".csv"):
         logger.error(
@@ -33,59 +34,49 @@ def generate_scoreboard(
         ]
         data = []
 
-    column_updated = [False] * len(columns)  # Record wether a score has been updated
+    submitter_found = False
-    # Update data
-    with open(score_file_path) as json_file:
-        scorefile: Dict[str, Any] = json.load(json_file)
-
-    student = f"{student_name} {student_id}"
-    student_found = False
     for row in data:
-        if row[0] == student:
+        if row[0] == submitter:
-            student_row = row  # This is a reference of the original data
+            submitter_row = row  # This is a reference of the original data
-            student_found = True
+            submitter_found = True
             break
-    if not student_found:
+    if not submitter_found:
-        student_row = [student, "", "0"] + [""] * (
+        submitter_row = [submitter, "", "0"] + [""] * (
             len(columns) - 3
         )  # FIXME: In formal version should be -2
-        data.append(student_row)
+        data.append(submitter_row)
 
-    for stagerecord in scorefile["stagerecords"]:
+    # Find if exercise in table:
-        stagename = stagerecord["stagename"]
+    if exercise_name not in columns:
-        for stageresult in stagerecord["stageresults"]:
+        column_tail = columns[3:]
-            name = stageresult["name"]
+        bisect.insort(column_tail, exercise_name)
-            for i, result in enumerate(stageresult["results"]):
+        columns[3:] = column_tail
-                score = result["score"]
+        index = columns.index(exercise_name)
-                colname = f"{stagename}/{name}"
+        for row in data:
-                if len(stageresult["results"]) != 1:
+            row.insert(index, "")
-                    colname = f"{colname}/{i}"
+
-                if colname not in columns:
+    # Update data
-                    columns.append(colname)
+    with open(score_file_path) as json_file:
-                    column_updated.append(True)
+        scorefile: List[Dict[str, Any]] = json.load(json_file)
-                    for row in data:
+
-                        row.append("")
+    exercise_total_score = 0
-                student_row[columns.index(colname)] = score
+    for stage in scorefile:
-                column_updated[columns.index(colname)] = True
+        for result in stage["results"]:
-    # Score of any unupdated columns should be cleared
+            exercise_total_score += result["score"]
-    for i, column in enumerate(columns):
+    submitter_row[columns.index(exercise_name)] = str(exercise_total_score)
-        if column in ["", "last_edit", "total"]:
-            continue
-        if column_updated[i] == False:
-            student_row[i] = ""
 
     total = 0
     for col in columns:
         if col in ["", "total", "last_edit"]:
             continue
         idx = columns.index(col)
-        if (student_row[idx] is not None) and (student_row[idx] != ""):
+        if (submitter_row[idx] is not None) and (submitter_row[idx] != ""):
-            total += int(student_row[idx])
+            total += int(submitter_row[idx])
 
-    student_row[columns.index("total")] = str(total)
+    submitter_row[columns.index("total")] = str(total)
 
     now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-    student_row[
+    submitter_row[
         columns.index("last_edit")
     ] = now  # FIXME: Delete this in formal version
 
@@ -99,6 +90,74 @@ def generate_scoreboard(
         writer.writerows(data)
 
 
+def get_failed_table_from_file(table_file_path: str) -> List[List[str]]:
+    data: List[List[str]] = []
+    if os.path.exists(table_file_path):
+        with open(table_file_path) as table_file:
+            for i, line in enumerate(table_file):
+                if i < 2:
+                    continue
+                stripped_line = line.strip().strip("|").split("|")
+                data.append(stripped_line)
+    return data
+
+
+def update_failed_table_from_score_file(
+    data: List[List[str]], score_file_path: str, repo_name: str, repo_link: str
+) -> None:
+    # get info from score file
+    with open(score_file_path) as json_file:
+        scorefile: List[Dict[str, Any]] = json.load(json_file)
+
+    failed_name = ""
+    for stage in scorefile:
+        if stage["force_quit"] == True:
+            failed_name = stage["name"]
+            break
+
+    # append to failed table
+    now = datetime.now().strftime("%Y-%m-%d %H:%M")
+    repo = f"[{repo_name}]({repo_link})"
+    failure = f"[{failed_name}]({'#'})"  # TODO: Update failure link
+    row_found = False
+    for i, row in enumerate(data[:]):
+        if row[1] == repo:
+            row_found = True
+            if failed_name == "":
+                data.remove(row)
+            else:
+                data[i][0] = now
+                data[i][2] = failure
+            break
+    if not row_found and failed_name != "":
+        data.append([now, repo, failure])
+
+
+def write_failed_table_into_file(data: List[List[str]], table_file_path: str) -> None:
+    data = sorted(data, key=lambda x: x[1])
+    text = "|date|repository|failure|\n"
+    text += "|----|----|----|\n"
+    for row in data:
+        text += f"|{row[0]}|{row[1]}|{row[2]}|\n"
+
+    with open(table_file_path, "w") as table_file:
+        table_file.write(text)
+
+
+def generate_failed_table(
+    score_file_path: str, repo_name: str, repo_link: str, table_file_path: str
+) -> None:
+    if not table_file_path.endswith(".md"):
+        logger.error(
+            f"Failed table file should be a .md file, but now it is {table_file_path}"
+        )
+        return
+
+    data = get_failed_table_from_file(table_file_path)
+    update_failed_table_from_score_file(data, score_file_path, repo_name, repo_link)
+    write_failed_table_into_file(data, table_file_path)
+
+
 def generate_comment(score_file_path: str) -> str:
     # TODO
     return ""