diff --git a/joint_teapot/app.py b/joint_teapot/app.py index de098d7..9bb8822 100644 --- a/joint_teapot/app.py +++ b/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"), - student_id: str = Argument("", help="id of student"), + submitter: str = Argument( + "", 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, - student_id, + submitter, + 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() diff --git a/joint_teapot/utils/joj3.py b/joint_teapot/utils/joj3.py index 6272581..c101aae 100644 --- a/joint_teapot/utils/joj3.py +++ b/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 - # 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 + submitter_found = False for row in data: - if row[0] == student: - student_row = row # This is a reference of the original data - student_found = True + if row[0] == submitter: + submitter_row = row # This is a reference of the original data + submitter_found = True break - if not student_found: - student_row = [student, "", "0"] + [""] * ( + if not submitter_found: + 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"]: - 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] = "" + # Find if exercise in table: + if exercise_name not in columns: + column_tail = columns[3:] + bisect.insort(column_tail, exercise_name) + columns[3:] = column_tail + index = columns.index(exercise_name) + for row in data: + row.insert(index, "") + + # Update data + with open(score_file_path) as json_file: + scorefile: List[Dict[str, Any]] = json.load(json_file) + + exercise_total_score = 0 + for stage in scorefile: + for result in stage["results"]: + exercise_total_score += result["score"] + submitter_row[columns.index(exercise_name)] = str(exercise_total_score) 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]) + if (submitter_row[idx] is not None) and (submitter_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 ""