66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
# adapter/telegram/converter.py
|
|
from __future__ import annotations
|
|
|
|
from aiogram.types import Message
|
|
|
|
from adapter.telegram import db
|
|
from core.protocol import Attachment, IncomingMessage, OutgoingEvent, OutgoingMessage, OutgoingUI
|
|
|
|
|
|
def from_message(message: Message, chat_id: str) -> IncomingMessage:
|
|
return IncomingMessage(
|
|
user_id=str(message.from_user.id),
|
|
chat_id=chat_id,
|
|
text=message.text or message.caption or "",
|
|
attachments=_extract_attachments(message),
|
|
platform="telegram",
|
|
)
|
|
|
|
|
|
def is_forum_message(message: Message) -> bool:
|
|
return getattr(message, "message_thread_id", None) is not None
|
|
|
|
|
|
def resolve_forum_chat_id(message: Message, tg_user_id: int) -> str | None:
|
|
thread_id = getattr(message, "message_thread_id", None)
|
|
if thread_id is None:
|
|
return None
|
|
|
|
chat = db.get_chat_by_thread(tg_user_id, thread_id)
|
|
if not chat:
|
|
return None
|
|
return chat["chat_id"]
|
|
|
|
|
|
def _extract_attachments(message: Message) -> list[Attachment]:
|
|
attachments: list[Attachment] = []
|
|
if message.photo:
|
|
file = message.photo[-1]
|
|
attachments.append(Attachment(
|
|
type="image",
|
|
url=f"tg://file/{file.file_id}",
|
|
mime_type="image/jpeg",
|
|
))
|
|
if message.document:
|
|
attachments.append(Attachment(
|
|
type="document",
|
|
url=f"tg://file/{message.document.file_id}",
|
|
mime_type=message.document.mime_type or "application/octet-stream",
|
|
filename=message.document.file_name,
|
|
))
|
|
if message.voice:
|
|
attachments.append(Attachment(
|
|
type="audio",
|
|
url=f"tg://file/{message.voice.file_id}",
|
|
mime_type="audio/ogg",
|
|
))
|
|
return attachments
|
|
|
|
|
|
def format_outgoing(chat_name: str, event: OutgoingEvent, *, prefix: bool = True) -> str:
|
|
rendered_prefix = f"[{chat_name}] " if prefix else ""
|
|
if isinstance(event, OutgoingMessage):
|
|
return rendered_prefix + event.text
|
|
if isinstance(event, OutgoingUI):
|
|
return rendered_prefix + event.text
|
|
return rendered_prefix + str(event)
|