本文整理汇总了Python中zerver.models.get_display_recipient函数的典型用法代码示例。如果您正苦于以下问题:Python get_display_recipient函数的具体用法?Python get_display_recipient怎么用?Python get_display_recipient使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_display_recipient函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: get_apns_alert_title
def get_apns_alert_title(message: Message) -> str:
"""
On an iOS notification, this is the first bolded line.
"""
if message.recipient.type == Recipient.HUDDLE:
recipients = cast(List[Dict[str, Any]], get_display_recipient(message.recipient))
return ', '.join(sorted(r['full_name'] for r in recipients))
elif message.is_stream_message():
return "#%s > %s" % (get_display_recipient(message.recipient), message.topic_name(),)
# For personal PMs, we just show the sender name.
return message.sender.full_name
开发者ID:showell,项目名称:zulip,代码行数:11,代码来源:push_notifications.py
示例2: export_messages_single_user
def export_messages_single_user(user_profile, chunk_size=1000, output_dir=None):
# type: (UserProfile, int, Path) -> None
user_message_query = UserMessage.objects.filter(user_profile=user_profile)
min_id = -1
dump_file_id = 1
while True:
actual_query = user_message_query.select_related("message", "message__sending_client").filter(id__gt=min_id)[0:chunk_size]
user_message_chunk = [um for um in actual_query]
user_message_ids = set(um.id for um in user_message_chunk)
if len(user_message_chunk) == 0:
break
message_chunk = []
for user_message in user_message_chunk:
item = model_to_dict(user_message.message)
item['flags'] = user_message.flags_list()
item['flags_mask'] = user_message.flags.mask
# Add a few nice, human-readable details
item['sending_client_name'] = user_message.message.sending_client.name
item['display_recipient'] = get_display_recipient(user_message.message.recipient)
message_chunk.append(item)
message_filename = os.path.join(output_dir, "messages-%06d.json" % (dump_file_id,))
logging.info("Fetched Messages for %s" % (message_filename,))
output = {'zerver_message': message_chunk}
floatify_datetime_fields(output, 'zerver_message')
write_message_export(message_filename, output)
min_id = max(user_message_ids)
dump_file_id += 1
开发者ID:HKingz,项目名称:zulip,代码行数:32,代码来源:export.py
示例3: message_header
def message_header(user_profile, message):
# type: (UserProfile, Message) -> Dict[str, Any]
disp_recipient = get_display_recipient(message.recipient)
if message.recipient.type == Recipient.PERSONAL:
header = u"You and %s" % (message.sender.full_name)
html_link = pm_narrow_url(user_profile.realm, [message.sender.email])
header_html = u"<a style='color: #ffffff;' href='%s'>%s</a>" % (html_link, header)
elif message.recipient.type == Recipient.HUDDLE:
assert not isinstance(disp_recipient, text_type)
other_recipients = [r['full_name'] for r in disp_recipient
if r['email'] != user_profile.email]
header = u"You and %s" % (", ".join(other_recipients),)
html_link = pm_narrow_url(user_profile.realm, [r["email"] for r in disp_recipient
if r["email"] != user_profile.email])
header_html = u"<a style='color: #ffffff;' href='%s'>%s</a>" % (html_link, header)
else:
assert isinstance(disp_recipient, text_type)
header = u"%s > %s" % (disp_recipient, message.topic_name())
stream_link = stream_narrow_url(user_profile.realm, disp_recipient)
topic_link = topic_narrow_url(user_profile.realm, disp_recipient, message.subject)
header_html = u"<a href='%s'>%s</a> > <a href='%s'>%s</a>" % (
stream_link, disp_recipient, topic_link, message.subject)
return {"plain": header,
"html": header_html,
"stream_message": message.recipient.type_name() == "stream"}
开发者ID:neurodynamic,项目名称:zulip,代码行数:25,代码来源:notifications.py
示例4: test_receive_stream_email_messages_success
def test_receive_stream_email_messages_success(self) -> None:
# build dummy messages for stream
# test valid incoming stream message is processed properly
user_profile = self.example_user('hamlet')
self.login(user_profile.email)
self.subscribe(user_profile, "Denmark")
stream = get_stream("Denmark", user_profile.realm)
stream_to_address = encode_email_address(stream)
incoming_valid_message = MIMEText('TestStreamEmailMessages Body') # type: Any # https://github.com/python/typeshed/issues/275
incoming_valid_message['Subject'] = 'TestStreamEmailMessages Subject'
incoming_valid_message['From'] = self.example_email('hamlet')
incoming_valid_message['To'] = stream_to_address
incoming_valid_message['Reply-to'] = self.example_email('othello')
process_message(incoming_valid_message)
# Hamlet is subscribed to this stream so should see the email message from Othello.
message = most_recent_message(user_profile)
self.assertEqual(message.content, "TestStreamEmailMessages Body")
self.assertEqual(get_display_recipient(message.recipient), stream.name)
self.assertEqual(message.topic_name(), incoming_valid_message['Subject'])
开发者ID:gnprice,项目名称:zulip,代码行数:26,代码来源:test_email_mirror.py
示例5: send_to_missed_message_address
def send_to_missed_message_address(address, message):
# type: (Text, message.Message) -> None
token = get_missed_message_token_from_address(address)
key = missed_message_redis_key(token)
result = redis_client.hmget(key, 'user_profile_id', 'recipient_id', 'subject')
if not all(val is not None for val in result):
raise ZulipEmailForwardError('Missing missed message address data')
user_profile_id, recipient_id, subject_b = result # type: (bytes, bytes, bytes)
user_profile = get_user_profile_by_id(user_profile_id)
recipient = Recipient.objects.get(id=recipient_id)
display_recipient = get_display_recipient(recipient)
# Testing with basestring so we don't depend on the list return type from
# get_display_recipient
if not isinstance(display_recipient, str):
recipient_str = u','.join([user['email'] for user in display_recipient])
else:
recipient_str = display_recipient
body = construct_zulip_body(message, user_profile.realm)
if recipient.type == Recipient.STREAM:
recipient_type_name = 'stream'
else:
recipient_type_name = 'private'
internal_send_message(user_profile.realm, user_profile.email,
recipient_type_name, recipient_str,
subject_b.decode('utf-8'), body)
logger.info("Successfully processed email from %s to %s" % (
user_profile.email, recipient_str))
开发者ID:brockwhittaker,项目名称:zulip,代码行数:32,代码来源:email_mirror.py
示例6: send_to_missed_message_address
def send_to_missed_message_address(address, message):
# type: (text_type, message.Message) -> None
token = get_missed_message_token_from_address(address)
key = missed_message_redis_key(token)
result = redis_client.hmget(key, "user_profile_id", "recipient_id", "subject")
if not all(val is not None for val in result):
raise ZulipEmailForwardError("Missing missed message address data")
user_profile_id, recipient_id, subject = result
user_profile = get_user_profile_by_id(user_profile_id)
recipient = Recipient.objects.get(id=recipient_id)
display_recipient = get_display_recipient(recipient)
# Testing with basestring so we don't depend on the list return type from
# get_display_recipient
if not isinstance(display_recipient, six.string_types):
recipient_str = ",".join([user["email"] for user in display_recipient])
else:
recipient_str = display_recipient
body = filter_footer(extract_body(message))
body += extract_and_upload_attachments(message, user_profile.realm)
if not body:
body = "(No email body)"
if recipient.type == Recipient.STREAM:
recipient_type_name = "stream"
else:
recipient_type_name = "private"
internal_send_message(user_profile.email, recipient_type_name, recipient_str, subject, body)
logging.info("Successfully processed email from %s to %s" % (user_profile.email, recipient_str))
开发者ID:galexrt,项目名称:zulip,代码行数:32,代码来源:email_mirror.py
示例7: message_header
def message_header(user_profile: UserProfile, message: Message) -> Dict[str, Any]:
if message.recipient.type == Recipient.PERSONAL:
header = "You and %s" % (message.sender.full_name,)
html_link = personal_narrow_url(
realm=user_profile.realm,
sender=message.sender,
)
header_html = "<a style='color: #ffffff;' href='%s'>%s</a>" % (html_link, header)
elif message.recipient.type == Recipient.HUDDLE:
disp_recipient = get_display_recipient(message.recipient)
assert not isinstance(disp_recipient, str)
other_recipients = [r['full_name'] for r in disp_recipient
if r['id'] != user_profile.id]
header = "You and %s" % (", ".join(other_recipients),)
other_user_ids = [r['id'] for r in disp_recipient
if r['id'] != user_profile.id]
html_link = huddle_narrow_url(
realm=user_profile.realm,
other_user_ids=other_user_ids,
)
header_html = "<a style='color: #ffffff;' href='%s'>%s</a>" % (html_link, header)
else:
stream = Stream.objects.only('id', 'name').get(id=message.recipient.type_id)
header = "%s > %s" % (stream.name, message.topic_name())
stream_link = stream_narrow_url(user_profile.realm, stream)
topic_link = topic_narrow_url(user_profile.realm, stream, message.topic_name())
header_html = "<a href='%s'>%s</a> > <a href='%s'>%s</a>" % (
stream_link, stream.name, topic_link, message.topic_name())
return {"plain": header,
"html": header_html,
"stream_message": message.recipient.type_name() == "stream"}
开发者ID:BakerWang,项目名称:zulip,代码行数:32,代码来源:notifications.py
示例8: test_receive_stream_email_messages_success
def test_receive_stream_email_messages_success(self):
# build dummy messages for stream
# test valid incoming stream message is processed properly
self.login("[email protected]")
user_profile = get_user_profile_by_email("[email protected]")
self.subscribe_to_stream(user_profile.email, "Denmark")
stream = get_stream("Denmark", user_profile.realm)
stream_to_address = encode_email_address(stream)
incoming_valid_message = MIMEText('TestStreamEmailMessages Body')
incoming_valid_message['Subject'] = 'TestStreamEmailMessages Subject'
incoming_valid_message['From'] = "[email protected]"
incoming_valid_message['To'] = stream_to_address
incoming_valid_message['Reply-to'] = "[email protected]"
process_message(incoming_valid_message)
# Hamlet is subscribed to this stream so should see the email message from Othello.
message = most_recent_message(user_profile)
self.assertEqual(message.content, "TestStreamEmailMessages Body")
self.assertEqual(get_display_recipient(message.recipient), stream.name)
self.assertEqual(message.subject, incoming_valid_message['Subject'])
开发者ID:seanly,项目名称:zulip,代码行数:26,代码来源:test_email_mirror.py
示例9: do_send_missedmessage_events
def do_send_missedmessage_events(user_profile, missed_messages, message_count):
"""
Send a reminder email and/or push notifications to a user if she's missed some PMs by being offline
`user_profile` is the user to send the reminder to
`missed_messages` is a list of Message objects to remind about
"""
# Disabled missedmessage emails internally
if not user_profile.enable_offline_email_notifications:
return
senders = set(m.sender.full_name for m in missed_messages)
sender_str = ", ".join(senders)
plural_messages = 's' if len(missed_messages) > 1 else ''
template_payload = {'name': user_profile.full_name,
'messages': build_message_list(user_profile, missed_messages),
'message_count': message_count,
'url': 'https://%s' % (settings.EXTERNAL_HOST,),
'reply_warning': False,
'external_host': settings.EXTERNAL_HOST}
headers = {}
if all(msg.recipient.type in (Recipient.HUDDLE, Recipient.PERSONAL)
for msg in missed_messages):
# If we have one huddle, set a reply-to to all of the members
# of the huddle except the user herself
disp_recipients = [", ".join(recipient['email']
for recipient in get_display_recipient(mesg.recipient)
if recipient['email'] != user_profile.email)
for mesg in missed_messages]
if all(msg.recipient.type == Recipient.HUDDLE for msg in missed_messages) and \
len(set(disp_recipients)) == 1:
headers['Reply-To'] = disp_recipients[0]
elif len(senders) == 1:
headers['Reply-To'] = missed_messages[0].sender.email
else:
template_payload['reply_warning'] = True
else:
# There are some @-mentions mixed in with personals
template_payload['mention'] = True
template_payload['reply_warning'] = True
headers['Reply-To'] = "Nobody <%s>" % (settings.NOREPLY_EMAIL_ADDRESS,)
# Give users a one-click unsubscribe link they can use to stop getting
# missed message emails without having to log in first.
unsubscribe_link = one_click_unsubscribe_link(user_profile, "missed_messages")
template_payload["unsubscribe_link"] = unsubscribe_link
subject = "Missed Zulip%s from %s" % (plural_messages, sender_str)
from_email = "%s (via Zulip) <%s>" % (sender_str, settings.NOREPLY_EMAIL_ADDRESS)
text_content = loader.render_to_string('zerver/missed_message_email.txt', template_payload)
html_content = loader.render_to_string('zerver/missed_message_email_html.txt', template_payload)
msg = EmailMultiAlternatives(subject, text_content, from_email, [user_profile.email],
headers = headers)
msg.attach_alternative(html_content, "text/html")
msg.send()
user_profile.last_reminder = datetime.datetime.now()
user_profile.save(update_fields=['last_reminder'])
开发者ID:Gabriel0402,项目名称:zulip,代码行数:60,代码来源:notifications.py
示例10: test_receive_stream_email_multiple_recipient_success
def test_receive_stream_email_multiple_recipient_success(self) -> None:
user_profile = self.example_user('hamlet')
self.login(user_profile.email)
self.subscribe(user_profile, "Denmark")
stream = get_stream("Denmark", user_profile.realm)
# stream address is angle-addr within multiple addresses
stream_to_addresses = ["A.N. Other <[email protected]>",
"Denmark <{}>".format(encode_email_address(stream))]
incoming_valid_message = MIMEText('TestStreamEmailMessages Body')
incoming_valid_message['Subject'] = 'TestStreamEmailMessages Subject'
incoming_valid_message['From'] = self.example_email('hamlet')
incoming_valid_message['To'] = ", ".join(stream_to_addresses)
incoming_valid_message['Reply-to'] = self.example_email('othello')
process_message(incoming_valid_message)
# Hamlet is subscribed to this stream so should see the email message from Othello.
message = most_recent_message(user_profile)
self.assertEqual(message.content, "TestStreamEmailMessages Body")
self.assertEqual(get_display_recipient(message.recipient), stream.name)
self.assertEqual(message.topic_name(), incoming_valid_message['Subject'])
开发者ID:deltay,项目名称:zulip,代码行数:25,代码来源:test_email_mirror.py
示例11: get_gcm_payload
def get_gcm_payload(user_profile, message):
# type: (UserProfile, Message) -> Dict[str, Any]
content = message.content
content_truncated = (len(content) > 200)
if content_truncated:
content = content[:200] + "..."
android_data = {
'user': user_profile.email,
'event': 'message',
'alert': get_alert_from_message(message),
'zulip_message_id': message.id, # message_id is reserved for CCS
'time': datetime_to_timestamp(message.pub_date),
'content': content,
'content_truncated': content_truncated,
'sender_email': message.sender.email,
'sender_full_name': message.sender.full_name,
'sender_avatar_url': absolute_avatar_url(message.sender),
}
if message.recipient.type == Recipient.STREAM:
android_data['recipient_type'] = "stream"
android_data['stream'] = get_display_recipient(message.recipient)
android_data['topic'] = message.subject
elif message.recipient.type in (Recipient.HUDDLE, Recipient.PERSONAL):
android_data['recipient_type'] = "private"
return android_data
开发者ID:JamesLinus,项目名称:zulip,代码行数:28,代码来源:push_notifications.py
示例12: send_to_missed_message_address
def send_to_missed_message_address(address, message):
# type: (text_type, message.Message) -> None
token = get_missed_message_token_from_address(address)
key = missed_message_redis_key(token)
result = redis_client.hmget(key, 'user_profile_id', 'recipient_id', 'subject')
if not all(val is not None for val in result):
raise ZulipEmailForwardError('Missing missed message address data')
user_profile_id, recipient_id, subject = result
user_profile = get_user_profile_by_id(user_profile_id)
recipient = Recipient.objects.get(id=recipient_id)
display_recipient = get_display_recipient(recipient)
# Testing with basestring so we don't depend on the list return type from
# get_display_recipient
if not isinstance(display_recipient, six.string_types):
recipient_str = ','.join([user['email'] for user in display_recipient])
else:
recipient_str = display_recipient
body = filter_footer(extract_body(message))
body += extract_and_upload_attachments(message, user_profile.realm)
if not body:
body = '(No email body)'
if recipient.type == Recipient.STREAM:
recipient_type_name = 'stream'
else:
recipient_type_name = 'private'
internal_send_message(user_profile.email, recipient_type_name,
recipient_str, subject, body)
开发者ID:150vb,项目名称:zulip,代码行数:32,代码来源:email_mirror.py
示例13: send_json_payload
def send_json_payload(self, user_profile: UserProfile, url: str,
payload: Union[str, Dict[str, Any]],
stream_name: Optional[str]=None, **post_params: Any) -> Message:
if stream_name is not None:
self.subscribe(user_profile, stream_name)
prior_msg = self.get_last_message()
result = self.client_post(url, payload, **post_params)
self.assert_json_success(result)
# Check the correct message was sent
msg = self.get_last_message()
if msg.id == prior_msg.id:
raise Exception('''
Your test code called an endpoint that did
not write any new messages. It is probably
broken (but still returns 200 due to exception
handling).
''') # nocoverage
self.assertEqual(msg.sender.email, user_profile.email)
if stream_name is not None:
self.assertEqual(get_display_recipient(msg.recipient), stream_name)
# TODO: should also validate recipient for private messages
return msg
开发者ID:BakerWang,项目名称:zulip,代码行数:28,代码来源:test_classes.py
示例14: get_streams
def get_streams(self, email: str, realm: Realm) -> List[str]:
"""
Helper function to get the stream names for a user
"""
user_profile = get_user(email, realm)
subs = get_stream_subscriptions_for_user(user_profile).filter(
active=True,
)
return [cast(str, get_display_recipient(sub.recipient)) for sub in subs]
开发者ID:BakerWang,项目名称:zulip,代码行数:9,代码来源:test_classes.py
示例15: test_stream_message_to_embedded_bot
def test_stream_message_to_embedded_bot(self) -> None:
self.send_stream_message(self.user_profile.email, "Denmark",
content="@**{}** foo".format(self.bot_profile.full_name),
topic_name="bar")
last_message = self.get_last_message()
self.assertEqual(last_message.content, "beep boop")
self.assertEqual(last_message.sender_id, self.bot_profile.id)
self.assertEqual(last_message.subject, "bar")
display_recipient = get_display_recipient(last_message.recipient)
self.assertEqual(display_recipient, "Denmark")
开发者ID:joydeep1701,项目名称:zulip,代码行数:10,代码来源:test_embedded_bot_system.py
示例16: test_stream_message_to_outgoing_webhook_bot
def test_stream_message_to_outgoing_webhook_bot(self, mock_requests_request: mock.Mock) -> None:
self.send_stream_message(self.user_profile.email, "Denmark",
content="@**{}** foo".format(self.bot_profile.full_name),
topic_name="bar")
last_message = self.get_last_message()
self.assertEqual(last_message.content, "Success! Hidley ho, I'm a webhook responding!")
self.assertEqual(last_message.sender_id, self.bot_profile.id)
self.assertEqual(last_message.subject, "bar")
display_recipient = get_display_recipient(last_message.recipient)
self.assertEqual(display_recipient, "Denmark")
开发者ID:gnprice,项目名称:zulip,代码行数:10,代码来源:test_outgoing_webhook_system.py
示例17: test_pm_to_outgoing_webhook_bot
def test_pm_to_outgoing_webhook_bot(self, mock_requests_request: mock.Mock) -> None:
self.send_personal_message(self.user_profile.email, self.bot_profile.email,
content="foo")
last_message = self.get_last_message()
self.assertEqual(last_message.content, "Success! Hidley ho, I'm a webhook responding!")
self.assertEqual(last_message.sender_id, self.bot_profile.id)
display_recipient = get_display_recipient(last_message.recipient)
# The next two lines error on mypy because the display_recipient is of type Union[Text, List[Dict[str, Any]]].
# In this case, we know that display_recipient will be of type List[Dict[str, Any]].
# Otherwise this test will error, which is wanted behavior anyway.
self.assert_length(display_recipient, 1) # type: ignore
self.assertEqual(display_recipient[0]['email'], self.user_profile.email) # type: ignore
开发者ID:gnprice,项目名称:zulip,代码行数:12,代码来源:test_outgoing_webhook_system.py
示例18: test_pm_to_embedded_bot
def test_pm_to_embedded_bot(self) -> None:
self.send_personal_message(self.user_profile.email, self.bot_profile.email,
content="help")
last_message = self.get_last_message()
self.assertEqual(last_message.content, "beep boop")
self.assertEqual(last_message.sender_id, self.bot_profile.id)
display_recipient = get_display_recipient(last_message.recipient)
# The next two lines error on mypy because the display_recipient is of type Union[Text, List[Dict[str, Any]]].
# In this case, we know that display_recipient will be of type List[Dict[str, Any]].
# Otherwise this test will error, which is wanted behavior anyway.
self.assert_length(display_recipient, 1) # type: ignore
self.assertEqual(display_recipient[0]['email'], self.user_profile.email) # type: ignore
开发者ID:joydeep1701,项目名称:zulip,代码行数:12,代码来源:test_embedded_bot_system.py
示例19: send_to_missed_message_address
def send_to_missed_message_address(address: str, message: message.Message) -> None:
token = get_missed_message_token_from_address(address)
key = missed_message_redis_key(token)
result = redis_client.hmget(key, 'user_profile_id', 'recipient_id', 'subject')
if not all(val is not None for val in result):
raise ZulipEmailForwardError('Missing missed message address data')
user_profile_id, recipient_id, subject_b = result # type: (bytes, bytes, bytes)
user_profile = get_user_profile_by_id(user_profile_id)
recipient = Recipient.objects.get(id=recipient_id)
body = construct_zulip_body(message, user_profile.realm)
if recipient.type == Recipient.STREAM:
stream = get_stream_by_id_in_realm(recipient.type_id, user_profile.realm)
internal_send_stream_message(
user_profile.realm, user_profile, stream,
subject_b.decode('utf-8'), body
)
recipient_str = stream.name
elif recipient.type == Recipient.PERSONAL:
display_recipient = get_display_recipient(recipient)
assert not isinstance(display_recipient, str)
recipient_str = display_recipient[0]['email']
recipient_user = get_user(recipient_str, user_profile.realm)
internal_send_private_message(user_profile.realm, user_profile,
recipient_user, body)
elif recipient.type == Recipient.HUDDLE:
display_recipient = get_display_recipient(recipient)
assert not isinstance(display_recipient, str)
emails = [user_dict['email'] for user_dict in display_recipient]
recipient_str = ', '.join(emails)
internal_send_huddle_message(user_profile.realm, user_profile,
emails, body)
else:
raise AssertionError("Invalid recipient type!")
logger.info("Successfully processed email from %s to %s" % (
user_profile.email, recipient_str))
开发者ID:deltay,项目名称:zulip,代码行数:39,代码来源:email_mirror.py
示例20: get_gcm_alert
def get_gcm_alert(message: Message) -> str:
"""
Determine what alert string to display based on the missed messages.
"""
sender_str = message.sender.full_name
if message.recipient.type == Recipient.HUDDLE and message.trigger == 'private_message':
return "New private group message from %s" % (sender_str,)
elif message.recipient.type == Recipient.PERSONAL and message.trigger == 'private_message':
return "New private message from %s" % (sender_str,)
elif message.is_stream_message() and message.trigger == 'mentioned':
return "New mention from %s" % (sender_str,)
else: # message.is_stream_message() and message.trigger == 'stream_push_notify'
return "New stream message from %s in %s" % (sender_str, get_display_recipient(message.recipient),)
开发者ID:showell,项目名称:zulip,代码行数:13,代码来源:push_notifications.py
注:本文中的zerver.models.get_display_recipient函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论