Coverage for src/history/views.py: 0%

1import json 

2from typing import TYPE_CHECKING, Any, Callable 

3from urllib.parse import urlencode 

4 

5from django.core.paginator import Paginator 

6from django.db.models import F 

7from django.http import HttpResponse, JsonResponse 

8from django.http.response import HttpResponseBadRequest 

9from django.urls import reverse_lazy 

10from django.utils import timezone 

11from django.views import View 

12from django.views.generic import TemplateView 

13from django.views.generic.base import ContextMixin 

14from django.views.generic.detail import BaseDetailView 

15from django.views.generic.edit import DeleteView, UpdateView 

16from matching import views as matching_views 

17from ptf import views as ptf_views 

18from ptf.exceptions import ServerUnderMaintenance 

19from ptf.models.classes.article import Article 

20from ptf.models.classes.collection import Collection 

21from requests.exceptions import Timeout 

22 

23from history.model_data import HistoryEventDict, HistoryEventStatus, HistoryEventType 

24from history.models import HistoryEvent, HistoryEventQuerySet 

25from history.utils import ( 

26 get_history_error_warning_counts, 

27 get_history_last_event_by, 

28 insert_history_event, 

29) 

30 

31if TYPE_CHECKING: 

32 from django import http 

33 from ptf.models.classes.resource import Resource 

34 

35 

36def manage_exceptions( 

37 event: HistoryEventDict, 

38 exception: BaseException, 

39): 

40 if type(exception).__name__ == "ServerUnderMaintenance": 

41 message = " - ".join([str(exception), "Please try again later"]) 

42 else: 

43 message = " ".join([str(exception)]) 

44 

45 event["message"] = message 

46 

47 insert_history_event(event) 

48 

49 

50class HistoryEventUpdateView(UpdateView): 

51 model = HistoryEvent 

52 fields = ["status", "message"] 

53 success_url = reverse_lazy("history") 

54 

55 

56def matching_decorator(func: Callable[[Any, str], str], is_article): 

57 def inner(*args, **kwargs): 

58 article: Article 

59 

60 if is_article: 

61 article = args[0] 

62 else: 

63 article = args[0].resource.cast() 

64 

65 event: HistoryEventDict = { 

66 "type": "matching", 

67 "pid": article.pid, 

68 "col": article.get_top_collection(), 

69 "status": HistoryEventStatus.OK, 

70 } 

71 # Merge matching ids (can be zbl, numdam...) 

72 try: 

73 id_value = func(*args, **kwargs) 

74 

75 insert_history_event(event) 

76 return id_value 

77 

78 except Timeout as exception: 

79 """ 

80 Exception caused by the requests module: store it as a warning 

81 """ 

82 event["status"] = HistoryEventStatus.WARNING 

83 manage_exceptions(event, exception) 

84 raise exception 

85 

86 except ServerUnderMaintenance as exception: 

87 event["status"] = HistoryEventStatus.ERROR 

88 event["type"] = "deploy" 

89 manage_exceptions(event, exception) 

90 raise exception 

91 

92 except Exception as exception: 

93 event["status"] = HistoryEventStatus.ERROR 

94 manage_exceptions(event, exception) 

95 raise exception 

96 

97 return inner 

98 

99 

100def execute_and_record_func( 

101 type: HistoryEventType, 

102 pid, 

103 colid, 

104 func, 

105 message="", 

106 record_error_only=False, 

107 user=None, 

108 type_error=None, 

109 *func_args, 

110 **func_kwargs, 

111): 

112 status = 200 

113 func_result = None 

114 collection = None 

115 if colid != "ALL": 

116 collection = Collection.objects.get(pid=colid) 

117 event: HistoryEventDict = { 

118 "type": type, 

119 "pid": pid, 

120 "col": collection, 

121 "status": HistoryEventStatus.OK, 

122 "message": message, 

123 } 

124 

125 try: 

126 func_result = func(*func_args, **func_kwargs) 

127 

128 if type_error: 

129 event["type_error"] = type_error 

130 

131 if not record_error_only: 

132 insert_history_event(event) 

133 except Timeout as exception: 

134 """ 

135 Exception caused by the requests module: store it as a warning 

136 """ 

137 event["status"] = HistoryEventStatus.WARNING 

138 

139 event["message"] = message 

140 

141 manage_exceptions(event, exception) 

142 raise exception 

143 except Exception as exception: 

144 event["status"] = HistoryEventStatus.ERROR 

145 

146 event["message"] = message 

147 manage_exceptions(event, exception) 

148 raise exception 

149 return func_result, status, message 

150 

151 

152def edit_decorator(func): 

153 def inner(self, action, *args, **kwargs): 

154 resource_obj: Resource = self.resource.cast() 

155 pid = resource_obj.pid 

156 colid = resource_obj.get_top_collection().pid 

157 

158 message = "" 

159 if hasattr(self, "obj"): 

160 # Edit 1 item (ExtId or BibItemId) 

161 obj = self.obj 

162 if hasattr(self, "parent"): 

163 parent = self.parent 

164 if parent: 

165 message += "[" + str(parent.sequence) + "] " 

166 list_ = obj.id_type.split("-") 

167 id_type = obj.id_type if len(list_) == 0 else list_[0] 

168 message += id_type + ":" + obj.id_value + " " + action 

169 else: 

170 message += "All " + action 

171 

172 args = (self, action) + args 

173 

174 execute_and_record_func( 

175 "edit", pid, colid, func, message, False, None, None, *args, **kwargs 

176 ) 

177 

178 return inner 

179 

