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 import transaction 

7from django.db.models import F 

8from django.http import HttpResponse, HttpResponseRedirect, JsonResponse 

9from django.http.response import HttpResponseBadRequest 

10from django.urls import reverse, reverse_lazy 

11from django.utils import timezone 

12from django.views import View 

13from django.views.generic import TemplateView 

14from django.views.generic.base import ContextMixin 

15from django.views.generic.detail import BaseDetailView 

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

17from matching_back import views as matching_views 

18from opentelemetry import trace 

19from ptf import views as ptf_views 

20from ptf.exceptions import ServerUnderMaintenance 

21from ptf.models.classes.article import Article 

22from ptf.models.classes.collection import Collection 

23from requests.exceptions import Timeout 

24 

25from history.model_data import HistoryEventDict, HistoryEventStatus, HistoryEventType 

26from history.models import HistoryEvent 

27from history.utils import ( 

28 get_history_error_warning_counts, 

29 get_history_last_event_by, 

30 insert_history_event, 

31) 

32 

33if TYPE_CHECKING: 

34 from django import http 

35 from ptf.models.classes.resource import Resource 

36 

37tracer = trace.get_tracer(__name__) 

38 

39 

40def manage_exceptions( 

41 event: HistoryEventDict, 

42 exception: BaseException, 

43): 

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

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

46 else: 

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

48 

49 event["message"] = message 

50 

51 insert_history_event(event) 

52 

53 

54class HistoryEventUpdateView(UpdateView): 

55 model = HistoryEvent 

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

57 success_url = reverse_lazy("history") 

58 

59 

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

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

62 article: Article 

63 

64 if is_article: 

65 article = args[0] 

66 else: 

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

68 

69 event: HistoryEventDict = { 

70 "type": "matching", 

71 "pid": article.pid, 

72 "col": article.get_top_collection(), 

73 "status": HistoryEventStatus.OK, 

74 } 

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

76 try: 

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

78 

79 insert_history_event(event) 

80 return id_value 

81 

82 except Timeout as exception: 

83 """ 

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

85 """ 

86 event["status"] = HistoryEventStatus.WARNING 

87 manage_exceptions(event, exception) 

88 raise exception 

89 

90 except ServerUnderMaintenance as exception: 

91 event["status"] = HistoryEventStatus.ERROR 

92 event["type"] = "deploy" 

93 manage_exceptions(event, exception) 

94 raise exception 

95 

96 except Exception as exception: 

97 event["status"] = HistoryEventStatus.ERROR 

98 manage_exceptions(event, exception) 

99 raise exception 

100 

101 return inner 

102 

103 

104def execute_and_record_func( 

105 type: HistoryEventType, 

106 pid, 

107 colid, 

108 func, 

109 message="", 

110 record_error_only=False, 

111 user=None, 

112 type_error=None, 

113 *func_args, 

114 **func_kwargs, 

115): 

116 status = 200 

117 func_result = None 

118 collection = None 

119 if colid not in ["ALL", "numdam"]: 

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

121 event: HistoryEventDict = { 

122 "type": type, 

123 "pid": pid, 

124 "col": collection, 

125 "status": HistoryEventStatus.OK, 

126 "message": message, 

127 } 

128 

129 try: 

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

131 

132 if type_error: 

133 event["type_error"] = type_error 

134 

135 if not record_error_only: 

136 insert_history_event(event) 

137 except Timeout as exception: 

138 """ 

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

140 """ 

141 event["status"] = HistoryEventStatus.WARNING 

142 

143 event["message"] = message 

144 

145 manage_exceptions(event, exception) 

146 raise exception 

147 except Exception as exception: 

148 event["status"] = HistoryEventStatus.ERROR 

149 

150 event["message"] = message 

151 manage_exceptions(event, exception) 

152 raise exception 

153 return func_result, status, message 

154 

155 

156def edit_decorator(func): 

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

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

159 pid = resource_obj.pid 

160 colid = resource_obj.get_top_collection().pid 

161 

162 message = "" 

163 if hasattr(self, "obj"): 

164 # Edit 1 item (ExtId or BibItemId) 

165 obj = self.obj 

166 if hasattr(self, "parent"): 

167 parent = self.parent 

168 if parent: 

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

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

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

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

173 else: 

174 message += "All " + action 

