feat: canvas score utils
This commit is contained in:
parent
7fa19c3cb9
commit
2dbe321d43
|
|
@ -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__":
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user