Coverage for app/backend/src/couchers/servicers/moderation.py: 95%

1import json 

2import logging 

3 

4import grpc 

5from sqlalchemy import and_, exists, not_, or_, select 

6from sqlalchemy.orm import Session 

7 

8from couchers import urls 

9from couchers.context import CouchersContext 

10from couchers.jobs.enqueue import queue_job 

11from couchers.metrics import ( 

12 observe_moderation_action, 

13 observe_moderation_queue_item_created, 

14 observe_moderation_queue_item_resolved, 

15 observe_moderation_queue_resolution_time, 

16 observe_moderation_visibility_transition, 

17) 

18from couchers.models import ( 

19 AdminActionLevel, 

20 Cluster, 

21 Comment, 

22 Discussion, 

23 Event, 

24 EventOccurrence, 

25 FriendRelationship, 

26 GroupChat, 

27 GroupChatSubscription, 

28 HostRequest, 

29 Message, 

30 MessageType, 

31 ModerationAction, 

32 ModerationLog, 

33 ModerationObjectType, 

34 ModerationQueueItem, 

35 ModerationState, 

36 ModerationTrigger, 

37 ModerationVisibility, 

38 Notification, 

39 NotificationDelivery, 

40 Page, 

41 PublicTrip, 

42 Reference, 

43 Reply, 

44 User, 

45 get_moderated_models, 

46) 

47from couchers.proto import moderation_pb2, moderation_pb2_grpc 

48from couchers.proto.internal import jobs_pb2 

49from couchers.utils import Timestamp_from_datetime, not_none, now 

50 

51logger = logging.getLogger(__name__) 

52 

53MAX_PAGINATION_LENGTH = 1_000 

54 

55# Moderation enum mappings 

56moderationvisibility2api = { 

57 None: moderation_pb2.MODERATION_VISIBILITY_UNSPECIFIED, 

58 ModerationVisibility.hidden: moderation_pb2.MODERATION_VISIBILITY_HIDDEN, 

59 ModerationVisibility.shadowed: moderation_pb2.MODERATION_VISIBILITY_SHADOWED, 

60 ModerationVisibility.visible: moderation_pb2.MODERATION_VISIBILITY_VISIBLE, 

61 ModerationVisibility.unlisted: moderation_pb2.MODERATION_VISIBILITY_UNLISTED, 

62} 

63 

64moderationvisibility2sql = { 

65 moderation_pb2.MODERATION_VISIBILITY_UNSPECIFIED: None, 

66 moderation_pb2.MODERATION_VISIBILITY_HIDDEN: ModerationVisibility.hidden, 

67 moderation_pb2.MODERATION_VISIBILITY_SHADOWED: ModerationVisibility.shadowed, 

68 moderation_pb2.MODERATION_VISIBILITY_VISIBLE: ModerationVisibility.visible, 

69 moderation_pb2.MODERATION_VISIBILITY_UNLISTED: ModerationVisibility.unlisted, 

70} 

71 

72moderationtrigger2api = { 

73 None: moderation_pb2.MODERATION_TRIGGER_UNSPECIFIED, 

74 ModerationTrigger.initial_review: moderation_pb2.MODERATION_TRIGGER_INITIAL_REVIEW, 

75 ModerationTrigger.user_flag: moderation_pb2.MODERATION_TRIGGER_USER_FLAG, 

76 ModerationTrigger.machine_flag: moderation_pb2.MODERATION_TRIGGER_MACHINE_FLAG, 

77 ModerationTrigger.moderator_review: moderation_pb2.MODERATION_TRIGGER_MODERATOR_REVIEW, 

78} 

79 

80moderationtrigger2sql = { 

81 moderation_pb2.MODERATION_TRIGGER_UNSPECIFIED: None, 

82 moderation_pb2.MODERATION_TRIGGER_INITIAL_REVIEW: ModerationTrigger.initial_review, 

83 moderation_pb2.MODERATION_TRIGGER_USER_FLAG: ModerationTrigger.user_flag, 

84 moderation_pb2.MODERATION_TRIGGER_MACHINE_FLAG: ModerationTrigger.machine_flag, 

85 moderation_pb2.MODERATION_TRIGGER_MODERATOR_REVIEW: ModerationTrigger.moderator_review, 

86} 

87 

