diff --git a/joint_teapot/__main__.py b/joint_teapot/__main__.py index b0f7526..2605c3e 100644 --- a/joint_teapot/__main__.py +++ b/joint_teapot/__main__.py @@ -1,6 +1,7 @@ __version__ = "0.0.0" from datetime import datetime +from pathlib import Path from typing import List from typer import Typer, echo @@ -66,22 +67,39 @@ def checkout_to_repos_by_release_name( "close-all-issues", help="close all issues and pull requests in gitea organization" ) def close_all_issues() -> None: - teapot.close_all_issues() + teapot.gitea.close_all_issues() @app.command("archieve-all-repos", help="archieve all repos in gitea organization") def archieve_all_repos() -> None: - teapot.archieve_all_repos() + teapot.gitea.archieve_all_repos() @app.command("get-no-collaborator-repos", help="list all repos with no collaborators") def get_no_collaborator_repos() -> None: - teapot.get_no_collaborator_repos() + teapot.gitea.get_no_collaborator_repos() @app.command("get-no-commit-repos", help="list all repos with no commit") def get_no_commit_repos() -> None: - teapot.get_no_commit_repos() + teapot.gitea.get_no_commit_repos() + + +@app.command( + "prepare-assignment-dir", + help='prepare assignment dir from extracted canvas "Download Submissions" zip', +) +def prepare_assignment_dir(dir: Path) -> None: + teapot.canvas.prepare_assignment_dir(str(dir)) + + +@app.command( + "upload-assignment-scores", + help="upload assignment scores to canvas from score file (SCORE.txt by default), " + + "read the first line as score, the rest as comments", +) +def upload_assignment_scores(dir: Path, assignment_name: str) -> None: + teapot.canvas.upload_assignment_scores(str(dir), assignment_name) if __name__ == "__main__": diff --git a/joint_teapot/teapot.py b/joint_teapot/teapot.py index 41678b7..5603d1f 100644 --- a/joint_teapot/teapot.py +++ b/joint_teapot/teapot.py @@ -68,8 +68,13 @@ class Teapot: ) def create_teams_and_repos_by_canvas_groups(self) -> List[str]: + def convertor(name: str) -> str: + team_name, number_str = name.split(" ") + number = int(number_str) + return f"{team_name}-{number:02}" + return self.gitea.create_teams_and_repos_by_canvas_groups( - self.canvas.students, self.canvas.groups + self.canvas.students, self.canvas.groups, convertor, convertor ) def get_public_key_of_all_canvas_students(self) -> List[str]: @@ -116,18 +121,6 @@ class Teapot: self.git.repo_clean_and_checkout(repo_name, f"tags/{release['tag_name']}") return failed_repos - def close_all_issues(self) -> None: - self.gitea.close_all_issues() - - def archieve_all_repos(self) -> None: - self.gitea.archieve_all_repos() - - def get_no_collaborator_repos(self) -> None: - self.gitea.get_no_collaborator_repos() - - def get_no_commit_repos(self) -> None: - self.gitea.get_no_commit_repos() - if __name__ == "__main__": teapot = Teapot() diff --git a/joint_teapot/workers/canvas.py b/joint_teapot/workers/canvas.py index 2ec8330..2725e69 100644 --- a/joint_teapot/workers/canvas.py +++ b/joint_teapot/workers/canvas.py @@ -1,9 +1,13 @@ +import os +from glob import glob + from canvasapi import Canvas as PyCanvas +from patoolib import extract_archive +from patoolib.util import PatoolError from joint_teapot.config import settings from joint_teapot.utils.logger import logger - -# from canvasapi.group import Group, GroupMembership +from joint_teapot.utils.main import first class Canvas: @@ -11,12 +15,13 @@ class Canvas: self, access_token: str = settings.canvas_access_token, course_id: int = settings.canvas_course_id, + score_filename: str = "SCORE.txt", ): self.canvas = PyCanvas("https://umjicanvas.com/", access_token) self.course = self.canvas.get_course(course_id) logger.info(f"Canvas course loaded. {self.course}") self.students = self.course.get_users( - enrollment_type=["student", "observer"], include=["email"] + enrollment_type=["student"], include=["email"] ) for attr in ["sis_login_id", "sortable_name", "name"]: if not hasattr(self.students[0], attr): @@ -28,8 +33,57 @@ class Canvas: logger.debug(f"Canvas assignments loaded") self.groups = self.course.get_groups() logger.debug(f"Canvas groups loaded") + self.score_filename = score_filename logger.debug("Canvas initialized") + def prepare_assignment_dir(self, dir: str, create_score_file: bool = True) -> None: + login_ids = {stu.id: stu.login_id for stu in self.students} + for v in login_ids.values(): + new_path = os.path.join(dir, v) + if not os.path.exists(new_path): + os.mkdir(new_path) + for path in glob(os.path.join(dir, "*")): + file_name = os.path.basename(path) + if "_" not in file_name: + continue + segments = file_name.split("_") + if segments[1] == "late": + file_id = int(segments[2]) + student = first( + self.students, lambda x: x.login_id == login_ids[file_id] + ) + logger.info(f"{student} submits late") + else: + file_id = int(segments[1]) + target_dir = os.path.join(dir, login_ids[file_id]) + try: + extract_archive(path, outdir=target_dir, verbosity=-1) + os.remove(path) + except PatoolError: + os.rename(path, os.path.join(target_dir, file_name)) + if create_score_file: + open(os.path.join(target_dir, self.score_filename), mode="w") + + def upload_assignment_scores(self, dir: str, assignment_name: str) -> None: + assignment = first(self.assignments, lambda x: x.name == assignment_name) + if assignment is None: + logger.info(f"Canvas assignment {assignment_name} not found") + return + for submission in assignment.get_submissions(): + student = first(self.students, lambda x: x.id == submission.user_id) + if student is None: + continue + score_file_path = os.path.join( + dir, student.sis_login_id, self.score_filename + ) + score, *comments = list(open(score_file_path)) + data = { + "submission": {"posted_grade": float(score)}, + "comment": {"text_comment": "".join(comments)}, + } + logger.info(f"{assignment} {student} {data.__repr__()}") + submission.edit(**data) + if __name__ == "__main__": canvas = Canvas() diff --git a/joint_teapot/workers/gitea.py b/joint_teapot/workers/gitea.py index 8799cdf..c1226a0 100644 --- a/joint_teapot/workers/gitea.py +++ b/joint_teapot/workers/gitea.py @@ -1,312 +1,320 @@ -import re -from enum import Enum -from functools import lru_cache -from typing import Any, Callable, Dict, List, Optional - -import focs_gitea -from canvasapi.group import Group, GroupMembership -from canvasapi.paginated_list import PaginatedList -from canvasapi.user import User -from focs_gitea.rest import ApiException - -from joint_teapot.config import settings -from joint_teapot.utils.logger import logger -from joint_teapot.utils.main import first - - -class PermissionEnum(Enum): - read = "read" - write = "write" - admin = "admin" - - -def default_repo_name_convertor(user: User) -> Optional[str]: - id, name = user.sis_login_id, user.name - eng = re.sub("[\u4e00-\u9fa5]", "", name) - eng = eng.replace(",", "") - eng = "".join([word[0].capitalize() + word[1:] for word in eng.split()]) - return f"{eng}{id}" - - -def list_all(method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: - all_res = [] - page = 1 - while True: - res = method(*args, **kwargs, page=page) - if not res: - break - for item in res: - all_res.append(item) - page += 1 - return all_res - - -class Gitea: - def __init__( - self, - access_token: str = settings.gitea_access_token, - org_name: str = settings.gitea_org_name, - ): - self.org_name = org_name - configuration = focs_gitea.Configuration() - configuration.api_key["access_token"] = access_token - self.api_client = focs_gitea.ApiClient(configuration) - self.admin_api = focs_gitea.AdminApi(self.api_client) - self.miscellaneous_api = focs_gitea.MiscellaneousApi(self.api_client) - self.organization_api = focs_gitea.OrganizationApi(self.api_client) - self.issue_api = focs_gitea.IssueApi(self.api_client) - self.repository_api = focs_gitea.RepositoryApi(self.api_client) - self.settings_api = focs_gitea.SettingsApi(self.api_client) - self.user_api = focs_gitea.UserApi(self.api_client) - logger.debug("Gitea initialized") - - @lru_cache() - def _get_team_id_by_name(self, name: str) -> int: - res = self.organization_api.team_search(self.org_name, q=str(name), limit=1) - if len(res["data"]) == 0: - raise Exception(f"{name} not found by name in Gitea") - return res["data"][0]["id"] - - @lru_cache() - def _get_username_by_canvas_student(self, student: User) -> str: - res = self.user_api.user_search(q=student.sis_login_id, limit=1) - if len(res["data"]) == 0: - raise Exception(f"{student} not found in Gitea") - return res["data"][0]["username"] - - def add_canvas_students_to_teams( - self, students: PaginatedList, team_names: List[str] - ) -> None: - for team_name in team_names: - team_id = self._get_team_id_by_name(team_name) - team_members = self.organization_api.org_list_team_members(team_id) - for student in students: - try: - username = self._get_username_by_canvas_student(student) - team_member = first(team_members, lambda x: x.login == username) - if team_member is None: - self.organization_api.org_add_team_member(team_id, username) - logger.info(f"{student} added to team {team_name}") - else: - team_members.remove(team_member) - logger.warning(f"{student} already in team {team_name}") - except Exception as e: - logger.error(e) - for team_member in team_members: - logger.error( - f"{team_member.full_name} found in team {team_name} " - + "but not found in Canvas students" - ) - - def create_personal_repos_for_canvas_students( - self, - students: PaginatedList, - repo_name_convertor: Callable[ - [User], Optional[str] - ] = default_repo_name_convertor, - ) -> List[str]: - repo_names = [] - for student in students: - repo_name = repo_name_convertor(student) - if repo_name is None: - continue - repo_names.append(repo_name) - body = { - "auto_init": False, - "default_branch": "master", - "name": repo_name, - "private": True, - "template": False, - "trust_model": "default", - } - try: - try: - repo = self.organization_api.create_org_repo( - self.org_name, body=body - ) - logger.info( - f"Personal repo {self.org_name}/{repo_name} for {student} created" - ) - except ApiException as e: - if e.status == 409: - logger.warning( - f"Personal repo {self.org_name}/{repo_name} for {student} already exists" - ) - else: - raise (e) - username = self._get_username_by_canvas_student(student) - self.repository_api.repo_add_collaborator( - self.org_name, repo_name, username - ) - except Exception as e: - logger.error(e) - return repo_names - - def create_teams_and_repos_by_canvas_groups( - self, - students: PaginatedList, - groups: PaginatedList, - team_name_convertor: Callable[[str], Optional[str]] = lambda name: name, - repo_name_convertor: Callable[[str], Optional[str]] = lambda name: name, - permission: PermissionEnum = PermissionEnum.write, - ) -> List[str]: - repo_names = [] - group: Group - for group in groups: - team_name = team_name_convertor(group.name) - repo_name = repo_name_convertor(group.name) - if team_name is None or repo_name is None: - continue - repo_names.append(repo_name) - team = self.organization_api.org_create_team( - self.org_name, - body={ - "can_create_org_repo": False, - "includes_all_repositories": False, - "name": team_name, - "permission": permission.value, - "units": [ - "repo.code", - "repo.issues", - "repo.ext_issues", - "repo.wiki", - "repo.pulls", - "repo.releases", - "repo.projects", - "repo.ext_wiki", - ], - }, - ) - repo = self.organization_api.create_org_repo( - self.org_name, - body={ - "auto_init": False, - "default_branch": "master", - "name": repo_name, - "private": True, - "template": False, - "trust_model": "default", - }, - ) - self.organization_api.org_add_team_repository( - team.id, self.org_name, repo_name - ) - membership: GroupMembership - for membership in group.get_memberships(): - student = first(students, lambda s: s.id == membership.user_id) - if student is None: - raise Exception( - f"student with user_id {membership.user_id} not found" - ) - username = self._get_username_by_canvas_student(student) - self.organization_api.org_add_team_member(team.id, username) - self.repository_api.repo_add_collaborator( - self.org_name, repo_name, username - ) - return repo_names - - def get_public_key_of_canvas_students(self, students: PaginatedList) -> List[str]: - res = [] - for student in students: - try: - username = self._get_username_by_canvas_student(student) - res.extend( - [ - item.key - for item in list_all(self.user_api.user_list_keys, username) - ] - ) - except Exception as e: - logger.error(e) - return res - - def get_repos_releases(self, repo_names: List[str]) -> List[List[Dict[str, Any]]]: - return [ - list_all(self.repository_api.repo_list_releases, self.org_name, repo_name) - for repo_name in repo_names - ] - - def get_all_repo_names(self) -> List[str]: - return [ - data.name - for data in list_all(self.organization_api.org_list_repos, self.org_name) - ] - - def get_no_collaborator_repos(self) -> List[str]: - res = [] - for data in list_all(self.organization_api.org_list_repos, self.org_name): - collaborators = self.repository_api.repo_list_collaborators( - self.org_name, data.name - ) - if collaborators: - continue - logger.info(f"{self.org_name}/{data.name} has no collaborators") - res.append(data.name) - return res - - def get_no_commit_repos(self) -> List[str]: - res = [] - for data in list_all(self.organization_api.org_list_repos, self.org_name): - try: - commits = self.repository_api.repo_get_all_commits( - self.org_name, data.name - ) - except ApiException as e: - if e.status == 409: - logger.info(f"{self.org_name}/{data.name} has no commits") - res.append(data.name) - else: - raise (e) - return res - - def create_issue( - self, - repo_name: str, - title: str, - body: str, - assign_every_collaborators: bool = True, - ) -> None: - assignees = [] - if assign_every_collaborators: - assignees = [ - item.username - for item in list_all( - self.repository_api.repo_list_collaborators, - self.org_name, - repo_name, - ) - ] - self.issue_api.issue_create_issue( - self.org_name, - repo_name, - body={"title": title, "body": body, "assignees": assignees}, - ) - - def check_exist_issue_by_title(self, repo_name: str, title: str) -> bool: - for issue in list_all( - self.issue_api.issue_list_issues, self.org_name, repo_name - ): - if issue.title == title: - return True - return False - - def close_all_issues(self) -> None: - for repo in list_all(self.organization_api.org_list_repos, self.org_name): - for issue in list_all( - self.issue_api.issue_list_issues, self.org_name, repo.name - ): - if issue.state != "closed": - self.issue_api.issue_edit_issue( - self.org_name, repo.name, issue.number, body={"state": "closed"} - ) - - def archieve_all_repos(self) -> None: - for repo in list_all(self.organization_api.org_list_repos, self.org_name): - self.repository_api.repo_edit( - self.org_name, repo.name, body={"archived": True} - ) - - -if __name__ == "__main__": - gitea = Gitea() - res = gitea.get_no_commit_repos() +import re +from enum import Enum +from functools import lru_cache +from typing import Any, Callable, Dict, List, Optional + +import focs_gitea +from canvasapi.group import Group, GroupMembership +from canvasapi.paginated_list import PaginatedList +from canvasapi.user import User +from focs_gitea.rest import ApiException + +from joint_teapot.config import settings +from joint_teapot.utils.logger import logger +from joint_teapot.utils.main import first + + +class PermissionEnum(Enum): + read = "read" + write = "write" + admin = "admin" + + +def default_repo_name_convertor(user: User) -> Optional[str]: + id, name = user.sis_login_id, user.name + eng = re.sub("[\u4e00-\u9fa5]", "", name) + eng = eng.replace(",", "") + eng = "".join([word[0].capitalize() + word[1:] for word in eng.split()]) + return f"{eng}{id}" + + +def list_all(method: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: + all_res = [] + page = 1 + while True: + res = method(*args, **kwargs, page=page) + if not res: + break + for item in res: + all_res.append(item) + page += 1 + return all_res + + +class Gitea: + def __init__( + self, + access_token: str = settings.gitea_access_token, + org_name: str = settings.gitea_org_name, + ): + self.org_name = org_name + configuration = focs_gitea.Configuration() + configuration.api_key["access_token"] = access_token + self.api_client = focs_gitea.ApiClient(configuration) + self.admin_api = focs_gitea.AdminApi(self.api_client) + self.miscellaneous_api = focs_gitea.MiscellaneousApi(self.api_client) + self.organization_api = focs_gitea.OrganizationApi(self.api_client) + self.issue_api = focs_gitea.IssueApi(self.api_client) + self.repository_api = focs_gitea.RepositoryApi(self.api_client) + self.settings_api = focs_gitea.SettingsApi(self.api_client) + self.user_api = focs_gitea.UserApi(self.api_client) + logger.debug("Gitea initialized") + + @lru_cache() + def _get_team_id_by_name(self, name: str) -> int: + res = self.organization_api.team_search(self.org_name, q=str(name), limit=1) + if len(res["data"]) == 0: + raise Exception(f"{name} not found by name in Gitea") + return res["data"][0]["id"] + + @lru_cache() + def _get_username_by_canvas_student(self, student: User) -> str: + res = self.user_api.user_search(q=student.sis_login_id, limit=1) + if len(res["data"]) == 0: + raise Exception(f"{student} not found in Gitea") + return res["data"][0]["username"] + + def add_canvas_students_to_teams( + self, students: PaginatedList, team_names: List[str] + ) -> None: + for team_name in team_names: + team_id = self._get_team_id_by_name(team_name) + team_members = self.organization_api.org_list_team_members(team_id) + for student in students: + try: + username = self._get_username_by_canvas_student(student) + team_member = first(team_members, lambda x: x.login == username) + if team_member is None: + self.organization_api.org_add_team_member(team_id, username) + logger.info(f"{student} added to team {team_name}") + else: + team_members.remove(team_member) + logger.warning(f"{student} already in team {team_name}") + except Exception as e: + logger.error(e) + for team_member in team_members: + logger.error( + f"{team_member.full_name} found in team {team_name} " + + "but not found in Canvas students" + ) + + def create_personal_repos_for_canvas_students( + self, + students: PaginatedList, + repo_name_convertor: Callable[ + [User], Optional[str] + ] = default_repo_name_convertor, + ) -> List[str]: + repo_names = [] + for student in students: + repo_name = repo_name_convertor(student) + if repo_name is None: + continue + repo_names.append(repo_name) + body = { + "auto_init": False, + "default_branch": "master", + "name": repo_name, + "private": True, + "template": False, + "trust_model": "default", + } + try: + try: + repo = self.organization_api.create_org_repo( + self.org_name, body=body + ) + logger.info( + f"Personal repo {self.org_name}/{repo_name} for {student} created" + ) + except ApiException as e: + if e.status == 409: + logger.warning( + f"Personal repo {self.org_name}/{repo_name} for {student} already exists" + ) + else: + raise (e) + username = self._get_username_by_canvas_student(student) + self.repository_api.repo_add_collaborator( + self.org_name, repo_name, username + ) + except Exception as e: + logger.error(e) + return repo_names + + def create_teams_and_repos_by_canvas_groups( + self, + students: PaginatedList, + groups: PaginatedList, + team_name_convertor: Callable[[str], Optional[str]] = lambda name: name, + repo_name_convertor: Callable[[str], Optional[str]] = lambda name: name, + permission: PermissionEnum = PermissionEnum.write, + ) -> List[str]: + repo_names = [] + teams = list_all(self.organization_api.org_list_teams, self.org_name) + repos = list_all(self.organization_api.org_list_repos, self.org_name) + group: Group + for group in groups: + team_name = team_name_convertor(group.name) + repo_name = repo_name_convertor(group.name) + if team_name is None or repo_name is None: + continue + team = first(teams, lambda team: team.name == team_name) + if team is None: + team = self.organization_api.org_create_team( + self.org_name, + body={ + "can_create_org_repo": False, + "includes_all_repositories": False, + "name": team_name, + "permission": permission.value, + "units": [ + "repo.code", + "repo.issues", + "repo.ext_issues", + "repo.wiki", + "repo.pulls", + "repo.releases", + "repo.projects", + "repo.ext_wiki", + ], + }, + ) + logger.info(f"{self.org_name}/{team_name} created") + repo = first(repos, lambda repo: repo.name == repo_name) + if repo is None: + repo_names.append(repo_name) + repo = self.organization_api.create_org_repo( + self.org_name, + body={ + "auto_init": False, + "default_branch": "master", + "name": repo_name, + "private": True, + "template": False, + "trust_model": "default", + }, + ) + logger.info(f"Team {team_name} created") + self.organization_api.org_add_team_repository( + team.id, self.org_name, repo_name + ) + membership: GroupMembership + for membership in group.get_memberships(): + student = first(students, lambda s: s.id == membership.user_id) + if student is None: + raise Exception( + f"student with user_id {membership.user_id} not found" + ) + username = self._get_username_by_canvas_student(student) + self.organization_api.org_add_team_member(team.id, username) + self.repository_api.repo_add_collaborator( + self.org_name, repo_name, username + ) + return repo_names + + def get_public_key_of_canvas_students(self, students: PaginatedList) -> List[str]: + res = [] + for student in students: + try: + username = self._get_username_by_canvas_student(student) + res.extend( + [ + item.key + for item in list_all(self.user_api.user_list_keys, username) + ] + ) + except Exception as e: + logger.error(e) + return res + + def get_repos_releases(self, repo_names: List[str]) -> List[List[Dict[str, Any]]]: + return [ + list_all(self.repository_api.repo_list_releases, self.org_name, repo_name) + for repo_name in repo_names + ] + + def get_all_repo_names(self) -> List[str]: + return [ + data.name + for data in list_all(self.organization_api.org_list_repos, self.org_name) + ] + + def get_no_collaborator_repos(self) -> List[str]: + res = [] + for data in list_all(self.organization_api.org_list_repos, self.org_name): + collaborators = self.repository_api.repo_list_collaborators( + self.org_name, data.name + ) + if collaborators: + continue + logger.info(f"{self.org_name}/{data.name} has no collaborators") + res.append(data.name) + return res + + def get_no_commit_repos(self) -> List[str]: + res = [] + for data in list_all(self.organization_api.org_list_repos, self.org_name): + try: + commits = self.repository_api.repo_get_all_commits( + self.org_name, data.name + ) + except ApiException as e: + if e.status == 409: + logger.info(f"{self.org_name}/{data.name} has no commits") + res.append(data.name) + else: + raise (e) + return res + + def create_issue( + self, + repo_name: str, + title: str, + body: str, + assign_every_collaborators: bool = True, + ) -> None: + assignees = [] + if assign_every_collaborators: + assignees = [ + item.username + for item in list_all( + self.repository_api.repo_list_collaborators, + self.org_name, + repo_name, + ) + ] + self.issue_api.issue_create_issue( + self.org_name, + repo_name, + body={"title": title, "body": body, "assignees": assignees}, + ) + + def check_exist_issue_by_title(self, repo_name: str, title: str) -> bool: + for issue in list_all( + self.issue_api.issue_list_issues, self.org_name, repo_name + ): + if issue.title == title: + return True + return False + + def close_all_issues(self) -> None: + for repo in list_all(self.organization_api.org_list_repos, self.org_name): + for issue in list_all( + self.issue_api.issue_list_issues, self.org_name, repo.name + ): + if issue.state != "closed": + self.issue_api.issue_edit_issue( + self.org_name, repo.name, issue.number, body={"state": "closed"} + ) + + def archieve_all_repos(self) -> None: + for repo in list_all(self.organization_api.org_list_repos, self.org_name): + self.repository_api.repo_edit( + self.org_name, repo.name, body={"archived": True} + ) + + +if __name__ == "__main__": + gitea = Gitea() + res = gitea.get_no_commit_repos() diff --git a/requirements.txt b/requirements.txt index 93d899d..fd49c60 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,5 +2,6 @@ canvasapi>=2.2.0 focs_gitea>=1.0.0 GitPython>=3.1.18 loguru>=0.5.3 +patool>=1.12 pydantic[dotenv]>=1.8.1 typer[all]>=0.3.2