93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
import logging
|
|
import re
|
|
from typing import Any, List, Optional, Tuple, cast
|
|
|
|
import attr
|
|
from fastapi import Request
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBasicCredentials
|
|
from jinja2 import TemplateNotFound
|
|
from mautrix.types import Format, MessageType, TextMessageEventContent
|
|
from mautrix.util.formatter import parse_html
|
|
|
|
from ..config import RoutingKey
|
|
from ..util.template import TemplateManager, TemplateUtil
|
|
from .types import OTHER_ENUMS, Action, EventParse # type: ignore
|
|
|
|
spaces = re.compile(" +")
|
|
space = " "
|
|
|
|
|
|
messages = TemplateManager("gitlab", "messages")
|
|
templates = TemplateManager("gitlab", "mixins")
|
|
|
|
|
|
async def authorize(
|
|
route: RoutingKey,
|
|
request: Request,
|
|
basic_credentials: Optional[HTTPBasicCredentials],
|
|
bearer_credentials: Optional[HTTPAuthorizationCredentials],
|
|
) -> bool:
|
|
provided: Optional[str] = request.headers.get("x-gitlab-token")
|
|
return provided == route.secret_token
|
|
|
|
|
|
async def handle_event(x_gitlab_event: str, payload: Any) -> List[Tuple[str, str]]:
|
|
evt = EventParse[x_gitlab_event].deserialize(payload)
|
|
try:
|
|
tpl = messages[evt.template_name]
|
|
except TemplateNotFound:
|
|
msg = f"Received unhandled gitlab event type {x_gitlab_event}"
|
|
logging.error(msg)
|
|
logging.debug(payload)
|
|
return []
|
|
|
|
aborted = False
|
|
|
|
def abort() -> None:
|
|
nonlocal aborted
|
|
aborted = True
|
|
|
|
action_fields = cast(Any, Action)
|
|
base_args = {
|
|
**{field.key: field for field in action_fields if field.key.isupper()},
|
|
**OTHER_ENUMS,
|
|
"util": TemplateUtil,
|
|
}
|
|
|
|
msgs = []
|
|
for subevt in evt.preprocess():
|
|
args = {
|
|
**attr.asdict(subevt, recurse=False),
|
|
**{key: getattr(subevt, key) for key in subevt.event_properties},
|
|
"abort": abort,
|
|
**base_args,
|
|
}
|
|
args["templates"] = templates.proxy(args)
|
|
|
|
html = tpl.render(**args)
|
|
if not html or aborted:
|
|
aborted = False
|
|
continue
|
|
html = spaces.sub(space, html.strip())
|
|
|
|
content = TextMessageEventContent(
|
|
msgtype=MessageType.TEXT,
|
|
format=Format.HTML,
|
|
formatted_body=html,
|
|
body=await parse_html(html),
|
|
)
|
|
content["xyz.maubot.gitlab.webhook"] = {
|
|
"event_type": x_gitlab_event,
|
|
**subevt.meta,
|
|
}
|
|
msgs.append((content.body, content.formatted_body))
|
|
return msgs
|
|
|
|
|
|
async def parse_event(
|
|
route: RoutingKey, payload: Any, request: Request
|
|
) -> List[Tuple[str, str]]:
|
|
x_gitlab_event = request.headers.get("x-gitlab-event")
|
|
if x_gitlab_event:
|
|
return await handle_event(x_gitlab_event, payload)
|
|
return []
|