88moderationaction2api = { 

89 None: moderation_pb2.MODERATION_ACTION_UNSPECIFIED, 

90 ModerationAction.create: moderation_pb2.MODERATION_ACTION_CREATE, 

91 ModerationAction.approve: moderation_pb2.MODERATION_ACTION_APPROVE, 

92 ModerationAction.hide: moderation_pb2.MODERATION_ACTION_HIDE, 

93 ModerationAction.flag: moderation_pb2.MODERATION_ACTION_FLAG, 

94 ModerationAction.unflag: moderation_pb2.MODERATION_ACTION_UNFLAG, 

95 ModerationAction.set_priority: moderation_pb2.MODERATION_ACTION_SET_PRIORITY, 

96 ModerationAction.bulk_set_visibility: moderation_pb2.MODERATION_ACTION_BULK_SET_VISIBILITY, 

97} 

98 

99moderationaction2sql = { 

100 moderation_pb2.MODERATION_ACTION_UNSPECIFIED: None, 

101 moderation_pb2.MODERATION_ACTION_CREATE: ModerationAction.create, 

102 moderation_pb2.MODERATION_ACTION_APPROVE: ModerationAction.approve, 

103 moderation_pb2.MODERATION_ACTION_HIDE: ModerationAction.hide, 

104 moderation_pb2.MODERATION_ACTION_FLAG: ModerationAction.flag, 

105 moderation_pb2.MODERATION_ACTION_UNFLAG: ModerationAction.unflag, 

106 moderation_pb2.MODERATION_ACTION_SET_PRIORITY: ModerationAction.set_priority, 

107 moderation_pb2.MODERATION_ACTION_BULK_SET_VISIBILITY: ModerationAction.bulk_set_visibility, 

108} 

109 

110moderationobjecttype2api = { 

111 None: moderation_pb2.MODERATION_OBJECT_TYPE_UNSPECIFIED, 

112 ModerationObjectType.host_request: moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST, 

113 ModerationObjectType.group_chat: moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT, 

114 ModerationObjectType.friend_request: moderation_pb2.MODERATION_OBJECT_TYPE_FRIEND_REQUEST, 

115 ModerationObjectType.event_occurrence: moderation_pb2.MODERATION_OBJECT_TYPE_EVENT_OCCURRENCE, 

116 ModerationObjectType.comment: moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT, 

117 ModerationObjectType.reply: moderation_pb2.MODERATION_OBJECT_TYPE_REPLY, 

118 ModerationObjectType.discussion: moderation_pb2.MODERATION_OBJECT_TYPE_DISCUSSION, 

119 ModerationObjectType.reference: moderation_pb2.MODERATION_OBJECT_TYPE_REFERENCE, 

120 ModerationObjectType.public_trip: moderation_pb2.MODERATION_OBJECT_TYPE_PUBLIC_TRIP, 

121 ModerationObjectType.user: moderation_pb2.MODERATION_OBJECT_TYPE_USER, 

122} 

123 

124moderationobjecttype2sql = { 

125 moderation_pb2.MODERATION_OBJECT_TYPE_UNSPECIFIED: None, 

126 moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST: ModerationObjectType.host_request, 

127 moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT: ModerationObjectType.group_chat, 

128 moderation_pb2.MODERATION_OBJECT_TYPE_FRIEND_REQUEST: ModerationObjectType.friend_request, 

129 moderation_pb2.MODERATION_OBJECT_TYPE_EVENT_OCCURRENCE: ModerationObjectType.event_occurrence, 

130 moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT: ModerationObjectType.comment, 

131 moderation_pb2.MODERATION_OBJECT_TYPE_REPLY: ModerationObjectType.reply, 

132 moderation_pb2.MODERATION_OBJECT_TYPE_DISCUSSION: ModerationObjectType.discussion, 

133 moderation_pb2.MODERATION_OBJECT_TYPE_REFERENCE: ModerationObjectType.reference, 

134 moderation_pb2.MODERATION_OBJECT_TYPE_PUBLIC_TRIP: ModerationObjectType.public_trip, 

135 moderation_pb2.MODERATION_OBJECT_TYPE_USER: ModerationObjectType.user, 

136} 

137 

138 

139def _resolve_queue_item( 

140 queue_item: ModerationQueueItem, 

141 log_entry: ModerationLog, 

142 action: ModerationAction, 

143 object_type: ModerationObjectType, 

144) -> None: 

145 """Resolve an open queue item against the given log entry and record metrics.""" 

146 queue_item.resolved_by_log_id = log_entry.id 

