feat: JOJ3 scoreboard (#28)

Co-authored-by: BoYanZh <boyanzh233@gmail.com>

joint_teapot/app.py

@@ -1,10 +1,13 @@
+import os
 from datetime import datetime
 from pathlib import Path
 from typing import List
 
+from git import Repo
 from typer import Argument, Option, Typer, echo
 
 from joint_teapot.teapot import Teapot
+from joint_teapot.utils import joj3
 from joint_teapot.utils.logger import logger
 
 app = Typer(add_completion=False)
@@ -192,6 +195,47 @@ def unsubscribe_from_repos(pattern: str = Argument("")) -> None:
     tea.pot.gitea.unsubscribe_from_repos(pattern)
 
 
+@app.command(
+    "JOJ3-scoreboard",
+    help="parse JOJ3 scoreboard json file 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"),
+    student_id: str = Argument("", help="id of student"),
+    repo_name: str = Argument(
+        "",
+        help="name of local gitea repo folder, or link to remote gitea repo, to push scoreboard file",
+    ),
+    scoreboard_file_name: str = Argument(
+        "", help="name of scoreboard 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_scoreboard(
+        scorefile_path,
+        student_name,
+        student_id,
+        os.path.join(repo_path, scoreboard_file_name),
+    )
+    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    tea.pot.git.add_commit_and_push(
+        repo_name, [scoreboard_file_name], f"test: JOJ3-dev testing at {now}"
+    )
+
+
 if __name__ == "__main__":
     try:
         app()

joint_teapot/utils/joj3.py

@@ -0,0 +1,104 @@
+import csv
+import json
+import os
+from datetime import datetime
+from typing import Any, Dict
+
+from joint_teapot.utils.logger import logger
+
+
+def generate_scoreboard(
+    score_file_path: str, student_name: str, student_id: str, scoreboard_file_path: str
+) -> None:
+    if not scoreboard_file_path.endswith(".csv"):
+        logger.error(
+            f"Scoreboard file should be a .csv file, but now it is {scoreboard_file_path}"
+        )
+        return
+
+    # Load the csv file if it already exists
+    if os.path.exists(scoreboard_file_path):
+        with open(scoreboard_file_path, newline="") as file:
+            reader = csv.reader(file)
+            rows = list(reader)
+        columns = rows[0]
+        data = rows[1:]
+    else:
+        columns = [
+            "",
+            "last_edit",  # FIXME:
+            # This is just to make changes in the file so that it can be pushed.
+            # Only used in development stage. Will be removed in the future.
+            "total",
+        ]
+        data = []
+
+    column_updated = [False] * len(columns)  # Record wether a score has been updated
+    # 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:
+            student_row = row  # This is a reference of the original data
+            student_found = True
+            break
+    if not student_found:
+        student_row = [student, "", "0"] + [""] * (
+            len(columns) - 3
+        )  # FIXME: In formal version should be -2
+        data.append(student_row)
+
+    for stagerecord in scorefile["stagerecords"]:
+        stagename = stagerecord["stagename"]
+        for stageresult in stagerecord["stageresults"]:
+            name = stageresult["name"]
+            for i, result in enumerate(stageresult["results"]):
+                score = result["score"]
+                colname = f"{stagename}/{name}"
+                if len(stageresult["results"]) != 1:
+                    colname = f"{colname}/{i}"
+                if colname not in columns:
+                    columns.append(colname)
+                    column_updated.append(True)
+                    for row in data:
+                        row.append("")
+                student_row[columns.index(colname)] = score
+                column_updated[columns.index(colname)] = True
+    # Score of any unupdated columns should be cleared
+    for i, column in enumerate(columns):
+        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] != ""):
+            total += int(student_row[idx])
+
+    student_row[columns.index("total")] = str(total)
+
+    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    student_row[
+        columns.index("last_edit")
+    ] = now  # FIXME: Delete this in formal version
+
+    # Sort data by total
+    data.sort(key=lambda x: int(x[columns.index("total")]), reverse=True)
+
+    # Write back to the csv file:
+    with open(scoreboard_file_path, mode="w", newline="") as file:
+        writer = csv.writer(file)
+        writer.writerow(columns)
+        writer.writerows(data)
+
+
+def generate_comment(score_file_path: str) -> str:
+    # TODO
+    return ""

joint_teapot/workers/git.py

@@ -1,7 +1,7 @@
 import os
 import sys
 from time import sleep
-from typing import Optional
+from typing import List, Optional
 
 from joint_teapot.utils.logger import logger
 
@@ -97,3 +97,19 @@ class Git:
                 else:
                     raise
         return repo_dir
+
+    def add_commit_and_push(
+        self, repo_name: str, files_to_add: List[str], commit_message: str
+    ) -> None:
+        repo: Repo = self.get_repo(repo_name)
+        for file in files_to_add:
+            try:
+                repo.index.add(file)
+            except OSError:
+                logger.warning(
+                    f'File path "{file}" does not exist. Skipping this file.'
+                )
+                continue
+        repo.index.commit(commit_message)
+        origin = repo.remote(name="origin")
+        origin.push()

requirements.txt

@@ -8,4 +8,4 @@ mattermostdriver>=7.3.2
 patool>=1.12
 pydantic>=2.0.2
 pydantic-settings>=2.0.1
-typer[all]>=0.3.2
+typer>=0.12.3