Coverage for mongo/homework.py: 100%
177 statements
« prev ^ index » next coverage.py v7.9.2, created at 2026-07-24 16:42 +0000
1from typing import List, Optional
2from . import engine
3from .user import User
4from .base import MongoBase
5from .course import Course
6from .utils import doc_required
7from .problem.problem import Problem
8from .ip_filter import IPFilter
9from datetime import datetime
11__all__ = ['Homework']
14class Error():
15 Illegal_penalty = 1
16 Invalid_penalty = 2
19def check_penalty(penalty: Optional[str]) -> int:
20 if penalty is None:
21 return 0
22 allowed_chars = ["+", "-", "*", "/", "=", ".", "(", ")", ":", ">", "<"]
23 allowed_words = ["score", "overtime", "if", "else"]
24 checkstring = ""
25 for i in penalty:
26 checkstring += (" " if i in allowed_chars else i)
27 for i in checkstring.split():
28 if i not in allowed_words:
29 try:
30 int(i)
31 except:
32 return Error.Illegal_penalty
33 try:
34 score = 0
35 overtime = 0
36 exec(penalty)
37 except:
38 return Error.Invalid_penalty
39 return 0
42# TODO: unittest for class `Homework`
43class Homework(MongoBase, engine=engine.Homework):
45 def is_valid_ip(self, ip: str) -> bool:
46 # no restriction, always valid
47 if not self.ip_filters:
48 return True
49 ip_filters = map(IPFilter, self.ip_filters)
50 return any(_filter.match(ip) for _filter in ip_filters)
52 @classmethod
53 @doc_required('course_name', 'course', Course)
54 def add(
55 cls,
56 user,
57 course: Course,
58 hw_name: str,
59 problem_ids: List[int] | None = None,
60 markdown: str = '',
61 scoreboard_status: int = 0,
62 start: Optional[float] = None,
63 end: Optional[float] = None,
64 penalty: Optional[str] = '',
65 ):
66 # check user is teacher or ta
67 if not course.permission(user, Course.Permission.GRADE):
68 raise PermissionError('user is not teacher or ta')
69 course_id = course.id
70 if cls.engine.objects(
71 course_id=str(course_id),
72 homework_name=hw_name,
73 ):
74 raise engine.NotUniqueError('homework exist')
75 # check problems exist
77 penalty_stat = check_penalty(penalty)
78 if penalty_stat == Error.Illegal_penalty:
79 raise ValueError("Illegal penalty")
80 elif penalty_stat == Error.Invalid_penalty:
81 raise ValueError("Invalid penalty")
83 problem_ids = problem_ids if problem_ids is not None else []
84 problems = [*map(Problem, problem_ids)]
85 if not all(problems):
86 raise engine.DoesNotExist('some problems not found!')
87 homework = cls.engine(
88 homework_name=hw_name,
89 course_id=str(course_id),
90 problem_ids=problem_ids,
91 scoreboard_status=scoreboard_status,
92 markdown=markdown,
93 )
94 if penalty:
95 homework.penalty = penalty
96 if start:
97 homework.duration.start = datetime.fromtimestamp(start)
98 if end:
99 homework.duration.end = datetime.fromtimestamp(end)
100 homework.save()
101 # init student status
102 user_problems = {}
103 for problem in problems:
104 problem_id = str(problem.problem_id)
105 user_problems[problem_id] = cls.default_problem_status()
106 problem.update(push__homeworks=homework)
107 homework.update(student_status={
108 s: user_problems
109 for s in course.student_nicknames
110 })
111 # add homework to course
112 course.update(push__homeworks=homework.id)
113 return homework
115 @classmethod
116 def update(
117 cls,
118 user,
119 homework_id: str,
120 markdown: str,
121 new_hw_name: str,
122 problem_ids: List[int],
123 penalty: str,
124 start: Optional[datetime] = None,
125 end: Optional[datetime] = None,
126 scoreboard_status: Optional[int] = None,
127 ):
128 homework = cls.engine.objects.get(id=homework_id)
129 course = Course(engine.Course.objects.get(id=homework.course_id))
130 # check user is teacher or ta
131 if not course.permission(user, Course.Permission.GRADE):
132 raise PermissionError('user is not teacher or ta')
133 # check the new_name hasn't been use in this course
135 if penalty is not None:
136 penalty_stat = check_penalty(penalty)
137 if penalty_stat == Error.Illegal_penalty:
138 raise ValueError("Illegal penalty")
139 elif penalty_stat == Error.Invalid_penalty:
140 raise ValueError("Invalid penalty")
141 else:
142 homework.penalty = penalty
144 if new_hw_name is not None:
145 if cls.engine.objects(
146 course_id=str(course.id),
147 homework_name=new_hw_name,
148 ):
149 raise engine.NotUniqueError('homework exist')
150 else:
151 homework.update(homework_name=new_hw_name)
152 # update fields
153 if start is not None:
154 homework.duration.start = datetime.fromtimestamp(start)
155 if end is not None:
156 homework.duration.end = datetime.fromtimestamp(end)
157 if scoreboard_status is not None:
158 homework.scoreboard_status = scoreboard_status
159 if markdown is not None:
160 homework.markdown = markdown
161 homework.save()
162 drop_ids = set(homework.problem_ids) - set(problem_ids)
163 new_ids = set(problem_ids) - set(homework.problem_ids)
164 student_status = homework.student_status
165 # add
166 for pid in new_ids:
167 problem = Problem(pid)
168 if not problem:
169 continue
170 homework.update(push__problem_ids=pid)
171 problem.update(push__homeworks=homework)
172 for key in course.student_nicknames:
173 student_status[key][str(pid)] = cls.default_problem_status()
174 # delete
175 for pid in drop_ids:
176 problem = Problem(pid)
177 if not problem:
178 continue
179 homework.update(pull__problem_ids=pid)
180 problem.update(pull__homeworks=homework)
181 for status in student_status.values():
182 del status[str(pid)]
183 homework.update(student_status=student_status)
184 return homework
186 # delete problems/paticipants in hw
187 @doc_required('course', 'course', Course)
188 def delete_problems(
189 self,
190 user,
191 course: Course,
192 ):
193 # check user is teacher or ta
194 if not course.permission(user, Course.Permission.GRADE):
195 raise PermissionError('user is not teacher or ta')
196 for pid in self.problem_ids:
197 problem = Problem(pid)
198 if not problem:
199 continue
200 problem.update(pull__homeworks=self.obj)
201 self.delete()
202 return self
204 @classmethod
205 @doc_required('course_name', 'course', Course)
206 def get_homeworks(cls, course: Course):
207 homeworks = course.homeworks or []
208 homeworks = sorted(homeworks, key=lambda h: h.duration.start)
209 return homeworks
211 @classmethod
212 def get_by_id(cls, homework_id):
213 try:
214 homework = cls.engine.objects.get(id=homework_id)
215 except engine.DoesNotExist:
216 raise engine.DoesNotExist('homework not exist')
217 return homework
219 @classmethod
220 def get_by_name(cls, course_name, homework_name):
221 try:
222 homework = cls.engine.objects.get(
223 course_id=str(Course(course_name).obj.id),
224 homework_name=homework_name,
225 )
226 except engine.DoesNotExist:
227 raise engine.DoesNotExist('homework not exist')
228 return homework
230 @classmethod
231 def default_problem_status(cls):
232 return {
233 'score': 0,
234 'problemStatus': None,
235 'submissionIds': [],
236 }
238 def add_student(self, students: List[User]):
239 if any(u.username in self.student_status for u in students):
240 raise ValueError('Student already in homework')
241 user_status = {
242 str(pid): self.default_problem_status()
243 for pid in self.problem_ids
244 }
245 for student in students:
246 self.student_status[student.username] = user_status
247 self.save()
249 def remove_student(self, students: List[User]):
250 if any(u.username not in self.student_status for u in students):
251 raise ValueError('Student not in homework')
252 for student in students:
253 del self.student_status[student.username]
254 self.save()
256 def do_penalty(self, submission, stat):
257 d = {}
259 d['score'] = submission.score - stat['rawScore']
260 if d['score'] > 0:
261 d['overtime'] = int((submission.timestamp.timestamp() -
262 self.duration.end.timestamp()) / 86400)
263 exec(self.penalty, d)
264 d['score'] = int(d['score'])
265 stat['score'] += d['score']
266 stat['rawScore'] = submission.score
268 return [stat['score'], stat['rawScore']]