147 observe_moderation_queue_item_resolved(queue_item.trigger, action, object_type) 

148 observe_moderation_queue_resolution_time( 

149 queue_item.trigger, 

150 action, 

151 object_type, 

152 (now() - queue_item.time_created).total_seconds(), 

153 ) 

154 

155 

156def bulk_set_user_content_visibility( 

157 session: Session, 

158 user: User, 

159 new_visibility: ModerationVisibility, 

160 moderator_user_id: int, 

161 from_visibilities: set[ModerationVisibility] | None = None, 

162 reason: str | None = None, 

163) -> int: 

164 """Set visibility on every UMS-governed object authored by the user. Returns count of updated states.""" 

165 final_reason = reason or f"Bulk visibility update for user {user.id} to {new_visibility.name}" 

166 

167 author_exists_clauses = [] 

168 for entry in get_moderated_models().values(): 

169 if entry.has_own_visibility_mechanism: 

170 continue 

171 author_exists_clauses.append( 

172 exists().where(and_(entry.moderation_state_id_column == ModerationState.id, entry.author_column == user.id)) 

173 ) 

174 

175 states = session.execute(select(ModerationState).where(or_(*author_exists_clauses))).scalars().all() 

176 

177 updated_count = 0 

178 for moderation_state in states: 

179 if from_visibilities and moderation_state.visibility not in from_visibilities: 

180 continue 

181 if moderation_state.visibility == new_visibility: 

182 continue 

183 

184 old_visibility = not_none(moderation_state.visibility) 

185 moderation_state.visibility = new_visibility 

186 moderation_state.updated = now() 

187 

188 log_entry = ModerationLog( 

189 moderation_state_id=moderation_state.id, 

190 action=ModerationAction.bulk_set_visibility, 

191 moderator_user_id=moderator_user_id, 

192 new_visibility=new_visibility, 

193 reason=final_reason, 

194 ) 

195 session.add(log_entry) 

196 session.flush() 

197 

198 open_items = ( 

199 session.execute( 

200 select(ModerationQueueItem) 

201 .where(ModerationQueueItem.moderation_state_id == moderation_state.id) 

202 .where(ModerationQueueItem.resolved_by_log_id.is_(None)) 

203 ) 

204 .scalars() 

205 .all() 

206 ) 

207 for queue_item in open_items: 

208 _resolve_queue_item( 

209 queue_item, log_entry, ModerationAction.bulk_set_visibility, moderation_state.object_type 

210 ) 

211 session.flush() 

212 

213 observe_moderation_action(ModerationAction.bulk_set_visibility, moderation_state.object_type) 

214 observe_moderation_visibility_transition(old_visibility, new_visibility, moderation_state.object_type) 

215 

216 if new_visibility in (ModerationVisibility.visible, ModerationVisibility.unlisted): 

217 _enqueue_pending_notifications(session, moderation_state.id) 

218 

219 updated_count += 1 

220 

221 return updated_count 

222 

223 

224def _enqueue_pending_notifications(session: Session, moderation_state_id: int) -> None: 

225 """Re-queue any pending notifications linked to the given moderation state whose deliveries were suppressed.""" 

226 pending_notifications = ( 

227 session.execute( 

228 select(Notification) 

229 .where(Notification.moderation_state_id == moderation_state_id) 

230 .where(not_(exists().where(NotificationDelivery.notification_id == Notification.id))) 

231 # they're delivered in the order they're queued, so the user reads them chronologically 

232 .order_by(Notification.id) 

233 ) 

234 .scalars() 

235 .all() 

236 ) 

237 

238 # Import here to avoid circular dependency 

239 from couchers.notifications.background import handle_notification # noqa: PLC0415 

240 

241 for notification in pending_notifications: 

242 queue_job( 

243 session, 

244 job=handle_notification, 

245 payload=jobs_pb2.HandleNotificationPayload(notification_id=notification.id), 

246 ) 

247 

248 

249def moderation_state_to_pb(state: ModerationState, session: Session) -> moderation_pb2.ModerationStateInfo: 

250 """Convert ModerationState model to proto message""" 

251 object_type = state.object_type 

252 object_id = state.object_id 

253 

254 # Get the author user ID and content based on object type 

255 if object_type == ModerationObjectType.host_request: 

256 author_user_id = session.execute( 

257 select(HostRequest.initiator_user_id).where(HostRequest.conversation_id == object_id) 

258 ).scalar_one() 

