feat: better extract assignment
This commit is contained in:
parent
449874afc3
commit
3d18d5c6f7
|
|
@ -1,4 +1,6 @@
|
|||
from joint_teapot.__main__ import app
|
||||
__version__ = "0.0.0"
|
||||
|
||||
from joint_teapot.app import app
|
||||
from joint_teapot.teapot import Teapot as Teapot
|
||||
from joint_teapot.utils.logger import logger as logger
|
||||
|
||||
|
|
|
|||
|
|
@ -1,134 +1,5 @@
|
|||
__version__ = "0.0.0"
|
||||
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from typer import Argument, Typer, echo
|
||||
|
||||
from joint_teapot.teapot import Teapot
|
||||
from joint_teapot.utils.logger import logger
|
||||
|
||||
app = Typer(add_completion=False)
|
||||
|
||||
|
||||
class Tea:
|
||||
_teapot = None
|
||||
|
||||
@property
|
||||
def pot(self) -> Teapot:
|
||||
if not self._teapot:
|
||||
self._teapot = Teapot()
|
||||
return self._teapot
|
||||
|
||||
|
||||
tea = Tea() # lazy loader
|
||||
|
||||
|
||||
@app.command(
|
||||
"invite-to-teams", help="invite all canvas students to gitea teams by team name"
|
||||
)
|
||||
def add_all_canvas_students_to_teams(team_names: List[str]) -> None:
|
||||
tea.pot.add_all_canvas_students_to_teams(team_names)
|
||||
|
||||
|
||||
@app.command(
|
||||
"create-personal-repos",
|
||||
help="create personal repos on gitea for all canvas students",
|
||||
)
|
||||
def create_personal_repos_for_all_canvas_students() -> None:
|
||||
tea.pot.create_personal_repos_for_all_canvas_students()
|
||||
|
||||
|
||||
@app.command("create-teams", help="create teams on gitea by canvas groups")
|
||||
def create_teams_and_repos_by_canvas_groups(group_prefix: str) -> None:
|
||||
tea.pot.create_teams_and_repos_by_canvas_groups(group_prefix)
|
||||
|
||||
|
||||
@app.command("get-public-keys", help="list all public keys on gitea")
|
||||
def get_public_key_of_all_canvas_students() -> None:
|
||||
res = []
|
||||
for k, v in tea.pot.get_public_key_of_all_canvas_students().items():
|
||||
keys = "\\n".join(v)
|
||||
res.append(f"{k},{keys}")
|
||||
echo("\n".join(res))
|
||||
|
||||
|
||||
@app.command("clone-all-repos", help="clone all gitea repos to local")
|
||||
def clone_all_repos() -> None:
|
||||
tea.pot.clone_all_repos()
|
||||
|
||||
|
||||
@app.command("create-issues", help="create issues on gitea")
|
||||
def create_issue_for_repos(repo_names: List[str], title: str, body: str) -> None:
|
||||
tea.pot.create_issue_for_repos(repo_names, title, body)
|
||||
|
||||
|
||||
@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("\n".join(tea.pot.check_exist_issue_by_title(repo_names, title)))
|
||||
|
||||
|
||||
@app.command(
|
||||
"checkout-releases",
|
||||
help="checkout git repo to git tag fetched from gitea by release name, with due date",
|
||||
)
|
||||
def checkout_to_repos_by_release_name(
|
||||
repo_names: List[str], release_name: str, due: datetime = Argument("3000-01-01")
|
||||
) -> None:
|
||||
failed_repos = []
|
||||
succeed_repos = []
|
||||
for repo_name in repo_names:
|
||||
succeed = tea.pot.checkout_to_repo_by_release_name(repo_name, release_name, due)
|
||||
if not succeed:
|
||||
failed_repos.append(repo_name)
|
||||
else:
|
||||
succeed_repos.append(repo_name)
|
||||
echo(f"succeed repos: {succeed_repos}")
|
||||
echo(f"failed repos: {failed_repos}")
|
||||
|
||||
|
||||
@app.command(
|
||||
"close-all-issues", help="close all issues and pull requests in gitea organization"
|
||||
)
|
||||
def close_all_issues() -> None:
|
||||
tea.pot.gitea.close_all_issues()
|
||||
|
||||
|
||||
@app.command("archieve-all-repos", help="archieve all repos in gitea organization")
|
||||
def archieve_all_repos() -> None:
|
||||
tea.pot.gitea.archieve_all_repos()
|
||||
|
||||
|
||||
@app.command("get-no-collaborator-repos", help="list all repos with no collaborators")
|
||||
def get_no_collaborator_repos() -> None:
|
||||
tea.pot.gitea.get_no_collaborator_repos()
|
||||
|
||||
|
||||
@app.command("get-repos-status", help="list status of all repos with conditions")
|
||||
def get_repos_status(
|
||||
commit_lt: int = Argument(100000, help="commit count less than"),
|
||||
issue_lt: int = Argument(100000, help="issue count less than"),
|
||||
) -> None:
|
||||
tea.pot.get_repos_status(commit_lt, issue_lt)
|
||||
|
||||
|
||||
@app.command(
|
||||
"prepare-assignment-dir",
|
||||
help='prepare assignment dir from extracted canvas "Download Submissions" zip',
|
||||
)
|
||||
def prepare_assignment_dir(dir_or_zip_file: Path) -> None:
|
||||
tea.pot.canvas.prepare_assignment_dir(str(dir_or_zip_file))
|
||||
|
||||
|
||||
@app.command(
|
||||
"upload-assignment-grades",
|
||||
help="upload assignment grades to canvas from grade file (GRADE.txt by default), "
|
||||
+ "read the first line as grade, the rest as comments",
|
||||
)
|
||||
def upload_assignment_grades(assignments_dir: Path, assignment_name: str) -> None:
|
||||
tea.pot.canvas.upload_assignment_grades(str(assignments_dir), assignment_name)
|
||||
|
||||
from joint_teapot.app import app
|
||||
from joint_teapot.utils.logger import logger as logger
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
|
|
|
|||
135
joint_teapot/app.py
Normal file
135
joint_teapot/app.py
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from typer import Argument, Typer, echo
|
||||
|
||||
from joint_teapot.teapot import Teapot
|
||||
from joint_teapot.utils.logger import logger
|
||||
|
||||
app = Typer(add_completion=False)
|
||||
|
||||
|
||||
class Tea:
|
||||
_teapot = None
|
||||
|
||||
@property
|
||||
def pot(self) -> Teapot:
|
||||
if not self._teapot:
|
||||
self._teapot = Teapot()
|
||||
return self._teapot
|
||||
|
||||
|
||||
tea = Tea() # lazy loader
|
||||
|
||||
|
||||
@app.command(
|
||||
"invite-to-teams", help="invite all canvas students to gitea teams by team name"
|
||||
)
|
||||
def add_all_canvas_students_to_teams(team_names: List[str]) -> None:
|
||||
tea.pot.add_all_canvas_students_to_teams(team_names)
|
||||
|
||||
|
||||
@app.command(
|
||||
"create-personal-repos",
|
||||
help="create personal repos on gitea for all canvas students",
|
||||
)
|
||||
def create_personal_repos_for_all_canvas_students() -> None:
|
||||
tea.pot.create_personal_repos_for_all_canvas_students()
|
||||
|
||||
|
||||
@app.command("create-teams", help="create teams on gitea by canvas groups")
|
||||
def create_teams_and_repos_by_canvas_groups(group_prefix: str) -> None:
|
||||
tea.pot.create_teams_and_repos_by_canvas_groups(group_prefix)
|
||||
|
||||
|
||||
@app.command("get-public-keys", help="list all public keys on gitea")
|
||||
def get_public_key_of_all_canvas_students() -> None:
|
||||
res = []
|
||||
for k, v in tea.pot.get_public_key_of_all_canvas_students().items():
|
||||
keys = "\\n".join(v)
|
||||
res.append(f"{k},{keys}")
|
||||
echo("\n".join(res))
|
||||
|
||||
|
||||
@app.command("clone-all-repos", help="clone all gitea repos to local")
|
||||
def clone_all_repos() -> None:
|
||||
tea.pot.clone_all_repos()
|
||||
|
||||
|
||||
@app.command("create-issues", help="create issues on gitea")
|
||||
def create_issue_for_repos(repo_names: List[str], title: str, body: str) -> None:
|
||||
tea.pot.create_issue_for_repos(repo_names, title, body)
|
||||
|
||||
|
||||
@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("\n".join(tea.pot.check_exist_issue_by_title(repo_names, title)))
|
||||
|
||||
|
||||
@app.command(
|
||||
"checkout-releases",
|
||||
help="checkout git repo to git tag fetched from gitea by release name, with due date",
|
||||
)
|
||||
def checkout_to_repos_by_release_name(
|
||||
repo_names: List[str], release_name: str, due: datetime = Argument("3000-01-01")
|
||||
) -> None:
|
||||
failed_repos = []
|
||||
succeed_repos = []
|
||||
for repo_name in repo_names:
|
||||
succeed = tea.pot.checkout_to_repo_by_release_name(repo_name, release_name, due)
|
||||
if not succeed:
|
||||
failed_repos.append(repo_name)
|
||||
else:
|
||||
succeed_repos.append(repo_name)
|
||||
echo(f"succeed repos: {succeed_repos}")
|
||||
echo(f"failed repos: {failed_repos}")
|
||||
|
||||
|
||||
@app.command(
|
||||
"close-all-issues", help="close all issues and pull requests in gitea organization"
|
||||
)
|
||||
def close_all_issues() -> None:
|
||||
tea.pot.gitea.close_all_issues()
|
||||
|
||||
|
||||
@app.command("archieve-all-repos", help="archieve all repos in gitea organization")
|
||||
def archieve_all_repos() -> None:
|
||||
tea.pot.gitea.archieve_all_repos()
|
||||
|
||||
|
||||
@app.command("get-no-collaborator-repos", help="list all repos with no collaborators")
|
||||
def get_no_collaborator_repos() -> None:
|
||||
tea.pot.gitea.get_no_collaborator_repos()
|
||||
|
||||
|
||||
@app.command("get-repos-status", help="list status of all repos with conditions")
|
||||
def get_repos_status(
|
||||
commit_lt: int = Argument(100000, help="commit count less than"),
|
||||
issue_lt: int = Argument(100000, help="issue count less than"),
|
||||
) -> None:
|
||||
tea.pot.get_repos_status(commit_lt, issue_lt)
|
||||
|
||||
|
||||
@app.command(
|
||||
"prepare-assignment-dir",
|
||||
help='prepare assignment dir from extracted canvas "Download Submissions" zip',
|
||||
)
|
||||
def prepare_assignment_dir(dir_or_zip_file: Path) -> None:
|
||||
tea.pot.canvas.prepare_assignment_dir(str(dir_or_zip_file))
|
||||
|
||||
|
||||
@app.command(
|
||||
"upload-assignment-grades",
|
||||
help="upload assignment grades to canvas from grade file (GRADE.txt by default), "
|
||||
+ "read the first line as grade, the rest as comments",
|
||||
)
|
||||
def upload_assignment_grades(assignments_dir: Path, assignment_name: str) -> None:
|
||||
tea.pot.canvas.upload_assignment_grades(str(assignments_dir), assignment_name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
app()
|
||||
except Exception:
|
||||
logger.exception("Unexpected error:")
|
||||
|
|
@ -61,6 +61,7 @@ class Canvas:
|
|||
if not os.path.exists(grade_file_path):
|
||||
open(grade_file_path, mode="w")
|
||||
late_students = set()
|
||||
error_students = set()
|
||||
submitted_ids = set()
|
||||
for path in glob(os.path.join(assignments_dir, "*")):
|
||||
filename = os.path.basename(path)
|
||||
|
|
@ -72,6 +73,7 @@ class Canvas:
|
|||
else:
|
||||
file_id = int(segments[1])
|
||||
login_id = login_ids[file_id]
|
||||
student = first(self.students, lambda x: x.login_id == login_id)
|
||||
target_dir = os.path.join(assignments_dir, login_id)
|
||||
if segments[1] == "late":
|
||||
# TODO: check the delay time of late submission
|
||||
|
|
@ -79,12 +81,15 @@ class Canvas:
|
|||
grade_file_path = os.path.join(path, self.grade_filename)
|
||||
if os.path.exists(grade_file_path):
|
||||
open(grade_file_path, mode="a").write("LATE SUBMISSION\n")
|
||||
student = first(self.students, lambda x: x.login_id == login_id)
|
||||
late_students.add(student)
|
||||
try:
|
||||
extract_archive(path, outdir=target_dir, verbosity=-1)
|
||||
logger.info(f"Extract succeed: {student}")
|
||||
os.remove(path)
|
||||
except PatoolError:
|
||||
except PatoolError as e:
|
||||
if not str(e).startswith("unknown archive format"):
|
||||
logger.exception(f"Extract failed: {student}")
|
||||
error_students.add(student)
|
||||
os.rename(path, os.path.join(target_dir, filename))
|
||||
submitted_ids.add(login_id)
|
||||
if login_ids:
|
||||
|
|
@ -98,6 +103,9 @@ class Canvas:
|
|||
if late_students:
|
||||
tmp = ", ".join([str(student) for student in late_students])
|
||||
logger.info(f"Late student(s): {tmp}")
|
||||
if error_students:
|
||||
tmp = ", ".join([str(student) for student in error_students])
|
||||
logger.info(f"Extract error student(s): {tmp}")
|
||||
|
||||
def upload_assignment_grades(
|
||||
self, assignments_dir: str, assignment_name: str
|
||||
|
|
|
|||
7
setup.py
7
setup.py
|
|
@ -7,9 +7,9 @@ from setuptools import find_packages, setup
|
|||
|
||||
def get_version(package: str) -> str:
|
||||
"""
|
||||
Return package version as listed in `__version__` in `__main__.py`.
|
||||
Return package version as listed in `__version__` in `__init__.py`.
|
||||
"""
|
||||
path = os.path.join(package, "__main__.py")
|
||||
path = os.path.join(package, "__init__.py")
|
||||
main_py = open(path, "r", encoding="utf8").read()
|
||||
match = re.search("__version__ = ['\"]([^'\"]+)['\"]", main_py)
|
||||
if match is None:
|
||||
|
|
@ -25,6 +25,9 @@ def get_long_description() -> str:
|
|||
|
||||
|
||||
def get_install_requires() -> List[str]:
|
||||
"""
|
||||
Return each line of requirements.txt.
|
||||
"""
|
||||
return open("requirements.txt").read().splitlines()
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user