180 

181ptf_views.UpdateExtIdView.update_obj = edit_decorator(ptf_views.UpdateExtIdView.update_obj) 

182matching_views.UpdateMatchingView.update_obj = edit_decorator( 

183 matching_views.UpdateMatchingView.update_obj 

184) 

185 

186 

187def getLastHistoryImport(pid): 

188 data = get_history_last_event_by(type="import", pid=pid) 

189 return data 

190 

191 

192class HistoryContextMixin(ContextMixin): 

193 def get_context_data(self, **kwargs): 

194 context = super().get_context_data(**kwargs) 

195 error_count, warning_count = get_history_error_warning_counts() 

196 context["warning_count"] = warning_count 

197 context["error_count"] = error_count 

198 

199 # if isinstance(last_clockss_event, datetime): 

200 # now = timezone.now() 

201 # td = now - last_clockss_event['created_on'] 

202 # context['last_clockss_event'] = td.days 

203 return context 

204 

205 

206class HistoryView(TemplateView, HistoryContextMixin): 

207 template_name = "history.html" 

208 accepted_params = ["type", "col", "status", "month"] 

209 

210 def get_context_data(self, **kwargs): 

211 context = super().get_context_data(**kwargs) 

212 

213 filters = {} 

214 

215 urlparams = {k: v[0] if isinstance(v, list) else v for k, v in self.request.GET.items()} 

216 if "page" in urlparams: 

217 del urlparams["page"] 

218 if "search" in urlparams: 

219 del urlparams["search"] 

220 

221 for filter in self.accepted_params: 

222 value = self.request.GET.get(filter, None) 

223 if value: 

224 filters[filter] = value 

225 

226 # Get current URL params without current filter 

227 _params = {**urlparams} 

228 if filter in _params: 

229 del _params[filter] 

230 

231 context[filter + "_link"] = "?" + urlencode(_params) 

232 

233 # filter_by_month = False # Ignore months for Now 

234 # filters.setdefault("status", Q(status="ERROR") | Q(status="WARNING")) 

235 

236 # if filter_by_month: 

237 # today = datetime.datetime.today() 

238 # filters["created_on__year"] = today.year 

239 # filters["created_on__month"] = today.month 

240 if "col" in filters: 

241 filters["col__pid"] = filters["col"] 

242 del filters["col"] 

243 qs = HistoryEvent.objects.filter(**filters).order_by("-created_on") 

244 

245 paginator = Paginator(qs, 50) 

246 try: 

247 page = int(self.request.GET.get("page", 1)) 

248 except ValueError: 

249 page = 1 

250 

251 if page > paginator.num_pages: 

252 page = paginator.num_pages 

253 

254 context["page_obj"] = paginator.get_page(page) 

255 context["paginator"] = { 

256 "page_range": paginator.get_elided_page_range(number=page, on_each_side=2, on_ends=1), 

257 } 

258 

259 context["now"] = timezone.now() 

260 

261 if "col__pid" in filters: 

262 del filters["col__pid"] 

263 context["collections"] = ( 

264 HistoryEvent.objects.filter(col__isnull=False, **filters) 

265 .values(colid=F("col__pid"), name=F("col__title_html")) 

266 .distinct("col__title_html") 

267 .order_by("col__title_html") 

268 ) 

269 return context 

270 

271 

272class HistoryClearView(DeleteView): 

273 model = HistoryEvent 

274 

275 template_name_suffix = "_confirm_clear" 

276 success_url = reverse_lazy("history") 

277 

278 def get_object(self, queryset=None): 

279 # for deleting events excluding latest 

280 _queryset: HistoryEventQuerySet = self.get_queryset() 

281 return _queryset.get_stale_events() 

282 # return self.get_queryset() 

283 

284 

285class HistoryEventDeleteView(DeleteView): 

286 model = HistoryEvent 

287 success_url = reverse_lazy("history") 

288 

289 

290class HistoryAPIView(BaseDetailView): 

291 model = HistoryEvent 

292 

293 http_method_names = ["get"] 

294 

295 def get(self, request: "http.HttpRequest", pk: int, datatype: str): 

296 object = self.get_object() 

297 if datatype == "message": 

298 return HttpResponse(object.message) 

299 if datatype == "table": 

300 data = [] 

301 for a in object.children.all(): 

302 entry = { 

303 "type": a.type, 

304 "status": a.status, 

305 "status_message": a.status_message, 

306 "message": a.message, 

307 "score": a.score, 

308 "url": a.url, 

309 } 

310 if a.resource: 

311 entry["resource_pid"] = a.resource.pid 

312 data.append(entry) 

313 return JsonResponse( 

314 { 

315 "data": data, 

316 "headers": [ 

317 "resource_pid", 

318 "type", 

319 "status", 

320 "status_message", 

321 "message", 

322 "score", 

323 "url", 

324 ], 

325 "source": object.source, 

326 "pid": object.pid, 

327 "col": object.col.pid if object.col else None, 

328 "date": object.created_on, 

329 } 

330 ) 

331 return HttpResponseBadRequest() 

332 

333 

334class HistoryEventInsert(View): 

335 def post(self, request: "http.HttpRequest"): 

336 # Todo : Sanitarization ? 

337 event = json.loads(request.body.decode("utf-8")) 

338 if "colid" in event: 

339 colid = event["colid"] 

340 del event["colid"] 

341 collection = Collection.objects.get(pid=colid) 

342 event["col"] = collection 

343 insert_history_event(event) 

344 response = HttpResponse() 

345 response.status_code = 201 

346 return response