259 # Get the first text message for this conversation 

260 content = session.execute( 

261 select(Message.text) 

262 .where(Message.conversation_id == object_id) 

263 .where(Message.message_type == MessageType.text) 

264 .order_by(Message.id.asc()) 

265 .limit(1) 

266 ).scalar_one_or_none() 

267 elif object_type == ModerationObjectType.group_chat: 

268 author_user_id = session.execute( 

269 select(GroupChat.creator_id).where(GroupChat.conversation_id == object_id) 

270 ).scalar_one() 

271 # Get the first text message for this conversation 

272 content = session.execute( 

273 select(Message.text) 

274 .where(Message.conversation_id == object_id) 

275 .where(Message.message_type == MessageType.text) 

276 .order_by(Message.id.asc()) 

277 .limit(1) 

278 ).scalar_one_or_none() 

279 elif object_type == ModerationObjectType.friend_request: 

280 author_user_id = session.execute( 

281 select(FriendRelationship.from_user_id).where(FriendRelationship.id == object_id) 

282 ).scalar_one() 

283 # Friend requests have no text content 

284 content = None 

285 elif object_type == ModerationObjectType.event_occurrence: 

286 author_user_id, title, description = session.execute( 

287 select(EventOccurrence.creator_user_id, Event.title, EventOccurrence.content) 

288 .join(Event, Event.id == EventOccurrence.event_id) 

289 .where(EventOccurrence.id == object_id) 

290 ).one() 

291 content = f"{title}\n\n{description}" 

292 elif object_type == ModerationObjectType.comment: 

293 author_user_id, content = session.execute( 

294 select(Comment.author_user_id, Comment.content).where(Comment.id == object_id) 

295 ).one() 

296 elif object_type == ModerationObjectType.reply: 

297 author_user_id, content = session.execute( 

298 select(Reply.author_user_id, Reply.content).where(Reply.id == object_id) 

299 ).one() 

300 elif object_type == ModerationObjectType.discussion: 

301 author_user_id, title, body = session.execute( 

302 select(Discussion.creator_user_id, Discussion.title, Discussion.content).where(Discussion.id == object_id) 

303 ).one() 

304 content = f"{title}\n\n{body}" 

305 elif object_type == ModerationObjectType.reference: 

306 author_user_id, content = session.execute( 

307 select(Reference.from_user_id, Reference.text).where(Reference.id == object_id) 

308 ).one() 

309 elif object_type == ModerationObjectType.public_trip: 

310 author_user_id, content = session.execute( 

311 select(PublicTrip.user_id, PublicTrip.description).where(PublicTrip.id == object_id) 

312 ).one() 

313 elif object_type == ModerationObjectType.user: 313 ↛ 318line 313 didn't jump to line 318 because the condition on line 313 was always true

314 author_user_id = object_id 

315 username = session.execute(select(User.username).where(User.id == object_id)).scalar_one() 

316 content = f"@{username} / {object_id}" 

317 else: 

318 raise ValueError(f"Unsupported moderation object type: {object_type}") 

319 

320 # Import here to avoid circular dependency 

321 from couchers.servicers.admin import _user_to_details # noqa: PLC0415 

322 

323 author = session.execute(select(User).where(User.id == author_user_id)).scalar_one() 

324 

325 state_pb = moderation_pb2.ModerationStateInfo( 

326 moderation_state_id=state.id, 

327 object_type=moderationobjecttype2api[state.object_type], 

328 object_id=state.object_id, 

329 visibility=moderationvisibility2api[state.visibility], 

330 created=Timestamp_from_datetime(state.created), 

331 updated=Timestamp_from_datetime(state.updated), 

332 author_user_id=author_user_id, 

333 author=_user_to_details(session, author), 

334 content=content or "", 

335 ) 

336 

337 return state_pb 

338 

339 

340def _community_pb(session: Session, node_id: int) -> moderation_pb2.ModeratedObjectCommunity: 

341 return moderation_pb2.ModeratedObjectCommunity( 

342 community_id=node_id, 

343 community_name=session.execute( 

344 select(Cluster.name).where(Cluster.parent_node_id == node_id).where(Cluster.is_official_cluster) 

345 ).scalar_one(), 

346 ) 

347 

348 

349def _cluster_community_pb(session: Session, cluster: Cluster) -> moderation_pb2.ModeratedObjectCommunity: 

350 if cluster.is_official_cluster: 

