+
-
+
`
+inside), a href (tg://user?id=N mentions), blockquote [expandable],
+tg-spoiler / span class="tg-spoiler", tg-emoji emoji-id, br.
+"""
+
+from html.parser import HTMLParser
+
+_SIMPLE = {
+ "b": "bold",
+ "strong": "bold",
+ "i": "italic",
+ "em": "italic",
+ "u": "underline",
+ "ins": "underline",
+ "s": "strike",
+ "strike": "strike",
+ "del": "strike",
+ "code": "code",
+ "pre": "pre",
+ "blockquote": "blockquote",
+ "tg-spoiler": "spoiler",
+}
+
+
+def utf16_len(text):
+ return len(text.encode("utf-16-le")) // 2
+
+
+class _Parser(HTMLParser):
+ def __init__(self):
+ super().__init__(convert_charrefs=True)
+ self.parts = []
+ self.length = 0 # UTF-16 units so far
+ self.open = [] # [(tag, entity or None)]
+ self.entities = []
+
+ def _entity(self, tag, attrs):
+ attrs = dict(attrs)
+ if tag == "code" and any(t == "pre" for t, _ in self.open):
+ pre = next(e for t, e in reversed(self.open) if t == "pre")
+ language = (attrs.get("class") or "").removeprefix("language-")
+ if pre is not None and language:
+ pre["language"] = language
+ return None
+ if tag == "span" and attrs.get("class") == "tg-spoiler":
+ return {"type": "spoiler"}
+ if tag in _SIMPLE:
+ entity = {"type": _SIMPLE[tag]}
+ if tag == "blockquote" and "expandable" in attrs:
+ entity["collapsed"] = True
+ return entity
+ if tag == "a" and attrs.get("href"):
+ href = attrs["href"]
+ if href.startswith("tg://user?id="):
+ try:
+ return {"type": "mention_name", "user_id": int(href.split("=", 1)[1])}
+ except ValueError:
+ pass
+ return {"type": "text_link", "url": href}
+ if tag == "tg-emoji":
+ try:
+ return {"type": "custom_emoji", "document_id": int(attrs.get("emoji-id"))}
+ except (TypeError, ValueError):
+ return None
+ return None
+
+ def handle_starttag(self, tag, attrs):
+ if tag == "br":
+ self.handle_data("\n")
+ return
+ entity = self._entity(tag, attrs)
+ if entity is not None:
+ entity["offset"] = self.length
+ self.open.append((tag, entity))
+
+ def handle_endtag(self, tag):
+ for index in range(len(self.open) - 1, -1, -1):
+ if self.open[index][0] == tag:
+ _, entity = self.open.pop(index)
+ if entity is not None:
+ entity["length"] = self.length - entity["offset"]
+ if entity["length"] > 0:
+ self.entities.append(entity)
+ return
+
+ def handle_data(self, data):
+ self.parts.append(data)
+ self.length += utf16_len(data)
+
+
+def parse(html):
+ """(plain text, entities) of `html`; entities ordered by offset."""
+ parser = _Parser()
+ parser.feed(html or "")
+ parser.close()
+ entities = sorted(parser.entities, key=lambda e: (e["offset"], -e["length"]))
+ return "".join(parser.parts), entities
+
+
+def escape(text):
+ """`text` made safe to put into a response."""
+ return str(text).replace("&", "&").replace("<", "<").replace(">", ">")
diff --git a/src/commands/model.py b/src/commands/model.py
new file mode 100644
index 0000000..548d920
--- /dev/null
+++ b/src/commands/model.py
@@ -0,0 +1,156 @@
+"""Commands and their parameters, as the panel lists them (and as plugins
+register them through the library). Pure Python: tested offline."""
+
+TYPES = ("string", "integer", "number", "boolean", "choice")
+REPLY_MODES = ("none", "optional", "required")
+
+# What the composer shows for a parameter's type.
+TYPE_LABELS = {
+ "string": "text",
+ "integer": "whole number",
+ "number": "number",
+ "boolean": "yes / no",
+ "choice": "choice",
+}
+
+
+class Param:
+ __slots__ = ("name", "type", "required", "description", "default", "choices")
+
+ def __init__(
+ self,
+ name,
+ type="string",
+ required=True,
+ description="",
+ default=None,
+ choices=(),
+ ):
+ if type not in TYPES:
+ raise ValueError(f"unknown parameter type {type!r}")
+ self.name = str(name)
+ self.type = type
+ self.required = bool(required)
+ self.description = str(description or "")
+ self.default = default
+ self.choices = tuple(str(c) for c in choices or ())
+ if self.type == "choice" and not self.choices:
+ raise ValueError(f"choice parameter {name!r} needs choices")
+
+ @classmethod
+ def from_json(cls, data):
+ kind = data.get("type") or "string"
+ return cls(
+ data["name"],
+ kind if kind in TYPES else "string",
+ data.get("required", True),
+ data.get("description") or "",
+ data.get("default"),
+ data.get("choices") or (),
+ )
+
+ def to_json(self):
+ return {
+ "name": self.name,
+ "type": self.type,
+ "required": self.required,
+ "description": self.description,
+ "default": self.default,
+ "choices": list(self.choices),
+ }
+
+ def placeholder(self):
+ """`` when required, `[name]` when optional."""
+ return f"<{self.name}>" if self.required else f"[{self.name}]"
+
+ def type_label(self):
+ if self.type == "choice":
+ return " | ".join(self.choices)
+ return TYPE_LABELS.get(self.type, self.type)
+
+
+class Command:
+ """A command: `source` is "server" (the panel runs it) or the id of the
+ plugin that registered it (then `handler` runs it on the device)."""
+
+ __slots__ = (
+ "name",
+ "aliases",
+ "summary",
+ "description",
+ "params",
+ "reply",
+ "source",
+ "handler",
+ )
+
+ def __init__(
+ self,
+ name,
+ params=(),
+ summary="",
+ description="",
+ aliases=(),
+ reply="none",
+ source="server",
+ handler=None,
+ ):
+ name = str(name).lstrip("/").lower()
+ if not name or not all(c.isalnum() and c.isascii() or c == "_" for c in name):
+ raise ValueError(f"bad command name {name!r}: use a-z, 0-9 and _")
+ if reply not in REPLY_MODES:
+ raise ValueError(f"reply must be one of {REPLY_MODES}")
+ self.name = name
+ self.aliases = tuple(str(a).lstrip("/").lower() for a in aliases or ())
+ self.summary = str(summary or "")
+ self.description = str(description or "")
+ self.params = tuple(params or ())
+ self.reply = reply
+ self.source = source
+ self.handler = handler
+ names = [p.name.lower() for p in self.params]
+ if len(set(names)) != len(names):
+ raise ValueError(f"/{name}: parameter names must be unique")
+
+ @property
+ def names(self):
+ return (self.name,) + self.aliases
+
+ @property
+ def local(self):
+ return self.source != "server"
+
+ @classmethod
+ def from_json(cls, data, source="server"):
+ return cls(
+ data["name"],
+ [Param.from_json(p) for p in data.get("params") or ()],
+ data.get("summary") or "",
+ data.get("description") or "",
+ data.get("aliases") or (),
+ data.get("reply") if data.get("reply") in REPLY_MODES else "none",
+ source,
+ )
+
+ def to_json(self):
+ return {
+ "name": self.name,
+ "aliases": list(self.aliases),
+ "summary": self.summary,
+ "description": self.description,
+ "params": [p.to_json() for p in self.params],
+ "reply": self.reply,
+ }
+
+ def usage(self):
+ return " ".join(["/" + self.name] + [p.placeholder() for p in self.params])
+
+ def hint(self):
+ """The suggestion row's second line: parameters, then the summary."""
+ parts = [p.placeholder() for p in self.params]
+ if self.reply == "required":
+ parts.append("↩ reply")
+ head = " ".join(parts)
+ if head and self.summary:
+ return f"{head} · {self.summary}"
+ return head or self.summary
diff --git a/src/commands/parse.py b/src/commands/parse.py
new file mode 100644
index 0000000..e5b4e82
--- /dev/null
+++ b/src/commands/parse.py
@@ -0,0 +1,199 @@
+"""Discord-style arguments of a `/command`. Pure Python: tested offline.
+
+Mirrors telegram-panel's `app/telegram/client_commands/parser.py`, which
+parses the same text again on the server: keep the two in step.
+
+- `/name args` (`/name@anything` too); the name is case-insensitive.
+- Arguments split like a shell (quotes group words).
+- `param:value` sets a parameter by name; other words fill the remaining
+ parameters in order, and the last remaining one, when a string, takes all
+ the words left over.
+- An empty `param:` counts as not given.
+"""
+
+import re
+import shlex
+
+_COMMAND = re.compile(r"^/([A-Za-z0-9_]{1,64})(?:@\S*)?(?:\s+|$)")
+_TRUE = {"true", "yes", "y", "on", "1", "+"}
+_FALSE = {"false", "no", "n", "off", "0", "-"}
+
+
+class CommandError(Exception):
+ """A user-facing command error (bad arguments, nothing to act on, ...)."""
+
+
+def split_command(text):
+ """(lowercased name, argument text) of a `/command`, else None."""
+ text = (text or "").strip()
+ match = _COMMAND.match(text)
+ if match is None:
+ return None
+ return match.group(1).lower(), text[match.end():]
+
+
+def command_name_at(text):
+ """The name being typed while the text is a bare `/prefix` (no space
+ yet), else None: suggestions are shown for it."""
+ match = re.match(r"^/([A-Za-z0-9_]{0,64})$", text or "")
+ return match.group(1).lower() if match else None
+
+
+def _convert(param, raw):
+ if param.type == "integer":
+ try:
+ return int(raw)
+ except ValueError:
+ raise CommandError(f"{param.name} must be a whole number") from None
+ if param.type == "number":
+ try:
+ return float(raw.replace(",", "."))
+ except ValueError:
+ raise CommandError(f"{param.name} must be a number") from None
+ if param.type == "boolean":
+ if raw.lower() in _TRUE:
+ return True
+ if raw.lower() in _FALSE:
+ return False
+ raise CommandError(f"{param.name} must be yes or no")
+ if param.type == "choice":
+ for choice in param.choices:
+ if choice.lower() == raw.lower():
+ return choice
+ raise CommandError(f"{param.name} must be one of: {', '.join(param.choices)}")
+ return raw
+
+
+def _assign(params, tokens):
+ """Raw values by parameter name, and the parameter each token went to
+ ([(token index, param name or None)])."""
+ by_name = {p.name.lower(): p for p in params}
+ raw, owners, positional = {}, [], []
+ for index, token in enumerate(tokens):
+ key, sep, value = token.partition(":")
+ param = by_name.get(key.lower()) if sep else None
+ if param is None:
+ positional.append(index)
+ continue
+ if param.name in raw:
+ raise CommandError(f"{param.name} is given twice")
+ owners.append((index, param.name))
+ if value:
+ raw[param.name] = value
+ free = [p for p in params if p.name not in raw]
+ greedy = bool(free) and free[-1].type == "string"
+ if len(positional) > len(free) and not greedy:
+ extra = " ".join(tokens[i] for i in positional[len(free):])
+ raise CommandError(f"Unexpected arguments: {extra}")
+ for slot, param in enumerate(free[: len(positional)]):
+ if slot == len(free) - 1 and greedy:
+ rest = positional[slot:]
+ raw[param.name] = " ".join(tokens[i] for i in rest)
+ owners += [(i, param.name) for i in rest]
+ else:
+ raw[param.name] = tokens[positional[slot]]
+ owners.append((positional[slot], param.name))
+ return raw, owners
+
+
+def assign(params, args):
+ """Raw string values by parameter name."""
+ try:
+ tokens = shlex.split(args)
+ except ValueError:
+ raise CommandError("Unclosed quote") from None
+ return _assign(params, tokens)[0]
+
+
+def parse(command, text):
+ """Handler keyword arguments for `text` (`/name ...`)."""
+ split = split_command(text)
+ if split is None or split[0] not in command.names:
+ raise CommandError(f"Not a /{command.name} command")
+ raw = assign(command.params, split[1])
+ values = {}
+ for param in command.params:
+ if param.name in raw:
+ values[param.name] = _convert(param, raw[param.name])
+ elif param.required:
+ raise CommandError(f"Missing required parameter: {param.name}")
+ else:
+ values[param.name] = param.default
+ return values
+
+
+def _tokens(args):
+ """Shell-like words of `args` with their spans; tolerant of an unclosed
+ quote (the word runs to the end), for analysing text being typed."""
+ out, start, word, quote = [], None, [], None
+ for index, char in enumerate(args):
+ if quote:
+ if char == quote:
+ quote = None
+ else:
+ word.append(char)
+ elif char in "\"'":
+ quote = char
+ start = index if start is None else start
+ elif char.isspace():
+ if start is not None:
+ out.append((start, index, "".join(word)))
+ start, word = None, []
+ else:
+ start = index if start is None else start
+ word.append(char)
+ if start is not None:
+ out.append((start, len(args), "".join(word)))
+ return out
+
+
+class Analysis:
+ """What the composer shows for a command being typed."""
+
+ __slots__ = ("values", "current", "missing", "error", "complete")
+
+ def __init__(self, values, current, missing, error):
+ self.values = values # raw values by parameter name
+ self.current = current # the parameter at the cursor, or None
+ self.missing = missing # required parameters not given yet
+ self.error = error # why it can't run (None when it can)
+ self.complete = error is None
+
+
+def analyze(command, text, cursor=None, has_reply=False):
+ """Values, the parameter at `cursor`, and whether `text` can run."""
+ split = split_command(text)
+ if split is None or split[0] not in command.names:
+ return Analysis({}, None, [p.name for p in command.params if p.required], "Not a command")
+ stripped = (text or "").strip()
+ args = split[1]
+ offset = len(text) - len(text.lstrip()) + len(stripped) - len(args)
+ cursor = len(text) if cursor is None else cursor
+ at = cursor - offset
+ tokens = _tokens(args)
+ error = None
+ try:
+ raw, owners = _assign(command.params, [t[2] for t in tokens])
+ except CommandError as exc:
+ raw, owners, error = {}, [], str(exc)
+ missing = [p.name for p in command.params if p.required and p.name not in raw]
+ current = None
+ if at >= 0:
+ inside = [i for i, (start, end, _) in enumerate(tokens) if start <= at <= end]
+ if inside:
+ current = dict(owners).get(inside[-1])
+ key, sep, _ = tokens[inside[-1]][2].partition(":")
+ if current is None and sep:
+ current = next((p.name for p in command.params if p.name.lower() == key.lower()), None)
+ else:
+ current = next((p.name for p in command.params if p.name not in raw), None)
+ if current is None and command.params and command.params[-1].type == "string":
+ current = command.params[-1].name # the greedy last one goes on
+ if error is None:
+ try:
+ parse(command, text)
+ except CommandError as exc:
+ error = str(exc)
+ if error is None and command.reply == "required" and not has_reply:
+ error = f"Reply to a message to use /{command.name}"
+ return Analysis(raw, current, missing, error)
diff --git a/src/config.py b/src/config.py
new file mode 100644
index 0000000..617880e
--- /dev/null
+++ b/src/config.py
@@ -0,0 +1,24 @@
+"""Setting keys, defaults and constants."""
+
+SERVER_URL = "server_url"
+API_TOKENS = "api_tokens"
+SEND_TO_CHAT = "send_to_chat"
+SERVER_COMMANDS = "server_commands"
+SHOW_PARAMS = "show_params"
+
+DEFAULTS = {
+ SERVER_URL: "",
+ API_TOKENS: "",
+ SEND_TO_CHAT: False,
+ SERVER_COMMANDS: True,
+ SHOW_PARAMS: True,
+}
+
+# The panel's list of an account's commands is asked for again after this
+# long (the server may say otherwise in `refresh_seconds`).
+REFRESH_SECONDS = 3600
+# A failed refresh is retried sooner.
+RETRY_SECONDS = 300
+HTTP_TIMEOUT_SECONDS = 15
+# Running a command may take a while on the server (it times out at 60 s).
+EXECUTE_TIMEOUT_SECONDS = 75
diff --git a/src/diag.py b/src/diag.py
new file mode 100644
index 0000000..50225ca
--- /dev/null
+++ b/src/diag.py
@@ -0,0 +1,54 @@
+"""Diagnostics mirrored to a file, since this device's logcat surfaces neither
+Chaquopy stdout nor android_utils.log output.
+
+The file lives under `get_files_dir()` (app external storage), so it is
+readable over adb:
+ adb shell cat "/sdcard/Android/data/com.exteragram.messenger/files/exteraGram/exteraGram Files/client_commands/diag.log"
+"""
+
+import os
+import time
+import traceback
+
+_MAX_BYTES = 256 * 1024
+_path = None
+
+
+def _file():
+ global _path
+ if _path is None:
+ try:
+ from file_utils import get_files_dir
+
+ directory = os.path.join(get_files_dir(), "client_commands")
+ os.makedirs(directory, exist_ok=True)
+ _path = os.path.join(directory, "diag.log")
+ except Exception:
+ _path = ""
+ return _path
+
+
+def log(message):
+ line = time.strftime("%H:%M:%S ") + str(message)
+ try:
+ from android_utils import log as android_log
+
+ android_log("[client_commands] " + line)
+ except Exception:
+ pass
+ path = _file()
+ if not path:
+ return
+ try:
+ mode = "a"
+ if os.path.exists(path) and os.path.getsize(path) > _MAX_BYTES:
+ mode = "w"
+ with open(path, mode, encoding="utf-8") as handle:
+ handle.write(line + "\n")
+ except Exception:
+ pass
+
+
+def error(where, exc=None):
+ detail = traceback.format_exc() if exc is not None else ""
+ log(f"ERROR {where}: {exc!r}\n{detail}")
diff --git a/src/greeting/__init__.py b/src/greeting/__init__.py
deleted file mode 100644
index f0d6578..0000000
--- a/src/greeting/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .template import render, TemplateError
diff --git a/src/greeting/template.py b/src/greeting/template.py
deleted file mode 100644
index 5ccc0e9..0000000
--- a/src/greeting/template.py
+++ /dev/null
@@ -1,13 +0,0 @@
-from textutil import normalize
-
-
-class TemplateError(ValueError):
- pass
-
-
-def render(template: str, name: str) -> str:
- normalized = normalize(name)
- try:
- return template.format(name=normalized)
- except (KeyError, IndexError, ValueError):
- raise TemplateError(f"unsupported placeholder in {template!r}")
diff --git a/src/hooks/__init__.py b/src/hooks/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/hooks/composer.py b/src/hooks/composer.py
new file mode 100644
index 0000000..5e9b7ef
--- /dev/null
+++ b/src/hooks/composer.py
@@ -0,0 +1,266 @@
+"""The message field: a command runs instead of going to the chat.
+
+Hooks on ChatActivityEnterView (the chat's input):
+- sendMessageInternal(notify, scheduleDate, ...), behind the send button, the
+ keyboard's send key and the send preview: a known command runs (on the
+ panel, or on the device for plugins' commands). One that can't run yet
+ (a required parameter missing, no reply) stays in the field and a bulletin
+ says why. Scheduled messages are sent as typed.
+- onSendLongClick(View): holding send on a command opens a menu: run it,
+ send it to the chat as a plain message (the panel then handles it from the
+ update, as without this plugin), and the "Send response in chat" check.
+- SendButton.onDraw(Canvas): while that check is on and the command can run,
+ a small "!" badge on the send button.
+
+Not installed, the same text is an ordinary message, which the panel handles
+from the account's outgoing update.
+"""
+
+from base_plugin import MethodHook
+
+import diag
+from commands.parse import analyze, split_command
+from config import SEND_TO_CHAT
+from jutil import field, hook_methods, java_class, runnable, unhook_all
+from runner import Invocation
+from hooks import responses
+from hooks.notify import bulletin
+
+ENTER = "org.telegram.ui.Components.ChatActivityEnterView"
+SEND_BUTTON = ENTER + "$SendButton"
+
+
+class Field:
+ """The command typed into one input, ready to run or not."""
+
+ __slots__ = ("enter", "chat", "account", "text", "command", "reply", "analysis")
+
+ def __init__(self, enter, chat, account, text, command, reply, analysis):
+ self.enter = enter
+ self.chat = chat
+ self.account = account
+ self.text = text
+ self.command = command
+ self.reply = reply
+ self.analysis = analysis
+
+
+def reply_of(chat):
+ """The message the chat is replying to (not one of our responses)."""
+ reply = field(chat, "replyingMessageObject")
+ if reply is None or responses.is_local(reply) or int(reply.getId()) <= 0:
+ return None
+ return reply
+
+
+def topic_of(chat):
+ try:
+ return int(chat.getTopicId()) if field(chat, "isTopic") else 0
+ except Exception:
+ return 0
+
+
+def _icon(*names):
+ from org.telegram.messenger import R
+
+ for name in names:
+ value = getattr(R.drawable, name, None)
+ if value:
+ return value
+ return 0
+
+
+class Composer:
+ def __init__(self, plugin, catalog, runner):
+ self.plugin = plugin
+ self.catalog = catalog
+ self.runner = runner
+ self.handles = []
+ self._badge = {} # (text, reply id) -> whether it can run
+ self._paints = []
+
+ def install(self):
+ composer = self
+
+ class Send(MethodHook):
+ def before_hooked_method(self, param):
+ try:
+ if int(param.args[1]) == 0 and composer.submit(param.thisObject):
+ param.setResult(False)
+ except Exception as exc:
+ diag.error("send", exc)
+
+ class LongSend(MethodHook):
+ def before_hooked_method(self, param):
+ try:
+ if composer.menu(param.thisObject, param.args[0]):
+ param.setResult(True)
+ except Exception as exc:
+ diag.error("send menu", exc)
+
+ class Badge(MethodHook):
+ def after_hooked_method(self, param):
+ try:
+ composer.draw_badge(param.thisObject, param.args[0])
+ except Exception as exc:
+ diag.error("send badge", exc)
+
+ self.handles += hook_methods(self.plugin, ENTER, "sendMessageInternal", Send(), arity=5)
+ self.handles += hook_methods(self.plugin, ENTER, "onSendLongClick", LongSend(), arity=1)
+ self.handles += hook_methods(self.plugin, SEND_BUTTON, "onDraw", Badge(), arity=1)
+
+ def uninstall(self):
+ unhook_all(self.plugin, self.handles)
+
+ # the field
+
+ def read(self, enter):
+ """The Field of a known command typed into `enter`, else None."""
+ chat = field(enter, "parentFragment")
+ if chat is None or int(chat.getChatMode()) != 0 or field(enter, "editingMessageObject") is not None:
+ return None
+ text = enter.getFieldText()
+ text = str(text) if text is not None else ""
+ split = split_command(text)
+ if split is None:
+ return None
+ account = int(chat.getCurrentAccount())
+ command = self.catalog.find(account, split[0])
+ if command is None:
+ return None
+ reply = reply_of(chat)
+ edit = enter.getEditField()
+ cursor = int(edit.getSelectionEnd()) if edit is not None else len(text)
+ analysis = analyze(command, text, cursor, reply is not None)
+ return Field(enter, chat, account, text, command, reply, analysis)
+
+ def set_text(self, enter, text):
+ enter.setFieldText(text)
+
+ def submit(self, enter):
+ """Runs the command typed into `enter`; False when it isn't one (or
+ should go to the chat as typed)."""
+ typed = self.read(enter)
+ if typed is None or self.runner.bypassed(typed.text):
+ return False
+ if not typed.analysis.complete:
+ bulletin(typed.analysis.error, error=True)
+ return True
+ chat = typed.chat
+ inv = Invocation(
+ typed.account,
+ chat.getDialogId(),
+ topic_of(chat),
+ typed.reply,
+ chat.getThreadMessage(),
+ typed.text,
+ typed.command,
+ self.plugin.get_setting(SEND_TO_CHAT, False),
+ )
+ enter.setFieldText("")
+ if typed.reply is not None:
+ chat.hideFieldPanel(False)
+ self.runner.run(inv)
+ return True
+
+ def send_as_text(self, enter):
+ typed = enter.getFieldText()
+ if typed is None:
+ return
+ self.runner.bypass(str(typed))
+ enter.sendMessage()
+
+ # holding send
+
+ def menu(self, enter, button):
+ typed = self.read(enter)
+ if typed is None:
+ return False
+ from org.telegram.ui.Components import ItemOptions
+
+ name = typed.command.name
+ checked = bool(self.plugin.get_setting(SEND_TO_CHAT, False))
+
+ def toggle():
+ self.plugin.set_setting(SEND_TO_CHAT, not checked)
+ self.refresh(enter)
+
+ options = ItemOptions.makeOptions(typed.chat, button)
+ options.add(_icon("msg_bot", "msg_bots", "msg_send"), f"Run /{name}", runnable(lambda: self.submit(enter)))
+ options.add(_icon("msg_send"), "Send as message", runnable(lambda: self.send_as_text(enter)))
+ options.addChecked(checked, _icon("msg_discussion", "msg_message", "msg_chats"), "Send response in chat", runnable(toggle))
+ options.show()
+ return True
+
+ # the badge
+
+ def refresh(self, enter):
+ """Redraws the send button (the field or the setting changed)."""
+ button = field(enter, "sendButton")
+ if button is not None:
+ button.invalidate()
+
+ def _enter_of(self, button):
+ enter_class = java_class(ENTER)
+ view = button.getParent()
+ while view is not None:
+ if enter_class is not None and enter_class.isInstance(view):
+ return view
+ view = view.getParent() if hasattr(view, "getParent") else None
+ return None
+
+ def _can_run(self, enter):
+ text = enter.getFieldText()
+ if text is None or not str(text).startswith("/"):
+ return False
+ chat = field(enter, "parentFragment")
+ reply = reply_of(chat) if chat is not None else None
+ key = (str(text), int(reply.getId()) if reply is not None else 0)
+ if key not in self._badge:
+ typed = self.read(enter)
+ if len(self._badge) > 64:
+ self._badge.clear()
+ self._badge[key] = typed is not None and typed.analysis.complete
+ return self._badge[key]
+
+ def draw_badge(self, button, canvas):
+ if not self.plugin.get_setting(SEND_TO_CHAT, False):
+ return
+ enter = self._enter_of(button)
+ if enter is None or not self._can_run(enter):
+ return
+ from android.graphics import Paint
+ from java import jint
+ from org.telegram.messenger import AndroidUtilities
+ from org.telegram.ui.ActionBar import Theme
+
+ if not self._paints:
+ self._paints = [Paint(Paint.ANTI_ALIAS_FLAG) for _ in range(3)]
+ self._paints[1].setTextAlign(Paint.Align.CENTER)
+ self._paints[1].setFakeBoldText(True)
+ fill, mark, ring = self._paints
+ dp = AndroidUtilities.dp
+ height = button.getMeasuredHeight()
+ rect = field(button, "backgroundRect")
+ if rect is not None and field(button, "isNewDesignSendButton") and rect.width() > 0:
+ cx, cy, r = rect.centerX(), rect.centerY(), rect.height() / 2
+ else:
+ cx, cy, r = button.getMeasuredWidth() - height / 2, height / 2, height / 2 - dp(6)
+ x, y, radius = cx + r * 0.62, cy - r * 0.62, dp(7)
+ fill.setColor(jint(Theme.getColor(Theme.key_text_RedBold)))
+ mark.setColor(jint(-1))
+ mark.setTextSize(dp(11))
+ # A ring of the input's background separates it from the button.
+ ring.setColor(jint(Theme.getColor(Theme.key_chat_messagePanelBackground)))
+ canvas.drawCircle(x, y, radius + dp(1.5), ring)
+ canvas.drawCircle(x, y, radius, fill)
+ canvas.drawText("!", x, y + dp(4), mark)
+
+ def forget(self):
+ """The commands changed: whether a text can run is asked again."""
+ self._badge.clear()
+
+ def text_changed(self, enter):
+ """The field's text changed: the badge may change."""
+ if self.plugin.get_setting(SEND_TO_CHAT, False):
+ self.refresh(enter)
diff --git a/src/hooks/menu.py b/src/hooks/menu.py
new file mode 100644
index 0000000..68f0a20
--- /dev/null
+++ b/src/hooks/menu.py
@@ -0,0 +1,87 @@
+""""Send in chat" first in the menu of a response shown only to you.
+
+Tapping a response (an ephemeral message) opens the app's message menu. The
+row is added to the built popup after ChatActivity.createMenu, above its
+first row (as messageLogger's "Edit history"); ChatActivity.selectedObject
+is the message. Sending posts the answer (without the command above it) as
+the account, replying where the command replied, and takes the local copy
+away.
+"""
+
+from base_plugin import MethodHook
+
+import diag
+from hooks import responses
+from jutil import field, hook_methods, method, unhook_all
+
+CHAT = "org.telegram.ui.ChatActivity"
+LABEL = "Send in chat"
+
+
+def _items(view, found):
+ """Every ActionBarMenuSubItem under `view`, in order."""
+ if view.getClass().getSimpleName() == "ActionBarMenuSubItem":
+ found.append(view)
+ elif hasattr(view, "getChildCount"):
+ for index in range(view.getChildCount()):
+ _items(view.getChildAt(index), found)
+ return found
+
+
+class ResponseMenu:
+ def __init__(self, plugin, runner):
+ self.plugin = plugin
+ self.runner = runner
+ self.handles = []
+
+ def install(self):
+ menu = self
+
+ class AfterCreate(MethodHook):
+ def after_hooked_method(self, param):
+ try:
+ menu.add_item(param.thisObject)
+ except Exception as exc:
+ diag.error("response menu", exc)
+
+ self.handles += hook_methods(self.plugin, CHAT, "createMenu", AfterCreate())
+
+ def uninstall(self):
+ unhook_all(self.plugin, self.handles)
+
+ def add_item(self, chat):
+ message = field(chat, "selectedObject")
+ popup = field(chat, "scrimPopupWindow")
+ body = responses.sendable(message) if message is not None else None
+ if popup is None or body is None:
+ return
+ items = _items(popup.getContentView(), [])
+ # The createMenu overloads call one another: add the row once.
+ if not items or any(str(i.getTextView().getText()) == LABEL for i in items if i.getTextView()):
+ return
+ from android_utils import OnClickListener
+ from org.telegram.messenger import R
+ from org.telegram.ui.ActionBar import ActionBarMenuSubItem
+
+ anchor = items[0]
+ parent = anchor.getParent()
+ item = ActionBarMenuSubItem(anchor.getContext(), False, False)
+ item.setTextAndIcon(LABEL, R.drawable.msg_send)
+ color = anchor.getTextView().getCurrentTextColor()
+ item.setColors(color, color)
+ item.setMinimumWidth(anchor.getMinimumWidth())
+
+ def chosen(_view):
+ method(CHAT, "closeMenu").invoke(chat)
+ self.runner.send(
+ body["account"],
+ body["dialog_id"],
+ body["reply"],
+ body["thread"],
+ body["text"],
+ body["entities"],
+ )
+ responses.remove(body["account"], body["dialog_id"], message.getId())
+
+ item.setOnClickListener(OnClickListener(chosen))
+ parent.addView(item, parent.indexOfChild(anchor), anchor.getLayoutParams())
diff --git a/src/hooks/notify.py b/src/hooks/notify.py
new file mode 100644
index 0000000..5b52b1b
--- /dev/null
+++ b/src/hooks/notify.py
@@ -0,0 +1,17 @@
+"""Bulletins (shown on the UI thread, from any thread)."""
+
+from jutil import on_ui
+
+
+def bulletin(text, error=False, success=False):
+ def show():
+ from ui.bulletin import BulletinHelper
+
+ if error:
+ BulletinHelper.show_error(text)
+ elif success:
+ BulletinHelper.show_success(text)
+ else:
+ BulletinHelper.show_info(text)
+
+ on_ui(show)
diff --git a/src/hooks/responses.py b/src/hooks/responses.py
new file mode 100644
index 0000000..75f37de
--- /dev/null
+++ b/src/hooks/responses.py
@@ -0,0 +1,358 @@
+"""Responses shown only on this device.
+
+They are Telegram's own ephemeral messages: a TL_message with an ephemeral
+id, sent by the account itself (an outgoing bubble), handed to the open chat
+the way new messages arrive (ChatActivity.processNewMessages) and never
+stored. The latest 20 come back when their chat is opened again, until the
+app restarts (memory only). The app labels an outgoing
+ephemeral message "Only visible to @"; the receiver is the account,
+and the label is made "Only visible to you" (ChatMessageCell's
+setMessageObjectInternal). Deleting one, or replying to one, stays on the
+device. When the chat isn't open, a bulletin shows the text instead.
+"""
+
+import random
+
+from base_plugin import MethodHook
+
+import diag
+from commands.markup import utf16_len
+from jutil import array_list, field, hook_methods, java_class, method, set_field, unhook_all
+
+CHAT = "org.telegram.ui.ChatActivity"
+
+_ENTITY_CLASSES = {
+ "bold": "TL_messageEntityBold",
+ "italic": "TL_messageEntityItalic",
+ "underline": "TL_messageEntityUnderline",
+ "strike": "TL_messageEntityStrike",
+ "code": "TL_messageEntityCode",
+ "pre": "TL_messageEntityPre",
+ "text_link": "TL_messageEntityTextUrl",
+ "url": "TL_messageEntityUrl",
+ "email": "TL_messageEntityEmail",
+ "blockquote": "TL_messageEntityBlockquote",
+ "custom_emoji": "TL_messageEntityCustomEmoji",
+ "mention_name": "TL_messageEntityMentionName",
+ "spoiler": "TL_messageEntitySpoiler",
+}
+
+# The latest responses are kept (in memory only) and shown again when their
+# chat is opened again, until the app restarts.
+HISTORY = 20
+
+
+def _state():
+ """Kept across reloads of this plugin (not app restarts):
+ shown: (account, message id) -> what "Send in chat" sends, or None;
+ history: the latest HISTORY responses, oldest first."""
+ from library import shared
+
+ return shared().setdefault(
+ "responses",
+ {"shown": {}, "history": [], "counter": [random.randint(0x01000000, 0x0E000000)]},
+ )
+
+
+def jboolean_type():
+ from java.lang import Boolean
+
+ return Boolean.TYPE
+
+
+def tl_entities(entities):
+ """ArrayList of entity dicts ({type, offset, length, ...})."""
+ from org.telegram.tgnet import TLRPC
+
+ out = []
+ for data in entities or ():
+ name = _ENTITY_CLASSES.get(data.get("type"))
+ if name is None:
+ continue
+ entity = getattr(TLRPC, name)()
+ entity.offset = int(data["offset"])
+ entity.length = int(data["length"])
+ if data.get("url") is not None and hasattr(entity, "url"):
+ entity.url = data["url"]
+ if data.get("language") is not None and hasattr(entity, "language"):
+ entity.language = data["language"]
+ if data.get("user_id") is not None and hasattr(entity, "user_id"):
+ entity.user_id = int(data["user_id"])
+ if data.get("document_id") is not None and hasattr(entity, "document_id"):
+ entity.document_id = int(data["document_id"])
+ if data.get("collapsed") and hasattr(entity, "collapsed"):
+ entity.collapsed = True
+ out.append(entity)
+ return array_list(out)
+
+
+def is_local(message_object):
+ """Whether a MessageObject is a response shown here."""
+ try:
+ return (int(message_object.currentAccount), int(message_object.getId())) in _state()["shown"]
+ except Exception:
+ return False
+
+
+def open_chat(account, dialog_id, topic_id=0):
+ """The ChatActivity showing that chat (and topic) now, else None."""
+ from client_utils import get_last_fragment
+
+ fragment = get_last_fragment()
+ if fragment is None or not java_class(CHAT).isInstance(fragment):
+ return None
+ if int(fragment.getCurrentAccount()) != int(account) or int(fragment.getDialogId()) != int(dialog_id):
+ return None
+ if topic_id and int(fragment.getTopicId()) != int(topic_id):
+ return None
+ return fragment
+
+
+def _build(account, dialog_id, topic_id, reply, text, entities, message_id, date):
+ from org.telegram.messenger import MessageObject, MessagesController, UserConfig
+ from org.telegram.tgnet import TLRPC
+
+ controller = MessagesController.getInstance(account)
+ own = UserConfig.getInstance(account).getClientUserId()
+ msg = TLRPC.TL_message()
+ msg.id = message_id
+ msg.dialog_id = dialog_id
+ msg.peer_id = controller.getPeer(dialog_id)
+ msg.from_id = controller.getPeer(own)
+ msg.flags |= 256
+ msg.date = date
+ msg.message = text
+ msg.entities = entities
+ if entities.size():
+ msg.flags |= 128
+ reply_id = int(reply.getId()) if reply is not None else 0
+ if reply_id or topic_id:
+ header = TLRPC.TL_messageReplyHeader()
+ header.reply_to_msg_id = reply_id or int(topic_id)
+ header.flags |= 16
+ if topic_id:
+ header.forum_topic = True
+ header.flags |= 8
+ if reply_id:
+ header.reply_to_top_id = int(topic_id)
+ header.flags |= 2
+ msg.reply_to = header
+ msg.flags |= 8
+ msg.out = True
+ msg.unread = False
+ msg.ephemeralReceiverBotId = own
+ obj = MessageObject(account, msg, True, True)
+ if reply is not None:
+ obj.replyMessageObject = reply
+ return obj
+
+
+def sendable(message_object):
+ """What "Send in chat" sends for a response shown here, else None:
+ {account, dialog_id, reply, thread, text, entities}."""
+ try:
+ return _state()["shown"].get((int(message_object.currentAccount), int(message_object.getId())))
+ except Exception:
+ return None
+
+
+def show(
+ account, dialog_id, topic_id, reply, text, entities=(), header=None, thread=None, answer=False, keep=True
+):
+ """Shows a response in the open chat; its id, or None (a bulletin was
+ shown instead). `header` (the command) goes first, as code. An `answer`
+ can be sent into the chat from its menu (without the header). Kept
+ responses come back when the chat is opened again. UI thread."""
+ from org.telegram.messenger import MessageObject
+ from org.telegram.tgnet import ConnectionsManager
+
+ state = _state()
+ body = {
+ "account": int(account),
+ "dialog_id": int(dialog_id),
+ "reply": reply,
+ "thread": thread,
+ "text": text,
+ "entities": list(entities or ()),
+ }
+ if header:
+ shift = utf16_len(header) + 1
+ entities = [{"type": "code", "offset": 0, "length": utf16_len(header)}] + [
+ dict(e, offset=int(e["offset"]) + shift) for e in entities or ()
+ ]
+ text = f"{header}\n{text}" if text else header
+ state["counter"][0] += 1
+ entry = {
+ "account": int(account),
+ "dialog_id": int(dialog_id),
+ "topic_id": int(topic_id or 0),
+ "id": int(MessageObject.ephemeralMessageIdPack(state["counter"][0])),
+ "date": int(ConnectionsManager.getInstance(account).getCurrentTime()),
+ "reply": reply,
+ "text": text,
+ "entities": list(entities or ()),
+ }
+ state["shown"][(entry["account"], entry["id"])] = body if answer else None
+ if keep:
+ history = state["history"]
+ history.append(entry)
+ del history[:-HISTORY]
+ fragment = open_chat(account, dialog_id, topic_id)
+ if fragment is None:
+ from hooks.notify import bulletin
+
+ bulletin(text[:300])
+ return None
+ _insert(fragment, [entry])
+ return entry["id"]
+
+
+def _insert(fragment, entries):
+ from java import jboolean
+
+ objects = [
+ _build(
+ e["account"], e["dialog_id"], e["topic_id"], e["reply"], e["text"],
+ tl_entities(e["entities"]), e["id"], e["date"],
+ )
+ for e in entries
+ ]
+ method(CHAT, "processNewMessages", "java.util.ArrayList", jboolean_type()).invoke(
+ fragment, array_list(objects), jboolean(False)
+ )
+
+
+def restore(fragment):
+ """Shows a just opened chat's kept responses it doesn't show yet."""
+ state = _state()
+ account, dialog_id = int(fragment.getCurrentAccount()), int(fragment.getDialogId())
+ topic_id = int(fragment.getTopicId())
+ dicts = field(fragment, "messagesDict")
+ present = dicts[0] if dicts is not None else None
+ missing = [
+ e
+ for e in state["history"]
+ if e["account"] == account
+ and e["dialog_id"] == dialog_id
+ and e["topic_id"] == topic_id
+ and (present is None or present.get(e["id"]) is None)
+ ]
+ if missing:
+ _insert(fragment, missing)
+
+
+def remove(account, dialog_id, message_id):
+ """Takes a response out of the chat showing it, and out of the kept
+ ones. UI thread."""
+ state = _state()
+ state["shown"].pop((int(account), int(message_id)), None)
+ state["history"][:] = [
+ e for e in state["history"] if (e["account"], e["id"]) != (int(account), int(message_id))
+ ]
+ fragment = open_chat(account, dialog_id)
+ if fragment is None:
+ return
+ from java import jboolean, jint, jlong
+ from java.lang import Long
+ from org.telegram.messenger import ChatObject
+
+ chat = fragment.getCurrentChat()
+ channel_id = int(chat.id) if chat is not None and ChatObject.isChannel(chat) else 0
+ method(CHAT, "processDeletedMessages", "java.util.ArrayList", Long.TYPE, jboolean_type()).invoke(
+ fragment, array_list([jint(int(message_id))]), jlong(channel_id), jboolean(False)
+ )
+
+
+def _label_only_you(cell):
+ """The cell's "Only visible to @…" label, as "Only visible to you"."""
+ from android.text import Layout, SpannableStringBuilder, StaticLayout
+ from org.telegram.messenger import AndroidUtilities, LocaleController, R
+ from org.telegram.ui.ActionBar import Theme
+
+ layout = field(cell, "ephemeralLayout")
+ if layout is None:
+ return
+ label = SpannableStringBuilder(layout.getText())
+ you = LocaleController.getString(R.string.EphemeralMessageVisibleToYou)
+ if str(label)[2:] == str(you):
+ return
+ label.replace(2, label.length(), you) # after the icon and a space
+ relabeled = StaticLayout(
+ label, Theme.chat_ephemeralPaint, AndroidUtilities.dp(300), Layout.Alignment.ALIGN_NORMAL, 0.0, 0.0, False
+ )
+ set_field(cell, "ephemeralLayout", relabeled)
+ if relabeled.getLineCount() > 0:
+ from java import jint
+
+ set_field(cell, "ephemeralWidth", jint(int(relabeled.getLineWidth(0)) + AndroidUtilities.dp(10)))
+
+
+class Responses:
+ """Keeps deleting and replying to responses on the device."""
+
+ def __init__(self, plugin):
+ self.plugin = plugin
+ self.handles = []
+ # Chats (class guids) already given their kept responses; a chat
+ # restored before a reload of this plugin is skipped by its ids.
+ self._restored = set()
+
+ def install(self):
+ class LocalDelete(MethodHook):
+ # MessagesController.deleteEphemeralMessage(long dialogId, int topicId, MessageObject)
+ def before_hooked_method(self, param):
+ try:
+ message = param.args[2]
+ if is_local(message):
+ remove(int(message.currentAccount), int(param.args[0]), int(message.getId()))
+ param.setResult(None)
+ except Exception as exc:
+ diag.error("local delete", exc)
+
+ class NoReply(MethodHook):
+ # ChatActivity.showFieldPanelForReply(MessageObject)
+ def before_hooked_method(self, param):
+ try:
+ if param.args and is_local(param.args[0]):
+ param.setResult(None)
+ except Exception as exc:
+ diag.error("local reply", exc)
+
+ self.handles += hook_methods(
+ self.plugin, "org.telegram.messenger.MessagesController", "deleteEphemeralMessage", LocalDelete()
+ )
+ class OnlyYou(MethodHook):
+ # ChatMessageCell.setMessageObjectInternal(MessageObject)
+ def after_hooked_method(self, param):
+ try:
+ if is_local(param.args[0]):
+ _label_only_you(param.thisObject)
+ except Exception as exc:
+ diag.error("local label", exc)
+
+ restored = self._restored
+
+ class Visible(MethodHook):
+ # ChatActivity.updateMessagesVisiblePart: its messages are shown
+ def after_hooked_method(self, param):
+ try:
+ chat = param.thisObject
+ guid = int(chat.getClassGuid())
+ if guid in restored:
+ return
+ messages = field(chat, "messages")
+ if messages is None or messages.isEmpty():
+ return
+ restored.add(guid)
+ restore(chat)
+ except Exception as exc:
+ diag.error("restore responses", exc)
+
+ self.handles += hook_methods(self.plugin, CHAT, "updateMessagesVisiblePart", Visible())
+ self.handles += hook_methods(self.plugin, CHAT, "showFieldPanelForReply", NoReply(), arity=1)
+ self.handles += hook_methods(
+ self.plugin, "org.telegram.ui.Cells.ChatMessageCell", "setMessageObjectInternal", OnlyYou(), arity=1
+ )
+
+ def uninstall(self):
+ unhook_all(self.plugin, self.handles)
diff --git a/src/hooks/suggest.py b/src/hooks/suggest.py
new file mode 100644
index 0000000..f943d23
--- /dev/null
+++ b/src/hooks/suggest.py
@@ -0,0 +1,386 @@
+"""Commands in Telegram's own suggestion list above the input.
+
+Typing `/` lists the account's commands as bot commands are listed (the
+account's avatar, `/name`, then its parameters and summary), before the
+chat's bot commands. Once a command is followed by a space, the list shows
+its parameters Discord-style instead: the one at the cursor highlighted,
+missing required ones in red, given ones with their value, and whether it
+needs a reply.
+
+MentionsAdapter.searchUsernameOrHashtag runs on every change of the field;
+afterwards its command lists are filled in. Tapping a row is handled here
+(the chat's item listeners are wrapped after ChatActivity.createView):
+a command without required parameters runs, like a bot command is sent;
+any other is put into the field to complete; a parameter row adds `name:`.
+"""
+
+from base_plugin import MethodHook
+
+import diag
+from commands.parse import analyze, command_name_at, split_command
+from config import SHOW_PARAMS
+from jutil import array_list, field, hook_methods, set_field, unhook_all
+from hooks.composer import reply_of
+from hooks.notify import bulletin
+
+ADAPTER = "org.telegram.ui.Adapters.MentionsAdapter"
+CHAT = "org.telegram.ui.ChatActivity"
+
+_CLEARED = (
+ "searchResultUsernames",
+ "searchResultUsernamesMap",
+ "searchResultHashtags",
+ "searchResultSuggestions",
+ "searchResultBotContext",
+ "stickers",
+ "hintHashtag",
+)
+
+_COMMAND_LISTS = (
+ "searchResultCommands",
+ "searchResultCommandsHelp",
+ "searchResultCommandsUsers",
+ "searchResultCommandsEphemeral",
+)
+
+
+def _identity(obj):
+ from java.lang import System
+
+ return int(System.identityHashCode(obj))
+
+
+class Row:
+ __slots__ = ("kind", "command", "param", "label", "help", "style")
+
+ def __init__(self, kind, command, label, help, param=None, style=None):
+ self.kind = kind # command | header | param | reply
+ self.command = command
+ self.param = param
+ self.label = label
+ self.help = help
+ self.style = style # current | missing | filled | None
+
+
+def param_rows(command, analysis, replying):
+ """The rows describing `command`'s parameters while it is typed."""
+ rows = [Row("header", command, "/" + command.name, command.summary or command.usage())]
+ for param in command.params:
+ value = analysis.values.get(param.name)
+ need = "required" if param.required else "optional"
+ about = f"{param.type_label()} · {need}"
+ if param.description:
+ about += f" · {param.description}"
+ if value is not None:
+ about = f"✓ {value} · {about}"
+ if param.name == analysis.current:
+ style = "current"
+ elif value is not None:
+ style = "filled"
+ else:
+ style = "missing" if param.required else None
+ rows.append(Row("param", command, param.name, about, param, style))
+ if command.reply != "none":
+ if replying:
+ rows.append(Row("reply", command, "↩ reply", "✓ replying to a message", style="filled"))
+ else:
+ need = "required" if command.reply == "required" else "optional"
+ style = "missing" if command.reply == "required" else None
+ rows.append(Row("reply", command, "↩ reply", f"{need} · swipe a message to reply to it", style=style))
+ return rows
+
+
+class Suggest:
+ def __init__(self, plugin, catalog, composer):
+ self.plugin = plugin
+ self.catalog = catalog
+ self.composer = composer
+ self.handles = []
+ self._states = {} # adapter identity -> ([Row] first in its list, that list's identity)
+ self._proxies = {}
+
+ def install(self):
+ suggest = self
+
+ class Search(MethodHook):
+ def before_hooked_method(self, param):
+ try:
+ if suggest.before_search(param.thisObject, param.args):
+ param.setResult(None)
+ except Exception as exc:
+ diag.error("suggest search", exc)
+
+ def after_hooked_method(self, param):
+ try:
+ suggest.after_search(param.thisObject, param.args)
+ except Exception as exc:
+ diag.error("suggest search", exc)
+
+ class Bind(MethodHook):
+ def after_hooked_method(self, param):
+ try:
+ suggest.on_bind(param.thisObject, param.args[0], int(param.args[1]))
+ except Exception as exc:
+ diag.error("suggest bind", exc)
+
+ class Created(MethodHook):
+ def after_hooked_method(self, param):
+ try:
+ suggest.wrap_listeners(param.thisObject)
+ except Exception as exc:
+ diag.error("suggest listeners", exc)
+
+ self.handles += hook_methods(self.plugin, ADAPTER, "searchUsernameOrHashtag", Search(), arity=5)
+ self.handles += hook_methods(self.plugin, ADAPTER, "onBindViewHolder", Bind(), arity=2)
+ self.handles += hook_methods(self.plugin, CHAT, "createView", Created(), arity=1)
+
+ def uninstall(self):
+ unhook_all(self.plugin, self.handles)
+ self._states.clear()
+
+ # filling the list
+
+ def _typed(self, adapter, args):
+ """(chat, text, cursor, commands) when a command may be being typed."""
+ chat = getattr(adapter, "parentFragment", None)
+ if chat is None or args[3] or args[4] or args[0] is None:
+ return None
+ text = str(args[0])
+ if not text.startswith("/") or int(chat.getChatMode()) != 0:
+ return None
+ commands = self.catalog.commands_for(int(chat.getCurrentAccount()))
+ return (chat, text, int(args[1]), commands) if commands else None
+
+ def before_search(self, adapter, args):
+ """Parameters of a known command: shown instead of the app's search
+ (whose `:` emoji search would replace them); True to skip it."""
+ self._states.pop(_identity(adapter), None)
+ typed = self._typed(adapter, args)
+ if typed is None:
+ return False
+ chat, text, cursor, commands = typed
+ for name in _COMMAND_LISTS: # left from an earlier search
+ set_field(adapter, name, None)
+ split = split_command(text)
+ if split is None or not self.plugin.get_setting(SHOW_PARAMS, True):
+ return False
+ command = next((c for c in commands if split[0] in c.names), None)
+ if command is None or (not command.params and command.reply == "none"):
+ return False
+ reply = reply_of(chat)
+ analysis = analyze(command, text, cursor, reply is not None)
+ self._fill(adapter, chat, param_rows(command, analysis, reply is not None), keep_bots=False)
+ self._text_changed(chat)
+ return True
+
+ def after_search(self, adapter, args):
+ """Commands matching a bare `/prefix`, before the chat's bot commands."""
+ typed = self._typed(adapter, args)
+ chat = getattr(adapter, "parentFragment", None)
+ if chat is not None:
+ self._text_changed(chat)
+ if typed is None:
+ return
+ chat, text, cursor, commands = typed
+ prefix = command_name_at(text) if cursor >= len(text) else None
+ if prefix is None:
+ return
+ found = [c for c in commands if any(n.startswith(prefix) for n in c.names)]
+ if found:
+ rows = [Row("command", c, "/" + c.name, c.hint()) for c in found]
+ self._fill(adapter, chat, rows, keep_bots=True)
+
+ def _text_changed(self, chat):
+ enter = chat.getChatActivityEnterView()
+ if enter is not None:
+ self.composer.text_changed(enter)
+
+ def _fill(self, adapter, chat, rows, keep_bots):
+ from org.telegram.messenger import MessagesController, UserConfig
+
+ account = int(chat.getCurrentAccount())
+ me = MessagesController.getInstance(account).getUser(UserConfig.getInstance(account).getClientUserId())
+ names, helps, users, ephemeral = [], [], [], []
+ # Quick replies go too: the app binds command rows as if they were
+ # first in the list.
+ for row in rows:
+ names.append(row.label)
+ helps.append(row.help)
+ users.append(me if row.kind in ("command", "header") else None)
+ ephemeral.append(False)
+ if keep_bots:
+ bots = field(adapter, "searchResultCommands")
+ if bots is not None:
+ bot_help = field(adapter, "searchResultCommandsHelp")
+ bot_users = field(adapter, "searchResultCommandsUsers")
+ bot_ephemeral = field(adapter, "searchResultCommandsEphemeral")
+ for i in range(bots.size()):
+ names.append(bots.get(i))
+ helps.append(bot_help.get(i) if bot_help is not None and i < bot_help.size() else "")
+ users.append(bot_users.get(i) if bot_users is not None and i < bot_users.size() else None)
+ ephemeral.append(
+ bool(bot_ephemeral.get(i)) if bot_ephemeral is not None and i < bot_ephemeral.size() else False
+ )
+ for name in _CLEARED + ("quickReplies",):
+ set_field(adapter, name, None)
+ shown = array_list(names)
+ set_field(adapter, "searchResultCommands", shown)
+ set_field(adapter, "searchResultCommandsHelp", array_list(helps))
+ set_field(adapter, "searchResultCommandsUsers", array_list(users))
+ set_field(adapter, "searchResultCommandsEphemeral", array_list(ephemeral))
+ self._states[_identity(adapter)] = (rows, _identity(shown))
+ self.wrap_listeners(chat)
+ adapter.notifyDataSetChanged()
+ # Its diff compares the labels only: rows whose help or style
+ # changed have to be rebound too.
+ adapter.notifyItemRangeChanged(0, len(rows))
+ delegate = field(adapter, "delegate")
+ if delegate is not None:
+ delegate.needChangePanelVisibility(True)
+
+ def _rows(self, adapter):
+ """Our rows while the adapter still shows them (an async search of
+ the app may have replaced them since)."""
+ state = self._states.get(_identity(adapter))
+ if state is None:
+ return ()
+ shown = field(adapter, "searchResultCommands")
+ if shown is None or _identity(shown) != state[1] or field(adapter, "searchResultSuggestions") is not None:
+ return ()
+ return state[0]
+
+ # styling rows
+
+ def on_bind(self, adapter, holder, position):
+ """Colors our rows' names; a recycled cell gets its own color back.
+ The original color is kept in the name view's tag, which outlives
+ this plugin being reloaded."""
+ cell = holder.itemView
+ if cell.getClass().getSimpleName() != "MentionCell":
+ return
+ name_view = field(cell, "nameTextView")
+ if name_view is None:
+ return
+ from java import jint
+
+ rows = self._rows(adapter)
+ row = rows[position] if 0 <= position < len(rows) else None
+ original = name_view.getTag()
+ if row is None or row.style is None:
+ if original is not None:
+ name_view.setTextColor(jint(int(original)))
+ name_view.setTag(None)
+ return
+ from org.telegram.ui.ActionBar import Theme
+
+ if original is None:
+ name_view.setTag(int(name_view.getCurrentTextColor()))
+ color = {
+ "current": Theme.key_windowBackgroundWhiteBlueText,
+ "missing": Theme.key_text_RedRegular,
+ "filled": Theme.key_windowBackgroundWhiteGreenText,
+ }[row.style]
+ name_view.setTextColor(jint(Theme.getColor(color)))
+
+ # tapping rows
+
+ def wrap_listeners(self, chat):
+ """Routes taps on the chat's suggestion list through `tapped`/`held`
+ (once per list and load of this plugin). The app's own listeners
+ handle every other row: the click one is ChatActivity's field, the
+ long-click one is kept across reloads (the list then holds ours)."""
+ from library import shared
+
+ container = field(chat, "mentionContainer")
+ if container is None:
+ return
+ list_view = container.getListView()
+ key = _identity(list_view)
+ mine = self._proxies.get(key)
+ current = field(list_view, "onItemClickListener")
+ if mine is not None and current is not None and _identity(current) == _identity(mine[0]):
+ return
+ click = field(chat, "mentionsOnItemClickListener")
+ long_clicks = shared().setdefault("long_clicks", {})
+ long_click = long_clicks.get(key) or field(list_view, "onItemLongClickListener")
+ long_clicks[key] = long_click
+ from java import dynamic_proxy
+ from org.telegram.ui.Components import RecyclerListView
+
+ suggest = self
+
+ class Click(dynamic_proxy(RecyclerListView.OnItemClickListener)):
+ def onItemClick(self, view, position):
+ handled = False
+ try:
+ handled = suggest.tapped(chat, container, position)
+ except Exception as exc:
+ diag.error("suggest tap", exc)
+ if not handled and click is not None:
+ click.onItemClick(view, position)
+
+ class LongClick(dynamic_proxy(RecyclerListView.OnItemLongClickListener)):
+ def onItemClick(self, view, position):
+ try:
+ if suggest.tapped(chat, container, position):
+ return True
+ except Exception as exc:
+ diag.error("suggest hold", exc)
+ return bool(long_click.onItemClick(view, position)) if long_click is not None else False
+
+ proxies = (Click(), LongClick())
+ self._proxies[key] = proxies # referenced as long as the list lives
+ if len(self._proxies) > 32:
+ self._proxies.pop(next(iter(self._proxies)))
+ list_view.setOnItemClickListener(proxies[0])
+ if long_click is not None:
+ list_view.setOnItemLongClickListener(proxies[1])
+
+ def row_at(self, container, position):
+ """Our row at a list position (the list's first item is padding)."""
+ rows = self._rows(container.getAdapter())
+ index = position - 1
+ return rows[index] if 0 <= index < len(rows) else None
+
+ def _our_command_at(self, chat, container, position):
+ """One of our commands at a list position though our rows are gone
+ (a stale list): its row is never left to the app, which would send
+ it."""
+ item = container.getAdapter().getItem(position - 1)
+ if not isinstance(item, str) or not item.startswith("/"):
+ return None
+ name = item[1:].split("@", 1)[0]
+ return self.catalog.find(int(chat.getCurrentAccount()), name)
+
+ def tapped(self, chat, container, position):
+ """A tap (or hold) on the list: True when it was one of our rows.
+ Rows only ever fill the field; sending is the send button's job."""
+ row = self.row_at(container, position)
+ if row is None:
+ command = self._our_command_at(chat, container, position)
+ if command is None:
+ return False
+ row = Row("command", command, "/" + command.name, "")
+ enter = chat.getChatActivityEnterView()
+ if enter is None:
+ return True
+ command = row.command
+ if row.kind in ("command", "header"):
+ takes = command.params or command.reply != "none"
+ text = str(enter.getFieldText() or "")
+ if row.kind == "command" or split_command(text) is None:
+ self.composer.set_text(enter, f"/{command.name} " if takes else f"/{command.name}")
+ elif command.description:
+ bulletin(command.description)
+ elif row.kind == "param":
+ text = str(enter.getFieldText() or "")
+ if row.param.name not in (self._current_values(chat, text) or {}):
+ gap = "" if not text or text.endswith(" ") else " "
+ self.composer.set_text(enter, f"{text}{gap}{row.param.name}:")
+ elif row.kind == "reply":
+ bulletin("Swipe a message to the left to reply to it")
+ return True
+
+ def _current_values(self, chat, text):
+ typed = self.composer.read(chat.getChatActivityEnterView())
+ return typed.analysis.values if typed is not None else None
diff --git a/src/jutil.py b/src/jutil.py
new file mode 100644
index 0000000..57241ee
--- /dev/null
+++ b/src/jutil.py
@@ -0,0 +1,150 @@
+"""Chaquopy/Java helpers shared by the hooks (after messageLogger's jutil)."""
+
+import threading
+
+import diag
+
+
+def java_list(items):
+ """A java.util.List as a Python list (Java lists aren't iterable here)."""
+ if items is None:
+ return []
+ return [items.get(i) for i in range(items.size())]
+
+
+def array_list(items):
+ from java.util import ArrayList
+
+ out = ArrayList()
+ for item in items:
+ out.add(item)
+ return out
+
+
+def java_class(name):
+ """The real java.lang.Class: find_class returns a Chaquopy wrapper whose
+ getClass() is the Class."""
+ from hook_utils import find_class
+
+ wrapper = find_class(name)
+ if wrapper is None:
+ return None
+ try:
+ return wrapper.getClass()
+ except Exception:
+ return wrapper
+
+
+def hook_methods(plugin, class_name, name, hook, arity=None):
+ """Hooks every overload of `class_name.name` (or those with `arity`
+ parameters); returns the unhook handles."""
+ cls = java_class(class_name)
+ if cls is None:
+ diag.log(f"class not found: {class_name}")
+ return []
+ handles = []
+ for method in cls.getDeclaredMethods():
+ if method.getName() != name:
+ continue
+ if arity is not None and len(method.getParameterTypes()) != arity:
+ continue
+ try:
+ handle = plugin.hook_method(method, hook)
+ if handle is not None:
+ handles.append(handle)
+ except Exception as exc:
+ diag.error(f"hook {class_name}.{name}", exc)
+ if not handles:
+ diag.log(f"nothing hooked for {class_name}.{name}")
+ return handles
+
+
+def unhook_all(plugin, handles):
+ for handle in handles:
+ try:
+ plugin.unhook_method(handle)
+ except Exception:
+ pass
+ del handles[:]
+
+
+def field(obj, name):
+ """A (possibly private) field of a Java object."""
+ value = getattr(obj, name, None)
+ if value is not None:
+ return value
+ try:
+ from hook_utils import get_private_field
+
+ return get_private_field(obj, name)
+ except Exception:
+ return None
+
+
+def set_field(obj, name, value):
+ from hook_utils import set_private_field
+
+ set_private_field(obj, name, value)
+
+
+_methods = {}
+
+
+def method(class_name, name, *parameter_types):
+ """A declared (possibly private) method, made accessible; cached.
+ `parameter_types` are java.lang.Class objects or class names."""
+ key = (class_name, name, tuple(str(t) for t in parameter_types))
+ if key not in _methods:
+ types = [java_class(t) if isinstance(t, str) else t for t in parameter_types]
+ found = java_class(class_name).getDeclaredMethod(name, *types)
+ found.setAccessible(True)
+ _methods[key] = found
+ return _methods[key]
+
+
+def on_ui(fn, delay_ms=0):
+ from android_utils import run_on_ui_thread
+
+ def run():
+ try:
+ fn()
+ except Exception as exc:
+ diag.error("ui task", exc)
+
+ run_on_ui_thread(run, delay_ms)
+
+
+def in_background(fn):
+ """Runs `fn` off the UI thread; exceptions are logged, not lost."""
+
+ def run():
+ try:
+ fn()
+ except Exception as exc:
+ diag.error("background task", exc)
+
+ threading.Thread(target=run, daemon=True).start()
+
+
+_Runnable = []
+
+
+def runnable(fn):
+ """A java.lang.Runnable calling `fn` (exceptions logged)."""
+ if not _Runnable:
+ from java import dynamic_proxy
+ from java.lang import Runnable
+
+ class PyRunnable(dynamic_proxy(Runnable)):
+ def __init__(self, fn):
+ super().__init__()
+ self.fn = fn
+
+ def run(self):
+ try:
+ self.fn()
+ except Exception as exc:
+ diag.error("runnable", exc)
+
+ _Runnable.append(PyRunnable)
+ return _Runnable[0](fn)
diff --git a/src/library.py b/src/library.py
new file mode 100644
index 0000000..d2f71f5
--- /dev/null
+++ b/src/library.py
@@ -0,0 +1,173 @@
+"""The plugin as a library: other plugins register their own commands.
+
+Published as the module `client_commands` (in `sys.modules`), so another
+plugin can do:
+
+ import sys, types
+
+ def setup(api):
+ api.register(
+ "roll",
+ roll,
+ summary="Rolls a die",
+ params=[api.Param("sides", "integer", required=False, default=6)],
+ owner=__id__,
+ )
+
+ def roll(ctx, sides=6):
+ import random
+ return f"🎲 {random.randint(1, sides)}"
+
+ # in on_plugin_load: works whichever plugin loads first
+ api = sys.modules.setdefault("client_commands", types.ModuleType("client_commands"))
+ if hasattr(api, "register"):
+ setup(api)
+ else:
+ api.__dict__.setdefault("pending", []).append(setup)
+
+ # in on_plugin_unload
+ sys.modules["client_commands"].unregister_owner(__id__)
+
+A handler gets a `CommandContext` and the parameters as keyword arguments,
+runs off the UI thread, and returns the response as Telegram-style HTML (or
+None). `raise api.CommandError("...")` shows a user-facing error.
+
+The registry outlives the module object (the plugin engine drops modules a
+plugin added when it unloads): it hangs off `builtins`, so registrations
+survive this plugin being updated or turned off and on, and an `api` kept
+from an earlier load still registers into it.
+"""
+
+import sys
+import types
+
+from commands.markup import escape
+from commands.model import Command, Param
+from commands.parse import CommandError
+
+MODULE = "client_commands"
+VERSION = 1
+_STATE = "__client_commands_state__"
+
+
+def shared():
+ """State kept across loads of this plugin: {"registry": {name: Command},
+ ...}; the hooks keep what they must not lose on a reload here too."""
+ import builtins
+
+ state = getattr(builtins, _STATE, None)
+ if state is None:
+ state = {"registry": {}}
+ setattr(builtins, _STATE, state)
+ return state
+
+
+class CommandContext:
+ """Where a command runs. `reply_to` is the replied MessageObject (or
+ None), `values` the parsed parameters, `send_to_chat` whether the
+ response goes into the chat (as the account) instead of only to you."""
+
+ def __init__(self, account, dialog_id, topic_id, reply_to, text, command, values, send_to_chat):
+ self.account = account
+ self.dialog_id = dialog_id
+ self.topic_id = topic_id
+ self.reply_to = reply_to
+ self.text = text
+ self.command = command
+ self.values = values
+ self.send_to_chat = send_to_chat
+
+ @property
+ def client(self):
+ """The SDK's AccountClient of the account (send_text, send_request...)."""
+ from client_utils import get_client
+
+ return get_client(self.account)
+
+
+class Library:
+ def __init__(self):
+ self.module = None
+
+ @property
+ def _registry(self):
+ return shared()["registry"]
+
+ def install(self):
+ """Publishes the API (keeping earlier registrations) and runs the
+ setups other plugins queued before this plugin loaded."""
+ module = sys.modules.get(MODULE)
+ if module is None:
+ module = types.ModuleType(MODULE)
+ sys.modules[MODULE] = module
+ self.module = module
+ module.__doc__ = __doc__
+ module.VERSION = VERSION
+ module.Param = Param
+ module.CommandError = CommandError
+ module.CommandContext = CommandContext
+ module.escape = escape
+ module.register = self.register
+ module.unregister = self.unregister
+ module.unregister_owner = self.unregister_owner
+ module.commands = self.commands
+ module.available = True
+ module._registry = self._registry
+ pending = module.__dict__.pop("pending", [])
+ for setup in pending:
+ try:
+ setup(module)
+ except Exception as exc:
+ import diag
+
+ diag.error("library pending setup", exc)
+
+ def uninstall(self):
+ if self.module is not None:
+ self.module.available = False
+
+ def register(
+ self,
+ name,
+ handler,
+ *,
+ params=(),
+ summary="",
+ description="",
+ aliases=(),
+ reply="none",
+ owner=None,
+ ):
+ """Adds (or replaces) a command run by `handler(ctx, **values)`.
+ `params` are `Param`s or dicts like the panel lists them."""
+ if not callable(handler):
+ raise TypeError("handler must be callable")
+ params = [p if isinstance(p, Param) else Param.from_json(p) for p in params]
+ command = Command(
+ name,
+ params,
+ summary,
+ description,
+ aliases,
+ reply,
+ source=str(owner or "plugin"),
+ handler=handler,
+ )
+ self._registry[command.name] = command
+ return command
+
+ def unregister(self, name):
+ return self._registry.pop(str(name).lstrip("/").lower(), None) is not None
+
+ def unregister_owner(self, owner):
+ names = [n for n, c in self._registry.items() if c.source == str(owner)]
+ for name in names:
+ self._registry.pop(name)
+ return len(names)
+
+ def commands(self):
+ """Every command plugins registered, as dicts."""
+ return [dict(c.to_json(), owner=c.source) for c in self.local_commands()]
+
+ def local_commands(self):
+ return sorted(self._registry.values(), key=lambda c: c.name)
diff --git a/src/main.py b/src/main.py
index f24e0fa..fa35a01 100644
--- a/src/main.py
+++ b/src/main.py
@@ -1,50 +1,204 @@
from base_plugin import BasePlugin, HookResult, HookStrategy
-from android_utils import log
-from ui.settings import Header, Input, Text
-from greeting import render, TemplateError
+from ui.settings import Divider, Header, Input, Switch, Text
-DEFAULT_TEMPLATE = "Hello, {name}!"
+import diag
+from api import Api, ApiError
+from catalog import Catalog
+from commands.parse import split_command
+from config import API_TOKENS, DEFAULTS, SEND_TO_CHAT, SERVER_COMMANDS, SERVER_URL, SHOW_PARAMS
+from jutil import in_background, on_ui
+from library import Library
+from runner import Invocation, Runner
+from hooks.composer import Composer
+from hooks.menu import ResponseMenu
+from hooks.notify import bulletin
+from hooks.responses import Responses
+from hooks.suggest import Suggest
-class HelloPlugin(BasePlugin):
+class ClientCommandsPlugin(BasePlugin):
def on_plugin_load(self):
- log("hello_bundle loaded")
+ self.library = Library()
+ self.library.install()
+ self.api = Api(self)
+ self.catalog = Catalog(self, self.api, self.library)
+ self.runner = Runner(self, self.api, self.catalog, self.library)
+ self.responses = Responses(self)
+ self.composer = Composer(self, self.catalog, self.runner)
+ self.suggest = Suggest(self, self.catalog, self.composer)
+ self.menu = ResponseMenu(self, self.runner)
+ self.catalog.on_change(self.composer.forget)
+ for part in (self.responses, self.composer, self.suggest, self.menu):
+ try:
+ part.install()
+ except Exception as exc:
+ diag.error(f"install {type(part).__name__}", exc)
self.add_on_send_message_hook()
+ self.catalog.start()
+ diag.log("loaded")
def on_plugin_unload(self):
- log("hello_bundle unloaded")
-
- def create_settings(self):
- return [
- Header(text="Hello Bundle"),
- Input(
- key="template",
- text="Greeting template",
- default=DEFAULT_TEMPLATE,
- subtext="Use {name} as placeholder.",
- icon="msg_edit",
- ),
- Text(
- text="Example",
- subtext=".hello alice smith -> Hello, Alice Smith!",
- icon="msg_info",
- ),
- ]
+ for part in ("menu", "suggest", "composer", "responses"):
+ if hasattr(self, part):
+ getattr(self, part).uninstall()
+ if hasattr(self, "catalog"):
+ self.catalog.stop()
+ if hasattr(self, "library"):
+ self.library.uninstall()
+ diag.log("unloaded")
def on_send_message_hook(self, account, params):
- message = getattr(params, "message", None)
- if not isinstance(message, str):
+ """Commands sent some other way than the field (the field's own
+ sending is handled by the composer): they run when complete, else go
+ back into the field. Never sent as text, unless "Send as message"."""
+ try:
+ return self._sent(int(account), params)
+ except Exception as exc:
+ diag.error("send hook", exc)
return HookResult()
- stripped = message.strip()
- if not stripped.startswith(".hello"):
+
+ def _sent(self, account, params):
+ text = getattr(params, "message", None)
+ if not isinstance(text, str) or not text.startswith("/"):
return HookResult()
- parts = stripped.split(" ", 1)
- name = parts[1] if len(parts) > 1 else ""
- if not name.strip():
- params.message = "Usage: .hello "
- else:
- try:
- params.message = render(self.get_setting("template", DEFAULT_TEMPLATE), name)
- except TemplateError as exc:
- params.message = "Bad template: " + str(exc)
- return HookResult(strategy=HookStrategy.MODIFY, params=params)
+ if self.runner.bypassed(text, consume=True) or int(getattr(params, "scheduleDate", 0) or 0):
+ return HookResult()
+ split = split_command(text)
+ command = self.catalog.find(account, split[0]) if split else None
+ if command is None:
+ return HookResult()
+ from commands.parse import analyze
+
+ reply = getattr(params, "replyToMsg", None)
+ thread = getattr(params, "replyToTopMsg", None)
+ if reply is not None and thread is not None and int(reply.getId()) == int(thread.getId()):
+ reply = None # the topic itself, not a message in it
+ analysis = analyze(command, text, None, reply is not None)
+ if not analysis.complete:
+ on_ui(lambda: self._refill(account, int(params.peer), text, analysis.error), 80)
+ return HookResult(strategy=HookStrategy.CANCEL)
+ topic = int(thread.getId()) if thread is not None else 0
+ inv = Invocation(
+ account,
+ int(params.peer),
+ topic,
+ reply,
+ thread,
+ text,
+ command,
+ self.get_setting(SEND_TO_CHAT, False),
+ )
+ on_ui(lambda: self.runner.run(inv))
+ return HookResult(strategy=HookStrategy.CANCEL)
+
+ def _refill(self, account, dialog_id, text, reason):
+ """Puts an incomplete command back into the open chat's field (after
+ the app cleared it for sending)."""
+ from hooks.responses import open_chat
+
+ chat = open_chat(account, dialog_id)
+ enter = chat.getChatActivityEnterView() if chat is not None else None
+ if enter is not None:
+ self.composer.set_text(enter, text if text.endswith(" ") else text + " ")
+ bulletin(reason, error=True)
+
+ # settings
+
+ def _tokens_changed(self, _value=None):
+ self.catalog.refresh(reset=True)
+
+ def _test_connection(self, _view=None):
+ def run():
+ if not self.api.configured:
+ bulletin("Client Commands: set the panel URL and API tokens", error=True)
+ return
+ lines, failed = [], False
+ for token in self.api.tokens():
+ try:
+ me = self.api.me(token)
+ state = "✓" if me.get("running") else "✓ (not running on the panel)"
+ lines.append(f"{me.get('account_name')} {state}")
+ except ApiError as exc:
+ failed = True
+ lines.append(f"{token[:10]}…: {exc}")
+ bulletin("Accounts: " + "; ".join(lines), error=failed)
+ self.catalog.refresh()
+
+ in_background(run)
+
+ def _refresh(self, _view=None):
+ self.catalog.refresh(reset=True)
+ bulletin("Client Commands: asking the panel for commands")
+
+ def _command_rows(self):
+ from client_utils import get_selected_account
+
+ account = get_selected_account()
+ commands = self.catalog.commands_for(account)
+ if not commands:
+ return [Text(text="No commands yet", subtext="Set up the panel above, or install a plugin that adds some")]
+ rows = []
+ for command in commands:
+ source = "panel" if not command.local else command.source
+ about = command.summary or command.description
+ rows.append(
+ Text(
+ text=command.usage(),
+ subtext=f"{about} · {source}" if about else source,
+ icon="msg_bot",
+ )
+ )
+ return rows
+
+ def create_settings(self):
+ rows = [
+ Header(text="Panel"),
+ Input(
+ key=SERVER_URL,
+ text="telegram-panel URL",
+ default=DEFAULTS[SERVER_URL],
+ subtext="e.g. https://panel.example.com",
+ icon="msg_link",
+ on_change=self._tokens_changed,
+ ),
+ Input(
+ key=API_TOKENS,
+ text="API tokens",
+ default=DEFAULTS[API_TOKENS],
+ subtext="Client commands tokens, one per account, separated by spaces "
+ "(panel: Settings → API Tokens). Not the message logger ones.",
+ icon="msg_permissions",
+ on_change=self._tokens_changed,
+ ),
+ Text(text="Test connection", icon="msg_info", on_click=self._test_connection),
+ Text(text="Refresh commands", icon="msg_retry", on_click=self._refresh),
+ Header(text="Commands"),
+ Switch(
+ key=SERVER_COMMANDS,
+ text="Panel commands",
+ default=DEFAULTS[SERVER_COMMANDS],
+ subtext="Run the panel's commands from here instead of sending them",
+ icon="msg_bot",
+ ),
+ Switch(
+ key=SEND_TO_CHAT,
+ text="Send response in chat",
+ default=DEFAULTS[SEND_TO_CHAT],
+ subtext="Responses go into the chat as you, not only to you. Also in the menu of a held send button",
+ icon="msg_discussion",
+ ),
+ Switch(
+ key=SHOW_PARAMS,
+ text="Parameter hints",
+ default=DEFAULTS[SHOW_PARAMS],
+ subtext="List a command's parameters while you type it",
+ icon="msg_help",
+ ),
+ Text(
+ text="Available commands",
+ icon="msg_bots",
+ create_sub_fragment=self._command_rows,
+ ),
+ Divider(text="Hold send on a command to send it as a plain message instead."),
+ ]
+ return rows
diff --git a/src/peers.py b/src/peers.py
new file mode 100644
index 0000000..13a5ff5
--- /dev/null
+++ b/src/peers.py
@@ -0,0 +1,22 @@
+"""Telegram Android dialog ids <-> the panel's marked (Telethon) chat ids.
+
+Android: users are positive, basic groups and channels both `-id`.
+Telethon: users positive, basic groups `-id`, channels `-(10**12 + id)`.
+"""
+
+CHANNEL_OFFSET = 10**12
+
+
+def marked_chat_id(account, dialog_id):
+ """The panel's chat id for a dialog, or None (secret chats)."""
+ from org.telegram.messenger import ChatObject, DialogObject, MessagesController
+
+ dialog_id = int(dialog_id)
+ if DialogObject.isEncryptedDialog(dialog_id):
+ return None
+ if dialog_id > 0:
+ return dialog_id
+ chat = MessagesController.getInstance(account).getChat(-dialog_id)
+ if chat is not None and ChatObject.isChannel(chat):
+ return -(CHANNEL_OFFSET + -dialog_id)
+ return dialog_id
diff --git a/src/runner.py b/src/runner.py
new file mode 100644
index 0000000..c932673
--- /dev/null
+++ b/src/runner.py
@@ -0,0 +1,190 @@
+"""Running a command: on the panel, or on the device for plugins' commands.
+
+While it runs, a "Running…" response stands in the chat; then the response
+(or the error) replaces it, shown only to you, or the response goes into the
+chat as the account when "Send response in chat" is on.
+"""
+
+import threading
+import time
+
+import diag
+import peers
+from api import ApiError
+from commands import markup
+from commands.parse import parse
+from jutil import array_list, in_background, on_ui
+from hooks import responses
+
+_RUNNING = "⏳ Running…"
+_DONE = "✓ Done"
+
+
+class Invocation:
+ """A command the composer is about to run (built on the UI thread)."""
+
+ def __init__(self, account, dialog_id, topic_id, reply, thread, text, command, send_to_chat):
+ self.account = int(account)
+ self.dialog_id = int(dialog_id)
+ self.topic_id = int(topic_id or 0)
+ self.reply = reply # the replied MessageObject, or None
+ self.thread = thread # the topic's MessageObject, or None
+ self.text = text.strip()
+ self.command = command
+ self.send_to_chat = bool(send_to_chat)
+ self.chat_id = peers.marked_chat_id(account, dialog_id)
+
+
+class Runner:
+ def __init__(self, plugin, api, catalog, library):
+ self.plugin = plugin
+ self.api = api
+ self.catalog = catalog
+ self.library = library
+ # Texts this plugin sends itself: the send hook lets them through.
+ self._bypass = {}
+ self._lock = threading.Lock()
+
+ # sending as plain messages
+
+ def bypass(self, text):
+ with self._lock:
+ self._bypass[text.strip()] = time.monotonic() + 30
+
+ def bypassed(self, text, consume=False):
+ key = (text or "").strip()
+ with self._lock:
+ until = self._bypass.get(key, 0)
+ if until < time.monotonic():
+ self._bypass.pop(key, None)
+ return False
+ if consume:
+ self._bypass.pop(key, None)
+ return True
+
+ # running
+
+ def run(self, inv):
+ """Starts `inv` (UI thread)."""
+ pending = responses.show(
+ inv.account, inv.dialog_id, inv.topic_id, inv.reply, _RUNNING, header=inv.text, keep=False
+ )
+
+ def work():
+ try:
+ result = self._local(inv) if inv.command.local else self._server(inv)
+ except Exception as exc:
+ diag.error(f"/{inv.command.name}", exc)
+ result = {"ok": False, "error": f"/{inv.command.name} failed: {exc}"}
+ on_ui(lambda: self._finish(inv, pending, result))
+
+ in_background(work)
+
+ def _server(self, inv):
+ if inv.chat_id is None:
+ return {"ok": False, "error": "Panel commands don't work in secret chats"}
+ token = self.catalog.token_for(inv.account)
+ if token is None:
+ return {
+ "ok": False,
+ "error": "No API token for this account: create a Client commands token "
+ "for it in the panel (Settings → API Tokens) and add it in the plugin settings",
+ }
+ from catalog import own_user_id
+
+ reply_id = int(inv.reply.getId()) if inv.reply is not None else None
+ try:
+ return self.api.execute(
+ token,
+ inv.text,
+ inv.chat_id,
+ reply_id,
+ inv.topic_id or None,
+ inv.send_to_chat,
+ own_user_id(inv.account),
+ )
+ except ApiError as exc:
+ if exc.refused:
+ self.catalog.refresh(reset=True)
+ return {"ok": False, "error": str(exc)}
+
+ def _local(self, inv):
+ from library import CommandContext
+
+ try:
+ values = parse(inv.command, inv.text)
+ ctx = CommandContext(
+ inv.account,
+ inv.dialog_id,
+ inv.topic_id,
+ inv.reply,
+ inv.text,
+ inv.command.name,
+ values,
+ inv.send_to_chat,
+ )
+ html = inv.command.handler(ctx, **values)
+ except Exception as exc:
+ if type(exc).__name__ == "CommandError":
+ return {"ok": False, "error": str(exc)}
+ raise
+ html = "" if html is None else str(html)
+ text, entities = markup.parse(html)
+ return {"ok": True, "html": html, "text": text, "entities": entities}
+
+ def _finish(self, inv, pending, result):
+ if pending is not None:
+ responses.remove(inv.account, inv.dialog_id, pending)
+ if not result.get("ok"):
+ error = result.get("error") or "Failed"
+ responses.show(inv.account, inv.dialog_id, inv.topic_id, inv.reply, f"❌ {error}", header=inv.text)
+ return
+ text = result.get("text") or ""
+ if inv.send_to_chat:
+ if inv.command.local and text.strip():
+ self.send(inv.account, inv.dialog_id, inv.reply, inv.thread, text, result.get("entities") or ())
+ return
+ if not text.strip():
+ responses.show(inv.account, inv.dialog_id, inv.topic_id, inv.reply, _DONE, header=inv.text)
+ return
+ responses.show(
+ inv.account,
+ inv.dialog_id,
+ inv.topic_id,
+ inv.reply,
+ text,
+ result.get("entities") or (),
+ header=inv.text,
+ thread=inv.thread,
+ answer=True,
+ )
+
+ def send(self, account, dialog_id, reply, thread, text, entities):
+ """Sends a response into the chat as the account: an ordinary text
+ message with formatting entities (not a rich or ephemeral one; a reply
+ to an ephemeral message would make it one). UI thread."""
+ from org.telegram.messenger import SendMessagesHelper
+
+ if reply is not None and reply.isEphemeral():
+ reply = None
+ self.bypass(text)
+ params = SendMessagesHelper.SendMessageParams.of(
+ text,
+ dialog_id,
+ reply,
+ thread,
+ None,
+ False,
+ responses.tl_entities(entities) if entities else array_list([]),
+ None,
+ None,
+ True,
+ 0,
+ 0,
+ None,
+ False,
+ )
+ params.richMessage = None
+ if hasattr(params, "ephemeralReceiverBotId"):
+ params.ephemeralReceiverBotId = 0
+ SendMessagesHelper.getInstance(account).sendMessage(params)
diff --git a/src/textutil.py b/src/textutil.py
deleted file mode 100644
index 7c3107b..0000000
--- a/src/textutil.py
+++ /dev/null
@@ -1,2 +0,0 @@
-def normalize(name: str) -> str:
- return " ".join(w[:1].upper() + w[1:] for w in name.split())
diff --git a/tools/verify_bundle.py b/tools/verify_bundle.py
index 2dbc171..8853a6a 100755
--- a/tools/verify_bundle.py
+++ b/tools/verify_bundle.py
@@ -1,6 +1,11 @@
#!/usr/bin/env python3
"""Offline verification harness for a built .plugin bundle. Runs without a phone.
+Executes the bundle against faked SDK modules and checks what can be checked
+off the device: metadata, the argument parser (in step with telegram-panel's),
+HTML responses, the library API other plugins use, the command catalog
+against a fake panel, and plugin commands run by the runner.
+
Usage: python3 tools/verify_bundle.py
"""
from __future__ import annotations
@@ -8,7 +13,7 @@ from __future__ import annotations
import ast
import json
import sys
-import traceback
+import tempfile
import types
from dataclasses import dataclass
from enum import Enum, auto
@@ -31,6 +36,14 @@ def check(name: str, condition: bool, detail: str = "") -> None:
_FAILURES.append(name)
+def raises(fn, error_type, text=None):
+ try:
+ fn()
+ except error_type as exc:
+ return text is None or text in str(exc)
+ return False
+
+
class HookStrategy(Enum):
DEFAULT = auto()
MODIFY = auto()
@@ -44,7 +57,8 @@ class HookResult:
params: object = None
-LOG_LINES: list[str] = []
+class MethodHook:
+ pass
class BasePlugin:
@@ -54,56 +68,85 @@ class BasePlugin:
def get_setting(self, key, default=None):
return self._settings.get(key, default)
- def set_setting(self, key, value):
+ def set_setting(self, key, value, reload_settings=False):
self._settings[key] = value
def add_on_send_message_hook(self):
pass
- def on_plugin_load(self):
- pass
-
- def on_plugin_unload(self):
- pass
-
- def create_settings(self):
- return []
-
-
-def fake_log(*args, **kwargs):
- LOG_LINES.append(" ".join(str(a) for a in args))
-
class _SettingsRow:
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
+class _UserConfig:
+ """org.telegram.messenger.UserConfig: account slot n is Telegram user 1000 + n."""
+
+ def __init__(self, slot):
+ self.slot = slot
+
+ @classmethod
+ def getInstance(cls, slot):
+ return cls(slot)
+
+ def getClientUserId(self):
+ return 1000 + self.slot
+
+
+FILES_DIR = tempfile.mkdtemp(prefix="cc-verify-")
+
+
def install_fakes() -> None:
- base_plugin_mod = types.ModuleType("base_plugin")
- base_plugin_mod.BasePlugin = BasePlugin
- base_plugin_mod.HookResult = HookResult
- base_plugin_mod.HookStrategy = HookStrategy
- sys.modules["base_plugin"] = base_plugin_mod
+ def module(name, **attrs):
+ mod = types.ModuleType(name)
+ mod.__dict__.update(attrs)
+ sys.modules[name] = mod
+ return mod
- android_utils_mod = types.ModuleType("android_utils")
- android_utils_mod.log = fake_log
- sys.modules["android_utils"] = android_utils_mod
-
- ui_mod = types.ModuleType("ui")
- ui_mod.__path__ = []
- sys.modules["ui"] = ui_mod
-
- ui_settings_mod = types.ModuleType("ui.settings")
- for cls_name in ("Header", "Input", "Text", "Switch"):
- setattr(ui_settings_mod, cls_name, type(cls_name, (_SettingsRow,), {}))
- sys.modules["ui.settings"] = ui_settings_mod
- ui_mod.settings = ui_settings_mod
+ module(
+ "base_plugin",
+ BasePlugin=BasePlugin,
+ HookResult=HookResult,
+ HookStrategy=HookStrategy,
+ MethodHook=MethodHook,
+ )
+ module("android_utils", log=lambda *a, **k: None, run_on_ui_thread=lambda fn, delay=0: fn())
+ module("file_utils", get_files_dir=lambda: FILES_DIR)
+ ui = module("ui")
+ ui.__path__ = []
+ ui.settings = module(
+ "ui.settings",
+ **{n: type(n, (_SettingsRow,), {}) for n in ("Header", "Input", "Text", "Switch", "Divider")},
+ )
+ for name in ("org", "org.telegram"):
+ module(name).__path__ = []
+ module("org.telegram.messenger", UserConfig=_UserConfig)
-class SimpleParams:
- def __init__(self, message):
- self.message = message
+class FakeApi:
+ """The panel, as api.Api's methods answer."""
+
+ def __init__(self, tokens):
+ self._tokens = tokens # token -> (tg id, [command json]) or an ApiError
+ self.configured = True
+
+ def tokens(self):
+ return list(self._tokens)
+
+ def _of(self, token):
+ found = self._tokens[token]
+ if isinstance(found, Exception):
+ raise found
+ return found
+
+ def me(self, token):
+ tg_id, _ = self._of(token)
+ return {"tg_id": tg_id, "account_name": f"acc{tg_id}", "running": True}
+
+ def commands(self, token):
+ _, commands = self._of(token)
+ return {"commands": commands, "refresh_seconds": 3600}
KEY_TO_DUNDER = {
@@ -133,6 +176,223 @@ def expected_dunders(meta: dict) -> dict:
return result
+def check_metadata(source, ns, bundle_path):
+ meta = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
+ expected = expected_dunders(meta)
+ mismatches = [
+ f"{dunder}: expected {value!r}, got {ns.get(dunder)!r}"
+ for dunder, value in expected.items()
+ if ns.get(dunder) != value
+ ]
+ check("metadata constants match plugin.json", not mismatches, "; ".join(mismatches))
+ tree = ast.parse(source, filename=str(bundle_path))
+ assigned = {
+ t.id for node in tree.body if isinstance(node, ast.Assign) for t in node.targets if isinstance(t, ast.Name)
+ }
+ check("metadata exposed as top-level literal assignments (AST)", set(expected) <= assigned)
+ plugins = [v for v in ns.values() if isinstance(v, type) and issubclass(v, BasePlugin) and v is not BasePlugin]
+ check("exactly one BasePlugin subclass defined", len(plugins) == 1, f"found {len(plugins)}")
+ return plugins[0] if len(plugins) == 1 else None
+
+
+def check_parser(load):
+ model, parse = load("commands.model"), load("commands.parse")
+ Param, Command = model.Param, model.Command
+ ban = Command(
+ "ban",
+ [
+ Param("days", "integer"),
+ Param("mode", "choice", False, default="soft", choices=("soft", "hard")),
+ Param("silent", "boolean", False, default=False),
+ Param("reason", "string", False),
+ ],
+ "Bans the replied user.",
+ aliases=["b"],
+ reply="required",
+ )
+ # The same cases as telegram-panel's tests/test_client_commands.py.
+ check("split /Ban@me", parse.split_command("/Ban@me 3") == ("ban", "3"))
+ check("no command mid-text", parse.split_command("hello /ban") is None)
+ check(
+ "positional, choice, boolean and the greedy last string",
+ parse.parse(ban, "/b 3 HARD yes for spamming links")
+ == {"days": 3, "mode": "hard", "silent": True, "reason": "for spamming links"},
+ )
+ check(
+ "named and quoted",
+ parse.parse(ban, '/ban reason:"too loud" days:7') == {"days": 7, "mode": "soft", "silent": False, "reason": "too loud"},
+ )
+ check("an empty name: is not given", parse.parse(ban, "/ban 1 mode:")["mode"] == "soft")
+ for text, error in (
+ ("/ban", "Missing required parameter: days"),
+ ("/ban x", "days must be a whole number"),
+ ("/ban 1 medium", "mode must be one of: soft, hard"),
+ ("/ban 1 soft maybe", "silent must be yes or no"),
+ ('/ban 1 "open', "Unclosed quote"),
+ ("/ban days:1 days:2", "days is given twice"),
+ ):
+ check(f"error for {text!r}", raises(lambda: parse.parse(ban, text), parse.CommandError, error))
+ add = Command("add", [Param("a", "integer"), Param("b", "integer")])
+ check("extra words without a string to take them", raises(lambda: parse.parse(add, "/add 1 2 3"), parse.CommandError, "Unexpected arguments: 3"))
+
+ check("bare /prefix is a name being typed", parse.command_name_at("/ba") == "ba" and parse.command_name_at("/ba ") is None)
+ typing = parse.analyze(ban, "/ban 3 ", None, has_reply=True)
+ check("analyze: next free parameter is current", typing.current == "mode" and typing.complete, typing.current)
+ named = parse.analyze(ban, "/ban reason:so", None, has_reply=True)
+ check("analyze: a named parameter at the cursor", named.current == "reason" and named.missing == ["days"])
+ check("analyze: missing days blocks running", not named.complete and "days" in named.error)
+ no_reply = parse.analyze(ban, "/ban 3", None, has_reply=False)
+ check("analyze: a required reply blocks running", no_reply.error == "Reply to a message to use /ban")
+ middle = parse.analyze(ban, "/ban 3 hard", 6, has_reply=True)
+ check("analyze: the cursor inside a word", middle.current == "days", middle.current)
+ check("usage and hint", ban.usage() == "/ban [mode] [silent] [reason]" and ban.hint().startswith(" [mode]"))
+ back = Command.from_json(ban.to_json())
+ check("command JSON round trip", back.to_json() == ban.to_json() and back.names == ("ban", "b"))
+ check("bad names are refused", raises(lambda: Command("no spaces"), ValueError))
+
+
+def check_markup(load):
+ markup = load("commands.markup")
+ text, entities = markup.parse('😀 bold link
s ')
+ check("HTML text", text == "😀 bold link\ns", repr(text))
+ check(
+ "HTML entities in UTF-16 units",
+ [(e["type"], e["offset"], e["length"]) for e in entities] == [("bold", 3, 4), ("text_link", 8, 4), ("spoiler", 13, 1)],
+ repr(entities),
+ )
+ _, pre = markup.parse('x = 1
')
+ check("pre with a language", pre == [{"type": "pre", "offset": 0, "length": 5, "language": "py"}], repr(pre))
+ _, mention = markup.parse('Ann')
+ check("tg://user links are mentions", mention[0]["type"] == "mention_name" and mention[0]["user_id"] == 42)
+ check("escape", markup.escape("") == "<a&b>")
+
+
+def check_library(load):
+ import builtins
+
+ sys.modules.pop("client_commands", None)
+ builtins.__dict__.pop("__client_commands_state__", None)
+ queued = []
+ early = types.ModuleType("client_commands")
+ early.pending = [lambda api: queued.append(api.register("early", lambda ctx: "hi", owner="other"))]
+ sys.modules["client_commands"] = early
+ lib = load("library").Library()
+ lib.install()
+ api = sys.modules["client_commands"]
+ check("library published as client_commands", api is early and api.available and hasattr(api, "register"))
+ check("setups queued before loading ran", len(queued) == 1 and "pending" not in api.__dict__)
+ api.register(
+ "roll",
+ lambda ctx, sides=6: f"🎲 {sides}",
+ params=[api.Param("sides", "integer", required=False, default=6)],
+ summary="Rolls a die",
+ owner="dice",
+ )
+ names = [c["name"] for c in api.commands()]
+ check("registered commands listed", names == ["early", "roll"], repr(names))
+ lib.uninstall()
+ sys.modules.pop("client_commands") # the engine drops modules a plugin added
+ again = load("library").Library()
+ again.install()
+ check("registrations survive a reload", [c.name for c in again.local_commands()] == ["early", "roll"])
+ api.register("late", lambda ctx: None, owner="other") # an api kept from before
+ check("an old api registers into the same registry", "late" in [c.name for c in again.local_commands()])
+ api = sys.modules["client_commands"]
+ check("unregister_owner", api.unregister_owner("other") == 2 and not api.unregister("early"))
+ return again
+
+
+def check_catalog(load, library):
+ from_api = load("api")
+ catalog_mod = load("catalog")
+ getid = {
+ "name": "getid",
+ "summary": "Ids",
+ "reply": "optional",
+ "params": [{"name": "target", "type": "string", "required": False}],
+ }
+ roll_shadow = {"name": "roll", "summary": "Panel roll", "params": []}
+ api = FakeApi(
+ {
+ "tok_a": (1000, [getid, roll_shadow]),
+ "tok_b": (1001, [{"name": "ping"}]),
+ "tok_bad": from_api.ApiError("403: Not a commands token", 403),
+ }
+ )
+ plugin = BasePlugin()
+ catalog = catalog_mod.Catalog(plugin, api, library)
+ catalog._refresh_due()
+ names0 = [c.name for c in catalog.commands_for(0)]
+ check("account 0: panel commands, then plugins'", names0 == ["getid", "roll"], repr(names0))
+ check("a panel command shadows a plugin's", catalog.find(0, "roll").summary == "Panel roll")
+ names1 = [c.name for c in catalog.commands_for(1)]
+ check("account 1 has its own panel commands", names1 == ["ping", "roll"], repr(names1))
+ check("tokens picked by Telegram id", catalog.token_for(0) == "tok_a" and catalog.token_for(1) == "tok_b")
+ check("a refused token is reported", "tok_bad" in catalog.errors())
+ check("accounts without a token only get plugins'", [c.name for c in catalog.commands_for(5)] == ["roll"])
+ cached = catalog_mod.Catalog(plugin, FakeApi({}), library)
+ check("commands cached on disk", [c.name for c in cached.commands_for(0)] == ["getid", "roll"])
+ plugin.set_setting("server_commands", False)
+ check("panel commands can be turned off", [c.name for c in catalog.commands_for(0)] == ["roll"])
+ plugin.set_setting("server_commands", True)
+ api._tokens.pop("tok_b")
+ api._tokens.pop("tok_bad")
+ catalog._refresh_due()
+ check("a removed token's commands go", [c.name for c in catalog.commands_for(1)] == ["roll"])
+
+
+def check_runner(load, library):
+ runner_mod = load("runner")
+ runner = runner_mod.Runner(BasePlugin(), None, None, library)
+ roll = next(c for c in library.local_commands() if c.name == "roll")
+ inv = types.SimpleNamespace(
+ account=0, dialog_id=5, topic_id=0, reply=None, text="/roll 20", command=roll, send_to_chat=False
+ )
+ result = runner._local(inv)
+ check(
+ "plugin command response parsed",
+ result["ok"] and result["text"] == "🎲 20" and result["entities"] == [{"type": "bold", "offset": 3, "length": 2}],
+ repr(result),
+ )
+ api = sys.modules["client_commands"]
+
+ def strict(ctx, n):
+ raise api.CommandError(f"no {n}")
+
+ api.register("strict", strict, params=[{"name": "n", "type": "integer", "required": True}], owner="t")
+ inv.command = next(c for c in library.local_commands() if c.name == "strict")
+ inv.text = "/strict 3"
+ check("CommandError is a user-facing error", runner._local(inv) == {"ok": False, "error": "no 3"})
+ inv.text = "/strict"
+ check("bad arguments are a user-facing error", runner._local(inv)["error"] == "Missing required parameter: n")
+ runner.bypass("/strict 1")
+ check("bypassed text goes through once", runner.bypassed("/strict 1", consume=True) and not runner.bypassed("/strict 1"))
+
+
+def check_send_hook(plugin_cls):
+ plugin = plugin_cls()
+ plugin.runner = types.SimpleNamespace(bypassed=lambda text, consume=False: False)
+ plugin.catalog = types.SimpleNamespace(find=lambda account, name: None)
+ params = types.SimpleNamespace(message="hello", scheduleDate=0)
+ check("plain messages pass the send hook", plugin.on_send_message_hook(0, params).strategy == HookStrategy.DEFAULT)
+ params.message = "/unknown"
+ check("unknown commands pass the send hook", plugin.on_send_message_hook(0, params).strategy == HookStrategy.DEFAULT)
+ model = sys.modules.get("_cc_model")
+ demo = model.Command("demo", [model.Param("count", "integer")])
+ refills = []
+ plugin.catalog = types.SimpleNamespace(find=lambda account, name: demo if name == "demo" else None)
+ plugin._refill = lambda *args: refills.append(args)
+ params = types.SimpleNamespace(message="/demo", scheduleDate=0, peer=5, replyToMsg=None, replyToTopMsg=None)
+ result = plugin.on_send_message_hook(0, params)
+ check(
+ "an incomplete command is never sent: it goes back into the field",
+ result.strategy == HookStrategy.CANCEL and refills and refills[0][2] == "/demo",
+ repr(refills),
+ )
+ keys = {getattr(r, "key", None) for r in plugin.create_settings()}
+ check("settings have the panel URL and tokens", {"server_url", "api_tokens", "send_to_chat"} <= keys, repr(keys))
+
+
def main(argv=None) -> int:
argv = sys.argv[1:] if argv is None else argv
if len(argv) != 1:
@@ -145,112 +405,26 @@ def main(argv=None) -> int:
install_fakes()
source = bundle_path.read_text(encoding="utf-8")
-
modules_before = set(sys.modules)
- ns = {"__name__": "hello_bundle"}
+ ns = {"__name__": "client_commands_plugin"}
exec(compile(source, str(bundle_path), "exec"), ns)
- new_modules = set(sys.modules) - modules_before
- meta = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
- expected = expected_dunders(meta)
- mismatches = [
- f"{dunder}: expected {value!r}, got {ns.get(dunder)!r}"
- for dunder, value in expected.items()
- if ns.get(dunder) != value
- ]
- check("metadata constants match plugin.json", not mismatches, "; ".join(mismatches))
-
- tree = ast.parse(source, filename=str(bundle_path))
- top_level_assigned = {
- target.id
- for node in tree.body
- if isinstance(node, ast.Assign)
- for target in node.targets
- if isinstance(target, ast.Name)
- }
- check(
- "metadata exposed as top-level literal assignments (AST)",
- set(expected) <= top_level_assigned,
- f"missing: {set(expected) - top_level_assigned}",
- )
-
- plugin_classes = [
- value
- for value in ns.values()
- if isinstance(value, type) and issubclass(value, BasePlugin) and value is not BasePlugin
- ]
- check(
- "exactly one BasePlugin subclass defined",
- len(plugin_classes) == 1,
- f"found {len(plugin_classes)}",
- )
-
- plugin_cls = plugin_classes[0] if len(plugin_classes) == 1 else None
+ plugin_cls = check_metadata(source, ns, bundle_path)
+ load = ns.get("_bundle_load")
+ check("bundle exposes its module loader", load is not None)
+ if load is not None:
+ sys.modules["_cc_model"] = load("commands.model")
+ check_parser(load)
+ check_markup(load)
+ library = check_library(load)
+ check_catalog(load, library)
+ check_runner(load, library)
if plugin_cls is not None:
- check(
- "plugin class __module__ matches executed namespace",
- plugin_cls.__module__ == ns["__name__"],
- f"__module__={plugin_cls.__module__!r} != {ns['__name__']!r}",
- )
+ check_send_hook(plugin_cls)
- plugin = plugin_cls()
- LOG_LINES.clear()
- plugin.on_plugin_load()
- check(
- "on_plugin_load emits load log line",
- any("hello_bundle loaded" in line for line in LOG_LINES),
- f"log lines: {LOG_LINES}",
- )
-
- params = SimpleParams(".hello alice smith")
- result = plugin.on_send_message_hook(account=0, params=params)
- check(
- "'.hello alice smith' renders greeting",
- params.message == "Hello, Alice Smith!",
- f"got {params.message!r}",
- )
- check(
- "on_send_message_hook returns MODIFY strategy",
- result.strategy == HookStrategy.MODIFY,
- f"got {result.strategy}",
- )
-
- plugin.set_setting("template", "{nope}")
- params2 = SimpleParams(".hello someone")
- plugin.on_send_message_hook(account=0, params=params2)
- check(
- "bad template produces 'Bad template:' message",
- params2.message.startswith("Bad template:"),
- f"got {params2.message!r}",
- )
-
- # main.py imports `render`/`TemplateError` by name, so they live directly
- # in the executed namespace — call render() directly to inspect the raised
- # traceback (proves linecache registration for bundled, non-file modules).
- render = ns.get("render")
- template_error = ns.get("TemplateError")
- traceback_text = ""
- if render is not None and template_error is not None:
- try:
- render("{nope}", "x")
- except template_error:
- traceback_text = traceback.format_exc()
- check(
- "render() traceback names src/greeting/template.py",
- "src/greeting/template.py" in traceback_text,
- f"traceback:\n{traceback_text}",
- )
- check(
- "traceback includes the raising source line text",
- "unsupported placeholder" in traceback_text,
- f"traceback:\n{traceback_text}",
- )
-
- check(
- "sys.modules gained no entry for bundled module names",
- not ({"greeting", "textutil", "main"} & new_modules),
- f"leaked: {({'greeting', 'textutil', 'main'} & new_modules)}",
- )
+ sys.modules.pop("_cc_model", None)
+ leaked = {"commands", "hooks", "catalog", "runner", "main"} & (set(sys.modules) - modules_before)
+ check("bundled modules stay out of sys.modules", not leaked, f"leaked: {leaked}")
if _FAILURES:
print(f"\n{len(_FAILURES)} check(s) failed: {', '.join(_FAILURES)}", file=sys.stderr)