messageLogger/src/views/removed_chat.py

"""The history of a removed chat, rebuilt from the logger: the logged messages
(`/chats/{id}/history`), each from its copy in a log group when an account
here can read it (header stripped, media included), else from the server's
text with its media marked unavailable. All of them are deleted for the
account; they are cached as such (store), so they draw like any deleted
message.

`History` holds one chat's messages, newest first, and answers the app's
getHistory/getMessages for it (hooks/removed_chats.py).
"""

import threading

import diag
import media_share
import peers
import store
import tg
from api import ApiError
from views.history import _MEDIA_NAMES, _entities, _epoch, _utf16

_PAGES = 3  # history pages (of 100) loaded when the chat opens
_COPIES_PER_REQUEST = 100


def stand_in_user(controller, peer_id, name):
    from org.telegram.tgnet import TLRPC

    user = controller.getUser(int(peer_id))
    if user is None:
        user = TLRPC.TL_user()
        user.id = int(peer_id)
        user.first_name = name or "Deleted"
        user.flags |= 2
    return user


def stand_in_chat(controller, chat_id, title, kind):
    """The app's chat (cached or stored), else a stand-in of the right kind."""
    from org.telegram.tgnet import TLRPC

    chat = controller.getChat(int(chat_id))
    if chat is not None:
        return chat
    if kind == "group":
        chat = TLRPC.TL_chat()
    else:
        chat = TLRPC.TL_channel()
        chat.broadcast = kind == "channel"
        chat.megagroup = kind != "channel"
    chat.id = int(chat_id)
    chat.title = title or "Chat"
    return chat


class History:
    def __init__(self, api, account, dialog_id, chat_id):
        self.api = api
        self.account = account
        self.dialog_id = dialog_id
        self.chat_id = chat_id
        self.messages = []  # TL messages, newest first
        self.users = {}
        self.chats = {}
        self.loaded = False
        self._lock = threading.Lock()

    # loading (background)

    def load(self):
        """Fetches and builds the chat's history once; later calls wait."""
        with self._lock:
            if self.loaded:
                return
            try:
                self._load()
            except Exception as exc:
                diag.error("removed_chat.load", exc)
            self.loaded = True

    def _load(self):
        me = peers.own_user_id(self.account)
        items, before, master = [], 0, None
        for _page in range(_PAGES):
            try:
                result = self.api.history(me, self.chat_id, before)
            except ApiError as exc:
                diag.log(f"removed chat {self.chat_id}: {exc}")
                break
            page = (result or {}).get("messages") or []
            if not page:
                break
            master = result.get("master_tg_id")
            items += page
            before = min(int(m["message_id"]) for m in page)
            if len(page) < 100:
                break
        revealed = store.is_revealed(self.account, self.dialog_id)
        items = [i for i in items if revealed or not i.get("hidden_at")]
        store.put_server_states(self.account, self.dialog_id, items)
        copies = self._copies(items, master)
        built = []
        for item in items:
            try:
                built.append(self._message(item, copies.get(self._pointer(item))))
            except Exception as exc:
                diag.error("removed_chat.message", exc)
        self.messages = sorted(built, key=lambda m: -int(m.id))
        diag.log(f"removed chat {self.chat_id}: {len(self.messages)} messages")

    @staticmethod
    def _pointer(item):
        logged = item.get("logged") or {}
        if not logged.get("chat_id") or not logged.get("message_id"):
            return None
        return int(logged["chat_id"]), int(logged["message_id"])

    def _copies(self, items, master):
        """Logged copies by (log chat, message), read through an account here
        that can, per log group."""
        by_group = {}
        for item in items:
            pointer = self._pointer(item)
            if pointer is not None:
                by_group.setdefault(pointer[0], set()).add(pointer[1])
        found = {}
        masters = [master, *self.api.masters()]
        for log_chat, ids in by_group.items():
            channel_id = peers.channel_id_from_marked(log_chat)
            reader = media_share.reader_for(self.account, log_chat, masters)
            if reader is None:
                continue
            ordered = sorted(ids)
            for start in range(0, len(ordered), _COPIES_PER_REQUEST):
                chunk = ordered[start : start + _COPIES_PER_REQUEST]
                for message_id, copy in _fetch(reader, -channel_id, chunk).items():
                    if reader != self.account:
                        media_share.copy_media(reader, copy)
                    found[(log_chat, message_id)] = copy
        return found

    # building

    def _message(self, item, copy):
        from org.telegram.messenger import MessagesController
        from org.telegram.tgnet import TLRPC

        from hooks.chat import _android_peer_id, build_from_copy

        controller = MessagesController.getInstance(self.account)
        message_id = int(item["message_id"])
        if copy is not None and not isinstance(copy, TLRPC.TL_messageEmpty):
            msg = build_from_copy(
                self.account, self.dialog_id, message_id, item, copy, True
            ).messageOwner
        else:
            msg = from_text(self.account, self.dialog_id, item)
        sender = item.get("sender_id")
        if sender:
            peer_id = _android_peer_id(int(sender))
            if peer_id > 0:
                user = stand_in_user(controller, peer_id, item.get("sender_name"))
                self.users[int(user.id)] = user
            else:
                chat = stand_in_chat(controller, -peer_id, item.get("sender_name"), "channel")
                self.chats[int(chat.id)] = chat
        if item.get("reply_to"):
            reply = TLRPC.TL_messageReplyHeader()
            reply.reply_to_msg_id = int(item["reply_to"])
            reply.flags |= 16
            msg.reply_to = reply
            msg.flags |= 8
        return msg

    # answering the app's requests

    def response(self, picked):
        from java.util import ArrayList
        from org.telegram.tgnet import TLRPC

        response = TLRPC.TL_messages_messages()
        response.messages = ArrayList()
        for message in picked:
            response.messages.add(message)
        response.users = ArrayList()
        for user in self.users.values():
            response.users.add(user)
        response.chats = ArrayList()
        for chat in self.chats.values():
            response.chats.add(chat)
        return response

    def page(self, offset_id, add_offset, limit):
        """getHistory's slice: newest first, from `offset_id` (exclusive; 0:
        the newest) moved by `add_offset` (negative: newer ones too)."""
        messages = self.messages
        if offset_id:
            start = next(
                (i for i, m in enumerate(messages) if int(m.id) < offset_id),
                len(messages),
            )
        else:
            start = 0
        start = max(0, start + add_offset)
        return messages[start : start + max(1, limit)]

    def by_ids(self, ids):
        wanted = set(int(i) for i in ids)
        return [m for m in self.messages if int(m.id) in wanted]