351 return moderation_pb2.ModeratedObjectCommunity(community_id=cluster.parent_node_id, community_name=cluster.name) 

352 community = _community_pb(session, cluster.parent_node_id) 

353 community.group_id = cluster.id 

354 community.group_name = cluster.name 

355 return community 

356 

357 

358def _owner_community_pb( 

359 session: Session, parent_node_id: int, owner_cluster: Cluster | None 

360) -> moderation_pb2.ModeratedObjectCommunity: 

361 if owner_cluster is not None: 

362 return _cluster_community_pb(session, owner_cluster) 

363 return _community_pb(session, parent_node_id) 

364 

365 

366def _event_occurrence_location( 

367 session: Session, occurrence: EventOccurrence 

368) -> tuple[str, moderation_pb2.ModeratedObjectCommunity]: 

369 event = occurrence.event 

370 return ( 

371 urls.event_link(occurrence_id=occurrence.id, slug=event.slug), 

372 _owner_community_pb(session, event.parent_node_id, event.owner_cluster), 

373 ) 

374 

375 

376def _discussion_location( 

377 session: Session, discussion: Discussion 

378) -> tuple[str, moderation_pb2.ModeratedObjectCommunity]: 

379 return ( 

380 urls.discussion_link(discussion_id=str(discussion.id), slug=discussion.slug), 

381 _cluster_community_pb(session, discussion.owner_cluster), 

382 ) 

383 

384 

385def _thread_parent( 

386 session: Session, thread_id: int 

387) -> tuple[str, moderation_pb2.ModeratedObjectCommunity | None, moderation_pb2.ModeratedThreadParent | None]: 

388 occurrence = session.execute( 

389 select(EventOccurrence).where(EventOccurrence.thread_id == thread_id) 

390 ).scalar_one_or_none() 

391 if occurrence is not None: 

392 url, community = _event_occurrence_location(session, occurrence) 

393 return ( 

394 url, 

395 community, 

396 moderation_pb2.ModeratedThreadParent(event_occurrence_id=occurrence.id, title=occurrence.event.title), 

397 ) 

398 

399 discussion = session.execute(select(Discussion).where(Discussion.thread_id == thread_id)).scalar_one_or_none() 

400 if discussion is not None: 

401 url, community = _discussion_location(session, discussion) 

402 return url, community, moderation_pb2.ModeratedThreadParent(discussion_id=discussion.id, title=discussion.title) 

403 

404 page = session.execute(select(Page).where(Page.thread_id == thread_id)).scalar_one_or_none() 

405 if page is not None: 405 ↛ 412line 405 didn't jump to line 412 because the condition on line 405 was always true

406 return ( 

407 "", 

408 _owner_community_pb(session, page.parent_node_id, page.owner_cluster), 

409 moderation_pb2.ModeratedThreadParent(page_id=page.id, title=page.versions[-1].title), 

410 ) 

411 

412 return "", None, None 

413 

414 

415def moderated_object_to_pb(state: ModerationState, session: Session) -> moderation_pb2.GetModeratedObjectRes: 

416 """Describe where a moderated object lives and what it's about""" 

417 state_pb = moderation_state_to_pb(state, session) 

418 object_type = state.object_type 

419 object_id = state.object_id 

420 

421 url = "" 

422 community: moderation_pb2.ModeratedObjectCommunity | None = None 

423 thread_parent: moderation_pb2.ModeratedThreadParent | None = None 

424 parent_comment: Comment | None = None 

425 involved_user_ids: list[int] = [] 

426 

427 if object_type == ModerationObjectType.host_request: 

428 involved_user_ids = [ 

429 session.execute( 

430 select(HostRequest.recipient_user_id).where(HostRequest.conversation_id == object_id) 

431 ).scalar_one() 

432 ] 

433 elif object_type == ModerationObjectType.group_chat: 

434 involved_user_ids = list( 

435 session.execute( 

436 select(GroupChatSubscription.user_id) 

437 .where(GroupChatSubscription.group_chat_id == object_id) 

438 .where(GroupChatSubscription.left.is_(None)) 

439 .where(GroupChatSubscription.user_id != state_pb.author_user_id) 

440 .order_by(GroupChatSubscription.joined.asc()) 

441 ) 

442 .scalars() 

443 .all() 

444 ) 

445 elif object_type == ModerationObjectType.friend_request: 

