import logging import re from typing import Any, List, Tuple import attr from jinja2 import TemplateNotFound from mautrix.types import Format, MessageType, TextMessageEventContent from mautrix.util.formatter import parse_html 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 parse_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 base_args = { **{field.key: field for field in Action 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, # type: ignore } args["templates"] = templates.proxy(args) # type: ignore 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