Coverage for app/backend/src/couchers/servicers/moderation.py: 93%
265 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +0000
1import logging
3import grpc
4from sqlalchemy import and_, exists, not_, or_, select
5from sqlalchemy.orm import Session
7from couchers.context import CouchersContext
8from couchers.jobs.enqueue import queue_job
9from couchers.metrics import (
10 observe_moderation_action,
11 observe_moderation_queue_item_created,
12 observe_moderation_queue_item_resolved,
13 observe_moderation_queue_resolution_time,
14 observe_moderation_visibility_transition,
15)
16from couchers.models import (
17 AdminActionLevel,
18 Comment,
19 Discussion,
20 Event,
21 EventOccurrence,
22 FriendRelationship,
23 GroupChat,
24 HostRequest,
25 Message,
26 MessageType,
27 ModerationAction,
28 ModerationLog,
29 ModerationObjectType,
30 ModerationQueueItem,
31 ModerationState,
32 ModerationTrigger,
33 ModerationVisibility,
34 Notification,
35 NotificationDelivery,
36 PublicTrip,
37 Reference,
38 Reply,
39 User,
40 get_moderated_models,
41)
42from couchers.proto import moderation_pb2, moderation_pb2_grpc
43from couchers.proto.internal import jobs_pb2
44from couchers.utils import Timestamp_from_datetime, now
46logger = logging.getLogger(__name__)
48MAX_PAGINATION_LENGTH = 1_000
50# Moderation enum mappings
51moderationvisibility2api = {
52 None: moderation_pb2.MODERATION_VISIBILITY_UNSPECIFIED,
53 ModerationVisibility.hidden: moderation_pb2.MODERATION_VISIBILITY_HIDDEN,
54 ModerationVisibility.shadowed: moderation_pb2.MODERATION_VISIBILITY_SHADOWED,
55 ModerationVisibility.visible: moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
56 ModerationVisibility.unlisted: moderation_pb2.MODERATION_VISIBILITY_UNLISTED,
57}
59moderationvisibility2sql = {
60 moderation_pb2.MODERATION_VISIBILITY_UNSPECIFIED: None,
61 moderation_pb2.MODERATION_VISIBILITY_HIDDEN: ModerationVisibility.hidden,
62 moderation_pb2.MODERATION_VISIBILITY_SHADOWED: ModerationVisibility.shadowed,
63 moderation_pb2.MODERATION_VISIBILITY_VISIBLE: ModerationVisibility.visible,
64 moderation_pb2.MODERATION_VISIBILITY_UNLISTED: ModerationVisibility.unlisted,
65}
67moderationtrigger2api = {
68 None: moderation_pb2.MODERATION_TRIGGER_UNSPECIFIED,
69 ModerationTrigger.initial_review: moderation_pb2.MODERATION_TRIGGER_INITIAL_REVIEW,
70 ModerationTrigger.user_flag: moderation_pb2.MODERATION_TRIGGER_USER_FLAG,
71 ModerationTrigger.machine_flag: moderation_pb2.MODERATION_TRIGGER_MACHINE_FLAG,
72 ModerationTrigger.moderator_review: moderation_pb2.MODERATION_TRIGGER_MODERATOR_REVIEW,
73}
75moderationtrigger2sql = {
76 moderation_pb2.MODERATION_TRIGGER_UNSPECIFIED: None,
77 moderation_pb2.MODERATION_TRIGGER_INITIAL_REVIEW: ModerationTrigger.initial_review,
78 moderation_pb2.MODERATION_TRIGGER_USER_FLAG: ModerationTrigger.user_flag,
79 moderation_pb2.MODERATION_TRIGGER_MACHINE_FLAG: ModerationTrigger.machine_flag,
80 moderation_pb2.MODERATION_TRIGGER_MODERATOR_REVIEW: ModerationTrigger.moderator_review,
81}
83moderationaction2api = {
84 None: moderation_pb2.MODERATION_ACTION_UNSPECIFIED,
85 ModerationAction.create: moderation_pb2.MODERATION_ACTION_CREATE,
86 ModerationAction.approve: moderation_pb2.MODERATION_ACTION_APPROVE,
87 ModerationAction.hide: moderation_pb2.MODERATION_ACTION_HIDE,
88 ModerationAction.flag: moderation_pb2.MODERATION_ACTION_FLAG,
89 ModerationAction.unflag: moderation_pb2.MODERATION_ACTION_UNFLAG,
90 ModerationAction.set_priority: moderation_pb2.MODERATION_ACTION_SET_PRIORITY,
91 ModerationAction.bulk_set_visibility: moderation_pb2.MODERATION_ACTION_BULK_SET_VISIBILITY,
92}
94moderationaction2sql = {
95 moderation_pb2.MODERATION_ACTION_UNSPECIFIED: None,
96 moderation_pb2.MODERATION_ACTION_CREATE: ModerationAction.create,
97 moderation_pb2.MODERATION_ACTION_APPROVE: ModerationAction.approve,
98 moderation_pb2.MODERATION_ACTION_HIDE: ModerationAction.hide,
99 moderation_pb2.MODERATION_ACTION_FLAG: ModerationAction.flag,
100 moderation_pb2.MODERATION_ACTION_UNFLAG: ModerationAction.unflag,
101 moderation_pb2.MODERATION_ACTION_SET_PRIORITY: ModerationAction.set_priority,
102 moderation_pb2.MODERATION_ACTION_BULK_SET_VISIBILITY: ModerationAction.bulk_set_visibility,
103}
105moderationobjecttype2api = {
106 None: moderation_pb2.MODERATION_OBJECT_TYPE_UNSPECIFIED,
107 ModerationObjectType.host_request: moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST,
108 ModerationObjectType.group_chat: moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT,
109 ModerationObjectType.friend_request: moderation_pb2.MODERATION_OBJECT_TYPE_FRIEND_REQUEST,
110 ModerationObjectType.event_occurrence: moderation_pb2.MODERATION_OBJECT_TYPE_EVENT_OCCURRENCE,
111 ModerationObjectType.comment: moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT,
112 ModerationObjectType.reply: moderation_pb2.MODERATION_OBJECT_TYPE_REPLY,
113 ModerationObjectType.discussion: moderation_pb2.MODERATION_OBJECT_TYPE_DISCUSSION,
114 ModerationObjectType.reference: moderation_pb2.MODERATION_OBJECT_TYPE_REFERENCE,
115 ModerationObjectType.public_trip: moderation_pb2.MODERATION_OBJECT_TYPE_PUBLIC_TRIP,
116}
118moderationobjecttype2sql = {
119 moderation_pb2.MODERATION_OBJECT_TYPE_UNSPECIFIED: None,
120 moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST: ModerationObjectType.host_request,
121 moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT: ModerationObjectType.group_chat,
122 moderation_pb2.MODERATION_OBJECT_TYPE_FRIEND_REQUEST: ModerationObjectType.friend_request,
123 moderation_pb2.MODERATION_OBJECT_TYPE_EVENT_OCCURRENCE: ModerationObjectType.event_occurrence,
124 moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT: ModerationObjectType.comment,
125 moderation_pb2.MODERATION_OBJECT_TYPE_REPLY: ModerationObjectType.reply,
126 moderation_pb2.MODERATION_OBJECT_TYPE_DISCUSSION: ModerationObjectType.discussion,
127 moderation_pb2.MODERATION_OBJECT_TYPE_REFERENCE: ModerationObjectType.reference,
128 moderation_pb2.MODERATION_OBJECT_TYPE_PUBLIC_TRIP: ModerationObjectType.public_trip,
129}
132def _resolve_queue_item(
133 queue_item: ModerationQueueItem,
134 log_entry: ModerationLog,
135 action: ModerationAction,
136 object_type: ModerationObjectType,
137) -> None:
138 """Resolve an open queue item against the given log entry and record metrics."""
139 queue_item.resolved_by_log_id = log_entry.id
140 observe_moderation_queue_item_resolved(queue_item.trigger, action, object_type)
141 observe_moderation_queue_resolution_time(
142 queue_item.trigger,
143 action,
144 object_type,
145 (now() - queue_item.time_created).total_seconds(),
146 )
149def bulk_set_user_content_visibility(
150 session: Session,
151 user: User,
152 new_visibility: ModerationVisibility,
153 moderator_user_id: int,
154 from_visibilities: set[ModerationVisibility] | None = None,
155 reason: str | None = None,
156) -> int:
157 """Set visibility on every UMS-governed object authored by the user. Returns count of updated states."""
158 final_reason = reason or f"Bulk visibility update for user {user.id} to {new_visibility.name}"
160 author_exists_clauses = []
161 for entry in get_moderated_models().values():
162 author_exists_clauses.append(
163 exists().where(and_(entry.moderation_state_id_column == ModerationState.id, entry.author_column == user.id))
164 )
166 states = session.execute(select(ModerationState).where(or_(*author_exists_clauses))).scalars().all()
168 updated_count = 0
169 for moderation_state in states:
170 if from_visibilities and moderation_state.visibility not in from_visibilities:
171 continue
172 if moderation_state.visibility == new_visibility:
173 continue
175 old_visibility = moderation_state.visibility
176 moderation_state.visibility = new_visibility
177 moderation_state.updated = now()
179 log_entry = ModerationLog(
180 moderation_state_id=moderation_state.id,
181 action=ModerationAction.bulk_set_visibility,
182 moderator_user_id=moderator_user_id,
183 new_visibility=new_visibility,
184 reason=final_reason,
185 )
186 session.add(log_entry)
187 session.flush()
189 open_items = (
190 session.execute(
191 select(ModerationQueueItem)
192 .where(ModerationQueueItem.moderation_state_id == moderation_state.id)
193 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
194 )
195 .scalars()
196 .all()
197 )
198 for queue_item in open_items:
199 _resolve_queue_item(
200 queue_item, log_entry, ModerationAction.bulk_set_visibility, moderation_state.object_type
201 )
202 session.flush()
204 observe_moderation_action(ModerationAction.bulk_set_visibility, moderation_state.object_type)
205 observe_moderation_visibility_transition(old_visibility, new_visibility, moderation_state.object_type)
207 if new_visibility in (ModerationVisibility.visible, ModerationVisibility.unlisted):
208 _enqueue_pending_notifications(session, moderation_state.id)
210 updated_count += 1
212 return updated_count
215def _enqueue_pending_notifications(session: Session, moderation_state_id: int) -> None:
216 """Re-queue any pending notifications linked to the given moderation state whose deliveries were suppressed."""
217 pending_notifications = (
218 session.execute(
219 select(Notification)
220 .where(Notification.moderation_state_id == moderation_state_id)
221 .where(not_(exists().where(NotificationDelivery.notification_id == Notification.id)))
222 )
223 .scalars()
224 .all()
225 )
227 # Import here to avoid circular dependency
228 from couchers.notifications.background import handle_notification # noqa: PLC0415
230 for notification in pending_notifications:
231 queue_job(
232 session,
233 job=handle_notification,
234 payload=jobs_pb2.HandleNotificationPayload(notification_id=notification.id),
235 )
238def moderation_state_to_pb(state: ModerationState, session: Session) -> moderation_pb2.ModerationStateInfo:
239 """Convert ModerationState model to proto message"""
240 object_type = state.object_type
241 object_id = state.object_id
243 # Get the author user ID and content based on object type
244 if object_type == ModerationObjectType.host_request:
245 author_user_id = session.execute(
246 select(HostRequest.initiator_user_id).where(HostRequest.conversation_id == object_id)
247 ).scalar_one()
248 # Get the first text message for this conversation
249 content = session.execute(
250 select(Message.text)
251 .where(Message.conversation_id == object_id)
252 .where(Message.message_type == MessageType.text)
253 .order_by(Message.id.asc())
254 .limit(1)
255 ).scalar_one_or_none()
256 elif object_type == ModerationObjectType.group_chat:
257 author_user_id = session.execute(
258 select(GroupChat.creator_id).where(GroupChat.conversation_id == object_id)
259 ).scalar_one()
260 # Get the first text message for this conversation
261 content = session.execute(
262 select(Message.text)
263 .where(Message.conversation_id == object_id)
264 .where(Message.message_type == MessageType.text)
265 .order_by(Message.id.asc())
266 .limit(1)
267 ).scalar_one_or_none()
268 elif object_type == ModerationObjectType.friend_request:
269 author_user_id = session.execute(
270 select(FriendRelationship.from_user_id).where(FriendRelationship.id == object_id)
271 ).scalar_one()
272 # Friend requests have no text content
273 content = None
274 elif object_type == ModerationObjectType.event_occurrence:
275 author_user_id, title, description = session.execute(
276 select(EventOccurrence.creator_user_id, Event.title, EventOccurrence.content)
277 .join(Event, Event.id == EventOccurrence.event_id)
278 .where(EventOccurrence.id == object_id)
279 ).one()
280 content = f"{title}\n\n{description}"
281 elif object_type == ModerationObjectType.comment:
282 author_user_id, content = session.execute(
283 select(Comment.author_user_id, Comment.content).where(Comment.id == object_id)
284 ).one()
285 elif object_type == ModerationObjectType.reply:
286 author_user_id, content = session.execute(
287 select(Reply.author_user_id, Reply.content).where(Reply.id == object_id)
288 ).one()
289 elif object_type == ModerationObjectType.discussion:
290 author_user_id, title, body = session.execute(
291 select(Discussion.creator_user_id, Discussion.title, Discussion.content).where(Discussion.id == object_id)
292 ).one()
293 content = f"{title}\n\n{body}"
294 elif object_type == ModerationObjectType.reference:
295 author_user_id, content = session.execute(
296 select(Reference.from_user_id, Reference.text).where(Reference.id == object_id)
297 ).one()
298 elif object_type == ModerationObjectType.public_trip: 298 ↛ 303line 298 didn't jump to line 303 because the condition on line 298 was always true
299 author_user_id, content = session.execute(
300 select(PublicTrip.user_id, PublicTrip.description).where(PublicTrip.id == object_id)
301 ).one()
302 else:
303 raise ValueError(f"Unsupported moderation object type: {object_type}")
305 # Import here to avoid circular dependency
306 from couchers.servicers.admin import _user_to_details # noqa: PLC0415
308 author = session.execute(select(User).where(User.id == author_user_id)).scalar_one()
310 state_pb = moderation_pb2.ModerationStateInfo(
311 moderation_state_id=state.id,
312 object_type=moderationobjecttype2api[state.object_type],
313 object_id=state.object_id,
314 visibility=moderationvisibility2api[state.visibility],
315 created=Timestamp_from_datetime(state.created),
316 updated=Timestamp_from_datetime(state.updated),
317 author_user_id=author_user_id,
318 author=_user_to_details(session, author),
319 content=content or "",
320 )
322 return state_pb
325class Moderation(moderation_pb2_grpc.ModerationServicer):
326 def GetModerationQueue(
327 self, request: moderation_pb2.GetModerationQueueReq, context: CouchersContext, session: Session
328 ) -> moderation_pb2.GetModerationQueueRes:
329 """Get moderation queue items with optional filtering"""
331 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
333 # Build query
334 statement = select(ModerationQueueItem)
336 # Apply page token filter based on ordering direction
337 if request.page_token:
338 page_token_id = int(request.page_token)
339 if request.newest_first: 339 ↛ 344line 339 didn't jump to line 344 because the condition on line 339 was always true
340 # Descending order: get items with smaller IDs
341 statement = statement.where(ModerationQueueItem.id < page_token_id)
342 else:
343 # Ascending order: get items with larger IDs
344 statement = statement.where(ModerationQueueItem.id > page_token_id)
346 # Apply filters
347 if request.triggers:
348 internal_triggers = [moderationtrigger2sql[t] for t in request.triggers]
349 statement = statement.where(ModerationQueueItem.trigger.in_(internal_triggers))
351 if request.object_type and request.object_type != moderation_pb2.MODERATION_OBJECT_TYPE_UNSPECIFIED: 351 ↛ 352line 351 didn't jump to line 352 because the condition on line 351 was never true
352 internal_object_type = moderationobjecttype2sql[request.object_type]
353 if internal_object_type:
354 statement = statement.join(ModerationState).where(ModerationState.object_type == internal_object_type)
356 if request.unresolved_only:
357 statement = statement.where(ModerationQueueItem.resolved_by_log_id.is_(None))
359 if request.HasField("priority_min"):
360 statement = statement.where(ModerationQueueItem.priority >= request.priority_min)
362 if request.HasField("priority_max"):
363 statement = statement.where(ModerationQueueItem.priority <= request.priority_max)
365 if request.HasField("created_before"):
366 created_before = request.created_before.ToDatetime()
367 statement = statement.where(ModerationQueueItem.time_created < created_before)
369 if request.HasField("created_after"):
370 created_after = request.created_after.ToDatetime()
371 statement = statement.where(ModerationQueueItem.time_created > created_after)
373 if request.item_author_user_id:
374 author_user_id = request.item_author_user_id
376 # Use EXISTS for efficient author filtering
377 author_exists_clauses = []
378 for entry in get_moderated_models().values():
379 author_exists_clauses.append(
380 exists().where(
381 and_(
382 entry.moderation_state_id_column == ModerationQueueItem.moderation_state_id,
383 entry.author_column == author_user_id,
384 )
385 )
386 )
387 statement = statement.where(or_(*author_exists_clauses))
389 # Order by time created
390 if request.newest_first:
391 statement = statement.order_by(ModerationQueueItem.time_created.desc(), ModerationQueueItem.id.desc())
392 else:
393 statement = statement.order_by(ModerationQueueItem.time_created.asc(), ModerationQueueItem.id.asc())
395 queue_items = session.execute(statement.limit(page_size + 1)).scalars().all()
397 # Convert to proto
398 queue_items_pb = []
399 for item in queue_items[:page_size]:
400 # Fetch the moderation state for this queue item
401 mod_state = session.execute(
402 select(ModerationState).where(ModerationState.id == item.moderation_state_id)
403 ).scalar_one()
405 queue_item_pb = moderation_pb2.ModerationQueueItemInfo(
406 queue_item_id=item.id,
407 moderation_state_id=item.moderation_state_id,
408 time_created=Timestamp_from_datetime(item.time_created),
409 trigger=moderationtrigger2api[item.trigger],
410 reason=item.reason,
411 is_resolved=item.resolved_by_log_id is not None,
412 resolved_by_log_id=item.resolved_by_log_id or 0,
413 moderation_state=moderation_state_to_pb(mod_state, session),
414 priority=item.priority,
415 )
417 queue_items_pb.append(queue_item_pb)
419 return moderation_pb2.GetModerationQueueRes(
420 queue_items=queue_items_pb,
421 # Use the ID of the last returned item (not the extra fetched item) as the cursor
422 next_page_token=str(queue_items[page_size - 1].id) if len(queue_items) > page_size else None,
423 )
425 def GetModerationState(
426 self, request: moderation_pb2.GetModerationStateReq, context: CouchersContext, session: Session
427 ) -> moderation_pb2.GetModerationStateRes:
428 """Get moderation state by object type and object ID"""
429 object_type = moderationobjecttype2sql[request.object_type]
430 if object_type is None:
431 context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Object type must be specified.")
433 moderation_state = session.execute(
434 select(ModerationState)
435 .where(ModerationState.object_type == object_type)
436 .where(ModerationState.object_id == request.object_id)
437 ).scalar_one_or_none()
438 if moderation_state is None:
439 context.abort(grpc.StatusCode.NOT_FOUND, "Moderation state not found.")
441 return moderation_pb2.GetModerationStateRes(
442 moderation_state=moderation_state_to_pb(moderation_state, session),
443 )
445 def GetModerationLog(
446 self, request: moderation_pb2.GetModerationLogReq, context: CouchersContext, session: Session
447 ) -> moderation_pb2.GetModerationLogRes:
448 """Get moderation log for a specific moderation state"""
449 # Get the moderation state
450 moderation_state = session.execute(
451 select(ModerationState).where(ModerationState.id == request.moderation_state_id)
452 ).scalar_one_or_none()
453 if moderation_state is None:
454 context.abort(grpc.StatusCode.NOT_FOUND, "Moderation state not found.")
456 # Get all log entries for this state, ordered by time (most recent first)
457 log_entries = (
458 session.execute(
459 select(ModerationLog)
460 .where(ModerationLog.moderation_state_id == request.moderation_state_id)
461 .order_by(ModerationLog.time.desc(), ModerationLog.id.desc())
462 )
463 .scalars()
464 .all()
465 )
467 # Convert moderation state to proto first (while still in session)
468 moderation_state_pb = moderation_state_to_pb(moderation_state, session)
470 # Convert to proto
471 log_entries_pb = []
472 for entry in log_entries:
473 log_entry_pb = moderation_pb2.ModerationLogEntryInfo(
474 log_entry_id=entry.id,
475 moderation_state_id=entry.moderation_state_id,
476 time=Timestamp_from_datetime(entry.time),
477 action=moderationaction2api[entry.action],
478 moderator_user_id=entry.moderator_user_id,
479 reason=entry.reason,
480 )
482 # Only include changed fields
483 if entry.new_visibility is not None:
484 log_entry_pb.new_visibility = moderationvisibility2api[entry.new_visibility]
485 if entry.new_priority is not None: 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true
486 log_entry_pb.new_priority = entry.new_priority
487 if entry.queue_item_id is not None:
488 log_entry_pb.queue_item_id = entry.queue_item_id
490 log_entries_pb.append(log_entry_pb)
492 return moderation_pb2.GetModerationLogRes(
493 log_entries=log_entries_pb,
494 moderation_state=moderation_state_pb,
495 )
497 def ModerateContent(
498 self, request: moderation_pb2.ModerateContentReq, context: CouchersContext, session: Session
499 ) -> moderation_pb2.ModerateContentRes:
500 """Single moderation entrypoint, dispatching on action.
502 APPROVE/HIDE act on the state's visibility; FLAG/SET_PRIORITY/UNFLAG act on a single
503 queue item. Every action appends a ModerationLog row.
504 """
506 moderation_state = session.execute(
507 select(ModerationState).where(ModerationState.id == request.moderation_state_id)
508 ).scalar_one_or_none()
509 if moderation_state is None:
510 context.abort(grpc.StatusCode.NOT_FOUND, "Moderation state not found.")
512 action = moderationaction2sql[request.action]
513 if action is None: 513 ↛ 514line 513 didn't jump to line 514 because the condition on line 513 was never true
514 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:action_must_be_specified")
516 reason = request.reason or "Moderated by admin"
517 object_type = moderation_state.object_type
519 if action in (ModerationAction.approve, ModerationAction.hide):
520 new_visibility = moderationvisibility2sql[request.visibility]
521 if new_visibility is None: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true
522 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:visibility_must_be_specified")
524 old_visibility = moderation_state.visibility
525 moderation_state.visibility = new_visibility
526 moderation_state.updated = now()
528 log_entry = ModerationLog(
529 moderation_state_id=moderation_state.id,
530 action=action,
531 moderator_user_id=context.user_id,
532 new_visibility=new_visibility,
533 reason=reason,
534 )
535 session.add(log_entry)
536 session.flush()
538 if request.clear_flags:
539 open_items = (
540 session.execute(
541 select(ModerationQueueItem)
542 .where(ModerationQueueItem.moderation_state_id == moderation_state.id)
543 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
544 )
545 .scalars()
546 .all()
547 )
548 for queue_item in open_items:
549 _resolve_queue_item(queue_item, log_entry, action, object_type)
550 session.flush()
552 observe_moderation_action(action, object_type)
553 observe_moderation_visibility_transition(old_visibility, new_visibility, object_type)
555 if new_visibility in (ModerationVisibility.visible, ModerationVisibility.unlisted):
556 _enqueue_pending_notifications(session, moderation_state.id)
558 elif action == ModerationAction.flag:
559 trigger = moderationtrigger2sql[request.trigger]
560 if trigger is None:
561 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:trigger_must_be_specified")
563 queue_item = ModerationQueueItem(
564 moderation_state_id=moderation_state.id,
565 trigger=trigger,
566 reason=reason,
567 priority=request.priority,
568 )
569 session.add(queue_item)
570 session.flush()
572 log_entry = ModerationLog(
573 moderation_state_id=moderation_state.id,
574 action=ModerationAction.flag,
575 moderator_user_id=context.user_id,
576 queue_item_id=queue_item.id,
577 reason=reason,
578 )
579 session.add(log_entry)
580 session.flush()
582 if request.supersede_queue_item_id:
583 superseded = session.execute(
584 select(ModerationQueueItem)
585 .where(ModerationQueueItem.id == request.supersede_queue_item_id)
586 .where(ModerationQueueItem.moderation_state_id == moderation_state.id)
587 ).scalar_one_or_none()
588 if superseded is None: 588 ↛ 589line 588 didn't jump to line 589 because the condition on line 588 was never true
589 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:queue_item_not_found")
590 if superseded.resolved_by_log_id is None: 590 ↛ 594line 590 didn't jump to line 594 because the condition on line 590 was always true
591 _resolve_queue_item(superseded, log_entry, ModerationAction.flag, object_type)
592 session.flush()
594 observe_moderation_action(ModerationAction.flag, object_type)
595 observe_moderation_queue_item_created(trigger, object_type)
597 elif action == ModerationAction.set_priority:
598 queue_item = self._get_queue_item_for_state(request, context, session, moderation_state.id)
599 queue_item.priority = request.priority
601 log_entry = ModerationLog(
602 moderation_state_id=moderation_state.id,
603 action=ModerationAction.set_priority,
604 moderator_user_id=context.user_id,
605 queue_item_id=queue_item.id,
606 new_priority=request.priority,
607 reason=reason,
608 )
609 session.add(log_entry)
610 session.flush()
612 observe_moderation_action(ModerationAction.set_priority, object_type)
614 elif action == ModerationAction.unflag: 614 ↛ 635line 614 didn't jump to line 635 because the condition on line 614 was always true
615 queue_item = self._get_queue_item_for_state(request, context, session, moderation_state.id)
616 moderation_state.updated = now()
618 log_entry = ModerationLog(
619 moderation_state_id=moderation_state.id,
620 action=ModerationAction.unflag,
621 moderator_user_id=context.user_id,
622 queue_item_id=queue_item.id,
623 reason=reason,
624 )
625 session.add(log_entry)
626 session.flush()
628 if queue_item.resolved_by_log_id is None: 628 ↛ 632line 628 didn't jump to line 632 because the condition on line 628 was always true
629 _resolve_queue_item(queue_item, log_entry, ModerationAction.unflag, object_type)
630 session.flush()
632 observe_moderation_action(ModerationAction.unflag, object_type)
634 else:
635 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:unsupported_action")
637 return moderation_pb2.ModerateContentRes(
638 moderation_state=moderation_state_to_pb(moderation_state, session),
639 )
641 def _get_queue_item_for_state(
642 self,
643 request: moderation_pb2.ModerateContentReq,
644 context: CouchersContext,
645 session: Session,
646 moderation_state_id: int,
647 ) -> ModerationQueueItem:
648 """Resolve the request's target queue item, asserting it belongs to the given state."""
649 if not request.queue_item_id:
650 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:queue_item_id_must_be_specified")
651 queue_item = session.execute(
652 select(ModerationQueueItem)
653 .where(ModerationQueueItem.id == request.queue_item_id)
654 .where(ModerationQueueItem.moderation_state_id == moderation_state_id)
655 ).scalar_one_or_none()
656 if queue_item is None: 656 ↛ 657line 656 didn't jump to line 657 because the condition on line 656 was never true
657 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:queue_item_not_found")
658 return queue_item
660 def SetUserContentVisibility(
661 self, request: moderation_pb2.SetUserContentVisibilityReq, context: CouchersContext, session: Session
662 ) -> moderation_pb2.SetUserContentVisibilityRes:
663 """Bulk-set visibility on every UMS-governed object authored by the given user.
665 If from_visibility is non-empty, only states currently at one of those visibilities are swept.
666 """
667 new_visibility = moderationvisibility2sql[request.visibility]
668 if new_visibility is None:
669 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:visibility_must_be_specified")
671 raw_from_visibilities = {moderationvisibility2sql.get(v) for v in request.from_visibility}
672 if None in raw_from_visibilities:
673 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:visibility_must_be_specified")
674 from_visibilities: set[ModerationVisibility] | None = {
675 v for v in raw_from_visibilities if v is not None
676 } or None
678 user = session.execute(select(User).where(User.id == request.user_id)).scalar_one_or_none()
679 if not user:
680 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
682 updated_count = bulk_set_user_content_visibility(
683 session=session,
684 user=user,
685 new_visibility=new_visibility,
686 moderator_user_id=context.user_id,
687 from_visibilities=from_visibilities,
688 reason=request.reason or None,
689 )
691 # Import here to avoid circular dependency
692 from couchers.servicers.admin import log_admin_action # noqa: PLC0415
694 log_admin_action(
695 session,
696 context,
697 user,
698 "set_user_content_visibility",
699 note=request.reason or None,
700 tag=new_visibility.name,
701 level=AdminActionLevel.high,
702 )
704 return moderation_pb2.SetUserContentVisibilityRes(updated_count=updated_count)
706 def ListModerationStates(
707 self, request: moderation_pb2.ListModerationStatesReq, context: CouchersContext, session: Session
708 ) -> moderation_pb2.ListModerationStatesRes:
709 """Chronological, paginated list of ModerationState rows. Optional author_user_id filter."""
710 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
712 statement = select(ModerationState)
714 if request.page_token:
715 page_token_id = int(request.page_token)
716 if request.newest_first: 716 ↛ 717line 716 didn't jump to line 717 because the condition on line 716 was never true
717 statement = statement.where(ModerationState.id < page_token_id)
718 else:
719 statement = statement.where(ModerationState.id > page_token_id)
721 if request.author_user_id:
722 author_exists_clauses = []
723 for entry in get_moderated_models().values():
724 author_exists_clauses.append(
725 exists().where(
726 and_(
727 entry.moderation_state_id_column == ModerationState.id,
728 entry.author_column == request.author_user_id,
729 )
730 )
731 )
732 statement = statement.where(or_(*author_exists_clauses))
734 if request.newest_first:
735 statement = statement.order_by(ModerationState.created.desc(), ModerationState.id.desc())
736 else:
737 statement = statement.order_by(ModerationState.created.asc(), ModerationState.id.asc())
739 states = session.execute(statement.limit(page_size + 1)).scalars().all()
741 state_pbs = [moderation_state_to_pb(state, session) for state in states[:page_size]]
743 return moderation_pb2.ListModerationStatesRes(
744 moderation_states=state_pbs,
745 next_page_token=str(states[page_size - 1].id) if len(states) > page_size else None,
746 )