def from_text(account, dialog_id, item):
    """A logged message from the server's text; each media is marked
    unavailable after it."""
    from org.telegram.messenger import MessagesController
    from org.telegram.tgnet import TLRPC

    from hooks.chat import _android_peer_id

    controller = MessagesController.getInstance(account)
    msg = TLRPC.TL_message()
    msg.id = int(item["message_id"])
    msg.dialog_id = dialog_id
    msg.peer_id = controller.getPeer(dialog_id)
    sender = item.get("sender_id")
    msg.from_id = (
        controller.getPeer(_android_peer_id(int(sender))) if sender else msg.peer_id
    )
    msg.flags |= 256
    msg.date = _epoch(item.get("source_date"))
    if int(item.get("edit_count") or 0) > 0:
        msg.edit_date = _epoch(item.get("edit_date")) or msg.date
        msg.flags |= 32768
    text = item.get("text") or ""
    entities = _entities(item.get("entities") or [])
    notes = [
        f"🚫 {_MEDIA_NAMES.get(kind, 'Media')} unavailable"
        for kind in item.get("media") or []
    ]
    if notes:
        note = "\n".join(notes)
        start = _utf16(text) + (2 if text else 0)
        text = (text + "\n\n" if text else "") + note
        italic = TLRPC.TL_messageEntityItalic()
        italic.offset = start
        italic.length = _utf16(note)
        entities.add(italic)
    msg.message = text or "(empty)"
    msg.entities = entities
    msg.out = sender is not None and int(sender) == peers.own_user_id(account)
    return msg


def _fetch(reader, dialog_id, ids):
    """Messages `ids` of a log group through `reader` (blocking)."""
    done = threading.Event()
    box = {}

    def callback(found, error):
        if found is None:
            diag.log(f"removed chat copies unreadable: {error}")
        box["found"] = found or {}
        done.set()

    tg.get_messages(reader, dialog_id, ids, callback)
    done.wait(20)
    return box.get("found") or {}