feat: logs & lists

joint_teapot/__main__.py

@@ -3,10 +3,10 @@ __version__ = "0.0.0"
 from datetime import datetime
 from typing import List
 
-from loguru import logger
 from typer import Typer, echo
 
 from joint_teapot.teapot import Teapot
+from joint_teapot.utils.logger import logger
 
 app = Typer(add_completion=False)
 teapot = Teapot()
@@ -20,7 +20,8 @@ def add_all_canvas_students_to_teams(team_names: List[str]) -> None:
 
 
 @app.command(
-    "create-personal", help="create personal repos on gitea for all canvas students"
+    "create-personal-repos",
+    help="create personal repos on gitea for all canvas students",
 )
 def create_personal_repos_for_all_canvas_students() -> None:
     teapot.create_personal_repos_for_all_canvas_students()
@@ -36,9 +37,9 @@ def get_public_key_of_all_canvas_students() -> None:
     echo("\n".join(teapot.get_public_key_of_all_canvas_students()))
 
 
-@app.command("archieve", help="clone all gitea repos to local")
+@app.command("clone-all-repos", help="clone all gitea repos to local")
-def archieve_all_repos() -> None:
+def clone_all_repos() -> None:
-    teapot.archieve_all_repos()
+    teapot.clone_all_repos()
 
 
 @app.command("create-issues", help="create issues on gitea")
@@ -52,7 +53,7 @@ def check_exist_issue_by_title(repo_names: List[str], title: str) -> None:
 
 
 @app.command(
-    "get-release",
+    "get-releases",
     help="checkout git repo to git tag fetched from gitea by release name, with due date",
 )
 def checkout_to_repos_by_release_name(
@@ -61,6 +62,18 @@ def checkout_to_repos_by_release_name(
     teapot.checkout_to_repos_by_release_name(repo_names, release_name, due)
 
 
+@app.command(
+    "close-all-issues", help="close all issues and pull requests in gitea organization"
+)
+def close_all_issues() -> None:
+    teapot.close_all_issues()
+
+
+@app.command("archieve-all-repos", help="archieve all repos in gitea organization")
+def archieve_all_repos() -> None:
+    teapot.archieve_all_repos()
+
+
 if __name__ == "__main__":
     try:
         app()

joint_teapot/config.py

@@ -19,6 +19,9 @@ class Settings(BaseSettings):
     # git
     repos_dir: str = "./repos"
 
+    # log file
+    log_file_path: str = "joint-teapot.log"
+
     class Config:
         env_file = ".env"
         env_file_encoding = "utf-8"

joint_teapot/teapot.py

@@ -2,16 +2,14 @@ import functools
 from datetime import datetime
 from typing import Any, Callable, List
 
-from loguru import logger
-
 from joint_teapot.config import settings
-from joint_teapot.utils import first
+from joint_teapot.utils.logger import logger
+from joint_teapot.utils.main import first
 from joint_teapot.workers import Canvas, Git, Gitea
 
 
 def for_all_methods(decorator: Callable[..., Any]) -> Callable[..., Any]:
     def decorate(cls: Any) -> Any:
-        print(type(cls))
         for attr in cls.__dict__:  # there's propably a better way to do this
             if callable(getattr(cls, attr)):
                 setattr(cls, attr, decorator(getattr(cls, attr)))
@@ -77,7 +75,7 @@ class Teapot:
     def get_public_key_of_all_canvas_students(self) -> List[str]:
         return self.gitea.get_public_key_of_canvas_students(self.canvas.students)
 
-    def archieve_all_repos(self) -> List[str]:
+    def clone_all_repos(self) -> List[str]:
         return [
             self.git.repo_clean_and_checkout(repo_name, "master")
             for repo_name in self.gitea.get_all_repo_names()
@@ -118,6 +116,12 @@ 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()
+
 
 if __name__ == "__main__":
     teapot = Teapot()

joint_teapot/utils/logger.py

@@ -0,0 +1,9 @@
+from sys import stderr
+
+from loguru import logger as logger
+
+from joint_teapot.config import settings
+
+logger.remove()
+logger.add(stderr)
+logger.add(settings.log_file_path)

joint_teapot/utils/main.py

@@ -7,7 +7,3 @@ 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)
-
-
-if __name__ == "__main__":
-    print(first([1, 2, 3, 4], lambda x: x == 5))

joint_teapot/workers/canvas.py

@@ -1,7 +1,7 @@
 from canvasapi import Canvas as PyCanvas
-from loguru import logger
 
 from joint_teapot.config import settings
+from joint_teapot.utils.logger import logger
 
 # from canvasapi.group import Group, GroupMembership
 

joint_teapot/workers/git.py

@@ -1,7 +1,7 @@
 import os
 import sys
 
-from loguru import logger
+from joint_teapot.utils.logger import logger
 
 current_path = sys.path[0]
 sys.path.remove(current_path)

joint_teapot/workers/gitea.py

@@ -8,10 +8,10 @@ from canvasapi.group import Group, GroupMembership
 from canvasapi.paginated_list import PaginatedList
 from canvasapi.user import User
 from focs_gitea.rest import ApiException
-from loguru import logger
 
 from joint_teapot.config import settings
-from joint_teapot.utils import first
+from joint_teapot.utils.logger import logger
+from joint_teapot.utils.main import first
 
 
 class PermissionEnum(Enum):
@@ -28,6 +28,19 @@ def default_repo_name_convertor(user: User) -> Optional[str]:
     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,
@@ -95,16 +108,19 @@ class Gitea:
                 "trust_model": "default",
             }
             try:
-                repo = self.organization_api.create_org_repo(self.org_name, body=body)
+                try:
-                username = self._get_username_by_canvas_student(student)
+                    repo = self.organization_api.create_org_repo(
-                self.repository_api.repo_add_collaborator(
+                        self.org_name, body=body
-                    self.org_name, repo.name, username
                     )
                 except ApiException as e:
                     if e.status == 409:
-                    logger.info(f"Peronsal repo for {student} already exists.")
+                        logger.warning(f"Peronsal repo for {student} already exists.")
                     else:
-                    logger.error(e)
+                        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
@@ -178,7 +194,10 @@ class Gitea:
             try:
                 username = self._get_username_by_canvas_student(student)
                 res.extend(
-                    [item.key for item in self.user_api.user_list_keys(username)]
+                    [
+                        item.key
+                        for item in list_all(self.user_api.user_list_keys, username)
+                    ]
                 )
             except Exception as e:
                 logger.error(e)
@@ -186,13 +205,14 @@ class Gitea:
 
     def get_repos_releases(self, repo_names: List[str]) -> List[List[Dict[str, Any]]]:
         return [
-            self.repository_api.repo_list_releases(self.org_name, repo_name)
+            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 self.organization_api.org_list_repos(self.org_name)
+            data.name
+            for data in list_all(self.organization_api.org_list_repos, self.org_name)
         ]
 
     def create_issue(
@@ -206,8 +226,10 @@ class Gitea:
         if assign_every_collaborators:
             assignees = [
                 item.username
-                for item in self.repository_api.repo_list_collaborators(
+                for item in list_all(
-                    self.org_name, repo_name
+                    self.repository_api.repo_list_collaborators,
+                    self.org_name,
+                    repo_name,
                 )
             ]
         self.issue_api.issue_create_issue(
@@ -217,11 +239,29 @@ class Gitea:
         )
 
     def check_exist_issue_by_title(self, repo_name: str, title: str) -> bool:
-        for issue in self.issue_api.issue_list_issues(self.org_name, repo_name):
+        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()