feat: implement discussion mute/unmute feature with user and staff-level controls

lms/djangoapps/discussion/forum_integration.py

@@ -0,0 +1,290 @@
+"""
+Integration utilities for connecting edx-platform with forum models.
+This module provides a bridge between the edx-platform discussion app and
+the forum service models for muting functionality.
+"""
+
+import logging
+from typing import List, Dict, Any, Optional, Set
+from datetime import datetime
+
+from django.contrib.auth import get_user_model
+from django.db.models import Q
+from opaque_keys.edx.keys import CourseKey
+
+# Import forum models
+try:
+    from forum.api.mutes import (
+        mute_user,
+        unmute_user,
+        mute_and_report_user,
+        get_user_mute_status,
+        get_all_muted_users_for_course,
+    )
+    FORUM_MODELS_AVAILABLE = True
+except ImportError as e:
+    log.warning(f"Forum models not available: {e}")
+    FORUM_MODELS_AVAILABLE = False
+
+log = logging.getLogger(__name__)
+User = get_user_model()
+
+
+class ForumMuteService:
+    """
+    Service class to handle mute operations using forum models.
+    Uses the existing backend selection pattern based on course configuration.
+    """
+
+    @staticmethod
+    def mute_user(muted_user_id: int, muted_by_id: int, course_id: str, 
+                  scope: str = "personal", reason: str = "") -> Dict[str, Any]:
+        """
+        Mute a user using forum service.
+
+        Args:
+            muted_user_id: ID of user to mute
+            muted_by_id: ID of user performing the mute
+            course_id: Course ID where mute applies
+            scope: Mute scope ('personal' or 'course')
+            reason: Optional reason for muting
+
+        Returns:
+            Dict containing mute operation result
+        """
+        if not FORUM_MODELS_AVAILABLE:
+            raise ImportError("Forum models not available")
+
+        try:
+            result = mute_user(
+                muted_user_id=str(muted_user_id),
+                muted_by_id=str(muted_by_id),
+                course_id=course_id,
+                scope=scope,
+                reason=reason
+            )
+            return result
+        except Exception as e:
+            log.error(f"Error muting user {muted_user_id}: {e}")
+            raise
+
+    @staticmethod
+    def unmute_user(muted_user_id: int, unmuted_by_id: int, course_id: str,
+                   scope: str = "personal", muted_by_id: Optional[int] = None) -> Dict[str, Any]:
+        """
+        Unmute a user using forum service.
+
+        Args:
+            muted_user_id: ID of user to unmute
+            unmuted_by_id: ID of user performing the unmute
+            course_id: Course ID where unmute applies
+            scope: Unmute scope ('personal' or 'course')
+            muted_by_id: Original muter ID (for personal unmutes)
+
+        Returns:
+            Dict containing unmute operation result
+        """
+        if not FORUM_MODELS_AVAILABLE:
+            raise ImportError("Forum models not available")
+
+        try:
+            result = unmute_user(
+                muted_user_id=str(muted_user_id),
+                unmuted_by_id=str(unmuted_by_id),
+                course_id=course_id,
+                scope=scope,
+                muted_by_id=str(muted_by_id) if muted_by_id else None
+            )
+            return result
+        except Exception as e:
+            log.error(f"Error unmuting user {muted_user_id}: {e}")
+            raise
+
+    @staticmethod
+    def mute_and_report_user(muted_user_id: int, muted_by_id: int, course_id: str,
+                            scope: str = "personal", reason: str = "") -> Dict[str, Any]:
+        """
+        Mute and report a user using forum service.
+
+        Args:
+            muted_user_id: ID of user to mute and report
+            muted_by_id: ID of user performing the action
+            course_id: Course ID where action applies
+            scope: Mute scope ('personal' or 'course')
+            reason: Reason for muting and reporting
+
+        Returns:
+            Dict containing operation result
+        """
+        if not FORUM_MODELS_AVAILABLE:
+            raise ImportError("Forum models not available")
+
+        try:
+            result = mute_and_report_user(
+                muted_user_id=str(muted_user_id),
+                muted_by_id=str(muted_by_id),
+                course_id=course_id,
+                scope=scope,
+                reason=reason
+            )
+            return result
+        except Exception as e:
+            log.error(f"Error muting and reporting user {muted_user_id}: {e}")
+            raise
+
+    @staticmethod
+    def get_user_mute_status(user_id: int, course_id: str, 
+                            viewer_id: int) -> Dict[str, Any]:
+        """
+        Get mute status for a user using forum service.
+
+        Args:
+            user_id: ID of user to check
+            course_id: Course ID
+            viewer_id: ID of user requesting the status
+
+        Returns:
+            Dict containing mute status information
+        """
+        if not FORUM_MODELS_AVAILABLE:
+            raise ImportError("Forum models not available")
+
+        try:
+            result = get_user_mute_status(
+                user_id=str(user_id),
+                course_id=course_id,
+                viewer_id=str(viewer_id)
+            )
+            return result
+        except Exception as e:
+            log.error(f"Error getting mute status for user {user_id}: {e}")
+            raise
+
+    @staticmethod
+    def get_all_muted_users_for_course(course_id: str, requester_id: Optional[int] = None,
+                                      scope: str = "all") -> Dict[str, Any]:
+        """
+        Get all muted users in a course using forum service.
+
+        Args:
+            course_id: Course ID
+            requester_id: ID of user requesting the list
+            scope: Scope filter ('personal', 'course', or 'all')
+
+        Returns:
+            Dict containing list of muted users
+        """
+        if not FORUM_MODELS_AVAILABLE:
+            raise ImportError("Forum models not available")
+
+        try:
+            result = get_all_muted_users_for_course(
+                course_id=course_id,
+                requester_id=str(requester_id) if requester_id else None,
+                scope=scope
+            )
+            return result
+        except Exception as e:
+            log.error(f"Error getting muted users for course {course_id}: {e}")
+            raise
+
+
+class ForumIntegrationService:
+    """
+    Service class for general forum integration operations.
+    Handles backend-agnostic forum operations.
+    """
+
+    @staticmethod
+    def is_user_muted_by_viewer(target_user_id: int, viewer_id: int, course_id: str) -> bool:
+        """
+        Check if a user is muted by the viewer.
+
+        Args:
+            target_user_id: ID of the user to check
+            viewer_id: ID of the viewing user
+            course_id: Course identifier
+
+        Returns:
+            True if target user is muted by viewer, False otherwise
+        """
+        try:
+            mute_status = ForumMuteService.get_user_mute_status(
+                user_id=target_user_id,
+                course_id=course_id,
+                viewer_id=viewer_id
+            )
+            return mute_status.get('is_muted', False)
+        except Exception as e:
+            log.warning(f"Error checking mute status: {e}")
+            return False
+
+    @staticmethod
+    def get_muted_user_ids_for_course(course_id: str, viewer_id: int) -> Set[int]:
+        """
+        Get set of user IDs that are muted in a course for the given viewer.
+        Used for content filtering.
+
+        Args:
+            course_id: Course identifier
+            viewer_id: ID of the viewing user
+
+        Returns:
+            Set of user IDs that should be filtered out for this viewer
+        """
+        try:
+            # Use the forum mute service to get muted users
+            muted_ids = set()
+
+            # Get course-wide mutes (apply to all users)
+            course_mutes = ForumMuteService.get_all_muted_users_for_course(
+                course_id=course_id,
+                requester_id=None,  # No specific requester for course-wide
+                scope="course"
+            )
+            course_muted_ids = {int(user['muted_user_id']) for user in course_mutes.get('muted_users', [])}
+            muted_ids.update(course_muted_ids)
+
+            # Get personal mutes done by this specific viewer
+            personal_mutes = ForumMuteService.get_all_muted_users_for_course(
+                course_id=course_id,
+                requester_id=viewer_id,
+                scope="personal"
+            )
+            # Filter to only include mutes done by this specific viewer
+            personal_muted_ids = set()
+            for user in personal_mutes.get('muted_users', []):
+                muted_by_id = user.get('muted_by_id')
+                muted_user_id = user.get('muted_user_id')
+                # Ensure both IDs are converted to int for comparison
+                try:
+                    muted_by_id = int(muted_by_id) if muted_by_id is not None else None
+                    muted_user_id = int(muted_user_id) if muted_user_id is not None else None
+
+                    if muted_by_id == viewer_id and muted_user_id is not None:
+                        personal_muted_ids.add(muted_user_id)
+                except (ValueError, TypeError):
+                    # Skip invalid data
+                    continue
+
+            muted_ids.update(personal_muted_ids)
+
+            # Ensure the viewer's own ID is never included in the muted list
+            # since users cannot mute themselves (self-mute prevention)
+            muted_ids.discard(viewer_id)
+
+            return muted_ids
+        except Exception as e:
+            log.warning(f"Error getting muted user IDs: {e}")
+            return set()
+
+
+# Legacy function aliases for backward compatibility
+def is_user_muted(target_user_id: int, viewer_id: int, course_id: str) -> bool:
+    """Legacy function - use ForumIntegrationService.is_user_muted_by_viewer instead."""
+    return ForumIntegrationService.is_user_muted_by_viewer(target_user_id, viewer_id, course_id)
+
+
+def get_muted_user_ids(course_id: str, viewer_id: int) -> Set[int]:
+    """Legacy function - use ForumIntegrationService.get_muted_user_ids_for_course instead."""
+    return ForumIntegrationService.get_muted_user_ids_for_course(course_id, viewer_id)