Coverage for model/copycat.py: 100%
116 statements
« prev ^ index » next coverage.py v7.3.2, created at 2024-11-05 04:22 +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[submission.user.username] = s.main_code_path
50 elif s.language in [2] \
51 and s.user.username not in last_python_submission:
52 last_python_submission[
53 submission.user.username] = s.main_code_path
55 moss_userid = 97089070
57 # get logger
58 logger = logging.getLogger('guincorn.error')
60 # Get problem object
61 problem = Problem(problem_id)
63 cpp_report_url = ''
64 python_report_url = ''
65 # check for c or cpp code
66 if problem.allowed_language != 4:
67 m1 = mosspy.Moss(moss_userid, "cc")
69 for user, code_path in last_cc_submission.items():
70 logger.info(f'send {user} {code_path}')
71 m1.addFile(code_path)
73 response = m1.send()
74 if is_valid_url(response):
75 cpp_report_url = response
76 else:
77 logger.info(f"[copycat] {response}")
78 cpp_report_url = ''
80 # check for python code
81 if problem.allowed_language >= 4:
82 m2 = mosspy.Moss(moss_userid, "python")
84 for user, code_path in last_python_submission.items():
85 logger.info(f'send {user} {code_path}')
86 m2.addFile(code_path)
88 response = m2.send()
89 if is_valid_url(response):
90 python_report_url = response
91 else:
92 logger.info(f"[copycat] {response}")
93 python_report_url = ''
95 # download report from moss
96 if cpp_report_url != '':
97 mosspy.download_report(
98 cpp_report_url,
99 f"submissions_report/{problem_id}/cpp_report/",
100 connections=8,
101 log_level=10,
102 )
103 if python_report_url != '':
104 mosspy.download_report(
105 python_report_url,
106 f"submissions_report/{problem_id}/python_report/",
107 connections=8,
108 log_level=10,
109 )
111 # insert report url into DB & update status
112 problem.obj.update(
113 cpp_report_url=cpp_report_url,
114 python_report_url=python_report_url,
115 moss_status=2,
116 )
119def get_report_by_url(url: str):
120 try:
121 response = requests.get(url)
122 return response.text
123 except (requests.exceptions.MissingSchema,
124 requests.exceptions.InvalidSchema):
125 return 'No report.'
128@copycat_api.route('/', methods=['GET'])
129@login_required
130@Request.args('course', 'problem_id')
131def get_report(user, course, problem_id):
132 if not (problem_id and course):
133 return HTTPError(
134 'missing arguments! (In HTTP GET argument format)',
135 400,
136 data={
137 'need': ['course', 'problemId'],
138 },
139 )
140 # some privilege or exist check
141 try:
142 problem = Problem(int(problem_id))
143 except ValueError:
144 return HTTPError('problemId must be integer', 400)
146 course = Course(course)
147 if not course.permission(user, Course.Permission.GRADE):
148 return HTTPError('Forbidden.', 403)
149 if not problem:
150 return HTTPError('Problem not exist.', 404)
151 if not course:
152 return HTTPError('Course not found.', 404)
154 cpp_report_url = problem.cpp_report_url
155 python_report_url = problem.python_report_url
157 if problem.moss_status == 0:
158 return HTTPError(
159 "No report found. Please make a post request to copycat api to generate a report",
160 404,
161 data={},
162 )
163 elif problem.moss_status == 1:
164 return HTTPResponse("Report generating...", data={})
165 else:
166 cpp_report = get_report_by_url(cpp_report_url)
167 python_report = get_report_by_url(python_report_url)
168 return HTTPResponse(
169 "Success.",
170 data={
171 "cpp_report": cpp_report,
172 "python_report": python_report
173 },
174 )
177@copycat_api.route('/', methods=['POST'])
178@login_required
179@Request.json('course', 'problem_id', 'student_nicknames')
180def detect(user, course, problem_id, student_nicknames):
181 if not (problem_id and course and type(student_nicknames) is dict):
182 return HTTPError(
183 'missing arguments! (In Json format)',
184 400,
185 data={
186 'need': ['course', 'problemId', 'studentNicknames'],
187 },
188 )
190 course = Course(course)
191 problem = Problem(problem_id)
193 # Check if student is in course
194 student_dict = {}
195 for student, nickname in student_nicknames.items():
196 if not User(student):
197 return HTTPResponse(f'User: {student} not found.', 404)
198 student_dict[student] = nickname
199 # Check student_dict
200 if not student_dict:
201 return HTTPResponse('Empty student list.', 404)
202 # some privilege or exist check
203 if not course.permission(user, Course.Permission.GRADE):
204 return HTTPError('Forbidden.', 403)
205 if not problem:
206 return HTTPError('Problem not exist.', 404)
207 if not course:
208 return HTTPError('Course not found.', 404)
210 problem = Problem(problem_id)
211 problem.update(
212 cpp_report_url="",
213 python_report_url="",
214 moss_status=1,
215 )
216 if not current_app.config['TESTING']:
217 threading.Thread(
218 target=get_report_task,
219 args=(
220 user,
221 problem_id,
222 student_dict,
223 ),
224 ).start()
226 # return Success
227 return HTTPResponse('Success.')