fix(regex): full match

The feature allows choosing templates when creating repos for individuals and groups.

Reviewed-on: JOJ/Joint-Teapot#1
Reviewed-by: 张泊明518370910136 <bomingzh@sjtu.edu.cn>
Co-authored-by: Min Zhengjie <minzhengjie@sjtu.edu.cn>
Co-committed-by: Min Zhengjie <minzhengjie@sjtu.edu.cn>

joint_teapot/app.py

@@ -50,13 +50,17 @@ def add_all_canvas_students_to_teams(team_names: List[str]) -> None:
     "create-personal-repos",
     help="create personal repos on gitea for all canvas students",
 )
-def create_personal_repos_for_all_canvas_students(suffix: str = Option("")) -> None:
-    tea.pot.create_personal_repos_for_all_canvas_students(suffix)
+def create_personal_repos_for_all_canvas_students(
+    suffix: str = Option(""), template: str = Option("", help="generate from template")
+) -> None:
+    tea.pot.create_personal_repos_for_all_canvas_students(suffix, template)
 
 
 @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)
+def create_teams_and_repos_by_canvas_groups(
+    group_prefix: str, template: str = Option("", help="generate from template")
+) -> None:
+    tea.pot.create_teams_and_repos_by_canvas_groups(group_prefix, template)
 
 
 @app.command("get-public-keys", help="list all public keys on gitea")
@@ -295,6 +299,23 @@ def joj3_all_env(
         True,
         help="whether to include submitter in issue title",
     ),
+    issue_label_name: str = Option(
+        "Kind/Testing",
+        help="label name for the issue created by this command",
+    ),
+    issue_label_color: str = Option(
+        "#795548",
+        help="label color for the issue created by this command",
+    ),
+    end_time: Optional[datetime] = Option(None),
+    penalty_config: str = Option(
+        "",
+        help=(
+            "Configuration for penalties in the format "
+            "'hours=factor'. "
+            "Example: --penalty-config 24=0.75,48=0.5"
+        ),
+    ),
 ) -> None:
     app.pretty_exceptions_enable = False
     set_settings(Settings(_env_file=env_path))
@@ -310,6 +331,7 @@ def joj3_all_env(
         logger.error("missing required env var")
         raise Exit(code=1)
     submitter_repo_name = env.github_repository.split("/")[-1]
+    penalty_factor = joj3.get_penalty_factor(end_time, penalty_config)
     total_score = joj3.get_total_score(env.joj3_output_path)
     res = {
         "totalScore": total_score,
@@ -336,6 +358,9 @@ def joj3_all_env(
             gitea_actions_url,
             submitter_in_issue_title,
             submitter_repo_name,
+            issue_label_name,
+            issue_label_color,
+            penalty_factor,
         )
         res["issue"] = issue_number
         gitea_issue_url = f"{submitter_repo_url}/issues/{issue_number}"
@@ -446,8 +471,16 @@ def joj3_check_env(
             "Example: --group-config joj=10:24,run=20:48"
         ),
     ),
-    valid_after: Optional[datetime] = Option(None),
-    valid_before: Optional[datetime] = Option(None),
+    begin_time: Optional[datetime] = Option(None),
+    end_time: Optional[datetime] = Option(None),
+    penalty_config: str = Option(
+        "",
+        help=(
+            "Configuration for penalties in the format "
+            "'hours=factor'. "
+            "Example: --penalty-config 24=0.75,48=0.5"
+        ),
+    ),
 ) -> None:
     app.pretty_exceptions_enable = False
     set_settings(Settings(_env_file=env_path))
