zulip/zerver/webhooks/papertrail/view.py

from typing import Any, Dict, Optional

from django.http import HttpRequest, HttpResponse
from django.utils.translation import ugettext as _

from zerver.decorator import api_key_only_webhook_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile

MATCHES_TEMPLATE = '[Search for "{name}"]({url}) found **{number}** matches:\n'
SEARCH_TEMPLATE = """
{timestamp} - {source} - {query}:
``` quote
{message}
```
""".strip()

def ensure_keys(name: str, data: Any) -> Optional[str]:
    if 'events' in data and 'saved_search' in data:
        saved_search = data['saved_search']
        if 'name' in saved_search and 'html_search_url' in saved_search:
            return None
    return _("Missing expected keys")

@api_key_only_webhook_view('Papertrail')
@has_request_variables
def api_papertrail_webhook(request: HttpRequest, user_profile: UserProfile,
                           payload: Dict[str, Any]=REQ(validator=ensure_keys)) -> HttpResponse:

    matches = MATCHES_TEMPLATE.format(
        name=payload["saved_search"]["name"],
        url=payload["saved_search"]["html_search_url"],
        number=str(len(payload["events"]))
    )
    message = [matches]

    for i, event in enumerate(payload["events"]):
        event_text = SEARCH_TEMPLATE.format(
            timestamp=event["display_received_at"],
            source=event["source_name"],
            query=payload["saved_search"]["query"],
            message=event["message"]
        )

        message.append(event_text)

        if i >= 3:
            message.append('[See more]({})'.format(payload["saved_search"]["html_search_url"]))
            break

    post = '\n'.join(message)
    topic = 'logs'

    check_send_webhook_message(request, user_profile, topic, post)
    return json_success()