diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 70e0690..7540520 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -29,3 +29,8 @@ repos: rev: 22.3.0 hooks: - id: black + - repo: https://github.com/Lucas-C/pre-commit-hooks + rev: v1.2.0 + hooks: + - id: remove-crlf + - id: remove-tabs diff --git a/joint_teapot/__main__.py b/joint_teapot/__main__.py index 53e124c..ab3b0c7 100644 --- a/joint_teapot/__main__.py +++ b/joint_teapot/__main__.py @@ -1,8 +1,8 @@ -from joint_teapot.app import app -from joint_teapot.utils.logger import logger as logger - -if __name__ == "__main__": - try: - app() - except Exception: - logger.exception("Unexpected error:") +from joint_teapot.app import app +from joint_teapot.utils.logger import logger as logger + +if __name__ == "__main__": + try: + app() + except Exception: + logger.exception("Unexpected error:") diff --git a/joint_teapot/teapot.py b/joint_teapot/teapot.py index 1aed329..8f2488d 100644 --- a/joint_teapot/teapot.py +++ b/joint_teapot/teapot.py @@ -1,167 +1,167 @@ -import functools -from datetime import datetime -from typing import Any, Callable, Dict, List, Optional, TypeVar - -from joint_teapot.config import settings -from joint_teapot.utils.logger import logger -from joint_teapot.utils.main import first -from joint_teapot.workers import Canvas, Git, Gitea, Mattermost -from joint_teapot.workers.joj import JOJ - -_T = TypeVar("_T") - - -def for_all_methods( - decorator: Callable[[Callable[[_T], _T]], Any] -) -> Callable[[_T], _T]: - @functools.wraps(decorator) - def decorate(cls: Any) -> Any: - for attr in cls.__dict__: # there's probably a better way to do this - if callable(getattr(cls, attr)): - setattr(cls, attr, decorator(getattr(cls, attr))) - return cls - - return decorate - - -def log_exception_in_loguru(func: Callable[..., Any]) -> Callable[..., Any]: - @functools.wraps(func) - def decorator(*args: Any, **kwargs: Any) -> Any: - try: - return func(*args, **kwargs) - except Exception as e: - logger.exception(e) - - return decorator - - -@for_all_methods(log_exception_in_loguru) -class Teapot: - _canvas = None - _gitea = None - _git = None - _joj = None - _mattermost = None - - @property - def canvas(self) -> Canvas: - if not self._canvas: - self._canvas = Canvas() - return self._canvas - - @property - def gitea(self) -> Gitea: - if not self._gitea: - self._gitea = Gitea() - return self._gitea - - @property - def git(self) -> Git: - if not self._git: - self._git = Git() - return self._git - - @property - def joj(self) -> JOJ: - if not self._joj: - self._joj = JOJ() - return self._joj - - @property - def mattermost(self) -> Mattermost: - if not self._mattermost: - self._mattermost = Mattermost() - return self._mattermost - - def __init__(self) -> None: - logger.info( - "Settings loaded. " - f"Canvas Course ID: {settings.canvas_course_id}, " - f"Gitea Organization name: {settings.gitea_org_name}, " - f"Mattermost Team name: {settings.mattermost_team}@{settings.mattermost_domain_name}{settings.mattermost_suffix}" - ) - logger.debug("Teapot initialized.") - - def add_all_canvas_students_to_teams(self, team_names: List[str]) -> None: - return self.gitea.add_canvas_students_to_teams(self.canvas.students, team_names) - - def create_personal_repos_for_all_canvas_students(self) -> List[str]: - return self.gitea.create_personal_repos_for_canvas_students( - self.canvas.students - ) - - def create_teams_and_repos_by_canvas_groups( - self, group_prefix: str = "" - ) -> List[str]: - def convertor(name: str) -> Optional[str]: - if group_prefix and not name.startswith(group_prefix): - return None - 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, convertor, convertor - ) - - def get_public_key_of_all_canvas_students(self) -> Dict[str, List[str]]: - return self.gitea.get_public_key_of_canvas_students(self.canvas.students) - - def clone_all_repos(self) -> None: - for i, repo_name in enumerate(self.gitea.get_all_repo_names()): - logger.info(f"{i}, {self.gitea.org_name}/{repo_name} cloning...") - self.git.repo_clean_and_checkout(repo_name, "master") - - def create_issue_for_repos( - self, repo_names: List[str], title: str, body: str - ) -> None: - for repo_name in repo_names: - self.gitea.create_issue(repo_name, title, body) - - def create_milestone_for_repos( - self, repo_names: List[str], title: str, description: str, due_on: datetime - ) -> None: - for repo_name in repo_names: - self.gitea.create_milestone(repo_name, title, description, due_on) - - def check_exist_issue_by_title( - self, repo_names: List[str], title: str - ) -> List[str]: - res = [] - for repo_name in repo_names: - if not self.gitea.check_exist_issue_by_title(repo_name, title): - res.append(repo_name) - return res - - def checkout_to_repo_by_release_name( - self, repo_name: str, release_name: str, due: datetime = datetime(3000, 1, 1) - ) -> bool: - repo_releases = self.gitea.get_repo_releases(repo_name) - release = first(repo_releases, lambda item: item.name == release_name) - if release is None or release.created_at.replace(tzinfo=None) >= due: - logger.warning( - f"{self.gitea.org_name}/{repo_name} checkout to " - f"release by name {release_name} fail" - ) - return False - self.git.repo_clean_and_checkout(repo_name, f"tags/{release.tag_name}") - logger.info( - f"{self.gitea.org_name}/{repo_name} checkout to " - f"tags/{release.tag_name} succeed" - ) - return True - - def get_repos_status(self, commit_lt: int, issue_lt: int) -> None: - for repo_name, ( - commit_count, - issue_count, - ) in self.gitea.get_repos_status().items(): - if commit_count < commit_lt or issue_count < issue_lt: - logger.info( - f"{self.gitea.org_name}/{repo_name} has " - f"{commit_count} commit(s), {issue_count} issue(s)" - ) - - -if __name__ == "__main__": - teapot = Teapot() +import functools +from datetime import datetime +from typing import Any, Callable, Dict, List, Optional, TypeVar + +from joint_teapot.config import settings +from joint_teapot.utils.logger import logger +from joint_teapot.utils.main import first +from joint_teapot.workers import Canvas, Git, Gitea, Mattermost +from joint_teapot.workers.joj import JOJ + +_T = TypeVar("_T") + + +def for_all_methods( + decorator: Callable[[Callable[[_T], _T]], Any] +) -> Callable[[_T], _T]: + @functools.wraps(decorator) + def decorate(cls: Any) -> Any: + for attr in cls.__dict__: # there's probably a better way to do this + if callable(getattr(cls, attr)): + setattr(cls, attr, decorator(getattr(cls, attr))) + return cls + + return decorate + + +def log_exception_in_loguru(func: Callable[..., Any]) -> Callable[..., Any]: + @functools.wraps(func) + def decorator(*args: Any, **kwargs: Any) -> Any: + try: + return func(*args, **kwargs) + except Exception as e: + logger.exception(e) + + return decorator + + +@for_all_methods(log_exception_in_loguru) +class Teapot: + _canvas = None + _gitea = None + _git = None + _joj = None + _mattermost = None + + @property + def canvas(self) -> Canvas: + if not self._canvas: + self._canvas = Canvas() + return self._canvas + + @property + def gitea(self) -> Gitea: + if not self._gitea: + self._gitea = Gitea() + return self._gitea + + @property + def git(self) -> Git: + if not self._git: + self._git = Git() + return self._git + + @property + def joj(self) -> JOJ: + if not self._joj: + self._joj = JOJ() + return self._joj + + @property + def mattermost(self) -> Mattermost: + if not self._mattermost: + self._mattermost = Mattermost() + return self._mattermost + + def __init__(self) -> None: + logger.info( + "Settings loaded. " + f"Canvas Course ID: {settings.canvas_course_id}, " + f"Gitea Organization name: {settings.gitea_org_name}, " + f"Mattermost Team name: {settings.mattermost_team}@{settings.mattermost_domain_name}{settings.mattermost_suffix}" + ) + logger.debug("Teapot initialized.") + + def add_all_canvas_students_to_teams(self, team_names: List[str]) -> None: + return self.gitea.add_canvas_students_to_teams(self.canvas.students, team_names) + + def create_personal_repos_for_all_canvas_students(self) -> List[str]: + return self.gitea.create_personal_repos_for_canvas_students( + self.canvas.students + ) + + def create_teams_and_repos_by_canvas_groups( + self, group_prefix: str = "" + ) -> List[str]: + def convertor(name: str) -> Optional[str]: + if group_prefix and not name.startswith(group_prefix): + return None + 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, convertor, convertor + ) + + def get_public_key_of_all_canvas_students(self) -> Dict[str, List[str]]: + return self.gitea.get_public_key_of_canvas_students(self.canvas.students) + + def clone_all_repos(self) -> None: + for i, repo_name in enumerate(self.gitea.get_all_repo_names()): + logger.info(f"{i}, {self.gitea.org_name}/{repo_name} cloning...") + self.git.repo_clean_and_checkout(repo_name, "master") + + def create_issue_for_repos( + self, repo_names: List[str], title: str, body: str + ) -> None: + for repo_name in repo_names: + self.gitea.create_issue(repo_name, title, body) + + def create_milestone_for_repos( + self, repo_names: List[str], title: str, description: str, due_on: datetime + ) -> None: + for repo_name in repo_names: + self.gitea.create_milestone(repo_name, title, description, due_on) + + def check_exist_issue_by_title( + self, repo_names: List[str], title: str + ) -> List[str]: + res = [] + for repo_name in repo_names: + if not self.gitea.check_exist_issue_by_title(repo_name, title): + res.append(repo_name) + return res + + def checkout_to_repo_by_release_name( + self, repo_name: str, release_name: str, due: datetime = datetime(3000, 1, 1) + ) -> bool: + repo_releases = self.gitea.get_repo_releases(repo_name) + release = first(repo_releases, lambda item: item.name == release_name) + if release is None or release.created_at.replace(tzinfo=None) >= due: + logger.warning( + f"{self.gitea.org_name}/{repo_name} checkout to " + f"release by name {release_name} fail" + ) + return False + self.git.repo_clean_and_checkout(repo_name, f"tags/{release.tag_name}") + logger.info( + f"{self.gitea.org_name}/{repo_name} checkout to " + f"tags/{release.tag_name} succeed" + ) + return True + + def get_repos_status(self, commit_lt: int, issue_lt: int) -> None: + for repo_name, ( + commit_count, + issue_count, + ) in self.gitea.get_repos_status().items(): + if commit_count < commit_lt or issue_count < issue_lt: + logger.info( + f"{self.gitea.org_name}/{repo_name} has " + f"{commit_count} commit(s), {issue_count} issue(s)" + ) + + +if __name__ == "__main__": + teapot = Teapot() diff --git a/joint_teapot/utils/main.py b/joint_teapot/utils/main.py index c44c776..74c96a0 100644 --- a/joint_teapot/utils/main.py +++ b/joint_teapot/utils/main.py @@ -1,37 +1,37 @@ -import math -import re -from typing import Callable, Iterable, Optional, TypeVar - -from canvasapi.user import User - -_T = TypeVar("_T") - - -def first( - iterable: Iterable[_T], condition: Callable[[_T], bool] = lambda x: True -) -> Optional[_T]: - return next((x for x in iterable if condition(x)), None) - - -def percentile( - N: Iterable[float], percent: float, key: Callable[[float], float] = lambda x: x -) -> Optional[float]: - if not N: - return None - N = sorted(N) - k = (len(N) - 1) * percent - f = math.floor(k) - c = math.ceil(k) - if f == c: - return key(N[int(k)]) - d0 = key(N[int(f)]) * (c - k) - d1 = key(N[int(c)]) * (k - f) - return d0 + d1 - - -def default_repo_name_convertor(user: User) -> str: - sis_login_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}{sis_login_id}" +import math +import re +from typing import Callable, Iterable, Optional, TypeVar + +from canvasapi.user import User + +_T = TypeVar("_T") + + +def first( + iterable: Iterable[_T], condition: Callable[[_T], bool] = lambda x: True +) -> Optional[_T]: + return next((x for x in iterable if condition(x)), None) + + +def percentile( + N: Iterable[float], percent: float, key: Callable[[float], float] = lambda x: x +) -> Optional[float]: + if not N: + return None + N = sorted(N) + k = (len(N) - 1) * percent + f = math.floor(k) + c = math.ceil(k) + if f == c: + return key(N[int(k)]) + d0 = key(N[int(f)]) * (c - k) + d1 = key(N[int(c)]) * (k - f) + return d0 + d1 + + +def default_repo_name_convertor(user: User) -> str: + sis_login_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}{sis_login_id}" diff --git a/requirements.txt b/requirements.txt index 71cd70d..41a6897 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,9 +1,9 @@ -canvasapi>=2.2.0 -focs_gitea>=1.0.0 -GitPython>=3.1.18 -joj-submitter>=0.0.8 -loguru>=0.5.3 -mattermostdriver>=7.3.2 -patool>=1.12 -pydantic[dotenv]>=1.8.1 -typer[all]>=0.3.2 +canvasapi>=2.2.0 +focs_gitea>=1.0.0 +GitPython>=3.1.18 +joj-submitter>=0.0.8 +loguru>=0.5.3 +mattermostdriver>=7.3.2 +patool>=1.12 +pydantic[dotenv]>=1.8.1 +typer[all]>=0.3.2