generated from exteraGramPlugins/pluginBase
lyuksovannyy
196da742de
Replace the sample plugin with an AI chat summarizer that renders its
output as a native RichMessage injected into the chat.
Summarization pipeline:
- Fetch chat/topic history via paginated TL requests.
- Send a system prompt, the user's instructions and an anonymized JSON
transcript as three separate roles. Only message text and attachment
names leave the device; senders and message ids are replaced by
indices, and out-of-range reply parents are included with negative ids
so replies keep their context.
- Stream the response over SSE, reading reasoning from reasoning_content,
reasoning or inline <think> tags.
- Substitute the model's {userN}/{msgN} placeholders back into real names
and deep links. Placeholders are used instead of Markdown links because
a half-streamed link renders as broken literal text.
Rendering:
- Compile Markdown into TL_iv page blocks (headings, lists, quotes, code,
dividers, inline styles) since the app's own converter works off
Android Spanned spans rather than Markdown source.
- Show progress stages immediately on click, animate reasoning while it
streams, then collapse it into a tappable details block.
- Attribute the card to a client-side mock sender and insert it directly
after the newest analyzed message, scrolling it into view.
Persistence:
- Store summaries in SQLite keyed by (dialog, topic, anchor) and
re-inject them when the chat is reopened.
Also add tools/release.py and a Gitea workflow that builds, verifies and
publishes dist/<id>.plugin to the Releases tab, and extend
tools/verify_bundle.py to 104 offline checks covering the transcript,
streaming, rendering, injection and storage behaviour.
366 lines
14 KiB
Python
366 lines
14 KiB
Python
"""Hooks two independent entry points so holding either opens the summarize
|
|
dialog, pre-filled with the chat's actual unread count/offset:
|
|
|
|
1. ChatActivity.onPageDownClicked() -- the floating "jump to newest" button.
|
|
This build exposes no long-click listener on that button; its only
|
|
stable hookable entry point is this method. To distinguish a hold from a
|
|
tap, touch DOWN/UP timing is captured globally at
|
|
LaunchActivity.dispatchTouchEvent and consulted here.
|
|
2. ChatUnreadCell -- the inline "N unread messages" divider shown in the
|
|
message list. Its constructor is hooked so a click/long-click listener
|
|
can be attached directly to each instantiated cell.
|
|
|
|
Reflection notes (grounded on-device): hook_utils.find_class(...) returns a
|
|
Chaquopy class wrapper, not a java.lang.Class -- only .getClass() reaches
|
|
getDeclaredMethods()/getDeclaredConstructors(). Methods are looked up by
|
|
enumerating getDeclaredMethods() and matching by name rather than by exact
|
|
getDeclaredMethod(name, *paramTypes) signature, since resolving the correct
|
|
Java parameter-type Class objects from Python is brittle and unnecessary
|
|
when the method name alone is unambiguous.
|
|
"""
|
|
import time
|
|
from typing import Any, List, Optional
|
|
|
|
import diagnostics
|
|
from config import LONG_PRESS_SECONDS, MAX_MSG_COUNT
|
|
|
|
_GESTURE = {"down_at": 0.0, "released_at": 0.0, "consumed": False}
|
|
_STALE_GESTURE_SECONDS = 0.75
|
|
|
|
|
|
def _held_duration() -> float:
|
|
down_at = _GESTURE["down_at"]
|
|
if not down_at:
|
|
return 0.0
|
|
released_at = _GESTURE["released_at"]
|
|
end = released_at or time.monotonic()
|
|
if released_at and (time.monotonic() - released_at) > _STALE_GESTURE_SECONDS:
|
|
return 0.0
|
|
return max(0.0, end - down_at)
|
|
|
|
|
|
def _class_candidates(cls: Any) -> List[Any]:
|
|
"""Returns every object that might expose real reflection members, best first."""
|
|
candidates: List[Any] = []
|
|
if cls is None:
|
|
return candidates
|
|
try:
|
|
inner = cls.getClass()
|
|
if inner is not None:
|
|
candidates.append(inner)
|
|
except Exception:
|
|
pass
|
|
if not any(cls is c for c in candidates):
|
|
candidates.append(cls)
|
|
return candidates
|
|
|
|
|
|
def _find_methods_by_name(cls: Any, name: str) -> List[Any]:
|
|
"""Enumerates declared/public methods matching `name` across candidates."""
|
|
for source in _class_candidates(cls):
|
|
for accessor in ("getDeclaredMethods", "getMethods"):
|
|
if not hasattr(source, accessor):
|
|
continue
|
|
try:
|
|
methods = getattr(source, accessor)()
|
|
except Exception as exc:
|
|
diagnostics.record_error(f"_find_methods_by_name:{accessor}", exc)
|
|
continue
|
|
found = [m for m in methods if _method_name(m) == name]
|
|
if found:
|
|
return found
|
|
return []
|
|
|
|
|
|
def _method_name(method: Any) -> Optional[str]:
|
|
try:
|
|
return method.getName()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _find_constructors(cls: Any) -> List[Any]:
|
|
for source in _class_candidates(cls):
|
|
if not hasattr(source, "getDeclaredConstructors"):
|
|
continue
|
|
try:
|
|
ctors = list(source.getDeclaredConstructors())
|
|
except Exception as exc:
|
|
diagnostics.record_error("_find_constructors", exc)
|
|
continue
|
|
if ctors:
|
|
return ctors
|
|
return []
|
|
|
|
|
|
class UnreadHook:
|
|
"""Installs the touch-timing tracker plus both pagedown-button and
|
|
unread-divider-cell interceptors.
|
|
"""
|
|
|
|
def __init__(self, plugin):
|
|
self.plugin = plugin
|
|
self._unhooks: List[Any] = []
|
|
self._hooked_view_ids: set = set()
|
|
|
|
def install(self) -> None:
|
|
try:
|
|
from base_plugin import MethodHook
|
|
from hook_utils import find_class
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.install:imports", exc)
|
|
return
|
|
|
|
self._install_touch_tracker(find_class, MethodHook)
|
|
self._install_pagedown_hook(find_class, MethodHook)
|
|
self._install_unread_cell_hook(find_class, MethodHook)
|
|
|
|
def _install_touch_tracker(self, find_class, MethodHook) -> None:
|
|
gesture_ref = _GESTURE
|
|
launch_activity_class = find_class("org.telegram.ui.LaunchActivity")
|
|
diagnostics.note(f"unread_hook.install: LaunchActivity found={bool(launch_activity_class)}")
|
|
if not launch_activity_class:
|
|
return
|
|
methods = _find_methods_by_name(launch_activity_class, "dispatchTouchEvent")
|
|
diagnostics.note(f"unread_hook.install: dispatchTouchEvent methods found={len(methods)}")
|
|
if not methods:
|
|
return
|
|
|
|
class _TouchTrackHook(MethodHook):
|
|
def before_hooked_method(self, param):
|
|
try:
|
|
event = param.args[0]
|
|
action = event.getActionMasked()
|
|
except Exception:
|
|
return
|
|
now = time.monotonic()
|
|
if action == 0: # ACTION_DOWN
|
|
gesture_ref["down_at"] = now
|
|
gesture_ref["released_at"] = 0.0
|
|
gesture_ref["consumed"] = False
|
|
elif action == 1: # ACTION_UP
|
|
gesture_ref["released_at"] = now
|
|
elif action == 3: # ACTION_CANCEL
|
|
gesture_ref["down_at"] = 0.0
|
|
gesture_ref["released_at"] = 0.0
|
|
gesture_ref["consumed"] = False
|
|
|
|
for method in methods:
|
|
try:
|
|
ref = self.plugin.hook_method(method, _TouchTrackHook())
|
|
diagnostics.note(f"unread_hook.install: dispatchTouchEvent hook ref={ref!r}")
|
|
if ref:
|
|
self._unhooks.append(ref)
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.install:dispatchTouchEvent", exc)
|
|
|
|
def _install_pagedown_hook(self, find_class, MethodHook) -> None:
|
|
chat_activity_class = find_class("org.telegram.ui.ChatActivity")
|
|
diagnostics.note(f"unread_hook.install: ChatActivity found={bool(chat_activity_class)}")
|
|
if not chat_activity_class:
|
|
diagnostics.record_error("unread_hook.install", RuntimeError("ChatActivity class not found"))
|
|
return
|
|
methods = _find_methods_by_name(chat_activity_class, "onPageDownClicked")
|
|
diagnostics.note(f"unread_hook.install: onPageDownClicked methods found={len(methods)}")
|
|
if not methods:
|
|
diagnostics.record_error(
|
|
"unread_hook.install", RuntimeError("onPageDownClicked method not found")
|
|
)
|
|
return
|
|
|
|
gesture_ref = _GESTURE
|
|
outer = self
|
|
|
|
class _PageDownHook(MethodHook):
|
|
def before_hooked_method(self, param):
|
|
try:
|
|
duration = _held_duration()
|
|
diagnostics.note(f"unread_hook: onPageDownClicked duration={duration:.3f}")
|
|
if duration < LONG_PRESS_SECONDS or gesture_ref["consumed"]:
|
|
return
|
|
chat_activity = param.thisObject
|
|
if chat_activity is None:
|
|
return
|
|
gesture_ref["consumed"] = True
|
|
param.setResult(None) # abort the normal scroll-to-newest
|
|
outer._trigger(chat_activity)
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook:onPageDownClicked", exc)
|
|
|
|
for method in methods:
|
|
try:
|
|
ref = self.plugin.hook_method(method, _PageDownHook())
|
|
diagnostics.note(f"unread_hook.install: onPageDownClicked hook ref={ref!r}")
|
|
if ref:
|
|
self._unhooks.append(ref)
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.install:onPageDownClicked", exc)
|
|
|
|
def _install_unread_cell_hook(self, find_class, MethodHook) -> None:
|
|
"""Hooks ChatUnreadCell's constructors to attach a click/long-click
|
|
listener to the inline "N unread messages" divider cell.
|
|
"""
|
|
unread_cell_class = find_class("org.telegram.ui.Cells.ChatUnreadCell")
|
|
diagnostics.note(f"unread_hook.install: ChatUnreadCell found={bool(unread_cell_class)}")
|
|
if not unread_cell_class:
|
|
return
|
|
ctors = _find_constructors(unread_cell_class)
|
|
diagnostics.note(f"unread_hook.install: ChatUnreadCell constructors found={len(ctors)}")
|
|
if not ctors:
|
|
return
|
|
|
|
outer = self
|
|
|
|
class _UnreadCellHook(MethodHook):
|
|
def after_hooked_method(self, param):
|
|
try:
|
|
cell = param.thisObject
|
|
if cell is not None:
|
|
outer._setup_unread_cell(cell)
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook:ChatUnreadCell.after", exc)
|
|
|
|
cell_hook = _UnreadCellHook()
|
|
attached = 0
|
|
for ctor in ctors:
|
|
try:
|
|
ref = self.plugin.hook_method(ctor, cell_hook)
|
|
if ref:
|
|
self._unhooks.append(ref)
|
|
attached += 1
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.install:ChatUnreadCell_ctor", exc)
|
|
diagnostics.note(f"unread_hook.install: ChatUnreadCell constructors hooked={attached}")
|
|
|
|
def _mark_hooked(self, view: Any) -> bool:
|
|
try:
|
|
key = id(view)
|
|
except Exception:
|
|
return True
|
|
if key in self._hooked_view_ids:
|
|
return False
|
|
if len(self._hooked_view_ids) > 64:
|
|
self._hooked_view_ids.clear()
|
|
self._hooked_view_ids.add(key)
|
|
return True
|
|
|
|
def _setup_unread_cell(self, unread_cell: Any) -> None:
|
|
"""Attaches click and long-click listeners to a ChatUnreadCell instance."""
|
|
try:
|
|
from android_utils import OnClickListener, OnLongClickListener, run_on_ui_thread
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.cell:imports", exc)
|
|
return
|
|
|
|
outer = self
|
|
|
|
def _init():
|
|
try:
|
|
if not outer._mark_hooked(unread_cell):
|
|
return
|
|
unread_cell.setClickable(True)
|
|
unread_cell.setFocusable(True)
|
|
unread_cell.setLongClickable(True)
|
|
|
|
def on_action(*_args):
|
|
outer._trigger(None)
|
|
return True
|
|
|
|
unread_cell.setOnClickListener(OnClickListener(lambda v=None: on_action()))
|
|
unread_cell.setOnLongClickListener(OnLongClickListener(lambda v=None: on_action()))
|
|
diagnostics.note("unread_hook: ChatUnreadCell listeners attached")
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.cell:_init", exc)
|
|
|
|
try:
|
|
run_on_ui_thread(_init)
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.cell:run_on_ui_thread", exc)
|
|
|
|
def uninstall(self) -> None:
|
|
try:
|
|
for ref in self._unhooks:
|
|
self.plugin.unhook_method(ref)
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook.uninstall", exc)
|
|
self._unhooks = []
|
|
self._hooked_view_ids.clear()
|
|
|
|
def _trigger(self, chat_activity: Optional[Any]) -> None:
|
|
"""Resolves the chat's unread count/offset and opens the summarize
|
|
dialog. `chat_activity` may be None (e.g. triggered from the unread
|
|
cell rather than the pagedown button), in which case the current
|
|
fragment is used to resolve the chat instead.
|
|
"""
|
|
diagnostics.note("unread_hook: _trigger fired")
|
|
try:
|
|
from client_utils import get_last_fragment, get_messages_controller
|
|
|
|
fragment = chat_activity or get_last_fragment()
|
|
if fragment is None:
|
|
diagnostics.record_error("unread_hook._trigger", RuntimeError("no fragment available"))
|
|
return
|
|
|
|
dialog_id = self._resolve_dialog_id(fragment)
|
|
diagnostics.note(f"unread_hook: resolved dialog_id={dialog_id}")
|
|
if not dialog_id:
|
|
diagnostics.record_error("unread_hook._trigger", RuntimeError("could not resolve dialog_id"))
|
|
return
|
|
|
|
raw_unread = 0
|
|
try:
|
|
controller = get_messages_controller()
|
|
dialog = controller.dialogs_dict.get(dialog_id) if controller else None
|
|
if dialog is not None:
|
|
raw_unread = int(getattr(dialog, "unread_count", 0) or 0)
|
|
except Exception:
|
|
raw_unread = 0
|
|
|
|
if raw_unread > 0:
|
|
count = min(MAX_MSG_COUNT, max(1, raw_unread))
|
|
offset = max(0, raw_unread - count)
|
|
else:
|
|
count = None # let caller fall back to configured defaults
|
|
offset = 0
|
|
|
|
topic_id = self._resolve_topic_id(fragment)
|
|
self.plugin.open_summarize_dialog(
|
|
fragment, dialog_id, topic_id, prefill_count=count, prefill_offset=offset
|
|
)
|
|
except Exception as exc:
|
|
diagnostics.record_error("unread_hook._trigger", exc)
|
|
|
|
@staticmethod
|
|
def _resolve_dialog_id(fragment: Any) -> Optional[int]:
|
|
try:
|
|
if hasattr(fragment, "getDialogId"):
|
|
return int(fragment.getDialogId())
|
|
except Exception:
|
|
pass
|
|
try:
|
|
from hook_utils import get_private_field
|
|
|
|
dialog_id = get_private_field(fragment, "dialog_id")
|
|
if dialog_id:
|
|
return int(dialog_id)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
@staticmethod
|
|
def _resolve_topic_id(fragment: Any) -> int:
|
|
try:
|
|
if hasattr(fragment, "getTopicId"):
|
|
return int(fragment.getTopicId() or 0)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
from hook_utils import get_private_field
|
|
|
|
topic_id = get_private_field(fragment, "topicId")
|
|
if topic_id:
|
|
return int(topic_id)
|
|
except Exception:
|
|
pass
|
|
return 0
|