Coverage for model/copycat.py: 100%

1import os 

2import re 

3import logging 

4import threading 

5from typing import Dict 

6import httpx 

7import mosspy 

8from fastapi import APIRouter, Depends 

9 

10from mongo import * 

11from mongo.utils import * 

12from .utils import * 

13from .auth import login_required 

14from .schemas import GetReportQuery, DetectBody 

15 

16__all__ = ['copycat_router'] 

17 

18copycat_router = APIRouter() 

19 

20 

21def is_valid_url(url): 

22 regex = re.compile( 

23 r'^https?://' 

24 r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|' 

25 r'localhost|' 

26 r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})' 

27 r'(?::\d+)?' 

28 r'(?:/?|[/?]\S+)$', 

29 re.IGNORECASE, 

30 ) 

31 return url is not None and regex.search(url) 

32 

33 

34def get_report_task(user, problem_id, student_dict: Dict): 

35 submissions = Submission.filter( 

36 user=user, 

37 offset=0, 

38 count=-1, 

39 status=0, 

40 problem=problem_id, 

41 ) 

42 

43 last_cc_submission = {} 

44 last_python_submission = {} 

45 for submission in submissions: 

46 s = Submission(submission.id) 

47 if s.user.username in student_dict: 

48 if s.language in [0, 1 

49 ] and s.user.username not in last_cc_submission: 

50 last_cc_submission[ 

51 submission.user.username] = s.main_code_path() 

52 elif s.language in [ 

53 2 

54 ] and s.user.username not in last_python_submission: 

55 last_python_submission[ 

56 submission.user.username] = s.main_code_path() 

57 

58 moss_userid = 97089070 

59 logger = logging.getLogger(__name__) 

60 problem = Problem(problem_id) 

61 

62 cpp_report_url = '' 

63 python_report_url = '' 

64 if problem.allowed_language != 4: 

65 m1 = mosspy.Moss(moss_userid, "cc") 

66 for user, code_path in last_cc_submission.items(): 

67 logger.info(f'send {user} {code_path}') 

68 m1.addFile(code_path) 

69 response = m1.send() 

70 if is_valid_url(response): 

71 cpp_report_url = response 

72 else: 

73 logger.info(f"[copycat] {response}") 

74 cpp_report_url = '' 

75 

76 if problem.allowed_language >= 4: 

77 m2 = mosspy.Moss(moss_userid, "python") 

78 for user, code_path in last_python_submission.items(): 

79 logger.info(f'send {user} {code_path}') 

80 m2.addFile(code_path) 

81 response = m2.send() 

82 if is_valid_url(response): 

83 python_report_url = response 

84 else: 

85 logger.info(f"[copycat] {response}") 

86 python_report_url = '' 

87 

88 if cpp_report_url != '': 

89 mosspy.download_report( 

90 cpp_report_url, 

91 f"submissions_report/{problem_id}/cpp_report/", 

92 connections=8, 

93 log_level=10, 

94 ) 

95 if python_report_url != '': 

96 mosspy.download_report( 

97 python_report_url, 

98 f"submissions_report/{problem_id}/python_report/", 

99 connections=8, 

100 log_level=10, 

101 ) 

102 

103 problem.obj.update( 

104 cpp_report_url=cpp_report_url, 

105 python_report_url=python_report_url, 

106 moss_status=2, 

107 ) 

108 

109 

110def get_report_by_url(url: str): 

111 try: 

112 response = httpx.get(url) 

113 return response.text 

114 except httpx.InvalidURL: 

115 return 'No report.' 

116 

117 

118@copycat_router.get('') 

119def get_report(query: GetReportQuery = Depends(), 

120 user=Depends(login_required)): 

121 course = query.course 

122 problem_id = query.problem_id 

123 if not (problem_id and course): 

124 return HTTPError( 

125 'missing arguments! (In HTTP GET argument format)', 

126 400, 

127 data={'need': ['course', 'problemId']}, 

128 ) 

129 try: 

130 problem = Problem(int(problem_id)) 

131 except ValueError: 

132 return HTTPError('problemId must be integer', 400) 

133 

134 course = Course(course) 

135 if not course.permission(user, Course.Permission.GRADE): 

136 return HTTPError('Forbidden.', 403) 

137 if not problem: 

138 return HTTPError('Problem not exist.', 404) 

139 if not course: 

140 return HTTPError('Course not found.', 404) 

141 

142 cpp_report_url = problem.cpp_report_url 

143 python_report_url = problem.python_report_url 

144 

145 if problem.moss_status == 0: 

146 return HTTPError( 

147 "No report found. Please make a post request to copycat api to generate a report", 

148 404, 

149 data={}, 

150 ) 

151 elif problem.moss_status == 1: 

152 return HTTPResponse("Report generating...", data={}) 

153 else: 

154 cpp_report = get_report_by_url(cpp_report_url) 

155 python_report = get_report_by_url(python_report_url) 

156 return HTTPResponse( 

157 "Success.", 

158 data={ 

159 "cpp_report": cpp_report, 

160 "python_report": python_report 

161 }, 

162 ) 

163 

164 

165@copycat_router.post('') 

166def detect(body: DetectBody, user=Depends(login_required)): 

167 course = body.course 

168 problem_id = body.problem_id 

169 student_nicknames = body.student_nicknames 

170 if not (problem_id and course and type(student_nicknames) is dict): 

171 return HTTPError( 

172 'missing arguments! (In Json format)', 

173 400, 

174 data={'need': ['course', 'problemId', 'studentNicknames']}, 

175 ) 

176 

177 course = Course(course) 

178 problem = Problem(problem_id) 

179 

180 student_dict = {} 

181 for student, nickname in student_nicknames.items(): 

182 if not User(student): 

183 return HTTPResponse(f'User: {student} not found.', 404) 

184 student_dict[student] = nickname 

185 if not student_dict: 

186 return HTTPResponse('Empty student list.', 404) 

187 if not course.permission(user, Course.Permission.GRADE): 

188 return HTTPError('Forbidden.', 403) 

189 if not problem: 

190 return HTTPError('Problem not exist.', 404) 

191 if not course: 

192 return HTTPError('Course not found.', 404) 

193 

194 problem = Problem(problem_id) 

195 problem.update(cpp_report_url="", python_report_url="", moss_status=1) 

196 if not is_testing(): 

197 threading.Thread( 

198 target=get_report_task, 

199 args=(user, problem_id, student_dict), 

200 ).start() 

201 

202 return HTTPResponse('Success.')