446 involved_user_ids = [ 

447 session.execute( 

448 select(FriendRelationship.to_user_id).where(FriendRelationship.id == object_id) 

449 ).scalar_one() 

450 ] 

451 elif object_type == ModerationObjectType.event_occurrence: 

452 occurrence = session.execute(select(EventOccurrence).where(EventOccurrence.id == object_id)).scalar_one() 

453 url, community = _event_occurrence_location(session, occurrence) 

454 elif object_type == ModerationObjectType.comment: 

455 thread_id = session.execute(select(Comment.thread_id).where(Comment.id == object_id)).scalar_one() 

456 url, community, thread_parent = _thread_parent(session, thread_id) 

457 elif object_type == ModerationObjectType.reply: 

458 parent_comment = session.execute( 

459 select(Comment).join(Reply, Reply.comment_id == Comment.id).where(Reply.id == object_id) 

460 ).scalar_one() 

461 url, community, thread_parent = _thread_parent(session, parent_comment.thread_id) 

462 elif object_type == ModerationObjectType.discussion: 

463 discussion = session.execute(select(Discussion).where(Discussion.id == object_id)).scalar_one() 

464 url, community = _discussion_location(session, discussion) 

465 elif object_type == ModerationObjectType.reference: 

466 to_user = session.execute( 

467 select(User).join(Reference, Reference.to_user_id == User.id).where(Reference.id == object_id) 

468 ).scalar_one() 

469 url = urls.user_link(username=to_user.username) 

470 involved_user_ids = [to_user.id] 

471 elif object_type == ModerationObjectType.public_trip: 

472 node_id = session.execute(select(PublicTrip.node_id).where(PublicTrip.id == object_id)).scalar_one() 

473 community = _community_pb(session, node_id) 

474 elif object_type == ModerationObjectType.user: 474 ↛ 477line 474 didn't jump to line 477 because the condition on line 474 was always true

475 url = urls.user_link(username=state_pb.author.username) 

476 else: 

477 raise ValueError(f"Unsupported moderation object type: {object_type}") 

478 

479 return moderation_pb2.GetModeratedObjectRes( 

480 moderation_state=state_pb, 

481 url=url, 

482 community=community, 

483 thread_parent=thread_parent, 

484 parent_comment_id=parent_comment.id if parent_comment else None, 

485 parent_comment_content=parent_comment.content if parent_comment else None, 

486 involved_user_ids=involved_user_ids, 

487 ) 

488 

489 

490class Moderation(moderation_pb2_grpc.ModerationServicer): 

491 def GetModerationQueue( 

492 self, request: moderation_pb2.GetModerationQueueReq, context: CouchersContext, session: Session 

493 ) -> moderation_pb2.GetModerationQueueRes: 

494 """Get moderation queue items with optional filtering""" 

495 

496 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

497 

498 # Build query 

499 statement = select(ModerationQueueItem) 

500 

501 # Apply page token filter based on ordering direction 

502 if request.page_token: 

503 page_token_id = int(request.page_token) 

504 if request.newest_first: 504 ↛ 509line 504 didn't jump to line 509 because the condition on line 504 was always true

505 # Descending order: get items with smaller IDs 

506 statement = statement.where(ModerationQueueItem.id < page_token_id) 

507 else: 

508 # Ascending order: get items with larger IDs 

509 statement = statement.where(ModerationQueueItem.id > page_token_id) 

510 

511 # Apply filters 

512 if request.triggers: 

513 internal_triggers = [moderationtrigger2sql[t] for t in request.triggers] 

514 statement = statement.where(ModerationQueueItem.trigger.in_(internal_triggers)) 

515 

516 if request.object_type and request.object_type != moderation_pb2.MODERATION_OBJECT_TYPE_UNSPECIFIED: 516 ↛ 517line 516 didn't jump to line 517 because the condition on line 516 was never true

517 internal_object_type = moderationobjecttype2sql[request.object_type] 

518 if internal_object_type: 

519 statement = statement.join(ModerationState).where(ModerationState.object_type == internal_object_type) 

520 

521 if request.unresolved_only: 

522 statement = statement.where(ModerationQueueItem.resolved_by_log_id.is_(None)) 

523 

524 if request.HasField("priority_min"): 

525 statement = statement.where(ModerationQueueItem.priority >= request.priority_min) 

526 

527 if request.HasField("priority_max"): 

528 statement = statement.where(ModerationQueueItem.priority <= request.priority_max) 

529 

