"""Сериализация Telethon-объектов в JSON-данные тулов.
Правила (pitfall): tl-объекты никогда не сериализуются через to_dict() —
только явные поля здесь. Entity наружу — id в bot-API-маркировке
(telethon.utils.get_peer_id: user > 0, chat > 0 без минуса по-telethon-ски
… см. get_peer_id), type, username, title; access_hash не светим — повторный
доступ к чату через client.get_input_entity(dialog_id) (резолвит из кеша
StringSession, вот зачем нужен persist).
"""
from telethon.utils import get_peer_id as _tg_peer_id
from app.tg.waveform import unpack_waveform
def marked_peer_id(peer) -> int:
"""Entity/Peer/int → bot-API-маркированный int id (Telethon.utils).
Юзер >0, малая группа -id, канал/супергруппа -100id; именно эта
маркировка принимается назад через get_input_entity. None-safe."""
if peer is None:
return 0
try:
return _tg_peer_id(peer)
except (ValueError, AttributeError, TypeError):
return 0
def entity_dict(entity, *, short: bool = False) -> dict:
"""User/Chat/Channel → плоский dict. id — маркированный int.
Меняются через get_input_entity(dialog_id): юзер >0, группа -id,
канал -100id (всё единообразно — маркировка Telethon)."""
if entity is None:
return {"id": 0, "type": "unknown"}
kind = type(entity).__name__ # User / Chat / Channel / ChatForbidden...
if kind == "User":
out = {
"id": marked_peer_id(entity), # сами объекты маркируются правильно
"type": "user",
"title": " ".join(p for p in (getattr(entity, "first_name", ""),
getattr(entity, "last_name", "")) if p),
"username": getattr(entity, "username", "") or "",
"is_bot": bool(getattr(entity, "bot", False)),
"is_self": bool(getattr(entity, "is_self", False)),
}
elif kind == "Channel":
out = {
"id": marked_peer_id(entity),
"type": "channel",
"title": getattr(entity, "title", "") or "",
"username": getattr(entity, "username", "") or "",
"is_broadcast": bool(getattr(entity, "broadcast", False)),
}
else: # Chat / ChatForbidden / ChatEmpty
out = {
"id": marked_peer_id(entity) if kind != "ChatForbidden" else getattr(entity, "id", 0),
"type": "chat",
"title": getattr(entity, "title", "") or "",
}
if short:
return {"id": out["id"], "type": out["type"], "title": out.get("title", ""),
"username": out.get("username", "")}
return out
def dialog_dict(dialog, entity) -> dict:
"""Диалог: entity + счётчики + превью последнего сообщения."""
d = entity_dict(entity)
d["unread_count"] = getattr(dialog, "unread_count", 0)
d["pinned"] = bool(getattr(dialog, "pinned", False))
d["is_muted"] = bool(getattr(dialog, "muted", False)) or bool(
getattr(getattr(dialog, "notify_settings", None), "mute_until", 0)
)
msg = getattr(dialog, "message", None)
if msg is not None:
d["last_message"] = message_dict(msg)
return d
def media_meta(message) -> dict:
"""Описание медиа без скачивания: тип, имя, размер, duration, dimensions.
voice = DocumentAttributeAudio(voice=True); round = DocumentAttributeVideo(
round_message=True) — оба Document; это важно и для download_media и для UI.
"""
media = getattr(message, "media", None)
if media is None:
return {}
kind = type(media).__name__
out = {"kind": kind}
doc = getattr(media, "document", None)
photo = getattr(media, "photo", None)
if kind == "MessageMediaDocument" and doc is not None:
attrs = list(getattr(doc, "attributes", []) or [])
for attr in attrs:
aname = type(attr).__name__
if aname == "DocumentAttributeAudio" and getattr(attr, "voice", False):
out["voice"] = True
out["duration"] = getattr(attr, "duration", None)
out["waveform"] = unpack_waveform(getattr(attr, "waveform", None) or b"")
elif aname == "DocumentAttributeAudio":
out["audio"] = True
out["duration"] = getattr(attr, "duration", None)
out["title"] = getattr(attr, "title", "") or ""
out["performer"] = getattr(attr, "performer", "") or ""
elif aname == "DocumentAttributeVideo":
if getattr(attr, "round_message", False):
out["round"] = True
else:
out["video"] = True
out["duration"] = getattr(attr, "duration", None)
w = getattr(attr, "w", 0) or 0
h = getattr(attr, "h", 0) or 0
if w or h:
out["size_hint"] = {"w": w, "h": h}
elif aname == "DocumentAttributeSticker":
out["sticker"] = True
out["emoji"] = getattr(attr, "alt", "") or ""
elif aname == "DocumentAttributeImageSize":
out["size_hint"] = {"w": getattr(attr, "w", 0), "h": getattr(attr, "h", 0)}
elif aname == "DocumentAttributeFilename":
out["file_name"] = getattr(attr, "file_name", "") or ""
out["mime_type"] = getattr(doc, "mime_type", "") or ""
out["size"] = getattr(doc, "size", 0) or 0
elif kind == "MessageMediaPhoto" and photo is not None:
out["photo"] = True
# размер фото — по самому большому size у photo.sizes (обычный int)
sizes = [s.size for s in (getattr(photo, "sizes", None) or [])
if isinstance(getattr(s, "size", None), int)]
out["size"] = max(sizes) if sizes else None
elif kind == "MessageMediaWebPage":
wp = getattr(media, "webpage", None)
out["webpage_url"] = getattr(wp, "url", "") or ""
out["webpage_title"] = getattr(wp, "title", "") or ""
elif kind == "MessageMediaGeo":
geo = getattr(media, "geo", None)
out["geo"] = {"lat": getattr(geo, "lat", None), "long": getattr(geo, "long", None)}
return out
def message_dict(message, *, with_media: bool = True) -> dict:
"""Message → плоский dict. MessageService → экшен-строка (join/leave)."""
if type(message).__name__ == "MessageService":
return {
"id": message.id,
"sender_id": marked_peer_id(getattr(message, "from_id", None) or getattr(message, "peer_id", None)),
"action": type(getattr(message, "action", None)).__name__,
}
out = {
"id": message.id,
"date": message.date.isoformat() if getattr(message, "date", None) else None,
"sender_id": marked_peer_id(message.from_id),
"out": bool(message.out),
"text": message.message or "",
"reply_to": getattr(message.reply_to, "reply_to_msg_id", None)
if message.reply_to is not None else None,
"edited": getattr(message, "edit_date", None).isoformat()
if getattr(message, "edit_date", None) else None,
"views": getattr(message, "views", None),
}
if with_media:
out["media"] = media_meta(message)
sender = getattr(message, "sender", None)
if sender is not None:
ent = entity_dict(sender, short=True)
out["sender"] = {"title": ent.get("title"), "username": ent.get("username", "")}
return out
