Coverage for mongo/homework.py: 100%
176 statements
« prev ^ index » next coverage.py v7.3.2, created at 2024-11-05 04:22 +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] = [],
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 problems = [*map(Problem, problem_ids)]
84 if not all(problems):
85 raise engine.DoesNotExist('some problems not found!')
86 homework = cls.engine(
87 homework_name=hw_name,
88 course_id=str(course_id),
89 problem_ids=problem_ids,
90 scoreboard_status=scoreboard_status,
91 markdown=markdown,
92 )
93 if penalty:
94 homework.penalty = penalty
95 if start:
96 homework.duration.start = datetime.fromtimestamp(start)
97 if end:
98 homework.duration.end = datetime.fromtimestamp(end)
99 homework.save()
100 # init student status
101 user_problems = {}
102 for problem in problems:
103 problem_id = str(problem.problem_id)
104 user_problems[problem_id] = cls.default_problem_status()
105 problem.update(push__homeworks=homework)
106 homework.update(student_status={
107 s: user_problems
108 for s in course.student_nicknames
109 })
110 # add homework to course
111 course.update(push__homeworks=homework.id)
112 return homework
114 @classmethod
115 def update(
116 cls,
117 user,
118 homework_id: str,
119 markdown: str,
120 new_hw_name: str,
121 problem_ids: List[int],
122 penalty: str,
123 start: Optional[datetime] = None,
124 end: Optional[datetime] = None,
125 scoreboard_status: Optional[int] = None,
126 ):
127 homework = cls.engine.objects.get(id=homework_id)
128 course = Course(engine.Course.objects.get(id=homework.course_id))
129 # check user is teacher or ta
130 if not course.permission(user, Course.Permission.GRADE):
131 raise PermissionError('user is not teacher or ta')
132 # check the new_name hasn't been use in this course
134 if penalty is not None:
135 penalty_stat = check_penalty(penalty)
136 if penalty_stat == Error.Illegal_penalty:
137 raise ValueError("Illegal penalty")
138 elif penalty_stat == Error.Invalid_penalty:
139 raise ValueError("Invalid penalty")
140 else:
141 homework.penalty = penalty
143 if new_hw_name is not None:
144 if cls.engine.objects(
145 course_id=str(course.id),
146 homework_name=new_hw_name,
147 ):
148 raise engine.NotUniqueError('homework exist')
149 else:
150 homework.update(homework_name=new_hw_name)
151 # update fields
152 if start is not None:
153 homework.duration.start = datetime.fromtimestamp(start)
154 if end is not None:
155 homework.duration.end = datetime.fromtimestamp(end)
156 if scoreboard_status is not None:
157 homework.scoreboard_status = scoreboard_status
158 if markdown is not None:
159 homework.markdown = markdown
160 homework.save()
161 drop_ids = set(homework.problem_ids) - set(problem_ids)
162 new_ids = set(problem_ids) - set(homework.problem_ids)
163 student_status = homework.student_status
164 # add
165 for pid in new_ids:
166 problem = Problem(pid)
167 if not problem:
168 continue
169 homework.update(push__problem_ids=pid)
170 problem.update(push__homeworks=homework)
171 for key in course.student_nicknames:
172 student_status[key][str(pid)] = cls.default_problem_status()
173 # delete
174 for pid in drop_ids:
175 problem = Problem(pid)
176 if not problem:
177 continue
178 homework.update(pull__problem_ids=pid)
179 problem.update(pull__homeworks=homework)
180 for status in student_status.values():
181 del status[str(pid)]
182 homework.update(student_status=student_status)
183 return homework
185 # delete problems/paticipants in hw
186 @doc_required('course', 'course', Course)
187 def delete_problems(
188 self,
189 user,
190 course: Course,
191 ):
192 # check user is teacher or ta
193 if not course.permission(user, Course.Permission.GRADE):
194 raise PermissionError('user is not teacher or ta')
195 for pid in self.problem_ids:
196 problem = Problem(pid)
197 if not problem:
198 continue
199 problem.update(pull__homeworks=self.obj)
200 self.delete()
201 return self
203 @classmethod
204 @doc_required('course_name', 'course', Course)
205 def get_homeworks(cls, course: Course):
206 homeworks = course.homeworks or []
207 homeworks = sorted(homeworks, key=lambda h: h.duration.start)
208 return homeworks
210 @classmethod
211 def get_by_id(cls, homework_id):
212 try:
213 homework = cls.engine.objects.get(id=homework_id)
214 except engine.DoesNotExist:
215 raise engine.DoesNotExist('homework not exist')
216 return homework
218 @classmethod
219 def get_by_name(cls, course_name, homework_name):
220 try:
221 homework = cls.engine.objects.get(
222 course_id=str(Course(course_name).obj.id),
223 homework_name=homework_name,
224 )
225 except engine.DoesNotExist:
226 raise engine.DoesNotExist('homework not exist')
227 return homework
229 @classmethod
230 def default_problem_status(cls):
231 return {
232 'score': 0,
233 'problemStatus': None,
234 'submissionIds': [],
235 }
237 def add_student(self, students: List[User]):
238 if any(u.username in self.student_status for u in students):
239 raise ValueError('Student already in homework')
240 user_status = {
241 str(pid): self.default_problem_status()
242 for pid in self.problem_ids
243 }
244 for student in students:
245 self.student_status[student.username] = user_status
246 self.save()
248 def remove_student(self, students: List[User]):
249 if any(u.username not in self.student_status for u in students):
250 raise ValueError('Student not in homework')
251 for student in students:
252 del self.student_status[student.username]
253 self.save()
255 def do_penalty(self, submission, stat):
256 d = {}
258 d['score'] = submission.score - stat['rawScore']
259 if d['score'] > 0:
260 d['overtime'] = int((submission.timestamp.timestamp() -
261 self.duration.end.timestamp()) / 86400)
262 exec(self.penalty, d)
263 d['score'] = int(d['score'])
264 stat['score'] += d['score']
265 stat['rawScore'] = submission.score
267 return [stat['score'], stat['rawScore']]