530 if request.HasField("created_before"): 

531 created_before = request.created_before.ToDatetime() 

532 statement = statement.where(ModerationQueueItem.time_created < created_before) 

533 

534 if request.HasField("created_after"): 

535 created_after = request.created_after.ToDatetime() 

536 statement = statement.where(ModerationQueueItem.time_created > created_after) 

537 

538 if request.item_author_user_id: 

539 author_user_id = request.item_author_user_id 

540 

541 # Use EXISTS for efficient author filtering 

542 author_exists_clauses = [] 

543 for entry in get_moderated_models().values(): 

544 author_exists_clauses.append( 

545 exists().where( 

546 and_( 

547 entry.moderation_state_id_column == ModerationQueueItem.moderation_state_id, 

548 entry.author_column == author_user_id, 

549 ) 

550 ) 

551 ) 

552 statement = statement.where(or_(*author_exists_clauses)) 

553 

554 if request.newest_first: 

555 statement = statement.order_by(ModerationQueueItem.id.desc()) 

556 else: 

557 statement = statement.order_by(ModerationQueueItem.id.asc()) 

558 

559 queue_items = session.execute(statement.limit(page_size + 1)).scalars().all() 

560 

561 # Convert to proto 

562 queue_items_pb = [] 

563 for item in queue_items[:page_size]: 

564 # Fetch the moderation state for this queue item 

565 mod_state = session.execute( 

566 select(ModerationState).where(ModerationState.id == item.moderation_state_id) 

567 ).scalar_one() 

568 

569 queue_item_pb = moderation_pb2.ModerationQueueItemInfo( 

570 queue_item_id=item.id, 

571 moderation_state_id=item.moderation_state_id, 

572 time_created=Timestamp_from_datetime(item.time_created), 

573 trigger=moderationtrigger2api[item.trigger], 

574 reason=item.reason, 

575 is_resolved=item.resolved_by_log_id is not None, 

576 resolved_by_log_id=item.resolved_by_log_id or 0, 

577 moderation_state=moderation_state_to_pb(mod_state, session), 

578 priority=item.priority, 

579 data=json.dumps(item.data) if item.data is not None else "", 

580 ) 

581 

582 queue_items_pb.append(queue_item_pb) 

583 

584 return moderation_pb2.GetModerationQueueRes( 

585 queue_items=queue_items_pb, 

586 # Use the ID of the last returned item (not the extra fetched item) as the cursor 

587 next_page_token=str(queue_items[page_size - 1].id) if len(queue_items) > page_size else None, 

588 ) 

589 

590 def GetModerationState( 

591 self, request: moderation_pb2.GetModerationStateReq, context: CouchersContext, session: Session 

592 ) -> moderation_pb2.GetModerationStateRes: 

593 """Get moderation state by object type and object ID""" 

594 object_type = moderationobjecttype2sql[request.object_type] 

595 if object_type is None: 

596 context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Object type must be specified.") 

597 

598 moderation_state = session.execute( 

599 select(ModerationState) 

600 .where(ModerationState.object_type == object_type) 

601 .where(ModerationState.object_id == request.object_id) 

602 ).scalar_one_or_none() 

603 if moderation_state is None: 

604 context.abort(grpc.StatusCode.NOT_FOUND, "Moderation state not found.") 

605 

606 return moderation_pb2.GetModerationStateRes( 

607 moderation_state=moderation_state_to_pb(moderation_state, session), 

608 ) 

609 

610 def GetModerationLog( 

611 self, request: moderation_pb2.GetModerationLogReq, context: CouchersContext, session: Session 

612 ) -> moderation_pb2.GetModerationLogRes: 

613 """Get moderation log for a specific moderation state""" 

614 # Get the moderation state 

615 moderation_state = session.execute( 

616 select(ModerationState).where(ModerationState.id == request.moderation_state_id) 

617 ).scalar_one_or_none() 

618 if moderation_state is None: 

619 context.abort(grpc.StatusCode.NOT_FOUND, "Moderation state not found.") 

620 

621 # Get all log entries for this state, ordered by time (most recent first) 

622 log_entries = ( 

623 session.execute( 

624 select(ModerationLog) 

625 .where(ModerationLog.moderation_state_id == request.moderation_state_id) 

626 .order_by(ModerationLog.time.desc(), ModerationLog.id.desc()) 

627 ) 

628 .scalars() 

629 .all() 

630 ) 

631 

