Coverage for model/copycat.py: 100%
116 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-03-14 03:01 +0000
1from typing import Dict
2from flask import Blueprint, request, current_app
3from mongo import *
4from mongo.utils import *
5from .utils import *
6from .auth import *
8import mosspy
9import threading
10import logging
11import requests
12import re
14__all__ = ['copycat_api']
16copycat_api = Blueprint('copycat_api', __name__)
19def is_valid_url(url):
20 import re
21 regex = re.compile(
22 r'^https?://' # http:// or https://
23 r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' # domain...
24 r'localhost|' # localhost...
25 r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' # ...or ip
26 r'(?::\d+)?' # optional port
27 r'(?:/?|[/?]\S+)$',
28 re.IGNORECASE)
29 return url is not None and regex.search(url)
32def get_report_task(user, problem_id, student_dict: Dict):
33 # select all ac code
34 submissions = Submission.filter(
35 user=user,
36 offset=0,
37 count=-1,
38 status=0,
39 problem=problem_id,
40 )
42 last_cc_submission = {}
43 last_python_submission = {}
44 for submission in submissions:
45 s = Submission(submission.id)
46 if s.user.username in student_dict:
47 if s.language in [0, 1] \
48 and s.user.username not in last_cc_submission:
49 last_cc_submission[
50 submission.user.username] = s.main_code_path()
51 elif s.language in [2] \
52 and s.user.username not in last_python_submission:
53 last_python_submission[
54 submission.user.username] = s.main_code_path()
56 moss_userid = 97089070
58 # get logger
59 logger = logging.getLogger('guincorn.error')
61 # Get problem object
62 problem = Problem(problem_id)
64 cpp_report_url = ''
65 python_report_url = ''
66 # check for c or cpp code
67 if problem.allowed_language != 4:
68 m1 = mosspy.Moss(moss_userid, "cc")
70 for user, code_path in last_cc_submission.items():
71 logger.info(f'send {user} {code_path}')
72 m1.addFile(code_path)
74 response = m1.send()
75 if is_valid_url(response):
76 cpp_report_url = response
77 else:
78 logger.info(f"[copycat] {response}")
79 cpp_report_url = ''
81 # check for python code
82 if problem.allowed_language >= 4:
83 m2 = mosspy.Moss(moss_userid, "python")
85 for user, code_path in last_python_submission.items():
86 logger.info(f'send {user} {code_path}')
87 m2.addFile(code_path)
89 response = m2.send()
90 if is_valid_url(response):
91 python_report_url = response
92 else:
93 logger.info(f"[copycat] {response}")
94 python_report_url = ''
96 # download report from moss
97 if cpp_report_url != '':
98 mosspy.download_report(
99 cpp_report_url,
100 f"submissions_report/{problem_id}/cpp_report/",
101 connections=8,
102 log_level=10,
103 )
104 if python_report_url != '':
105 mosspy.download_report(
106 python_report_url,
107 f"submissions_report/{problem_id}/python_report/",
108 connections=8,
109 log_level=10,
110 )
112 # insert report url into DB & update status
113 problem.obj.update(
114 cpp_report_url=cpp_report_url,
115 python_report_url=python_report_url,
116 moss_status=2,
117 )
120def get_report_by_url(url: str):
121 try:
122 response = requests.get(url)
123 return response.text
124 except (requests.exceptions.MissingSchema,
125 requests.exceptions.InvalidSchema):
126 return 'No report.'
129@copycat_api.route('/', methods=['GET'])
130@login_required
131@Request.args('course', 'problem_id')
132def get_report(user, course, problem_id):
133 if not (problem_id and course):
134 return HTTPError(
135 'missing arguments! (In HTTP GET argument format)',
136 400,
137 data={
138 'need': ['course', 'problemId'],
139 },
140 )
141 # some privilege or exist check
142 try:
143 problem = Problem(int(problem_id))
144 except ValueError:
145 return HTTPError('problemId must be integer', 400)
147 course = Course(course)
148 if not course.permission(user, Course.Permission.GRADE):
149 return HTTPError('Forbidden.', 403)
150 if not problem:
151 return HTTPError('Problem not exist.', 404)
152 if not course:
153 return HTTPError('Course not found.', 404)
155 cpp_report_url = problem.cpp_report_url
156 python_report_url = problem.python_report_url
158 if problem.moss_status == 0:
159 return HTTPError(
160 "No report found. Please make a post request to copycat api to generate a report",
161 404,
162 data={},
163 )
164 elif problem.moss_status == 1:
165 return HTTPResponse("Report generating...", data={})
166 else:
167 cpp_report = get_report_by_url(cpp_report_url)
168 python_report = get_report_by_url(python_report_url)
169 return HTTPResponse(
170 "Success.",
171 data={
172 "cpp_report": cpp_report,
173 "python_report": python_report
174 },
175 )
178@copycat_api.route('/', methods=['POST'])
179@login_required
180@Request.json('course', 'problem_id', 'student_nicknames')
181def detect(user, course, problem_id, student_nicknames):
182 if not (problem_id and course and type(student_nicknames) is dict):
183 return HTTPError(
184 'missing arguments! (In Json format)',
185 400,
186 data={
187 'need': ['course', 'problemId', 'studentNicknames'],
188 },
189 )
191 course = Course(course)
192 problem = Problem(problem_id)
194 # Check if student is in course
195 student_dict = {}
196 for student, nickname in student_nicknames.items():
197 if not User(student):
198 return HTTPResponse(f'User: {student} not found.', 404)
199 student_dict[student] = nickname
200 # Check student_dict
201 if not student_dict:
202 return HTTPResponse('Empty student list.', 404)
203 # some privilege or exist check
204 if not course.permission(user, Course.Permission.GRADE):
205 return HTTPError('Forbidden.', 403)
206 if not problem:
207 return HTTPError('Problem not exist.', 404)
208 if not course:
209 return HTTPError('Course not found.', 404)
211 problem = Problem(problem_id)
212 problem.update(
213 cpp_report_url="",
214 python_report_url="",
215 moss_status=1,
216 )
217 if not current_app.config['TESTING']:
218 threading.Thread(
219 target=get_report_task,
220 args=(
221 user,
222 problem_id,
223 student_dict,
224 ),
225 ).start()
227 # return Success
228 return HTTPResponse('Success.')