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 opentelemetry import trace 

18from ptf import views as ptf_views 

19from ptf.exceptions import ServerUnderMaintenance 

20from ptf.models.classes.article import Article 

21from ptf.models.classes.collection import Collection 

22from requests.exceptions import Timeout 

23 

24from history.model_data import HistoryEventDict, HistoryEventStatus, HistoryEventType 

25from history.models import HistoryEvent 

26from history.utils import ( 

27 get_history_error_warning_counts, 

28 get_history_last_event_by, 

29 insert_history_event, 

30) 

31 

32if TYPE_CHECKING: 

33 from django import http 

34 from ptf.models.classes.resource import Resource 

35 

36tracer = trace.get_tracer(__name__) 

37 

38 

39def manage_exceptions( 

40 event: HistoryEventDict, 

41 exception: BaseException, 

42): 

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

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

45 else: 

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

47 

48 event["message"] = message 

49 

50 insert_history_event(event) 

51 

52 

53class HistoryEventUpdateView(UpdateView): 

54 model = HistoryEvent 

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

56 success_url = reverse_lazy("history") 

57 

58 

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

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

61 article: Article 

62 

63 if is_article: 

64 article = args[0] 

65 else: 

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

67 

68 event: HistoryEventDict = { 

69 "type": "matching", 

70 "pid": article.pid, 

71 "col": article.get_top_collection(), 

72 "status": HistoryEventStatus.OK, 

73 } 

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

75 try: 

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

77 

78 insert_history_event(event) 

79 return id_value 

80 

81 except Timeout as exception: 

82 """ 

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

84 """ 

85 event["status"] = HistoryEventStatus.WARNING 

86 manage_exceptions(event, exception) 

87 raise exception 

88 

89 except ServerUnderMaintenance as exception: 

90 event["status"] = HistoryEventStatus.ERROR 

91 event["type"] = "deploy" 

92 manage_exceptions(event, exception) 

93 raise exception 

94 

95 except Exception as exception: 

96 event["status"] = HistoryEventStatus.ERROR 

97 manage_exceptions(event, exception) 

98 raise exception 

99 

100 return inner 

101 

102 

103def execute_and_record_func( 

104 type: HistoryEventType, 

105 pid, 

106 colid, 

107 func, 

108 message="", 

109 record_error_only=False, 

110 user=None, 

111 type_error=None, 

112 *func_args, 

113 **func_kwargs, 

114): 

115 status = 200 

116 func_result = None 

117 collection = None 

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

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

120 event: HistoryEventDict = { 

121 "type": type, 

122 "pid": pid, 

123 "col": collection, 

124 "status": HistoryEventStatus.OK, 

125 "message": message, 

126 } 

127 

128 try: 

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

130 

131 if type_error: 

132 event["type_error"] = type_error 

133 

134 if not record_error_only: 

135 insert_history_event(event) 

136 except Timeout as exception: 

137 """ 

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

139 """ 

140 event["status"] = HistoryEventStatus.WARNING 

141 

142 event["message"] = message 

143 

144 manage_exceptions(event, exception) 

145 raise exception 

146 except Exception as exception: 

147 event["status"] = HistoryEventStatus.ERROR 

148 

149 event["message"] = message 

150 manage_exceptions(event, exception) 

151 raise exception 

152 return func_result, status, message 

153 

154 

155def edit_decorator(func): 

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

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

158 pid = resource_obj.pid 

159 colid = resource_obj.get_top_collection().pid 

160 

161 message = "" 

162 if hasattr(self, "obj"): 

163 # Edit 1 item (ExtId or BibItemId) 

164 obj = self.obj 

165 if hasattr(self, "parent"): 

166 parent = self.parent 

167 if parent: 

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

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

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

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

172 else: 

173 message += "All " + action 

174 

175 args = (self, action) + args 

176 

177 execute_and_record_func( 

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

179 ) 

180 

181 return inner 

182 

183 

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

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 filterurlpatterns 

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["element_count"] = HistoryEvent.objects.all().count() 

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

256 context["paginator"] = { 

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

258 } 

259 

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

261 

262 if "col__pid" in filters: 

263 del filters["col__pid"] 

264 context["collections"] = ( 

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

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

267 .distinct("col__title_html") 

268 .order_by("col__title_html") 

269 ) 

270 return context 

271 

272 

273class HistoryClearView(TemplateView): 

274 model = HistoryEvent 

275 http_method_names = ["post"] 

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

277 success_url = reverse_lazy("history") 

278 

279 @transaction.atomic 

280 def post(self, *args): 

281 HistoryEvent.objects.all().delete() 

282 return HttpResponseRedirect(reverse("history")) 

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 @tracer.start_as_current_span("HistoryAPIView.get") 

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

297 object: HistoryEvent = self.get_object() 

298 if datatype == "message": 

299 return HttpResponse(object.message) 

300 if datatype == "table": 

301 data = [] 

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

303 entry = { 

304 "type": a.type, 

305 "status": a.status, 

306 "status_message": a.status_message, 

307 "message": a.message, 

308 "score": a.score, 

309 "url": a.url, 

310 } 

311 if a.resource: 

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

313 data.append(entry) 

314 return JsonResponse( 

315 { 

316 "data": data, 

317 "headers": [ 

318 "resource_pid", 

319 "type", 

320 "status", 

321 "status_message", 

322 "message", 

323 "score", 

324 "url", 

325 ], 

326 "source": object.source, 

327 "pid": object.pid, 

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

329 "date": object.created_on, 

330 } 

331 ) 

332 return HttpResponseBadRequest() 

333 

334 

335class HistoryEventInsert(View): 

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

337 # Todo : Sanitarization ? 

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

339 if "colid" in event: 

340 colid = event["colid"] 

341 del event["colid"] 

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

343 event["col"] = collection 

344 insert_history_event(event) 

345 response = HttpResponse() 

346 response.status_code = 201 

347 return response