632 # Convert moderation state to proto first (while still in session) 

633 moderation_state_pb = moderation_state_to_pb(moderation_state, session) 

634 

635 # Convert to proto 

636 log_entries_pb = [] 

637 for entry in log_entries: 

638 log_entry_pb = moderation_pb2.ModerationLogEntryInfo( 

639 log_entry_id=entry.id, 

640 moderation_state_id=entry.moderation_state_id, 

641 time=Timestamp_from_datetime(entry.time), 

642 action=moderationaction2api[entry.action], 

643 moderator_user_id=entry.moderator_user_id, 

644 reason=entry.reason, 

645 ) 

646 

647 # Only include changed fields 

648 if entry.new_visibility is not None: 

649 log_entry_pb.new_visibility = moderationvisibility2api[entry.new_visibility] 

650 if entry.new_priority is not None: 650 ↛ 651line 650 didn't jump to line 651 because the condition on line 650 was never true

651 log_entry_pb.new_priority = entry.new_priority 

652 if entry.queue_item_id is not None: 

653 log_entry_pb.queue_item_id = entry.queue_item_id 

654 

655 log_entries_pb.append(log_entry_pb) 

656 

657 return moderation_pb2.GetModerationLogRes( 

658 log_entries=log_entries_pb, 

659 moderation_state=moderation_state_pb, 

660 ) 

661 

662 def ModerateContent( 

663 self, request: moderation_pb2.ModerateContentReq, context: CouchersContext, session: Session 

664 ) -> moderation_pb2.ModerateContentRes: 

665 """Single moderation entrypoint, dispatching on action. 

666 

667 APPROVE/HIDE act on the state's visibility; FLAG/SET_PRIORITY/UNFLAG act on a single 

668 queue item. Every action appends a ModerationLog row. 

669 """ 

670 

671 moderation_state = session.execute( 

672 select(ModerationState).where(ModerationState.id == request.moderation_state_id) 

673 ).scalar_one_or_none() 

674 if moderation_state is None: 

675 context.abort(grpc.StatusCode.NOT_FOUND, "Moderation state not found.") 

676 

677 action = moderationaction2sql[request.action] 

678 if action is None: 678 ↛ 679line 678 didn't jump to line 679 because the condition on line 678 was never true

679 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:action_must_be_specified") 

680 

681 reason = request.reason or "Moderated by admin" 

682 object_type = moderation_state.object_type 

683 

684 data = None 

685 if request.data.strip(): 

686 if action != ModerationAction.flag: 

687 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:data_only_allowed_on_flag") 

688 try: 

689 data = json.loads(request.data) 

690 except json.JSONDecodeError: 

691 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:data_must_be_valid_json") 

692 

693 if action in (ModerationAction.approve, ModerationAction.hide): 

694 if get_moderated_models()[object_type].has_own_visibility_mechanism: 

695 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:cannot_set_visibility_on_state") 

696 

697 new_visibility = moderationvisibility2sql[request.visibility] 

698 if new_visibility is None: 698 ↛ 699line 698 didn't jump to line 699 because the condition on line 698 was never true

699 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:visibility_must_be_specified") 

700 

701 old_visibility = not_none(moderation_state.visibility) 

702 moderation_state.visibility = new_visibility 

703 moderation_state.updated = now() 

704 

705 log_entry = ModerationLog( 

706 moderation_state_id=moderation_state.id, 

707 action=action, 

708 moderator_user_id=context.user_id, 

709 new_visibility=new_visibility, 

710 reason=reason, 

711 ) 

712 session.add(log_entry) 

713 session.flush() 

714 

715 if request.clear_flags: 

716 open_items = ( 

717 session.execute( 

718 select(ModerationQueueItem) 

719 .where(ModerationQueueItem.moderation_state_id == moderation_state.id) 

720 .where(ModerationQueueItem.resolved_by_log_id.is_(None)) 

721 ) 

722 .scalars() 

723 .all() 

724 ) 

725 for queue_item in open_items: 

726 _resolve_queue_item(queue_item, log_entry, action, object_type) 

727 session.flush() 

728 

729 observe_moderation_action(action, object_type) 

730 observe_moderation_visibility_transition(old_visibility, new_visibility, object_type) 

731 

732 if new_visibility in (ModerationVisibility.visible, ModerationVisibility.unlisted): 

733 _enqueue_pending_notifications(session, moderation_state.id) 

734 

735 elif action == ModerationAction.flag: