Coverage for sites/ptf_tools/comments_moderation/utils.py: 97%

135 statements  

« prev     ^ index     » next       coverage.py v7.3.2, created at 2024-07-18 09:02 +0000

1from itertools import chain 

2 

3from django.contrib import messages 

4from django.contrib.auth.models import User 

5from django.db.models import Q 

6from django.http import HttpRequest 

7from django.urls import reverse 

8 

9from comments_api.constants import PARAM_ACTION 

10from comments_api.constants import PARAM_ACTION_CREATE 

11from comments_api.constants import PARAM_COMMENT 

12from comments_api.constants import PARAM_MODERATOR 

13from comments_api.constants import STATUS_SUBMITTED 

14from comments_views.core.rights import AbstractUserRights 

15from comments_views.core.utils import comments_credentials 

16from comments_views.core.utils import comments_server_url 

17from comments_views.core.utils import format_comment 

18from comments_views.core.utils import get_comment 

19from comments_views.core.utils import get_user_dict 

20from comments_views.core.utils import make_api_request 

21from ptf.model_helpers import get_article_by_doi 

22from ptf.models import Collection 

23from ptf.utils import send_email_from_template 

24from ptf_tools.models import Invitation 

25from ptf_tools.models import InvitationExtraData 

26 

27from .rights import ModeratorUserRights 

28 

29 

30def is_comment_moderator(user: User) -> bool: 

31 """Whether the user is a comment moderator.""" 

32 return hasattr(user, "comment_moderator") and user.comment_moderator.is_moderator 

33 

34 

35def update_moderation_right( 

36 comment_id: int, 

37 moderators: int | str, 

38 rights: AbstractUserRights, 

39 action: str = PARAM_ACTION_CREATE, 

40 request_for_message: HttpRequest | None = None, 

41) -> tuple[bool, dict]: 

42 """ 

43 Makes a POST request to the comment server to create a new moderation entry or 

44 delete an existing one. 

45 """ 

46 post_data = {PARAM_COMMENT: comment_id, PARAM_MODERATOR: moderators} 

47 query_params = rights.comment_rights_query_params() 

48 query_params[PARAM_ACTION] = action 

49 return make_api_request( 

50 "POST", 

51 comments_server_url(query_params, "moderators"), 

52 request_for_message=request_for_message, 

53 auth=comments_credentials(), 

54 json=post_data, 

55 timeout=4, 

56 ) 

57 

58 

59def get_all_moderators(rights: AbstractUserRights, request: HttpRequest): 

60 """ 

61 Gets all the existing moderators available to the given rights. 

62 

63 Returns: 

64 - all_moderators: The queryset of all available moderators. 

65 - moderation_rights: The moderation rights from the comment server. 

66 """ 

67 query_params = rights.comment_rights_query_params() 

68 

69 error, moderation_rights = make_api_request( 

70 "GET", 

71 comments_server_url(query_params, "moderators"), 

72 request_for_message=request, 

73 auth=comments_credentials(), 

74 ) 

75 

76 base_moderators = [] 

77 if error: 

78 moderation_rights = [] 

79 else: 

80 # mod_right struct: {"moderator_id": x, "comment_id": x, "comment__status": x} 

81 base_moderators = [m["moderator_id"] for m in moderation_rights] 

82 

83 user_collections = rights.get_user_admin_collections() + rights.get_user_staff_collections() 

84 

85 # Remove duplicates 

86 user_collections = set(user_collections) 

87 base_moderators = set(base_moderators) 

88 

89 moderator_filter = Q(comment_moderator__is_moderator=True) & ( 

90 Q(comment_moderator__collections__pid__in=user_collections) | Q(pk__in=base_moderators) 

91 ) 

92 

93 all_moderators = ( 

94 User.objects.prefetch_related("comment_moderator", "comment_moderator__collections") 

95 .filter(moderator_filter) 

96 .distinct() 

97 .order_by("first_name", "pk") 

98 ) 

99 

100 return all_moderators, moderation_rights 

101 

102 

103def update_moderator_collections( 

104 rights: AbstractUserRights, 

105 moderator: User, 

106 collections_to_add: list[str] = [], 

107 collections_to_delete: list[str] = [], 

108): 

109 """ 

110 Updates a moderator's attached collections. 

111 The modified collections are limited to the rights's admin collections. 

112 

113 Params: 

114 - rights: the rights of the user updating the moderator's 

115 collections 

116 - moderator: the moderator 

117 - collections_to_add: the PIDs of the collections to add 

118 - collections_to_delete: the PIDs of the collections to delete 

119 

120 """ 

121 if not is_comment_moderator(moderator) or not rights.is_admin_moderator(): 

122 return 

123 

124 user_collections = rights.get_user_admin_collections() 

125 collections = Collection.objects.all().values("pk", "pid") 

126 collections_to_delete = [ 

127 c["pk"] 

128 for c in collections 

129 if (c["pid"] in collections_to_delete and c["pid"] in user_collections) 

130 ] 

131 if collections_to_delete: 

132 moderator.comment_moderator.collections.remove(*collections_to_delete) 

133 

134 collections_to_add = [ 

135 c["pk"] 

136 for c in collections 

137 if (c["pid"] in collections_to_add and c["pid"] in user_collections) 

138 ] 

139 if collections_to_add: 

140 moderator.comment_moderator.collections.add(*collections_to_add) 

141 

142 

143def merge_dict(existing: dict, update: dict): 

144 """ 

145 Recursively merges 2 dicts by creating potential missing keys. 

146 When a key is common between the 2 dicts: 

147 - If the value is a dict: recursive call to merge_dict 

148 - If the value is a list: the 2 lists are concatenated 

149 TODO: what if list of dict/list ? 

150 - If the value is something else: the value of the update overwrites the 

151 existing one. 

152 """ 

153 for k, update_v in update.items(): 

154 existing_v = existing.get(k) 

155 

156 if isinstance(update_v, dict): 

157 if not existing_v: 

158 existing_v = {} 

159 elif existing_v and not isinstance(existing_v, dict): 

160 raise TypeError("Cannot merge dict and non-dict objects.") 

161 existing[k] = merge_dict(existing_v, update_v) if existing_v else update_v 

162 

163 elif isinstance(update_v, list): 

164 if not existing_v: 

165 existing_v = [] 

166 elif existing_v and not isinstance(existing_v, list): 

167 raise TypeError("Cannot merge list and non-list objects.") 

168 existing[k] = existing_v + update_v 

169 

170 else: 

171 if isinstance(existing_v, dict): 

172 raise TypeError("Cannot merge dict and non-dict objects.") 

173 if isinstance(existing_v, list): 

174 raise TypeError("Cannot merge list and non-list objects.") 

175 existing[k] = update_v 

176 

177 return existing 

178 

179 

180def get_comment_and_can_manage_moderators( 

181 request: HttpRequest, rights: AbstractUserRights, comment_id: int 

182) -> tuple[bool, dict]: 

183 """ 

184 GET the requested comment and check the given rights allow moderator management. 

185 Adds a message to the request if an error occurs of if the rights are insufficient. 

186 """ 

187 error, comment = get_comment( 

188 rights.comment_rights_query_params(), comment_id, request_for_message=request 

189 ) 

190 if error: 

191 return True, {} 

192 

193 users = get_user_dict() 

194 

195 format_comment(comment, rights, users) 

196 if not rights.comment_can_manage_moderators(comment) or comment["status"] != STATUS_SUBMITTED: 

197 messages.error( 

198 request, 

199 "Error: You don't have enough rights to manage the moderators of " 

200 "the given comment.", 

201 ) 

202 return True, {} 

203 

204 return False, comment 

205 

206 

207def email_moderator_assigned(request: HttpRequest, moderator: User, comment: dict): 

208 """ 

209 Send the "Comment moderation request" mail to the provided moderator. 

210 Add messages to the provided request. 

211 """ 

212 messages.success( 

213 request, 

214 f"The moderator {moderator.first_name} {moderator.last_name} " 

215 "has been assigned to the comment.", 

216 ) 

217 # Send an e-mail to the moderator assigned to the comment - 

218 # TODO: Maybe this should be done few times a day by a cron 

219 # to avoid too many e-mails? 

220 article = get_article_by_doi(comment["doi"]) 

221 if not article: 

222 messages.warning( 

223 request, 

224 "The comment's article was not found on Trammel. " 

225 + "No e-mail was sent to the moderator.", 

226 ) 

227 return 

228 

229 article_title = article.title_tex if article else "" 