175 

176 args = (self, action) + args 

177 

178 execute_and_record_func( 

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

180 ) 

181 

182 return inner 

183 

184 

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

186matching_views.UpdateMatchingView.update_obj = edit_decorator( 

187 matching_views.UpdateMatchingView.update_obj 

188) 

189 

190 

191def getLastHistoryImport(pid): 

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

193 return data 

194 

195 

196class HistoryContextMixin(ContextMixin): 

197 def get_context_data(self, **kwargs): 

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

199 error_count, warning_count = get_history_error_warning_counts() 

200 context["warning_count"] = warning_count 

201 context["error_count"] = error_count 

202 

203 # if isinstance(last_clockss_event, datetime): 

204 # now = timezone.now() 

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

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

207 return context 

208 

209 

210class HistoryView(TemplateView, HistoryContextMixin): 

211 template_name = "history.html" 

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

213 

214 def get_context_data(self, **kwargs): 

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

216 

217 filters = {} 

218 

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

220 if "page" in urlparams: 

221 del urlparams["page"] 

222 if "search" in urlparams: 

223 del urlparams["search"] 

224 

225 for filter in self.accepted_params: 

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

227 if value: 

228 filters[filter] = value 

229 

230 # Get current URL params without current filter 

231 _params = {**urlparams} 

232 if filter in _params: 

233 del _params[filter] 

234 

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

236 

237 # filter_by_month = False # Ignore months for Now 

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

239 

240 # if filter_by_month: 

241 # today = datetime.datetime.today() 

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

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

244 if "col" in filters: 

245 filters["col__pid"] = filters["col"] 

246 del filters["col"] 

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

248 

249 paginator = Paginator(qs, 50) 

250 try: 

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

252 except ValueError: 

253 page = 1 

254 

255 if page > paginator.num_pages: 

256 page = paginator.num_pages 

257 

258 context["element_count"] = HistoryEvent.objects.all().count() 

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

260 context["paginator"] = { 

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

262 } 

263 

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

265 

266 if "col__pid" in filters: 

267 del filters["col__pid"] 

268 context["collections"] = ( 

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

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

271 .distinct("col__title_html") 

272 .order_by("col__title_html") 

273 ) 

274 return context 

275 

276 

277class HistoryClearView(TemplateView): 

278 model = HistoryEvent 

279 http_method_names = ["post"] 

280 template_name = "blocks/history/historyevent_confirm_clear.html" 

281 success_url = reverse_lazy("history") 

282 

283 @transaction.atomic 

284 def post(self, *args): 

285 HistoryEvent.objects.all().delete() 

286 return HttpResponseRedirect(reverse("history")) 

287 

288 

289class HistoryEventDeleteView(DeleteView): 

290 model = HistoryEvent 

291 success_url = reverse_lazy("history") 

292 

293 

294class HistoryAPIView(BaseDetailView): 

295 model = HistoryEvent 

296 

297 http_method_names = ["get"] 

298 

299 @tracer.start_as_current_span("HistoryAPIView.get") 

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

301 object: HistoryEvent = self.get_object() 

302 if datatype == "message": 

303 return HttpResponse(object.message) 

304 if datatype == "table": 

305 data = [] 

306 for a in object.children.all().prefetch_related("resource"): 

307 entry = { 

308 "type": a.type, 

309 "status": a.status, 

310 "status_message": a.status_message, 

311 "message": a.message, 

312 "score": a.score, 

313 "url": a.url, 

314 } 

315 if a.resource: 

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

317 data.append(entry) 

318 return JsonResponse( 

319 { 

320 "data": data, 

321 "headers": [ 

322 "resource_pid", 

323 "type", 

324 "status", 

325 "status_message", 

326 "message", 

327 "score", 

328 "url", 

329 ], 

330 "source": object.source, 

331 "pid": object.pid, 

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

333 "date": object.created_on, 

334 } 

335 ) 

336 return HttpResponseBadRequest() 

337 

338 

339class HistoryEventInsert(View): 

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

341 # Todo : Sanitarization ? 

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

343 if "colid" in event: 

344 colid = event["colid"] 

345 del event["colid"] 

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

347 event["col"] = collection 

348 insert_history_event(event) 

349 response = HttpResponse() 

350 response.status_code = 201 

351 return response