import re
from datetime import datetime
from typing import TYPE_CHECKING, Awaitable, Callable, TypeVar, Union
import emoji as emoji_lib
from mattermost_api_reference_client.models import FileInfo
from mattermost_api_reference_client.types import Unset
from slidge.core.mixins.message import ContentMessageMixin
from slidge.util.types import LegacyAttachment
from .events import Post, get_slack_attachments
if TYPE_CHECKING:
from .session import Session
[docs]
def _emoji_name_conversion(x: str):
return (
x.replace("_3_", "_three_")
.replace("thumbsup", "+1")
.replace("star_struck", "star-struck")
)
[docs]
def emojize_single(x: str, add_delimiters=True):
if add_delimiters:
x = f":{x}:"
char = emoji_lib.emojize(f"{_emoji_name_conversion(x)}", language="alias")
if not emoji_lib.is_emoji(char):
return "❓"
return char
[docs]
def emojize_body(body: str):
char = emoji_lib.emojize(f"{_emoji_name_conversion(body)}", language="alias")
return char
[docs]
def demojize(emoji_char: str):
return _emoji_name_conversion(
emoji_lib.demojize(emoji_char, delimiters=("", ""), language="alias")
)
[docs]
class UserMixin(ContentMessageMixin):
[docs]
mm_id: Callable[[], Awaitable[str]]
[docs]
async def send_mm_post(self, post: Post, carbon=False, archive_only=False):
assert post.metadata
assert post.update_at
file_metas = post.metadata.files
if not (isinstance(file_metas, list)):
file_metas = []
if isinstance(m := post.message, str):
text = emojize_body(m)
else:
text = ""
post_id = post.id
when = datetime.fromtimestamp(post.update_at / 1000)
for slack_attachment in get_slack_attachments(post):
if slack_attachment.fallback:
text += emojize_body(slack_attachment.fallback) + " (BOT)"
await self.send_files(
[
Attachment.from_mm(x, await self.session.mm_client.get_file(x.id))
for x in file_metas
],
post_id,
body=text,
when=when,
carbon=carbon,
thread=post.root_id or post_id,
body_first=True,
archive_only=archive_only,
)
[docs]
async def update_reactions(self, legacy_msg_id: str, carbon: bool = False):
self.react(
legacy_msg_id,
await self.session.get_mm_reactions(
legacy_msg_id,
await self.session.mm_client.mm_id if carbon else await self.mm_id(),
),
carbon=carbon,
)
[docs]
class Attachment(LegacyAttachment):
@staticmethod
[docs]
def from_mm(info: FileInfo, data: bytes):
return Attachment(
name=unset_to_none(info.name),
legacy_file_id=unset_to_none(info.id),
# TODO: data could be an awaitable of bytes in slidge core so we don't
# have to fetch if legacy_file_id has already been seen
data=data,
)
[docs]
def unset_to_none(x: Union[T, Unset]) -> Union[T, None]:
if isinstance(x, Unset):
return None
return x
[docs]
def ensure_newline_after_quote(text: str) -> str:
"""
Adds an empty newline after lines prepended by ">" unless an empty newline already exists.
We need this because without the extra empty newline, mattermost collapse everything in the quote
"""
return re.sub(_NEW_LINE_AFTER_QUOTE_REGEX, r"\1\n\n", text)
[docs]
_NEW_LINE_AFTER_QUOTE_REGEX = re.compile(r"(>*>.*)\n(?=[^\n>])", flags=re.MULTILINE)