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 joint_teapot.config import settings from joint_teapot.utils 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.sortable_name eng = re.sub("[\u4e00-\u9fa5]", "", name) eng = "".join([word[0].capitalize() + word[1:] for word in eng.split()]) return f"{eng}{id}" class Gitea: def __init__( self, access_token: str = settings.gitea_access_token, org_name: str = settings.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) @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) return res["data"][0]["id"] @lru_cache def __get_username_by_student_id(self, student_id: str) -> str: res = self.user_api.user_search(q=student_id, limit=1) 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) for student in students: username = self.__get_username_by_student_id(student.sis_login_id) self.organization_api.org_add_team_member(team_id, username) 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) repo: Dict[str, Any] = 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.repository_api.repo_add_collaborator( self.org_name, repo["name"], self.__get_username_by_student_id(student.sis_login_id), ) 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: Dict[str, Any] = 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: Dict[str, Any] = 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" ) self.organization_api.org_add_team_member( team["id"], self.__get_username_by_student_id(student.sis_login_id) ) return repo_names def get_public_key_of_students( self, students: PaginatedList ) -> List[List[Dict[str, Any]]]: return [ self.user_api.user_list_keys( self.__get_username_by_student_id(student.sis_login_id) ) for student in students ] if __name__ == "__main__": gitea = Gitea()