mirror of
https://github.com/zulip/zulip.git
synced 2025-11-03 21:43:21 +00:00
915c2b2fd9
Rather than check if the topic exists and then try to delete it, just try to delete it, and catch the lack of matching rows.
119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
import datetime
|
|
from typing import Optional
|
|
|
|
from django.http import HttpRequest, HttpResponse
|
|
from django.utils.timezone import now as timezone_now
|
|
from django.utils.translation import gettext as _
|
|
|
|
from zerver.lib.actions import do_mute_topic, do_mute_user, do_unmute_topic, do_unmute_user
|
|
from zerver.lib.exceptions import JsonableError
|
|
from zerver.lib.request import REQ, has_request_variables
|
|
from zerver.lib.response import json_success
|
|
from zerver.lib.streams import (
|
|
access_stream_by_id,
|
|
access_stream_by_name,
|
|
access_stream_for_unmute_topic_by_id,
|
|
access_stream_for_unmute_topic_by_name,
|
|
check_for_exactly_one_stream_arg,
|
|
)
|
|
from zerver.lib.topic_mutes import topic_is_muted
|
|
from zerver.lib.user_mutes import get_mute_object
|
|
from zerver.lib.users import access_user_by_id
|
|
from zerver.lib.validator import check_int
|
|
from zerver.models import UserProfile
|
|
|
|
|
|
def mute_topic(
|
|
user_profile: UserProfile,
|
|
stream_id: Optional[int],
|
|
stream_name: Optional[str],
|
|
topic_name: str,
|
|
date_muted: datetime.datetime,
|
|
) -> HttpResponse:
|
|
if stream_name is not None:
|
|
(stream, sub) = access_stream_by_name(user_profile, stream_name)
|
|
else:
|
|
assert stream_id is not None
|
|
(stream, sub) = access_stream_by_id(user_profile, stream_id)
|
|
|
|
if topic_is_muted(user_profile, stream.id, topic_name):
|
|
raise JsonableError(_("Topic already muted"))
|
|
|
|
do_mute_topic(user_profile, stream, topic_name, date_muted)
|
|
return json_success()
|
|
|
|
|
|
def unmute_topic(
|
|
user_profile: UserProfile, stream_id: Optional[int], stream_name: Optional[str], topic_name: str
|
|
) -> HttpResponse:
|
|
error = _("Topic is not muted")
|
|
|
|
if stream_name is not None:
|
|
stream = access_stream_for_unmute_topic_by_name(user_profile, stream_name, error)
|
|
else:
|
|
assert stream_id is not None
|
|
stream = access_stream_for_unmute_topic_by_id(user_profile, stream_id, error)
|
|
|
|
do_unmute_topic(user_profile, stream, topic_name)
|
|
return json_success()
|
|
|
|
|
|
@has_request_variables
|
|
def update_muted_topic(
|
|
request: HttpRequest,
|
|
user_profile: UserProfile,
|
|
stream_id: Optional[int] = REQ(json_validator=check_int, default=None),
|
|
stream: Optional[str] = REQ(default=None),
|
|
topic: str = REQ(),
|
|
op: str = REQ(),
|
|
) -> HttpResponse:
|
|
|
|
check_for_exactly_one_stream_arg(stream_id=stream_id, stream=stream)
|
|
|
|
if op == "add":
|
|
return mute_topic(
|
|
user_profile=user_profile,
|
|
stream_id=stream_id,
|
|
stream_name=stream,
|
|
topic_name=topic,
|
|
date_muted=timezone_now(),
|
|
)
|
|
elif op == "remove":
|
|
return unmute_topic(
|
|
user_profile=user_profile,
|
|
stream_id=stream_id,
|
|
stream_name=stream,
|
|
topic_name=topic,
|
|
)
|
|
|
|
|
|
def mute_user(request: HttpRequest, user_profile: UserProfile, muted_user_id: int) -> HttpResponse:
|
|
if user_profile.id == muted_user_id:
|
|
raise JsonableError(_("Cannot mute self"))
|
|
|
|
muted_user = access_user_by_id(
|
|
user_profile, muted_user_id, allow_bots=False, allow_deactivated=True, for_admin=False
|
|
)
|
|
date_muted = timezone_now()
|
|
|
|
if get_mute_object(user_profile, muted_user) is not None:
|
|
raise JsonableError(_("User already muted"))
|
|
|
|
do_mute_user(user_profile, muted_user, date_muted)
|
|
return json_success()
|
|
|
|
|
|
def unmute_user(
|
|
request: HttpRequest, user_profile: UserProfile, muted_user_id: int
|
|
) -> HttpResponse:
|
|
muted_user = access_user_by_id(
|
|
user_profile, muted_user_id, allow_bots=False, allow_deactivated=True, for_admin=False
|
|
)
|
|
mute_object = get_mute_object(user_profile, muted_user)
|
|
|
|
if mute_object is None:
|
|
raise JsonableError(_("User is not muted"))
|
|
|
|
do_unmute_user(mute_object)
|
|
return json_success()
|