Coverage for model/problem.py: 85%
240 statements
« prev ^ index » next coverage.py v7.9.2, created at 2026-07-24 16:42 +0000
1import asyncio
2import json
3import hashlib
4import statistics
5from dataclasses import asdict
6from typing import Optional
7from fastapi import APIRouter, Depends, Request, UploadFile, File
8from fastapi.responses import StreamingResponse
9from urllib import parse
10from zipfile import BadZipFile
11from mongo import *
12from mongo import engine
13from mongo import sandbox
14from mongo.utils import drop_none
15from mongo.problem import *
16from .auth import identity_verify, login_required
17from .utils import *
18from .schemas import (
19 ViewProblemListQuery,
20 ProblemBody,
21 InitiateTestCaseUploadBody,
22 CompleteTestCaseUploadBody,
23 GetTestdataQuery,
24 CloneProblemBody,
25 PublishProblemBody,
26)
28__all__ = ['problem_router']
30problem_router = APIRouter()
33def permission_error_response():
34 return HTTPError('Not enough permission', 403)
37def online_error_response():
38 return HTTPError('Problem is unavailable', 403)
41@problem_router.get('')
42def view_problem_list(
43 query: ViewProblemListQuery = Depends(),
44 user=Depends(login_required),
45):
46 offset = query.offset
47 count = query.count
48 tags = query.tags
49 problem_id = query.problem_id
50 name = query.name
51 course = query.course
52 try:
53 if offset is not None:
54 offset = int(offset)
55 if count is not None:
56 count = int(count)
57 except (TypeError, ValueError):
58 return HTTPError('offset and count must be integer!', 400)
59 problem_id, name, tags, course = (parse.unquote(p or '') or None
60 for p in (problem_id, name, tags,
61 course))
62 try:
63 ks = {
64 'user': user,
65 'offset': offset,
66 'count': count,
67 'tags': tags and tags.split(','),
68 'problem_id': problem_id,
69 'name': name,
70 'course': course,
71 }
72 ks = {k: v for k, v in ks.items() if v is not None}
73 data = Problem.get_problem_list(**ks)
74 except IndexError:
75 return HTTPError('invalid offset', 400)
76 data = [{
77 'problemId': p.problem_id,
78 'problemName': p.problem_name,
79 'status': p.problem_status,
80 'ACUser': p.ac_user,
81 'submitter': p.submitter,
82 'tags': p.tags,
83 'type': p.problem_type,
84 'quota': p.quota,
85 'submitCount': Problem(p.problem_id).submit_count(user),
86 } for p in data]
87 return HTTPResponse('Success.', data=data)
90@problem_router.get('/manage/{problem_id}')
91def get_problem_detailed(
92 user: User = identity_verify(0, 1),
93 problem: Problem = get_doc('problem_id', Problem, int),
94):
95 if not problem.permission(user, problem.Permission.MANAGE):
96 return permission_error_response()
97 if not problem.permission(user=user, req=problem.Permission.ONLINE):
98 return online_error_response()
99 info = problem.detailed_info(
100 'courses',
101 'problemName',
102 'description',
103 'tags',
104 'testCase',
105 'ACUser',
106 'submitter',
107 'allowedLanguage',
108 'canViewStdout',
109 'quota',
110 status='problemStatus',
111 type='problemType',
112 )
113 info['submitCount'] = problem.submit_count(user)
114 return HTTPResponse('Success.', data=info)
117@problem_router.post('/manage')
118def create_problem(body: ProblemBody, user: User = identity_verify(0, 1)):
119 try:
120 pid = Problem.add(user=user, **body.model_dump())
121 except ValidationError as e:
122 return HTTPError('Invalid or missing arguments.',
123 400,
124 data=e.to_dict())
125 except DoesNotExist:
126 return HTTPError('Course not found', 404)
127 except ValueError as e:
128 return HTTPError(str(e), 400)
129 return HTTPResponse(data={'problemId': pid})
132@problem_router.delete('/manage/{problem_id}')
133def delete_problem(
134 user: User = identity_verify(0, 1),
135 problem: Problem = get_doc('problem_id', Problem, int),
136):
137 if not problem.permission(user, problem.Permission.MANAGE):
138 return permission_error_response()
139 if not problem.permission(user=user, req=problem.Permission.ONLINE):
140 return online_error_response()
141 problem.delete()
142 return HTTPResponse()
145@problem_router.put('/manage/{problem_id}')
146async def manage_problem(
147 problem_id: int,
148 request: Request,
149 user: User = identity_verify(0, 1),
150):
151 problem = await asyncio.to_thread(Problem, problem_id)
152 if not problem:
153 raise NOJException('Problem not found', 404)
154 if not await asyncio.to_thread(problem.permission, user,
155 problem.Permission.MANAGE):
156 return permission_error_response()
157 if not await asyncio.to_thread(
158 problem.permission, user=user, req=problem.Permission.ONLINE):
159 return online_error_response()
161 content_type = request.headers.get('content-type', '')
162 if content_type.startswith('application/json'):
163 body_data = await request.json()
164 try:
165 body = ProblemBody(**body_data)
166 except Exception:
167 return HTTPError('Invalid or missing arguments.', 400)
168 try:
169 await asyncio.to_thread(
170 Problem.edit_problem,
171 user=user,
172 problem_id=problem.id,
173 **drop_none(body.model_dump()),
174 )
175 except ValidationError as ve:
176 return HTTPError('Invalid or missing arguments.',
177 400,
178 data=ve.to_dict())
179 except engine.DoesNotExist:
180 return HTTPError('Course not found.', 404)
181 return HTTPResponse()
182 elif content_type.startswith('multipart/form-data'):
183 from io import BytesIO
184 form = await request.form()
185 case_upload = form.get('case')
186 if case_upload is None or not hasattr(case_upload, 'read'):
187 return HTTPError('missing or invalid form field: case', 400)
188 case = BytesIO(await case_upload.read())
189 try:
190 await asyncio.to_thread(problem.update_test_case, case)
191 except engine.DoesNotExist as e:
192 return HTTPError(str(e), 404)
193 except (ValueError, BadZipFile) as e:
194 return HTTPError(str(e), 400)
195 except BadTestCase as e:
196 return HTTPError(str(e), 400)
197 return HTTPResponse('Success.')
198 else:
199 return HTTPError('Unknown content type',
200 400,
201 data={'contentType': content_type})
204@problem_router.post('/{problem_id}/initiate-test-case-upload')
205def initiate_test_case_upload(
206 body: InitiateTestCaseUploadBody,
207 user: User = identity_verify(0, 1),
208 problem: Problem = get_doc('problem_id', Problem, int),
209):
210 if not problem.permission(user, problem.Permission.MANAGE):
211 return permission_error_response()
212 if not problem.permission(user=user, req=problem.Permission.ONLINE):
213 return online_error_response()
214 upload_info = problem.generate_urls_for_uploading_test_case(
215 body.length, body.part_size)
216 return HTTPResponse(data=asdict(upload_info))
219@problem_router.post('/{problem_id}/complete-test-case-upload')
220def complete_test_case_upload(
221 body: CompleteTestCaseUploadBody,
222 user: User = identity_verify(0, 1),
223 problem: Problem = get_doc('problem_id', Problem, int),
224):
225 if not problem.permission(user, problem.Permission.MANAGE):
226 return permission_error_response()
227 if not problem.permission(user=user, req=problem.Permission.ONLINE):
228 return online_error_response()
229 from minio.datatypes import Part
230 parts = [
231 Part(part_number=part['PartNumber'], etag=part['ETag'])
232 for part in body.parts
233 ]
234 try:
235 problem.complete_test_case_upload(body.upload_id, parts)
236 except BadTestCase as e:
237 return HTTPError(str(e), 400)
238 return HTTPResponse(status_code=201)
241@problem_router.get('/{problem_id}/test-case')
242@problem_router.get('/{problem_id}/testcase')
243def get_test_case(
244 problem_id: int,
245 user=Depends(login_required),
246 problem: Problem = get_doc('problem_id', Problem, int),
247):
248 if not problem.permission(user, problem.Permission.MANAGE):
249 return permission_error_response()
250 if not problem.permission(user=user, req=problem.Permission.ONLINE):
251 return online_error_response()
252 return StreamingResponse(
253 problem.get_test_case(),
254 media_type='application/zip',
255 headers={
256 'Content-Disposition':
257 f'attachment; filename="testdata-{problem.id}.zip"'
258 },
259 )
262@problem_router.get('/{problem_id}/testdata')
263def get_testdata(
264 query: GetTestdataQuery = Depends(),
265 problem: Problem = get_doc('problem_id', Problem, int),
266):
267 if sandbox.find_by_token(query.token) is None:
268 return HTTPError('Invalid sandbox token', 401)
269 return StreamingResponse(
270 problem.get_test_case(),
271 media_type='application/zip',
272 headers={
273 'Content-Disposition':
274 f'attachment; filename="testdata-{problem.id}.zip"'
275 },
276 )
279@problem_router.get('/{problem_id}/checksum')
280def get_checksum(problem_id: int, query: GetTestdataQuery = Depends()):
281 if sandbox.find_by_token(query.token) is None:
282 return HTTPError('Invalid sandbox token', 401)
283 problem = Problem(problem_id)
284 if not problem:
285 return HTTPError(f'{problem} not found', 404)
286 meta = json.dumps({
287 'tasks':
288 [json.loads(task.to_json()) for task in problem.test_case.tasks]
289 }).encode()
290 content = problem.get_test_case().read() + meta
291 digest = hashlib.md5(content).hexdigest()
292 return HTTPResponse(data=digest)
295@problem_router.get('/{problem_id}/meta')
296def get_meta(problem_id: int, query: GetTestdataQuery = Depends()):
297 if sandbox.find_by_token(query.token) is None:
298 return HTTPError('Invalid sandbox token', 401)
299 problem = Problem(problem_id)
300 if not problem:
301 return HTTPError(f'{problem} not found', 404)
302 meta = {
303 'tasks':
304 [json.loads(task.to_json()) for task in problem.test_case.tasks]
305 }
306 return HTTPResponse(data=meta)
309@problem_router.get('/{problem_id}/high-score')
310def high_score(
311 user=Depends(login_required),
312 problem: Problem = get_doc('problem_id', Problem, int),
313):
314 return HTTPResponse(data={'score': problem.get_high_score(user=user)})
317@problem_router.post('/clone')
318@problem_router.post('/copy')
319def clone_problem(body: CloneProblemBody, user: User = identity_verify(0, 1)):
320 try:
321 problem = Problem(body.problem_id)
322 if not problem:
323 return HTTPError('Problem not found', 404)
324 except engine.DoesNotExist as e:
325 return HTTPError(str(e), 404)
326 if not problem.permission(user, problem.Permission.VIEW):
327 return HTTPError('Problem can not view.', 403)
328 override = drop_none({'status': body.status})
329 new_problem_id = problem.copy_to(user=user, target=body.target, **override)
330 return HTTPResponse('Success.', data={'problemId': new_problem_id})
333@problem_router.post('/publish')
334def publish_problem(body: PublishProblemBody,
335 user: User = identity_verify(0, 1)):
336 try:
337 problem = Problem(body.problem_id)
338 if not problem:
339 return HTTPError('Problem not found', 404)
340 except engine.DoesNotExist as e:
341 return HTTPError(str(e), 404)
342 if user.role == 1 and problem.owner != user.username:
343 return HTTPError('Not the owner.', 403)
344 Problem.release_problem(problem.problem_id)
345 return HTTPResponse('Success.')
348@problem_router.get('/{problem_id}/stats')
349def problem_stats(
350 user=Depends(login_required),
351 problem: Problem = get_doc('problem_id', Problem, int),
352):
353 if not problem.permission(user, problem.Permission.VIEW):
354 return permission_error_response()
355 if not problem.permission(user=user, req=problem.Permission.ONLINE):
356 return online_error_response()
357 ret = {}
358 students = []
359 for course in problem.courses:
360 students += [User(name) for name in course.student_nicknames.keys()]
361 students_high_scores = [problem.get_high_score(user=u) for u in students]
362 ret['acUserRatio'] = [problem.get_ac_user_count(), len(students)]
363 ret['triedUserCount'] = problem.get_tried_user_count()
364 ret['average'] = None if len(students) == 0 else statistics.mean(
365 students_high_scores)
366 ret['std'] = None if len(students) <= 1 else statistics.pstdev(
367 students_high_scores)
368 ret['scoreDistribution'] = students_high_scores
369 ret['statusCount'] = problem.get_submission_status()
370 params = {
371 'user': user,
372 'offset': 0,
373 'count': 10,
374 'problem': problem.id,
375 'status': 0
376 }
377 ret['top10RunTime'] = [
378 s.to_dict() for s in Submission.filter(**params, sort_by='runTime')
379 ]
380 ret['top10MemoryUsage'] = [
381 s.to_dict() for s in Submission.filter(**params, sort_by='memoryUsage')
382 ]
383 return HTTPResponse('Success.', data=ret)
386@problem_router.get('/{problem_id}')
387@problem_router.get('/view/{problem_id}')
388def view_problem(
389 problem_id: int,
390 ip: str = Depends(get_ip),
391 user=Depends(login_required),
392 problem: Problem = get_doc('problem_id', Problem, int),
393):
394 if not problem.permission(user=user, req=problem.Permission.VIEW):
395 return permission_error_response()
396 if not problem.permission(user=user, req=problem.Permission.ONLINE):
397 return online_error_response()
398 if not problem.is_valid_ip(ip):
399 return HTTPError('Invalid IP address.', 403)
400 data = problem.detailed_info(
401 'problemName',
402 'description',
403 'owner',
404 'tags',
405 'allowedLanguage',
406 'courses',
407 'quota',
408 defaultCode='defaultCode',
409 status='problemStatus',
410 type='problemType',
411 testCase='testCase__tasks',
412 )
413 if problem.obj.problem_type == 1:
414 data['fillInTemplate'] = problem.obj.test_case.fill_in_template
415 data.update({
416 'submitCount': problem.submit_count(user),
417 'highScore': problem.get_high_score(user=user),
418 })
419 return HTTPResponse('Problem can view.', data=data)
422@problem_router.post('/{problem_id}/migrate-test-case')
423def problem_migrate_test_case(
424 user: User = identity_verify(0),
425 problem: Problem = get_doc('problem_id', Problem, int),
426):
427 if not problem.permission(user, problem.Permission.MANAGE):
428 return permission_error_response()
429 problem.migrate_gridfs_to_minio()
430 return HTTPResponse('Success.')