Message Logger: show logged edits and deletions from telegram-panel [no bump]

Keeps deleted messages (75% opacity, red icon-pack trash icon), restores
logged messages deleted while away, reports deletions, and shows edit
history as a mock chat (Edit history in the message menu, Open topic).
Message links and history open as any signed-in account with access.
Per-master API tokens; chat state synced per checked range with an hour's
validity, cached in SQLite with a configurable lifetime.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/views/history.py

@@ -0,0 +1,368 @@
+"""Edit history as a mock chat: every version of a message drawn as a native
+message bubble, in a bottom sheet over the chat wallpaper.
+
+The versions live in the logger's activity group, re-posted without a header.
+Only a member can read them (and load their media): the master account owning
+the logger groups, or another account in that group. When one is signed in on
+this device, the versions are fetched and drawn through it (the account viewing
+the history if it can); otherwise the server's text is shown and each media is
+marked unavailable. Posts of the logger groups themselves (saved posts, history
+messages) open the history of the message they belong to.
+"""
+
+import accounts
+import diag
+import peers
+import tg
+from api import ApiError
+from jutil import in_background, on_ui
+
+_MEDIA_NAMES = {
+    "photo": "Photo",
+    "video": "Video",
+    "animation": "GIF",
+    "audio": "Audio",
+    "voice": "Voice message",
+    "video_note": "Video message",
+    "sticker": "Sticker",
+    "document": "File",
+}
+
+
+def _bulletin(text, error=False):
+    def show():
+        from ui.bulletin import BulletinHelper
+
+        (BulletinHelper.show_error if error else BulletinHelper.show_info)(text)
+
+    on_ui(show)
+
+
+def _when(iso):
+    if not iso:
+        return ""
+    from datetime import datetime
+
+    try:
+        return datetime.fromisoformat(iso).astimezone().strftime("%d %b %Y, %H:%M")
+    except ValueError:
+        return iso
+
+
+def _epoch(iso):
+    if not iso:
+        return 0
+    from datetime import datetime
+
+    try:
+        return int(datetime.fromisoformat(iso).timestamp())
+    except ValueError:
+        return 0
+
+
+def _entities(items):
+    """Telethon entity dicts (server) -> TLRPC entities."""
+    from java.util import ArrayList
+    from org.telegram.tgnet import TLRPC
+
+    out = ArrayList()
+    for item in items:
+        name = str(item.get("_", ""))
+        cls = getattr(TLRPC, "TL_" + name[:1].lower() + name[1:], None) if name else None
+        if cls is None:
+            continue
+        entity = cls()
+        entity.offset = int(item["offset"])
+        entity.length = int(item["length"])
+        for key in ("url", "language", "document_id"):
+            if item.get(key) is not None and hasattr(entity, key):
+                setattr(entity, key, item[key])
+        out.add(entity)
+    return out
+
+
+def _utf16(text):
+    return len(text.encode("utf-16-le")) // 2
+
+
+class History:
+    def __init__(self, plugin, api):
+        self.plugin = plugin
+        self.api = api
+
+    def open(self, message):
+        """Shows the edit history of a MessageObject."""
+        if message is None:
+            return
+        if not self.api.configured:
+            _bulletin("Message Logger: set the server URL and API token first", True)
+            return
+        account = int(message.currentAccount)
+        chat_id = peers.marked_chat_id(account, int(message.getDialogId()))
+        if chat_id is None or int(message.getId()) <= 0:
+            _bulletin("No edit history for this message")
+            return
+        in_background(lambda: self._load(message, account, chat_id))
+
+    def _load(self, original, account, chat_id):
+        try:
+            result = self.api.versions(
+                peers.own_user_id(account), chat_id, int(original.getId())
+            )
+        except ApiError as exc:
+            _bulletin(f"Message Logger: {exc}", True)
+            return
+        versions = (result or {}).get("versions") or []
+        if len(versions) < 2:
+            _bulletin("No edits logged for this message")
+            return
+        ids = [int(v["message_id"]) for v in versions if v.get("message_id")]
+        if not result.get("chat_id") or not ids:
+            on_ui(lambda: self._show(original, versions, None, {}, None))
+            return
+        channel_id = peers.channel_id_from_marked(result["chat_id"])
+        # Whoever can read the activity group: this account, the master, or
+        # any other signed-in member.
+        readers = accounts.with_access(
+            -channel_id, prefer=(account, accounts.slot_of(result.get("master_tg_id")))
+        )
+        if not readers:
+            on_ui(lambda: self._show(original, versions, None, {}, None))
+            return
+        reader = readers[0]
+        # The history topic, at its newest version.
+        topic = (channel_id, result.get("topic_id"), ids[-1])
+
+        def done(found, error):
+            if found is None:
+                diag.log(f"history: activity group unreadable: {error}")
+                found = {}
+            on_ui(lambda: self._show(original, versions, reader, found, topic))
+
+        tg.get_messages(reader, -channel_id, ids, done)
+
+    # building the mock messages
+
+    def _base(self, original, version):
+        from org.telegram.tgnet import TLRPC
+
+        owner = original.messageOwner
+        msg = TLRPC.TL_message()
+        # Its own (negative) id: the original's would give the bubbles the
+        # original's marks (a deleted message's trash icon and opacity).
+        msg.id = -(int(version.get("version") or 0) + 1)
+        msg.dialog_id = int(original.getDialogId())
+        msg.peer_id = owner.peer_id
+        msg.from_id = owner.from_id
+        msg.flags |= 256
+        msg.out = bool(owner.out)
+        msg.date = _epoch(version.get("edit_date")) or int(owner.date)
+        return msg
+
+    def _from_copy(self, original, version, copy):
+        from org.telegram.tgnet import TLRPC
+
+        msg = self._base(original, version)
+        msg.message = copy.message or ""
+        msg.entities = copy.entities
+        media = getattr(copy, "media", None)
+        if media is not None and not isinstance(media, TLRPC.TL_messageMediaEmpty):
+            msg.media = media
+            msg.flags |= 512
+        rich = getattr(copy, "rich_message", None)
+        if rich is not None:
+            msg.rich_message = rich
+            msg.flags2 |= 1 << 12
+        return msg
+
+    def _from_text(self, original, version):
+        """The server's text; each media marked unavailable after it."""
+        from org.telegram.tgnet import TLRPC
+
+        msg = self._base(original, version)
+        text = version.get("text") or ""
+        entities = _entities(version.get("entities") or [])
+        notes = [
+            f"🚫 {_MEDIA_NAMES.get(kind, 'Media')} unavailable"
+            for kind in version.get("media") or []
+        ]
+        if notes:
+            note = "\n".join(notes) + "\n(sign in an account of the logger's groups to load media)"
+            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
+        return msg
+
+    def _message_object(self, account, original, msg):
+        from org.telegram.messenger import MessageObject, MessagesController
+
+        viewer = int(original.currentAccount)
+        if account != viewer:
+            # The bubble is drawn as the reading account (media); names come
+            # from the viewer's cache.
+            source = MessagesController.getInstance(viewer)
+            target = MessagesController.getInstance(account)
+            sender = original.getFromChatId()
+            if sender > 0 and source.getUser(sender) is not None:
+                target.putUser(source.getUser(sender), True)
+            elif sender < 0 and source.getChat(-sender) is not None:
+                target.putChat(source.getChat(-sender), True)
+        return MessageObject(account, msg, True, True)
+
+    # the sheet
+
+    def _show(self, original, versions, reader, found, topic):
+        from android.view import Gravity
+        from android.widget import LinearLayout, ScrollView, TextView
+        from client_utils import get_last_fragment
+        from java import jint
+        from org.telegram.messenger import AndroidUtilities
+        from org.telegram.ui.ActionBar import BottomSheet, Theme
+
+        fragment = get_last_fragment()
+        context = fragment.getParentActivity() if fragment is not None else None
+        if context is None:
+            return
+        dp = AndroidUtilities.dp
+        column = LinearLayout(context)
+        column.setOrientation(LinearLayout.VERTICAL)
+        column.setPadding(0, dp(8), 0, dp(12))
+        last = len(versions) - 1
+        for index, version in enumerate(versions):
+            label = "Original" if version["version"] == 0 else f"Edit {version['version']}"
+            if index == last:
+                label += " · current"
+            when = _when(version.get("edit_date"))
+            column.addView(self._service(context, f"{label} · {when}" if when else label))
+            copy = found.get(int(version.get("message_id") or 0)) if reader is not None else None
+            if copy is not None and not isinstance(copy, _message_empty()):
+                msg, account = self._from_copy(original, version, copy), reader
+            else:
+                msg, account = self._from_text(original, version), int(original.currentAccount)
+            try:
+                column.addView(
+                    self._bubble(context, account, self._message_object(account, original, msg), version)
+                )
+            except Exception as exc:
+                diag.error("history.bubble", exc)
+
+        scroll = ScrollView(context)
+        scroll.addView(column)
+        wallpaper = Theme.getCachedWallpaper()
+        if wallpaper is not None:
+            scroll.setBackground(wallpaper)
+        else:
+            scroll.setBackgroundColor(Theme.getColor(Theme.key_windowBackgroundGray))
+
+        root = LinearLayout(context)
+        root.setOrientation(LinearLayout.VERTICAL)
+        header = LinearLayout(context)
+        header.setOrientation(LinearLayout.HORIZONTAL)
+        header.setGravity(Gravity.CENTER_VERTICAL)
+        title = TextView(context)
+        title.setText("Edit history" + ("" if reader is not None else " · media unavailable"))
+        title.setTextSize(17)
+        title.setTextColor(jint(Theme.getColor(Theme.key_dialogTextBlack)))
+        title.setPadding(dp(20), dp(14), dp(12), dp(12))
+        header.addView(title, LinearLayout.LayoutParams(0, -2, 1.0))
+        sheet = BottomSheet(context, False)
+        if reader is not None and topic is not None and topic[1]:
+            header.addView(self._topic_button(context, sheet, fragment, reader, topic))
+        root.addView(header)
+        root.addView(
+            scroll,
+            LinearLayout.LayoutParams(-1, int(AndroidUtilities.displaySize.y * 0.7)),
+        )
+
+        sheet.setCustomView(root)
+        sheet.show()
+
+    def _topic_button(self, context, sheet, fragment, reader, topic):
+        """Opens the history topic of the activity group, as `reader` (an
+        account that can read it) when the current one can't."""
+        from android.widget import TextView
+        from android_utils import OnClickListener
+        from java import jint
+        from org.telegram.messenger import AndroidUtilities
+        from org.telegram.ui.ActionBar import Theme
+
+        from views import open_as
+
+        dp = AndroidUtilities.dp
+        button = TextView(context)
+        button.setText("Open topic")
+        button.setTextSize(15)
+        button.setTextColor(jint(Theme.getColor(Theme.key_dialogTextBlue2)))
+        button.setPadding(dp(12), dp(14), dp(20), dp(12))
+        channel_id, topic_id, message_id = topic
+
+        def open_topic(_view):
+            sheet.dismiss()
+            open_as.open_topic(fragment, reader, channel_id, topic_id, message_id)
+
+        button.setOnClickListener(OnClickListener(open_topic))
+        return button
+
+    def _service(self, context, text):
+        """A date-like service pill above each version."""
+        from android.view import Gravity
+        from android.widget import FrameLayout, TextView
+        from java import jint
+        from org.telegram.messenger import AndroidUtilities
+        from org.telegram.ui.ActionBar import Theme
+
+        dp = AndroidUtilities.dp
+        pill = TextView(context)
+        pill.setText(text)
+        pill.setTextSize(13)
+        pill.setTextColor(jint(Theme.getColor(Theme.key_chat_serviceText)))
+        pill.setPadding(dp(10), dp(3), dp(10), dp(4))
+        pill.setBackground(
+            Theme.createRoundRectDrawable(
+                dp(12), jint(Theme.getColor(Theme.key_chat_serviceBackground))
+            )
+        )
+        holder = FrameLayout(context)
+        holder.setPadding(0, dp(10), 0, dp(4))
+        params = FrameLayout.LayoutParams(-2, -2)
+        params.gravity = Gravity.CENTER_HORIZONTAL
+        holder.addView(pill, params)
+        return holder
+
+    def _bubble(self, context, account, message_object, version):
+        """A ChatMessageCell for the version, under a layer taking its touches
+        (no chat delegate behind it); long-press copies the text."""
+        from android.view import View
+        from android.widget import FrameLayout
+        from android_utils import OnClickListener, OnLongClickListener, copy_to_clipboard
+        from org.telegram.ui.Cells import ChatMessageCell
+
+        cell = ChatMessageCell(context, account)
+        cell.setMessageObject(message_object, None, False, False, False)
+        holder = FrameLayout(context)
+        holder.addView(cell, FrameLayout.LayoutParams(-1, -2))
+        shield = View(context)
+        text = str(message_object.messageOwner.message or version.get("text") or "")
+
+        def copy(_view):
+            if text:
+                copy_to_clipboard(text)
+                _bulletin("Copied")
+            return True
+
+        shield.setOnClickListener(OnClickListener(lambda _view: None))
+        shield.setOnLongClickListener(OnLongClickListener(copy))
+        holder.addView(shield, FrameLayout.LayoutParams(-1, -1))
+        return holder
+
+
+def _message_empty():
+    from org.telegram.tgnet import TLRPC
+
+    return TLRPC.TL_messageEmpty
+