@@ -461,8 +494,9 @@ def joj3_check_env(
         logger.error("missing required env var")
         raise Exit(code=1)
     time_msg, time_failed = tea.pot.joj3_check_submission_time(
-        valid_after,
-        valid_before,
+        begin_time,
+        end_time,
+        penalty_config,
     )
     count_msg, count_failed = tea.pot.joj3_check_submission_count(
         env, grading_repo_name, group_config, scoreboard_filename

joint_teapot/teapot.py

@@ -96,15 +96,16 @@ class Teapot:
         return self.gitea.add_canvas_students_to_teams(self.canvas.students, team_names)
 
     def create_personal_repos_for_all_canvas_students(
-        self, suffix: str = ""
+        self, suffix: str = "", template: str = ""
     ) -> List[str]:
         return self.gitea.create_personal_repos_for_canvas_students(
             self.canvas.students,
             lambda user: default_repo_name_convertor(user) + suffix,
+            template,
         )
 
     def create_teams_and_repos_by_canvas_groups(
-        self, group_prefix: str = ""
+        self, group_prefix: str = "", template: str = ""
     ) -> List[str]:
         def convertor(name: str) -> Optional[str]:
             if group_prefix and not name.startswith(group_prefix):
@@ -114,7 +115,7 @@ class Teapot:
             return f"{team_name}{number:02}"
 
         return self.gitea.create_teams_and_repos_by_canvas_groups(
-            self.canvas.students, self.canvas.groups, convertor, convertor
+            self.canvas.students, self.canvas.groups, convertor, convertor, template
         )
 
     def get_public_key_of_all_canvas_students(self) -> Dict[str, List[str]]:
@@ -237,6 +238,9 @@ class Teapot:
         gitea_actions_url: str,
         submitter_in_issue_title: bool,
         submitter_repo_name: str,
+        issue_label_name: str,
+        issue_label_color: str,
+        penalty_factor: float,
     ) -> int:
         title, comment = joj3.generate_title_and_comment(
             env.joj3_output_path,
@@ -248,6 +252,7 @@ class Teapot:
             submitter_in_issue_title,
             env.joj3_run_id,
             max_total_score,
+            penalty_factor,
         )
         title_prefix = joj3.get_title_prefix(
             env.joj3_conf_name, env.github_actor, submitter_in_issue_title
@@ -264,10 +269,24 @@ class Teapot:
                 break
         else:
             new_issue = True
+            labels = self.gitea.issue_api.issue_list_labels(
+                self.gitea.org_name, submitter_repo_name
+            )
+            label_id = 0
+            label = first(labels, lambda label: label.name == issue_label_name)
+            if label:
+                label_id = label.id
+            else:
+                label = self.gitea.issue_api.issue_create_label(
+                    self.gitea.org_name,
+                    submitter_repo_name,
+                    body={"name": issue_label_name, "color": issue_label_color},
+                )
+                label_id = label.id
             joj3_issue = self.gitea.issue_api.issue_create_issue(
                 self.gitea.org_name,
                 submitter_repo_name,
-                body={"title": title, "body": comment},
+                body={"title": title, "body": comment, "labels": [label_id]},
             )
             logger.info(f"created joj3 issue: #{joj3_issue.number}")
         gitea_issue_url = joj3_issue.html_url
@@ -283,21 +302,49 @@ class Teapot:
 
     def joj3_check_submission_time(
         self,
-        valid_after: Optional[datetime] = None,
-        valid_before: Optional[datetime] = None,
+        begin_time: Optional[datetime] = None,
+        end_time: Optional[datetime] = None,
+        penalty_config: str = "",
     ) -> Tuple[str, bool]:
         now = datetime.now()
-        if (valid_after and now < valid_after) or (valid_before and now > valid_before):
-            return (
-                "### Submission Time Check Failed:\n"
-                f"Current time {now} is not in the valid range "
-                f"[{valid_after}, {valid_before}].\n",
-                True,
-            )
+        penalties = joj3.parse_penalty_config(penalty_config)
+        if penalties and end_time:
+            penalty_end_time = end_time + timedelta(hours=penalties[-1][0])
+            if begin_time and now < begin_time:
+                return (
+                    "### Submission Time Check Failed\n"
+                    f"Current time {now} is not in the valid range "
+                    f"[{begin_time}, {end_time}].\n",
+                    True,
+                )
+            elif now > penalty_end_time:
+                return (
+                    "### Submission Time Check Failed\n"
+                    f"Current time {now} is not in the valid range "
+                    f"[{begin_time}, {end_time}], and the penalty range "
+                    f"[{end_time + timedelta(seconds=1)}, {penalty_end_time}].\n",
+                    True,
+                )
+            else:
+                return (
+                    "### Submission Time Check Passed\n"
+                    f"Current time {now} is not in the valid range "
+                    f"[{begin_time}, {end_time}], but in the penalty range "
+                    f"[{end_time + timedelta(seconds=1)}, {penalty_end_time}].\n",
+                    False,
+                )
+        else:
+            if (begin_time and now < begin_time) or (end_time and now > end_time):
+                return (
+                    "### Submission Time Check Failed\n"
+                    f"Current time {now} is not in the valid range "
+                    f"[{begin_time}, {end_time}].\n",
+                    True,
+                )
         return (
-            "### Submission Time Check Passed:\n"
+            "### Submission Time Check Passed\n"
             f"Current time {now} is in the valid range "
-            f"[{valid_after}, {valid_before}].\n",
+            f"[{begin_time}, {end_time}].\n",
             False,
         )
 
@@ -322,7 +369,11 @@ class Teapot:
         time_windows = []
         valid_items = []
         for item in items:
+            if "=" not in item:
+                continue
             name, values = item.split("=")
+            if ":" not in values:
+                continue
             max_count, time_period = map(int, values.split(":"))
             if max_count < 0 or time_period < 0:
                 continue
@@ -394,9 +445,9 @@ class Teapot:
                 comment += "."
             comment += "\n"
         if failed:
-            title = "### Submission Count Check Failed:"
+            title = "### Submission Count Check Failed"
         else:
-            title = "### Submission Count Check Passed:"
+            title = "### Submission Count Check Passed"
         msg = f"{title}\n{comment}\n"
         return msg, failed
 

joint_teapot/utils/joj3.py

@@ -2,8 +2,8 @@ import bisect
 import csv
 import json
 import os
-from datetime import datetime
-from typing import Any, Dict, List, Tuple
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional, Tuple
 
 from pydantic_settings import BaseSettings
 
@@ -215,6 +215,7 @@ def generate_title_and_comment(
     submitter_in_title: bool = True,
     run_id: str = "unknown",
     max_total_score: int = -1,
+    penalty_factor: float = 1.0,
 ) -> Tuple[str, str]:
     with open(score_file_path) as json_file:
         stages: List[Dict[str, Any]] = json.load(json_file)
@@ -234,6 +235,10 @@ def generate_title_and_comment(
         "Powered by [JOJ3](https://github.com/joint-online-judge/JOJ3) and "
         "[Joint-Teapot](https://github.com/BoYanZh/Joint-Teapot) with ❤️.\n"
     )
+    if penalty_factor != 1.0:
+        comment += (
+            f"## Total Score Penalty Warning\nThe total score is multiplied by 0.75.\n"
+        )
     for stage in stages:
         if all(
             result["score"] == 0 and result["comment"].strip() == ""
@@ -254,6 +259,8 @@ def generate_title_and_comment(
             comment += "</details>\n\n"
             total_score += result["score"]
         comment += "\n"
+    if penalty_factor != 1.0:
+        total_score = round(total_score * penalty_factor)
     title = get_title_prefix(exercise_name, submitter, submitter_in_title)
     if max_total_score >= 0:
         title += f"{total_score} / {max_total_score}"
@@ -281,3 +288,31 @@ def get_title_prefix(
     if not submitter_in_title:
         title = f"JOJ3 Result for {exercise_name} - Score: "
     return title
+
+
+def parse_penalty_config(penalty_config: str) -> List[Tuple[float, float]]:
+    res = []
+    for penalty in penalty_config.split(","):
+        if "=" not in penalty:
+            continue
+        hour, factor = map(float, penalty.split("="))
+        res.append((hour, factor))
+    res.sort(key=lambda x: x[0])
+    return res
+
+
+def get_penalty_factor(
+    end_time: Optional[datetime],
+    penalty_config: str,
+) -> float:
+    if not end_time or not penalty_config:
+        return 1.0
+    penalties = parse_penalty_config(penalty_config)
+    now = datetime.now()
+    res = 0.0
+    for hour, factor in penalties[::-1]:
+        if now < end_time + timedelta(hours=hour):
+            res = factor
+        else:
+            break
+    return res

joint_teapot/workers/gitea.py

@@ -116,6 +116,7 @@ class Gitea:
         repo_name_convertor: Callable[
             [User], Optional[str]
         ] = default_repo_name_convertor,
+        template: str = "",
     ) -> List[str]:
         repo_names = []
         for student in students:
@@ -123,17 +124,32 @@ class Gitea:
             if repo_name is None:
                 continue
             repo_names.append(repo_name)
-            body = {
-                "auto_init": False,
-                "default_branch": settings.default_branch,
-                "name": repo_name,
-                "private": True,
-                "template": False,
-                "trust_model": "default",
-            }
             try:
                 try:
-                    self.organization_api.create_org_repo(self.org_name, body=body)
+                    if template == "":
+                        body = {
+                            "auto_init": False,
+                            "default_branch": settings.default_branch,
+                            "name": repo_name,
+                            "private": True,
+                            "template": False,
+                            "trust_model": "default",
+                        }
+                        self.organization_api.create_org_repo(self.org_name, body=body)
+                    else:
+                        body = {
+                            "default_branch": settings.default_branch,
+                            "git_content": True,
+                            "git_hooks": True,
+                            "labels": True,
+                            "name": repo_name,
+                            "owner": self.org_name,
+                            "private": True,
+                            "protected_branch": True,
+                        }
+                        self.repository_api.generate_repo(
+                            self.org_name, template, body=body
+                        )
                     logger.info(
                         f"Personal repo {self.org_name}/{repo_name} for {student} created"
                     )
@@ -158,6 +174,7 @@ class Gitea:
         groups: PaginatedList,
         team_name_convertor: Callable[[str], Optional[str]] = lambda name: name,
         repo_name_convertor: Callable[[str], Optional[str]] = lambda name: name,
+        template: str = "",
         permission: PermissionEnum = PermissionEnum.write,
     ) -> List[str]:
         repo_names = []
@@ -190,21 +207,37 @@ class Gitea:
                         ],
                     },
                 )