230 context_data = { 

231 "full_name": f"{moderator.first_name} {moderator.last_name}", 

232 "inviter_name": f"{request.user.first_name} {request.user.last_name}", 

233 "article_url": comment.get("base_url"), 

234 "article_title": article_title, 

235 "comment_dashboard_url": request.build_absolute_uri(reverse("comment_list")), 

236 "email_signature": "The editorial team", 

237 } 

238 subject = f"[{comment['site_name'].upper()}] Comment moderation request" 

239 send_email_from_template( 

240 "mail/comment_moderator_assigned.html", 

241 context_data, 

242 subject, 

243 to=[moderator.email], 

244 from_collection=comment["site_name"], 

245 ) 

246 

247 

248def get_comments_for_home(user: User) -> tuple[bool, dict]: 

249 """ 

250 Query the comments server to get the summary of the comments data per collection. 

251 

252 Returns: 

253 - `error` Whether the HTTP query was successful 

254 - `data` The processed comments summary: 

255 ` 

256 { 

257 "COLID": submitted_comments_nb 

258 } 

259 ` 

260 

261 There is no entry for a collection without any comments. 

262 """ 

263 rights = ModeratorUserRights(user) 

264 query_params = rights.comment_rights_query_params() 

265 

266 error, comments_summary = make_api_request( 

267 "GET", comments_server_url(query_params, "comments-summary"), auth=comments_credentials() 

268 ) 

269 if error: 

270 return True, {} 

271 

272 data = {} 

273 for col_id, col_data in comments_summary.items(): 

274 data[col_id] = col_data.get(STATUS_SUBMITTED, 0) 

275 

276 return False, data 

277 

278 

279def get_pending_invitations(rights: AbstractUserRights) -> list: 

280 """ 

281 Returns the list of pending invitations to comment moderators available to the 

282 given user rights. 

283 """ 

284 pending_invites = Invitation.objects.filter(accepted=False, extra_data__has_key="moderator") 

285 invites = [] 

286 for invite in pending_invites: 

287 collections_to_add = [] 

288 extra_data = InvitationExtraData(**invite.extra_data) 

289 

290 # Get all collections matching the user_collections 

291 if rights.user.is_superuser: 291 ↛ 292line 291 didn't jump to line 292, because the condition on line 291 was never true

292 collections_to_add = chain.from_iterable( 

293 c.pid for c in extra_data.moderator.collections 

294 ) 

295 elif rights.is_admin_moderator(): 

296 collections = chain.from_iterable(c.pid for c in extra_data.moderator.collections) 

297 collections_to_add = [ 

298 c for c in collections if c in rights.get_user_admin_collections() 

299 ] 

300 

301 # Get all comments whose collection matches the user_collections 

302 comments_to_add = [] 

303 user_collections = ( 

304 rights.get_user_admin_collections() or rights.get_user_staff_collections() 

305 ) 

306 for comment in extra_data.moderator.comments: 

307 if rights.user.is_superuser: 307 ↛ 308line 307 didn't jump to line 308, because the condition on line 307 was never true

308 comments_to_add.append(comment.id) 

309 elif comment.pid in user_collections: 

310 comments_to_add.append(comment.id) 

311 

312 if collections_to_add or comments_to_add: 

313 invites.append( 

314 { 

315 "name": f"{invite.first_name} {invite.last_name}", 

316 "email": invite.email, 

317 "sent": invite.sent, 

318 "date_expired": invite.date_expired(), 

319 "key_expired": invite.key_expired(), 

320 "collections": collections_to_add, 

321 "comments": comments_to_add, 

322 } 

323 ) 

324 

325 return invites 

326 

327 

328def get_comment_pending_invitations(comment_id: int) -> list: 

329 """ 

330 Returns the invitations linked to the given comment ID. 

331 """ 

332 # Not working on CI with SQLite DB backend. 

333 # return Invitation.objects.filter( 

334 # accepted=False, 

335 # extra_data__moderator__comments__contains=[{"id": comment_id}] 

336 # ) 

337 base_invites = Invitation.objects.filter( 

338 accepted=False, extra_data__moderator__has_key="comments" 

339 ) 

340 invites = [] 

341 for invite in base_invites: 

342 extra_data = InvitationExtraData(**invite.extra_data) 

343 if any(c.id == comment_id for c in extra_data.moderator.comments): 

344 invites.append(invite) 

345 

346 return invites