generated from exteraGramPlugins/pluginBase
lyuksovannyyandClaude Opus 5.5
a2e4075a03
Publish plugin / publish (push) Successful in 4s
The comments button of a deleted post opens the app's own comments section (discussion thread), its history request answered locally from the logged comments; the input bar is replaced by a "Viewing deleted comments" notice. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
384 lines
14 KiB
Python
384 lines
14 KiB
Python
"""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):
|
|
entries = []
|
|
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"))
|
|
entries.append((f"{label} · {when}" if when else label, version))
|
|
self._sheet(original, "Edit history", entries, reader, found, topic)
|
|
|
|
def _message_for(self, original, item, copy, reader):
|
|
"""(TL message, account to draw it as) of an entry: from its copy in
|
|
the activity group when readable, else from the server's text."""
|
|
if copy is not None and not isinstance(copy, _message_empty()):
|
|
return self._from_copy(original, item, copy), reader
|
|
return self._from_text(original, item), int(original.currentAccount)
|
|
|
|
def _sheet(self, original, title_text, entries, reader, found, topic):
|
|
"""The mock chat: a service pill (label) and a bubble per entry."""
|
|
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))
|
|
for label, item in entries:
|
|
column.addView(self._service(context, label))
|
|
copy = (
|
|
found.get(int(item.get("message_id") or 0)) if reader is not None else None
|
|
)
|
|
try:
|
|
msg, account = self._message_for(original, item, copy, reader)
|
|
column.addView(
|
|
self._bubble(
|
|
context, account, self._message_object(account, original, msg), item
|
|
)
|
|
)
|
|
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(title_text + ("" 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
|
|
|