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