generated from exteraGramPlugins/pluginBase
lyuksovannyyandClaude Opus 5.5
46a1f6ee1f
Lists the panel account's client commands as Telegram slash commands (hourly refresh, cached on the device), shows their parameters while typing, runs complete commands on the panel instead of sending them and shows the answer as an "Only visible to you" message (latest 20 kept in memory). Holding send offers Run / Send as message / Send response in chat (red ! badge); an answer's menu starts with Send in chat. Other plugins register their own commands through the client_commands library module. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
151 lines
3.8 KiB
Python
151 lines
3.8 KiB
Python
"""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)
|