import json import re import time import traceback from typing import Any, List from base_plugin import BasePlugin, HookResult, HookStrategy, MenuItemData, MenuItemType, MethodHook from ui.settings import Header, Text, Divider, Input, Switch from ui.bulletin import BulletinHelper from ui.alert import AlertDialogBuilder from android_utils import log, run_on_ui_thread from client_utils import get_last_fragment, get_messages_controller from hook_utils import find_class, get_private_field from org.telegram.messenger import NotificationCenter, UserConfig, SendMessagesHelper from java.util import ArrayList from java.lang import Integer, Long try: from java.chaquopy import dynamic_proxy NotificationCenterDelegate = dynamic_proxy(NotificationCenter.NotificationCenterDelegate) except Exception: NotificationCenterDelegate = object __id__ = "mute_ls" __name__ = "Мут в ЛС" __description__ = ( "Мьютит собеседника в личке: все входящие сообщения от него удаляются. " "При желании отправляет уведомления о муте и размьюте." ) __author__ = "@shit_plugins" __version__ = "1.1" __icon__ = "exteraPlugins/1" __app_version__ = ">=12.5.1" __sdk_version__ = ">=1.4.3.3" MUTED_KEY = "muted_users" DEFAULT_PREFIX = "!" DEFAULT_DURATION = "1h" DURATION_RE = re.compile(r"(\d+)\s*([smhdw])", re.IGNORECASE) UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800} MENU_ITEM_ID = "mute_ls_chat_action" class _UpdateInterfaceHook(MethodHook): def __init__(self, plugin): super().__init__() self.plugin = plugin def after_hooked_method(self, param): try: dialog_id = int(param.args[0]) messages = param.args[1] account = get_private_field(param.thisObject, "currentAccount") self.plugin._scan_incoming(account, dialog_id, messages) except Exception: self.plugin.log("[mute_ls] updateInterface hook error: " + traceback.format_exc()) class MuteLsPlugin(BasePlugin): _muted_cache = None # ------------------------------------------------------------------ lifecycle def on_plugin_load(self): self._observers = [] self._menu_item_id = None self._last_menu_state = "init" self._muted_cache = None self._expiry_generation = 0 self._remove_expired() self._schedule_expiry_cleanup() steps = ( ("send hook", self.add_on_send_message_hook), ("update hook", lambda: self.add_hook("TL_updateNewMessage")), ("java hook", self._install_update_interface_hook), ("observers", self._register_observers), ("chat menu", self._refresh_chat_menu), ) for name, step in steps: try: step() except Exception: self.log("[mute_ls] %s setup error: %s" % (name, traceback.format_exc())) def on_plugin_unload(self): for account, observer in getattr(self, "_observers", []): try: nc = NotificationCenter.getInstance(account) nc.removeObserver(observer, NotificationCenter.didReceiveNewMessages) opened_id = getattr(NotificationCenter, "openedChatChanged", None) if opened_id is not None: nc.removeObserver(observer, opened_id) except Exception: pass self._observers = [] self._menu_item_id = None self._muted_cache = None self._expiry_generation = getattr(self, "_expiry_generation", 0) + 1 try: if getattr(self, "_update_interface_unhook", None): self.unhook_method(self._update_interface_unhook) self._update_interface_unhook = None except Exception: pass # --------------------------------------------------------------- observers def _register_observers(self): max_accounts = 4 try: max_accounts = int(UserConfig.MAX_ACCOUNT_COUNT) except Exception: pass for account in range(max_accounts): try: observer = self._make_observer() nc = NotificationCenter.getInstance(account) nc.addObserver(observer, NotificationCenter.didReceiveNewMessages) opened_id = getattr(NotificationCenter, "openedChatChanged", None) if opened_id is not None: nc.addObserver(observer, opened_id) self._observers.append((account, observer)) except Exception: self.log("[mute_ls] observer %s error: %s" % (account, traceback.format_exc())) def _make_observer(self): plugin = self class Observer(NotificationCenterDelegate): def __init__(self): try: super().__init__() except Exception: pass def didReceivedNotification(self, id, account, args): try: plugin._on_notification(account, id, args) except Exception: log("[mute_ls] observer error: " + traceback.format_exc()) return Observer() def _on_notification(self, account, notif_id, args): if notif_id == NotificationCenter.didReceiveNewMessages: self._handle_new_messages(account, args) return opened_id = getattr(NotificationCenter, "openedChatChanged", None) if opened_id is not None and notif_id == opened_id: self._refresh_chat_menu() def _handle_new_messages(self, account, args): if not self.get_setting("enabled", True): return try: if args is None or len(args) < 2: return messages = args[1] if messages is None: return except Exception: return pending = {} try: count = messages.size() except Exception: return for i in range(count): try: message_object = messages.get(i) owner = getattr(message_object, "messageOwner", None) if owner is None: continue if bool(getattr(owner, "out", False)): continue dialog_id = int(getattr(owner, "dialog_id", 0) or 0) if dialog_id <= 0: continue if not self._is_muted(account, dialog_id): continue message_id = int(getattr(owner, "id", 0) or 0) if message_id > 0: pending.setdefault(dialog_id, []).append(message_id) except Exception: self.log("[mute_ls] scan error: " + traceback.format_exc()) for dialog_id, ids in pending.items(): self._delete_messages(account, dialog_id, ids) def _install_update_interface_hook(self): controller_class = find_class("org.telegram.messenger.MessagesController") array_list_class = find_class("java.util.ArrayList") if controller_class is None or array_list_class is None: return method = controller_class.getDeclaredMethod( "updateInterfaceWithMessages", Long.TYPE, array_list_class, Integer.TYPE ) method.setAccessible(True) self._update_interface_unhook = self.hook_method(method, _UpdateInterfaceHook(self)) def _scan_incoming(self, account, dialog_id, messages): try: if not self.get_setting("enabled", True): return if messages is None: return dialog_id = int(dialog_id or 0) if dialog_id <= 0: return account = int(account) if account is not None else self._selected_account() if not self._is_muted(account, dialog_id): return ids = [] for i in range(messages.size()): message_object = messages.get(i) owner = getattr(message_object, "messageOwner", None) if owner is None: continue if bool(getattr(owner, "out", False)): continue message_id = int(getattr(owner, "id", 0) or 0) if message_id > 0: ids.append(message_id) if ids: self._delete_messages(account, dialog_id, ids, delay=100) except Exception: self.log("[mute_ls] scan incoming error: " + traceback.format_exc()) def _delete_messages(self, account, dialog_id, ids, delay=0): for_all = bool(self.get_setting("delete_for_everyone", True)) def _do(): try: id_list = ArrayList() for message_id in ids: id_list.add(Integer(int(message_id))) controller = self._messages_controller(account) if controller is None: return attempts = [ (id_list, None, None, int(dialog_id), 0, for_all, 0), (id_list, None, None, int(dialog_id), for_all, 0), (id_list, None, None, int(dialog_id), for_all, False), ] for call_args in attempts: try: controller.deleteMessages(*call_args) return except Exception: continue self.log("[mute_ls] deleteMessages failed for dialog %s" % dialog_id) except Exception: self.log("[mute_ls] delete error: " + traceback.format_exc()) run_on_ui_thread(_do, delay) # -------------------------------------------------------------- outgoing cmd def on_send_message_hook(self, account: int, params: Any) -> HookResult: try: text = getattr(params, "message", None) if not isinstance(text, str) or not text.strip(): return HookResult() text = text.strip() prefix = self.get_setting("prefix", DEFAULT_PREFIX) or DEFAULT_PREFIX parts = text.split(None, 1) command = parts[0].lower() argument = parts[1].strip() if len(parts) > 1 else "" mute_cmd = (prefix + "mute").lower() unmute_cmd = (prefix + "unmute").lower() status_cmd = (prefix + "status").lower() if command not in (mute_cmd, unmute_cmd, status_cmd): return HookResult() dialog_id = self._peer_dialog_id(params) if command == mute_cmd: self._command_mute(account, dialog_id, argument) elif command == unmute_cmd: self._command_unmute(account, dialog_id) else: self._command_status(account, dialog_id) return HookResult(strategy=HookStrategy.CANCEL) except Exception: self.log("[mute_ls] send hook error: " + traceback.format_exc()) return HookResult() def on_update_hook(self, update_name: str, account: int, update: Any) -> HookResult: try: if update_name != "TL_updateNewMessage": return HookResult() if not self.get_setting("enabled", True): return HookResult() message = getattr(update, "message", None) if message is None: return HookResult() if bool(getattr(message, "out", False)): return HookResult() dialog_id = int(getattr(message, "dialog_id", 0) or 0) if dialog_id <= 0: return HookResult() if not self._is_muted(account, dialog_id): return HookResult() message_id = int(getattr(message, "id", 0) or 0) if message_id > 0: self._delete_messages(account, dialog_id, [message_id], delay=250) return HookResult(strategy=HookStrategy.CANCEL) except Exception: self.log("[mute_ls] update hook error: " + traceback.format_exc()) return HookResult() def _peer_dialog_id(self, params): try: return int(params.peer) except Exception: return None def _command_mute(self, account, dialog_id, argument): if dialog_id is None or dialog_id <= 0: self._bulletin("error", "Мут работает только в личных чатах.") return until = 0 duration_text = argument if not duration_text: duration_text = (self.get_setting("default_duration", DEFAULT_DURATION) or "").strip() if duration_text: seconds = self._parse_duration(duration_text) if not seconds: self._bulletin("error", "Неверный формат времени. Примеры: 30m, 2h, 1d.") return until = int(time.time() * 1000) + seconds * 1000 was_muted = self._is_muted(account, dialog_id) name = self._display_name(account, dialog_id) muted = self._load_muted() muted[self._key(account, dialog_id)] = { "name": name, "until": until, "account": account, "id": dialog_id, } self._persist(muted, reload_ui=True) self._refresh_chat_menu() if not was_muted: self._send_mute_notice(account, dialog_id) if until: self._bulletin("success", "Замутил %s до %s" % (name, self._fmt_until(until))) else: self._bulletin("success", "Замутил %s навсегда" % name) def _send_mute_notice(self, account, dialog_id): self._send_notice(account, dialog_id, "notify_text", "Ты в муте.") def _send_unmute_notice(self, account, dialog_id): self._send_notice(account, dialog_id, "notify_unmute_text", "Ты больше не в муте.") def _send_notice(self, account, dialog_id, setting_key, default_text): if not self.get_setting("notify_peer", False): return text = (self.get_setting(setting_key, default_text) or "").strip() if not text: return def _do(): try: params = SendMessagesHelper.SendMessageParams() params.peer = int(dialog_id) params.message = text SendMessagesHelper.getInstance(account).sendMessage(params) except Exception: self.log("[mute_ls] notice error: " + traceback.format_exc()) run_on_ui_thread(_do) def _command_unmute(self, account, dialog_id): if dialog_id is None or dialog_id <= 0: self._bulletin("error", "Размутить можно только в личном чате.") return muted = self._load_muted() key = self._key(account, dialog_id) entry = muted.pop(key, None) if entry is None: self._bulletin("info", "Этот пользователь не в муте.") return self._persist(muted, reload_ui=True) self._refresh_chat_menu() self._send_unmute_notice(account, dialog_id) name = entry.get("name") or ("id %s" % dialog_id) self._bulletin("success", "Размутил %s" % name) def _command_status(self, account, dialog_id): muted = self._remove_expired() if dialog_id is None or dialog_id <= 0: self._bulletin("info", "В муте: %d" % len(muted)) return entry = muted.get(self._key(account, dialog_id)) if entry is None: self._bulletin("info", "Этот пользователь не в муте.") return name = entry.get("name") or ("id %s" % dialog_id) until = int(entry.get("until", 0) or 0) if until: self._bulletin("info", "%s в муте до %s" % (name, self._fmt_until(until))) else: self._bulletin("info", "%s в муте навсегда" % name) # ------------------------------------------------------------- chat menu def _on_menu_click(self, context): try: dialog_id = context.get("dialog_id") if dialog_id is None: dialog_id = context.get("dialogId") if dialog_id is None: dialog_id = context.get("chatId") dialog_id = int(dialog_id) if dialog_id <= 0: self._bulletin("error", "Мут работает только в личных чатах.") return account = context.get("account") account = int(account) if account is not None else self._selected_account() if self._is_muted(account, dialog_id): self._command_unmute(account, dialog_id) else: self._command_mute(account, dialog_id, "") except Exception: self.log("[mute_ls] menu click error: " + traceback.format_exc()) def _refresh_chat_menu(self): def _do(): try: state = self._menu_state() if state == self._last_menu_state: return self._last_menu_state = state self._remove_menu_item() if state is None: return _account, _dialog_id, is_muted = state text = "Размутить" if is_muted else "Замутить" icon = "msg_unmute" if is_muted else "msg_mute" self._menu_item_id = self.add_menu_item( MenuItemData( menu_type=MenuItemType.CHAT_ACTION_MENU, text=text, icon=icon, item_id=MENU_ITEM_ID, on_click=self._on_menu_click, ) ) except Exception: self.log("[mute_ls] menu refresh error: " + traceback.format_exc()) run_on_ui_thread(_do) def _menu_state(self): try: fragment = get_last_fragment() if fragment is None or not hasattr(fragment, "getDialogId"): return None dialog_id = int(fragment.getDialogId()) except Exception: return None if dialog_id <= 0: return None account = self._selected_account() return (account, dialog_id, self._is_muted(account, dialog_id)) def _remove_menu_item(self): for candidate in (self._menu_item_id, MENU_ITEM_ID): if not candidate: continue try: self.remove_menu_item(candidate) break except Exception: continue self._menu_item_id = None # --------------------------------------------------------------- settings def create_settings(self) -> List[Any]: items = [ Header(text="Мут в ЛС"), Switch( key="enabled", text="Включить мут", subtext="Удалять входящие сообщения от замьюченных.", default=True, icon="msg_mute", ), Switch( key="delete_for_everyone", text="Удалять у всех", subtext="Удалять сообщение и у собеседника.", default=True, icon="msg_delete", ), Header(text="Уведомление собеседнику"), Switch( key="notify_peer", text="Уведомить собеседника", subtext="Отправлять уведомления при муте и индивидуальном размьюте.", default=False, icon="msg_info", ), Input( key="notify_text", text="Текст при муте", default="Ты в муте.", subtext="Отправляется один раз и не повторяется при продлении мута.", icon="msg_edit", ), Input( key="notify_unmute_text", text="Текст при размьюте", default="Ты больше не в муте.", subtext="Отправляется при размьюте одного собеседника.", icon="msg_edit", ), Divider(), Input( key="default_duration", text="Длительность по умолчанию", default=DEFAULT_DURATION, subtext="Для команды мута без времени. Пример: 30m, 2h, 1d. Пусто — навсегда.", icon="msg_mute_period", ), Input( key="prefix", text="Префикс команд", default=DEFAULT_PREFIX, subtext="Например: %smute, %sunmute, %sstatus" % (DEFAULT_PREFIX, DEFAULT_PREFIX, DEFAULT_PREFIX), icon="msg_edit", ), Text( text="Команды", subtext="Показать список команд", icon="msg_help", create_sub_fragment=self._create_commands_page, ), Divider(), Header(text="Замьюченные"), ] muted = self._remove_expired() if not muted: items.append( Text( text="Никого нет в муте", subtext="Напишите %smute в личном чате, чтобы добавить." % DEFAULT_PREFIX, icon="msg_info", ) ) else: for key, entry in sorted(muted.items(), key=lambda kv: kv[1].get("name") or ""): name = entry.get("name") or ("id %s" % entry.get("id", "?")) until = entry.get("until", 0) subtext = ("до %s" % self._fmt_until(until)) if until else "навсегда" items.append( Text( text=name, subtext=subtext + " • нажмите, чтобы размьютить", icon="msg_mute", on_click=lambda view, k=key: self._confirm_unmute(k), ) ) items.append(Divider()) items.append( Text( text="Размьютить всех", subtext="Убрать всех из мут-листа.", icon="msg_unmute", red=True, on_click=lambda view: self._confirm_unmute_all(), ) ) return items def _create_commands_page(self) -> List[Any]: prefix = self.get_setting("prefix", DEFAULT_PREFIX) or DEFAULT_PREFIX return [ Header(text="Команды"), Text(text="%smute" % prefix, subtext="Замутить на время по умолчанию", icon="msg_mute"), Text(text="%smute 30m" % prefix, subtext="Замутить на 30 минут (m — минуты, h — часы, d — дни, w — недели)", icon="msg_mute_period"), Text(text="%sunmute" % prefix, subtext="Размьютить", icon="msg_unmute"), Text(text="%sstatus" % prefix, subtext="Показать статус мута в этом чате", icon="msg_info"), Divider(), Text( text="Команды пишутся в личном чате и не отправляются собеседнику.", icon="msg_info", ), ] def _confirm_unmute(self, key): def _do(): try: entry = self._load_muted().get(key, {}) name = entry.get("name") or key fragment = get_last_fragment() activity = fragment.getParentActivity() if fragment else None if activity is None: self._remove_key(key) return builder = AlertDialogBuilder(activity) builder.set_title("Размутить") builder.set_message("Размутить %s?" % name) builder.set_positive_button( "Размутить", lambda b, w: (self._remove_key(key), b.dismiss()) ) builder.set_negative_button("Отмена", lambda b, w: b.dismiss()) builder.show() except Exception: self.log("[mute_ls] unmute dialog error: " + traceback.format_exc()) run_on_ui_thread(_do) def _confirm_unmute_all(self): def _do(): try: muted = self._load_muted() if not muted: return fragment = get_last_fragment() activity = fragment.getParentActivity() if fragment else None if activity is None: self._persist({}, reload_ui=True) self._refresh_chat_menu() return builder = AlertDialogBuilder(activity) builder.set_title("Размьютить всех") builder.set_message("Убрать всех из мут-листа (%d)?" % len(muted)) builder.set_positive_button("Размьютить всех", lambda b, w: (self._clear_all(), b.dismiss())) builder.set_negative_button("Отмена", lambda b, w: b.dismiss()) builder.show() except Exception: self.log("[mute_ls] unmute all error: " + traceback.format_exc()) run_on_ui_thread(_do) def _remove_key(self, key): muted = self._load_muted() entry = muted.pop(key, None) if entry is None: return self._persist(muted, reload_ui=True) self._refresh_chat_menu() name = entry.get("name") or entry.get("id") self._bulletin("success", "Размутил %s" % name) def _clear_all(self): self._persist({}, reload_ui=True) self._refresh_chat_menu() self._bulletin("success", "Все размьючены") # ------------------------------------------------------------------ storage def _load_muted(self): if self._muted_cache is not None: return dict(self._muted_cache) data = {} try: raw = self.get_setting(MUTED_KEY, "{}") or "{}" parsed = json.loads(raw) if isinstance(parsed, dict): data = parsed except Exception: data = {} self._muted_cache = dict(data) return dict(data) def _persist(self, muted, reload_ui=False): self._muted_cache = dict(muted) try: payload = json.dumps(muted, ensure_ascii=False) except Exception: payload = "{}" try: self.set_setting(MUTED_KEY, payload, reload_settings=reload_ui) except TypeError: self.set_setting(MUTED_KEY, payload) if reload_ui: self.set_setting(MUTED_KEY, self.get_setting(MUTED_KEY, "{}") or "{}", reload_settings=True) self._schedule_expiry_cleanup() def _remove_expired(self, muted=None, reload_ui=False): muted = dict(muted) if muted is not None else self._load_muted() now = int(time.time() * 1000) active = {} changed = False for key, entry in muted.items(): if not isinstance(entry, dict): changed = True continue try: until = int(entry.get("until", 0) or 0) except Exception: until = 0 if until and until <= now: changed = True continue active[key] = entry if changed: self._persist(active, reload_ui=reload_ui) return active def _schedule_expiry_cleanup(self): if not hasattr(self, "_expiry_generation"): return now = int(time.time() * 1000) deadlines = [] for entry in self._load_muted().values(): if not isinstance(entry, dict): continue try: until = int(entry.get("until", 0) or 0) except Exception: continue if until > now: deadlines.append(until) self._expiry_generation += 1 generation = self._expiry_generation if not deadlines: return # Android delayed callbacks use millisecond delays. Cap very long # durations and reschedule when the callback eventually runs. delay = min(max(1, min(deadlines) - now), 2147000000) run_on_ui_thread(lambda: self._expire_due(generation), delay) def _expire_due(self, generation): if generation != getattr(self, "_expiry_generation", 0): return self._remove_expired(reload_ui=True) self._refresh_chat_menu() self._schedule_expiry_cleanup() def _key(self, account, dialog_id): return "%s:%s" % (account, dialog_id) def _is_muted(self, account, dialog_id): muted = self._load_muted() key = self._key(account, dialog_id) entry = muted.get(key) if not entry: return False try: until = int(entry.get("until", 0) or 0) except Exception: until = 0 if until and until <= int(time.time() * 1000): muted.pop(key, None) self._persist(muted) return False return True # -------------------------------------------------------------------- utils def _parse_duration(self, text): text = str(text).strip().lower() if not text: return None total = 0 found = False for number, unit in DURATION_RE.findall(text): total += int(number) * UNITS[unit.lower()] found = True leftover = re.sub(r"[0-9\s]", "", text) leftover = re.sub(r"[smhdw]", "", leftover) if leftover: return None if not found: if text.isdigit(): return int(text) * 60 return None return total or None def _fmt_until(self, until_ms): try: return time.strftime("%d.%m %H:%M", time.localtime(until_ms / 1000.0)) except Exception: return str(until_ms) def _selected_account(self): try: return int(UserConfig.selectedAccount) except Exception: return 0 def _messages_controller(self, account): try: return get_messages_controller(account) except TypeError: try: return get_messages_controller() except Exception: return None except Exception: return None def _display_name(self, account, user_id): try: controller = self._messages_controller(account) if controller is not None: user = controller.getUser(int(user_id)) if user is not None: parts = [] first = getattr(user, "first_name", None) last = getattr(user, "last_name", None) if first: parts.append(first) if last: parts.append(last) if parts: return " ".join(parts) username = getattr(user, "username", None) if username: return "@" + username except Exception: pass return "id %s" % user_id def _bulletin(self, kind, text): try: fragment = get_last_fragment() if kind == "error": BulletinHelper.show_error(text, fragment) elif kind == "success": BulletinHelper.show_success(text, fragment) else: BulletinHelper.show_info(text, fragment) except Exception: self.log("[mute_ls] bulletin error: %s | %s" % (text, traceback.format_exc()))