-                logger.info(f"{self.org_name}/{team_name} created")
+                logger.info(f"Team {team_name} created")
             if first(repos, lambda repo: repo.name == repo_name) is None:
                 repo_names.append(repo_name)
-                self.organization_api.create_org_repo(
-                    self.org_name,
-                    body={
-                        "auto_init": False,
-                        "default_branch": settings.default_branch,
-                        "name": repo_name,
-                        "private": True,
-                        "template": False,
-                        "trust_model": "default",
-                    },
-                )
-                logger.info(f"Team {team_name} created")
+                if template == "":
+                    self.organization_api.create_org_repo(
+                        self.org_name,
+                        body={
+                            "auto_init": False,
+                            "default_branch": settings.default_branch,
+                            "name": repo_name,
+                            "private": True,
+                            "template": False,
+                            "trust_model": "default",
+                        },
+                    )
+                else:
+                    self.repository_api.generate_repo(
+                        self.org_name,
+                        template,
+                        body={
+                            "default_branch": settings.default_branch,
+                            "git_content": True,
+                            "git_hooks": True,
+                            "labels": True,
+                            "name": repo_name,
+                            "owner": self.org_name,
+                            "private": True,
+                            "protected_branch": True,
+                        },
+                    )
+                logger.info(f"{self.org_name}/{team_name} created")
             try:
                 self.organization_api.org_add_team_repository(
                     team.id, self.org_name, repo_name
@@ -449,7 +482,7 @@ class Gitea:
             logger.info("Dry run enabled. No changes will be made to the repositories.")
         logger.info(f"Archiving repos with name matching {regex}")
         for repo_name in self.get_all_repo_names():
-            if re.match(regex, repo_name):
+            if re.fullmatch(regex, repo_name):
                 logger.info(f"Archived {repo_name}")
                 if not dry_run:
                     self.repository_api.repo_edit(
@@ -504,7 +537,7 @@ class Gitea:
         self, milestone: str, regex: str, due_date: str, description: str
     ) -> None:
         for repo_name in self.get_all_repo_names():
-            if not re.match(regex, repo_name):
+            if not re.fullmatch(regex, repo_name):
                 continue
             milestone_list = self.issue_api.issue_get_milestones_list(
                 self.org_name, repo_name