mirror of
https://github.com/zulip/zulip.git
synced 2025-11-03 21:43:21 +00:00
85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
from collections.abc import Collection
|
|
from typing import Any, Protocol
|
|
|
|
from django.utils.translation import gettext as _
|
|
|
|
from zerver.lib.exceptions import JsonableError
|
|
from zerver.lib.narrow_helpers import NarrowTerm
|
|
from zerver.lib.topic import RESOLVED_TOPIC_PREFIX, get_topic_from_message_info
|
|
|
|
# "stream" is a legacy alias for "channel"
|
|
channel_operators: list[str] = ["channel", "stream"]
|
|
# "streams" is a legacy alias for "channels"
|
|
channels_operators: list[str] = ["channels", "streams"]
|
|
|
|
|
|
def check_narrow_for_events(narrow: Collection[NarrowTerm]) -> None:
|
|
supported_operators = [*channel_operators, "topic", "sender", "is"]
|
|
unsupported_is_operands = ["followed"]
|
|
for narrow_term in narrow:
|
|
operator = narrow_term.operator
|
|
if operator not in supported_operators:
|
|
raise JsonableError(_("Operator {operator} not supported.").format(operator=operator))
|
|
if operator == "is" and narrow_term.operand in unsupported_is_operands:
|
|
raise JsonableError(
|
|
_("Operand {operand} not supported.").format(operand=narrow_term.operand)
|
|
)
|
|
|
|
|
|
class NarrowPredicate(Protocol):
|
|
def __call__(self, *, message: dict[str, Any], flags: list[str]) -> bool: ...
|
|
|
|
|
|
def build_narrow_predicate(
|
|
narrow: Collection[NarrowTerm],
|
|
) -> NarrowPredicate:
|
|
"""Changes to this function should come with corresponding changes to
|
|
NarrowLibraryTest."""
|
|
check_narrow_for_events(narrow)
|
|
|
|
def narrow_predicate(*, message: dict[str, Any], flags: list[str]) -> bool:
|
|
def satisfies_operator(*, operator: str, operand: str) -> bool:
|
|
if operator in channel_operators:
|
|
if message["type"] != "stream":
|
|
return False
|
|
if operand.lower() != message["display_recipient"].lower():
|
|
return False
|
|
elif operator == "topic":
|
|
if message["type"] != "stream":
|
|
return False
|
|
topic_name = get_topic_from_message_info(message)
|
|
if operand.lower() != topic_name.lower():
|
|
return False
|
|
elif operator == "sender":
|
|
if operand.lower() != message["sender_email"].lower():
|
|
return False
|
|
elif operator == "is" and operand in ["dm", "private"]:
|
|
# "is:private" is a legacy alias for "is:dm"
|
|
if message["type"] != "private":
|
|
return False
|
|
elif operator == "is" and operand in ["starred"]:
|
|
if operand not in flags:
|
|
return False
|
|
elif operator == "is" and operand == "unread":
|
|
if "read" in flags:
|
|
return False
|
|
elif operator == "is" and operand in ["alerted", "mentioned"]:
|
|
if "mentioned" not in flags:
|
|
return False
|
|
elif operator == "is" and operand == "resolved":
|
|
if message["type"] != "stream":
|
|
return False
|
|
topic_name = get_topic_from_message_info(message)
|
|
if not topic_name.startswith(RESOLVED_TOPIC_PREFIX):
|
|
return False
|
|
return True
|
|
|
|
for narrow_term in narrow:
|
|
# TODO: Eventually handle negated narrow terms.
|
|
if not satisfies_operator(operator=narrow_term.operator, operand=narrow_term.operand):
|
|
return False
|
|
|
|
return True
|
|
|
|
return narrow_predicate
|