本文整理汇总了Python中zerver.lib.webhooks.common.check_send_webhook_message函数的典型用法代码示例。如果您正苦于以下问题:Python check_send_webhook_message函数的具体用法?Python check_send_webhook_message怎么用?Python check_send_webhook_message使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了check_send_webhook_message函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: api_gocd_webhook
def api_gocd_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body'),
) -> HttpResponse:
modifications = payload['build_cause']['material_revisions'][0]['modifications'][0]
result = payload['stages'][0]['result']
material = payload['build_cause']['material_revisions'][0]['material']
if result == "Passed":
emoji = ':thumbs_up:'
elif result == "Failed":
emoji = ':thumbs_down:'
build_details_file = os.path.join(os.path.dirname(__file__), 'fixtures/build_details.json')
with open(build_details_file, 'r') as f:
contents = json.load(f)
build_link = contents["build_details"]["_links"]["pipeline"]["href"]
body = MESSAGE_TEMPLATE.format(
modifications['user_name'],
result,
emoji,
build_link,
modifications['comment']
)
branch = material['description'].split(",")
topic = branch[0].split(" ")[1]
check_send_webhook_message(request, user_profile, topic, body)
return json_success()
开发者ID:284928489,项目名称:zulip,代码行数:32,代码来源:view.py
示例2: api_papertrail_webhook
def api_papertrail_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
# construct the message of the message
message_template = '**"{}"** search found **{}** matches - {}\n```'
message = [message_template.format(payload["saved_search"]["name"],
str(len(payload["events"])),
payload["saved_search"]["html_search_url"])]
for i, event in enumerate(payload["events"]):
event_text = '{} {} {}:\n {}'.format(event["display_received_at"],
event["source_name"],
payload["saved_search"]["query"],
event["message"])
message.append(event_text)
if i >= 3:
message.append('```\n[See more]({})'.format(payload["saved_search"]["html_search_url"]))
break
else:
message.append('```')
post = '\n'.join(message)
topic = 'logs'
# send the message
check_send_webhook_message(request, user_profile, topic, post)
# return json result
return json_success()
开发者ID:gnprice,项目名称:zulip,代码行数:28,代码来源:view.py
示例3: api_hellosign_webhook
def api_hellosign_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Dict[str, Any]]=REQ(argument_type='body')) -> HttpResponse:
model_payload = ready_payload(payload['signature_request']['signatures'], payload)
body = format_body(payload['signature_request']['signatures'], model_payload)
topic = model_payload['contract_title']
check_send_webhook_message(request, user_profile, topic, body)
return json_success()
开发者ID:284928489,项目名称:zulip,代码行数:7,代码来源:view.py
示例4: api_insping_webhook
def api_insping_webhook(
request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Dict[str, Any]]=REQ(argument_type='body')
) -> HttpResponse:
data = payload['webhook_event_data']
state_name = data['check_state_name']
url_tested = data['request_url']
response_time = data['response_time']
timestamp = data['request_start_time']
time_formatted = time.strftime("%c", time.strptime(timestamp,
"%Y-%m-%dT%H:%M:%S.%f+00:00"))
body = """State changed: {}
URL: {}
Response time: {} ms
Timestamp: {}
""".format(state_name, url_tested, response_time, time_formatted)
topic = 'insping'
check_send_webhook_message(request, user_profile, topic, body)
return json_success()
开发者ID:284928489,项目名称:zulip,代码行数:25,代码来源:view.py
示例5: send_formated_pagerduty
def send_formated_pagerduty(request: HttpRequest,
user_profile: UserProfile,
message_type: Text,
format_dict: Dict[str, Any]) -> None:
if message_type in ('incident.trigger', 'incident.unacknowledge'):
template = (u':imp: Incident '
u'[{incident_num}]({incident_url}) {action} by '
u'[{service_name}]({service_url}) and assigned to '
u'[{assigned_to_username}@]({assigned_to_url})\n\n>{trigger_message}')
elif message_type == 'incident.resolve' and format_dict['resolved_by_url']:
template = (u':grinning: Incident '
u'[{incident_num}]({incident_url}) resolved by '
u'[{resolved_by_username}@]({resolved_by_url})\n\n>{trigger_message}')
elif message_type == 'incident.resolve' and not format_dict['resolved_by_url']:
template = (u':grinning: Incident '
u'[{incident_num}]({incident_url}) resolved\n\n>{trigger_message}')
else:
template = (u':no_good: Incident [{incident_num}]({incident_url}) '
u'{action} by [{assigned_to_username}@]({assigned_to_url})\n\n>{trigger_message}')
subject = u'incident {incident_num}'.format(**format_dict)
body = template.format(**format_dict)
check_send_webhook_message(request, user_profile, subject, body)
开发者ID:gnprice,项目名称:zulip,代码行数:25,代码来源:view.py
示例6: api_jira_webhook
def api_jira_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
event = get_event_type(payload)
if event == 'jira:issue_created':
subject = get_issue_subject(payload)
content = handle_created_issue_event(payload)
elif event == 'jira:issue_deleted':
subject = get_issue_subject(payload)
content = handle_deleted_issue_event(payload)
elif event == 'jira:issue_updated':
subject = get_issue_subject(payload)
content = handle_updated_issue_event(payload, user_profile)
elif event == 'comment_created':
subject = get_issue_subject(payload)
content = handle_updated_issue_event(payload, user_profile)
elif event in IGNORED_EVENTS:
return json_success()
else:
raise UnexpectedWebhookEventType('Jira', event)
check_send_webhook_message(request, user_profile,
subject, content,
unquote_url_parameters=True)
return json_success()
开发者ID:BakerWang,项目名称:zulip,代码行数:25,代码来源:view.py
示例7: api_freshdesk_webhook
def api_freshdesk_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
ticket_data = payload["freshdesk_webhook"]
required_keys = [
"triggered_event", "ticket_id", "ticket_url", "ticket_type",
"ticket_subject", "ticket_description", "ticket_status",
"ticket_priority", "requester_name", "requester_email",
]
for key in required_keys:
if ticket_data.get(key) is None:
logging.warning("Freshdesk webhook error. Payload was:")
logging.warning(request.body)
return json_error(_("Missing key %s in JSON") % (key,))
ticket = TicketDict(ticket_data)
subject = "#%s: %s" % (ticket.id, ticket.subject)
event_info = parse_freshdesk_event(ticket.triggered_event)
if event_info[1] == "created":
content = format_freshdesk_ticket_creation_message(ticket)
elif event_info[0] == "note_type":
content = format_freshdesk_note_message(ticket, event_info)
elif event_info[0] in ("status", "priority"):
content = format_freshdesk_property_change_message(ticket, event_info)
else:
# Not an event we know handle; do nothing.
return json_success()
check_send_webhook_message(request, user_profile, subject, content)
return json_success()
开发者ID:BakerWang,项目名称:zulip,代码行数:33,代码来源:view.py
示例8: api_wordpress_webhook
def api_wordpress_webhook(request: HttpRequest, user_profile: UserProfile,
hook: str=REQ(default="WordPress Action"),
post_title: str=REQ(default="New WordPress Post"),
post_type: str=REQ(default="post"),
post_url: str=REQ(default="WordPress Post URL"),
display_name: str=REQ(default="New User Name"),
user_email: str=REQ(default="New User Email"),
user_login: str=REQ(default="Logged in User")) -> HttpResponse:
# remove trailing whitespace (issue for some test fixtures)
hook = hook.rstrip()
if hook == 'publish_post' or hook == 'publish_page':
data = PUBLISH_POST_OR_PAGE_TEMPLATE.format(type=post_type, title=post_title, url=post_url)
elif hook == 'user_register':
data = USER_REGISTER_TEMPLATE.format(name=display_name, email=user_email)
elif hook == 'wp_login':
data = WP_LOGIN_TEMPLATE.format(name=user_login)
else:
return json_error(_("Unknown WordPress webhook action: " + hook))
topic = 'WordPress Notification'
check_send_webhook_message(request, user_profile, topic, data)
return json_success()
开发者ID:284928489,项目名称:zulip,代码行数:27,代码来源:view.py
示例9: api_bitbucket2_webhook
def api_bitbucket2_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body'),
branches: Optional[str]=REQ(default=None),
user_specified_topic: Optional[str]=REQ("topic", default=None)) -> HttpResponse:
type = get_type(request, payload)
if type == 'push':
# ignore push events with no changes
if not payload['push']['changes']:
return json_success()
branch = get_branch_name_for_push_event(payload)
if branch and branches:
if branches.find(branch) == -1:
return json_success()
subject = get_subject_based_on_type(payload, type)
body_function = get_body_based_on_type(type)
if 'include_title' in signature(body_function).parameters:
body = body_function(
payload,
include_title=user_specified_topic is not None
)
else:
body = body_function(payload)
if type != 'push':
check_send_webhook_message(request, user_profile, subject, body)
else:
for b, s in zip(body, subject):
check_send_webhook_message(request, user_profile, s, b)
return json_success()
开发者ID:brainwane,项目名称:zulip,代码行数:31,代码来源:view.py
示例10: api_zapier_webhook
def api_zapier_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
if payload.get('type') == 'auth':
# The bot's details are used by our Zapier app to format a connection
# label for users to be able to distinguish between different Zulip
# bots and API keys in their UI
return json_success({
'full_name': user_profile.full_name,
'email': user_profile.email,
'id': user_profile.id
})
topic = payload.get('topic')
content = payload.get('content')
if topic is None:
topic = payload.get('subject') # Backwards-compatibility
if topic is None:
return json_error(_("Topic can't be empty"))
if content is None:
return json_error(_("Content can't be empty"))
check_send_webhook_message(request, user_profile, topic, content)
return json_success()
开发者ID:jdherg,项目名称:zulip,代码行数:25,代码来源:view.py
示例11: api_basecamp_webhook
def api_basecamp_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
event = get_event_type(payload)
if event not in SUPPORT_EVENTS:
raise UnexpectedWebhookEventType('Basecamp', event)
subject = get_project_name(payload)
if event.startswith('document_'):
body = get_document_body(event, payload)
elif event.startswith('question_answer_'):
body = get_questions_answer_body(event, payload)
elif event.startswith('question_'):
body = get_questions_body(event, payload)
elif event.startswith('message_'):
body = get_message_body(event, payload)
elif event.startswith('todolist_'):
body = get_todo_list_body(event, payload)
elif event.startswith('todo_'):
body = get_todo_body(event, payload)
elif event.startswith('comment_'):
body = get_comment_body(event, payload)
else:
raise UnexpectedWebhookEventType('Basecamp', event)
check_send_webhook_message(request, user_profile, subject, body)
return json_success()
开发者ID:284928489,项目名称:zulip,代码行数:27,代码来源:view.py
示例12: api_raygun_webhook
def api_raygun_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any] = REQ(argument_type='body')) -> HttpResponse:
# The payload contains 'event' key. This 'event' key has a value of either
# 'error_notification' or 'error_activity'. 'error_notification' happens
# when an error is caught in an application, where as 'error_activity'
# happens when an action is being taken for the error itself
# (ignored/resolved/assigned/etc.).
event = payload['event']
# Because we wanted to create a message for all of the payloads, it is best
# to handle them separately. This is because some payload keys don't exist
# in the other event.
if event == 'error_notification':
message = compose_notification_message(payload)
elif event == 'error_activity':
message = compose_activity_message(payload)
else:
message = "Unsupported event type: {}".format(event)
topic = 'test'
check_send_webhook_message(request, user_profile, topic, message)
return json_success()
开发者ID:gnprice,项目名称:zulip,代码行数:25,代码来源:view.py
示例13: api_gogs_webhook
def api_gogs_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body'),
branches: Optional[Text]=REQ(default=None)) -> HttpResponse:
repo = payload['repository']['name']
event = request.META['HTTP_X_GOGS_EVENT']
if event == 'push':
branch = payload['ref'].replace('refs/heads/', '')
if branches is not None and branches.find(branch) == -1:
return json_success()
body = format_push_event(payload)
topic = SUBJECT_WITH_BRANCH_TEMPLATE.format(
repo=repo,
branch=branch
)
elif event == 'create':
body = format_new_branch_event(payload)
topic = SUBJECT_WITH_BRANCH_TEMPLATE.format(
repo=repo,
branch=payload['ref']
)
elif event == 'pull_request':
body = format_pull_request_event(payload)
topic = SUBJECT_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=repo,
type='PR',
id=payload['pull_request']['id'],
title=payload['pull_request']['title']
)
else:
return json_error(_('Invalid event "{}" in request headers').format(event))
check_send_webhook_message(request, user_profile, topic, body)
return json_success()
开发者ID:gnprice,项目名称:zulip,代码行数:34,代码来源:view.py
示例14: api_bitbucket_webhook
def api_bitbucket_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Mapping[str, Any]=REQ(validator=check_dict([])),
branches: Optional[str]=REQ(default=None)) -> HttpResponse:
repository = payload['repository']
commits = [
{
'name': payload.get('user'),
'sha': commit.get('raw_node'),
'message': commit.get('message'),
'url': u'{}{}commits/{}'.format(
payload.get('canon_url'),
repository.get('absolute_url'),
commit.get('raw_node'))
}
for commit in payload['commits']
]
if len(commits) == 0:
# Bitbucket doesn't give us enough information to really give
# a useful message :/
subject = repository['name']
content = (u"%s [force pushed](%s)"
% (payload['user'],
payload['canon_url'] + repository['absolute_url']))
else:
branch = payload['commits'][-1]['branch']
if branches is not None and branches.find(branch) == -1:
return json_success()
content = get_push_commits_event_message(payload['user'], None, branch, commits)
subject = SUBJECT_WITH_BRANCH_TEMPLATE.format(repo=repository['name'], branch=branch)
check_send_webhook_message(request, user_profile, subject, content)
return json_success()
开发者ID:284928489,项目名称:zulip,代码行数:34,代码来源:view.py
示例15: api_travis_webhook
def api_travis_webhook(request: HttpRequest, user_profile: UserProfile,
ignore_pull_requests: bool = REQ(validator=check_bool, default=True),
message: Dict[str, str]=REQ('payload', validator=check_dict([
('author_name', check_string),
('status_message', check_string),
('compare_url', check_string),
]))) -> HttpResponse:
message_status = message['status_message']
if ignore_pull_requests and message['type'] == 'pull_request':
return json_success()
if message_status in GOOD_STATUSES:
emoji = ':thumbs_up:'
elif message_status in BAD_STATUSES:
emoji = ':thumbs_down:'
else:
emoji = "(No emoji specified for status '{}'.)".format(message_status)
body = MESSAGE_TEMPLATE.format(
message['author_name'],
message_status,
emoji,
message['compare_url'],
message['build_url']
)
topic = 'builds'
check_send_webhook_message(request, user_profile, topic, body)
return json_success()
开发者ID:BakerWang,项目名称:zulip,代码行数:30,代码来源:view.py
示例16: api_beanstalk_webhook
def api_beanstalk_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(validator=check_dict([])),
branches: Optional[str]=REQ(default=None)) -> HttpResponse:
# Beanstalk supports both SVN and git repositories
# We distinguish between the two by checking for a
# 'uri' key that is only present for git repos
git_repo = 'uri' in payload
if git_repo:
if branches is not None and branches.find(payload['branch']) == -1:
return json_success()
# To get a linkable url,
for commit in payload['commits']:
commit['author'] = {'username': commit['author']['name']}
subject, content = build_message_from_gitlog(user_profile, payload['repository']['name'],
payload['ref'], payload['commits'],
payload['before'], payload['after'],
payload['repository']['url'],
payload['pusher_name'])
else:
author = payload.get('author_full_name')
url = payload.get('changeset_url')
revision = payload.get('revision')
(short_commit_msg, _, _) = payload['message'].partition("\n")
subject = "svn r%s" % (revision,)
content = "%s pushed [revision %s](%s):\n\n> %s" % (author, revision, url, short_commit_msg)
check_send_webhook_message(request, user_profile, subject, content)
return json_success()
开发者ID:BakerWang,项目名称:zulip,代码行数:30,代码来源:view.py
示例17: api_jira_webhook
def api_jira_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
event = get_event_type(payload)
if event == 'jira:issue_created':
subject = get_issue_subject(payload)
content = handle_created_issue_event(payload)
elif event == 'jira:issue_deleted':
subject = get_issue_subject(payload)
content = handle_deleted_issue_event(payload)
elif event == 'jira:issue_updated':
subject = get_issue_subject(payload)
content = handle_updated_issue_event(payload, user_profile)
elif event in IGNORED_EVENTS:
return json_success()
else:
if event is None:
if not settings.TEST_SUITE:
message = u"Got JIRA event with None event type: {}".format(payload)
logging.warning(message)
return json_error(_("Event is not given by JIRA"))
else:
if not settings.TEST_SUITE:
logging.warning("Got JIRA event type we don't support: {}".format(event))
return json_success()
check_send_webhook_message(request, user_profile, subject, content)
return json_success()
开发者ID:gnprice,项目名称:zulip,代码行数:28,代码来源:view.py
示例18: api_gogs_webhook
def api_gogs_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body'),
branches: Optional[str]=REQ(default=None)) -> HttpResponse:
repo = payload['repository']['name']
event = validate_extract_webhook_http_header(request, 'X_GOGS_EVENT', 'Gogs')
if event == 'push':
branch = payload['ref'].replace('refs/heads/', '')
if branches is not None and branches.find(branch) == -1:
return json_success()
body = format_push_event(payload)
topic = SUBJECT_WITH_BRANCH_TEMPLATE.format(
repo=repo,
branch=branch
)
elif event == 'create':
body = format_new_branch_event(payload)
topic = SUBJECT_WITH_BRANCH_TEMPLATE.format(
repo=repo,
branch=payload['ref']
)
elif event == 'pull_request':
body = format_pull_request_event(payload)
topic = SUBJECT_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(
repo=repo,
type='PR',
id=payload['pull_request']['id'],
title=payload['pull_request']['title']
)
else:
raise UnexpectedWebhookEventType('Gogs', event)
check_send_webhook_message(request, user_profile, topic, body)
return json_success()
开发者ID:phansen01,项目名称:zulip,代码行数:34,代码来源:view.py
示例19: api_splunk_webhook
def api_splunk_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
# use default values if expected data is not provided
search_name = payload.get('search_name', 'Missing search_name')
results_link = payload.get('results_link', 'Missing results_link')
host = payload.get('result', {}).get('host', 'Missing host')
source = payload.get('result', {}).get('source', 'Missing source')
raw = payload.get('result', {}).get('_raw', 'Missing _raw')
# for the default topic, use search name but truncate if too long
if len(search_name) >= MAX_TOPIC_NAME_LENGTH:
topic = "{}...".format(search_name[:(MAX_TOPIC_NAME_LENGTH - 3)])
else:
topic = search_name
# construct the message body
body = "Splunk alert from saved search"
body_template = ('\n[{search}]({link})\nhost: {host}'
'\nsource: {source}\n\nraw: {raw}')
body += body_template.format(search = search_name, link = results_link,
host = host, source = source, raw = raw)
# send the message
check_send_webhook_message(request, user_profile, topic, body)
return json_success()
开发者ID:akashnimare,项目名称:zulip,代码行数:27,代码来源:view.py
示例20: api_basecamp_webhook
def api_basecamp_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type='body')) -> HttpResponse:
event = get_event_type(payload)
if event not in SUPPORT_EVENTS:
logging.warning("Basecamp {} event is not supported".format(event))
return json_success()
subject = get_project_name(payload)
if event.startswith('document_'):
body = get_document_body(event, payload)
elif event.startswith('question_answer_'):
body = get_questions_answer_body(event, payload)
elif event.startswith('question_'):
body = get_questions_body(event, payload)
elif event.startswith('message_'):
body = get_message_body(event, payload)
elif event.startswith('todolist_'):
body = get_todo_list_body(event, payload)
elif event.startswith('todo_'):
body = get_todo_body(event, payload)
elif event.startswith('comment_'):
body = get_comment_body(event, payload)
else:
logging.warning("Basecamp handling of {} event is not implemented".format(event))
return json_success()
check_send_webhook_message(request, user_profile, subject, body)
return json_success()
开发者ID:gnprice,项目名称:zulip,代码行数:29,代码来源:view.py
注:本文中的zerver.lib.webhooks.common.check_send_webhook_message函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论