JOJ/Joint-Teapot
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: better logs

Browse Source
This commit is contained in:
张泊明518370910136 2021-09-15 04:00:23 +08:00
parent 48cbf4871d
commit c75caa3990
No known key found for this signature in database
GPG Key ID: FBEF5DE8B9F4C629
9 changed files with 150 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.env.example
View File

@ -1,4 +1,4 @@
CANVAS_ACCESS_TOKEN=CiBD8fLtEhUOkcjdKhIj18Fx13WV1e3hwffewrfN3whAcS1IKPeJ41fBKq1EzkGd
COURSE_ID=1445
CANVAS_COURSE_ID=1445
GITEA_ACCESS_TOKEN=10cdf70f8fe1b7a5d3321e5a10d2d743e4818d4a
ORG_NAME=VG101
GITEA_ORG_NAME=VG101

2
README.md
View File

@ -11,7 +11,7 @@ python3 -m venv env
source env/Scripts/activate
```
### Install
### Install & Run
```bash
pip3 install -e .

10
joint_teapot/__main__.py
View File

@ -3,6 +3,7 @@ __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
@ -32,7 +33,7 @@ def create_teams_and_repos_by_canvas_groups() -> None:
@app.command("get-public-keys", help="get all public keys on gitea")
def get_public_key_of_all_canvas_students() -> None:
echo(teapot.get_public_key_of_all_canvas_students())
echo("\n".join(teapot.get_public_key_of_all_canvas_students()))
@app.command("archieve", help="clone all gitea repos to local")
@ -47,7 +48,7 @@ def create_issue_for_repos(repo_names: List[str], title: str, body: str) -> None
@app.command("check-issues", help="check the existence of issue by title on gitea")
def check_exist_issue_by_title(repo_names: List[str], title: str) -> None:
echo(teapot.check_exist_issue_by_title(repo_names, title))
echo("\n".join(teapot.check_exist_issue_by_title(repo_names, title)))
@app.command(
@ -61,4 +62,7 @@ def checkout_to_repos_by_release_name(
if __name__ == "__main__":
app()
try:
app()
except Exception:
logger.exception("Unexpected error:")

4
joint_teapot/config.py
View File

@ -10,11 +10,11 @@ class Settings(BaseSettings):
# canvas
canvas_access_token: str = ""
course_id: int = 0
canvas_course_id: int = 0
# gitea
gitea_access_token: str = ""
org_name: str = ""
gitea_org_name: str = ""
# git
repos_dir: str = "./repos"

189
joint_teapot/teapot.py
View File

@ -1,90 +1,99 @@
from datetime import datetime
from typing import Any, Dict, List
from joint_teapot.utils import first
from joint_teapot.workers import Canvas, Git, Gitea
class Teapot:
_canvas = None
_gitea = None
_git = 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
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) -> List[str]:
return self.gitea.create_teams_and_repos_by_canvas_groups(
self.canvas.students, self.canvas.groups
)
def get_public_key_of_all_canvas_students(self) -> List[List[Dict[str, Any]]]:
return self.gitea.get_public_key_of_canvas_students(self.canvas.students)
def archieve_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()
]
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 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_repos_by_release_name(
self,
repo_names: List[str],
release_name: str,
due: datetime = datetime(3000, 1, 1),
) -> List[str]:
failed_repos = []
repos_releases = self.gitea.get_repos_releases(repo_names)
for repo_name, repo_releases in zip(repo_names, repos_releases):
release = first(repo_releases, lambda item: item["name"] == release_name)
if (
release is None
or datetime.strptime(release["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
>= due
):
failed_repos.append(repo_name)
continue
self.git.repo_clean_and_checkout(repo_name, f"tags/{release['tag_name']}")
return failed_repos
if __name__ == "__main__":
teapot = Teapot()
from datetime import datetime
from typing import List
from loguru import logger
from joint_teapot.config import settings
from joint_teapot.utils import first
from joint_teapot.workers import Canvas, Git, Gitea
class Teapot:
_canvas = None
_gitea = None
_git = 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
def __init__(self) -> None:
logger.info(
f"Settings loaded. Canvas Course ID: {settings.canvas_course_id}, Gitea Organization name: {settings.gitea_org_name}"
)
logger.info("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) -> List[str]:
return self.gitea.create_teams_and_repos_by_canvas_groups(
self.canvas.students, self.canvas.groups
)
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]:
return [
self.git.repo_clean_and_checkout(repo_name, "master")
for repo_name in self.gitea.get_all_repo_names()
]
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 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_repos_by_release_name(
self,
repo_names: List[str],
release_name: str,
due: datetime = datetime(3000, 1, 1),
) -> List[str]:
failed_repos = []
repos_releases = self.gitea.get_repos_releases(repo_names)
for repo_name, repo_releases in zip(repo_names, repos_releases):
release = first(repo_releases, lambda item: item["name"] == release_name)
if (
release is None
or datetime.strptime(release["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ")
>= due
):
failed_repos.append(repo_name)
continue
self.git.repo_clean_and_checkout(repo_name, f"tags/{release['tag_name']}")
return failed_repos
if __name__ == "__main__":
teapot = Teapot()

21
joint_teapot/workers/canvas.py
View File

@ -1,4 +1,5 @@
from canvasapi import Canvas as PyCanvas
from loguru import logger
from joint_teapot.config import settings
@ -9,27 +10,25 @@ class Canvas:
def __init__(
self,
access_token: str = settings.canvas_access_token,
courseID: int = settings.course_id,
course_id: int = settings.canvas_course_id,
):
self.canvas = PyCanvas("https://umjicanvas.com/", access_token)
self.course = self.canvas.get_course(courseID)
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"], include=["email"]
)
self.assignments = self.course.get_assignments()
self.groups = self.course.get_groups()
for attr in ["sis_login_id", "sortable_name"]:
if not hasattr(self.students[0], attr):
raise Exception(
f"Unable to gather students' {attr}, please contact the Canvas site admin"
)
# group: Group
# for group in self.groups:
# membership: GroupMembership
# print(group.__dict__)
# for membership in group.get_memberships():
# print(membership.user_id, end=", ")
# print("")
logger.info(f"Canvas students loaded.")
self.assignments = self.course.get_assignments()
logger.info(f"Canvas assignments loaded.")
self.groups = self.course.get_groups()
logger.info(f"Canvas groups loaded.")
logger.info("Canvas initialized.")
if __name__ == "__main__":

7
joint_teapot/workers/git.py
View File

@ -1,6 +1,8 @@
import os
import sys
from loguru import logger
current_path = sys.path[0]
sys.path.remove(current_path)
from git import Repo
@ -12,12 +14,15 @@ from joint_teapot.config import settings
class Git:
def __init__(
self, org_name: str = settings.org_name, repos_dir: str = settings.repos_dir
self,
org_name: str = settings.gitea_org_name,
repos_dir: str = settings.repos_dir,
):
self.org_name = org_name
if not os.path.isdir(repos_dir):
raise Exception(f"{repos_dir} does not exist! Create it first.")
self.repos_dir = repos_dir
logger.info("Git initialized.")
def clone_repo(self, repo_name: str, branch: str = "master") -> Repo:
repo_dir = os.path.join(self.repos_dir, repo_name)

34
joint_teapot/workers/gitea.py
View File

@ -7,6 +7,7 @@ import focs_gitea
from canvasapi.group import Group, GroupMembership
from canvasapi.paginated_list import PaginatedList
from canvasapi.user import User
from loguru import logger
from joint_teapot.config import settings
from joint_teapot.utils import first
@ -29,7 +30,7 @@ class Gitea:
def __init__(
self,
access_token: str = settings.gitea_access_token,
org_name: str = settings.org_name,
org_name: str = settings.gitea_org_name,
):
self.org_name = org_name
configuration = focs_gitea.Configuration()
@ -42,6 +43,7 @@ class Gitea:
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.info("Gitea initialized.")
@lru_cache()
def _get_team_id_by_name(self, name: str) -> int:
@ -67,7 +69,7 @@ class Gitea:
username = self._get_username_by_canvas_student(student)
self.organization_api.org_add_team_member(team_id, username)
except Exception as e:
print(e)
logger.error(e)
def create_personal_repos_for_canvas_students(
self,
@ -93,9 +95,13 @@ class Gitea:
"trust_model": "default",
},
)
self.repository_api.repo_add_collaborator(
self.org_name, repo.name, self._get_username_by_canvas_student(student)
)
try:
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(
@ -161,13 +167,17 @@ class Gitea:
)
return repo_names
def get_public_key_of_canvas_students(
self, students: PaginatedList
) -> List[List[Dict[str, Any]]]:
return [
self.user_api.user_list_keys(self._get_username_by_canvas_student(student))
for student in students
]
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 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 [

1
requirements.txt
View File

@ -1,5 +1,6 @@
canvasapi>=2.2.0
focs_gitea>=1.0.0
GitPython>=3.1.18
loguru>=0.5.3
pydantic[dotenv]>=1.8.1
typer[all]>=0.3.2