CactusUtils

Класс CactusUtils предоставляет набор вспомогательных методов для различных задач, включая генерацию динамических прокси, операции с файловой системой, сжатие и кодирование данных, манипуляции со строками, логирование и взаимодействие со специфическими функциями Android.


Методы класса

gen(java_class, method_name, return_value: bool = False)

Этот метод генерирует новый прокси-класс, который расширяет данный java_class и переопределяет определенный метод.

  • java_class: Java-класс, для которого создается прокси.

  • method_name: Имя метода, который будет переопределен в прокси-классе.

  • return_value (bool, optional): Если True, переопределенный метод будет возвращать значение из оригинального вызова метода. По умолчанию False.

Пример использования:

from org.telegram.messenger import Utilities

# Функция для переопределения
def function(arg1, arg2, test):
    ...

# Это создает прокси, который переопределяет 'run'
MyProxyClass = CactusUtils.gen(Utilities.Callback2, "run")

# Создание экземпляра прокси
proxy_instance = MyProxyClass(function, test="value")

# Можете дальше использовать этот класс
...

gen2(java_class, return_value: bool = False, **methods)

Этот метод похож на gen, но позволяет переопределять несколько методов в сгенерированном прокси-классе.

  • java_class: Java-класс, для которого создается прокси.

  • return_value (bool, optional): Если True, переопределенные методы будут возвращать свои соответствующие значения. По умолчанию False.

  • **methods: Именованные аргументы, где ключ - это имя метода (строка), а значение - это вызываемый объект Python, который заменит исходную реализацию метода.

Пример использования:

from com.example import AnotherJavaClass

# Предположим, что AnotherJavaClass имеет методы 'methodA' и 'methodB'б которые нам нужно переопределить
MyMultiProxyClass = CactusUtils.gen2(
    AnotherJavaClass,
    return_value=True,
    methodA=lambda *args: print(f"Метод A вызван с: {args}"),
    methodB=lambda *args, **kwargs: print(f"Метод B вызван с: {args}, {kwargs}")
)

proxy_instance = MyMultiProxyClass("аргумент1", test="аргумент2")

Классы Callback2 и Callback5

Эти классы являются удобными обертками для Utilities.Callback2 и Utilities.Callback5, позволяя легко определять вызываемые объекты Python в качестве их методов run.

Пример использования:


def my_callback_function(*args):
    print(f"Коллбэк выполнен с: {args}")

# Использование Callback2
callback2_instance = CactusUtils.Callback2(my_callback_function, "дополнительный_аргумент")
# В контексте Java, где ожидается Utilities.Callback2:
# java_object.setCallback(callback2_instance)
callback2_instance.run("данные_события")

# Использование Callback5
callback5_instance = CactusUtils.Callback5(lambda: print("Еще один коллбэк!"))
# В контексте Java, где ожидается Utilities.Callback5:
# java_object.setAnotherCallback(callback5_instance)

# Также вы можете создать свой такой класс
from org.telegram.messenger import Utilities

callback3_instance = CactusUtils.gen(Utilities.Callback3, "run")(my_callback_function, "дополнительный_аргумент")

Класс FileSystem (вложенный в CactusUtils)

Класс FileSystem предоставляет статические методы для взаимодействия с файловой системой на устройстве Android.


FileSystem.basedir(*path: str)

Возвращает базовый каталог приложения. Если указаны аргументы path, он строит подкаталоги внутри базового каталога и гарантирует их существование.

  • *path (str): Необязательные компоненты пути для добавления к базовому каталогу.

Пример использования:

# Получить базовый каталог
base_dir = CactusUtils.FileSystem.basedir()
print(f"Базовый каталог: {base_dir.getAbsolutePath()}")

# Получить и создать подкаталог
my_data_dir = CactusUtils.FileSystem.basedir("my_app_data", "configs")
print(f"Каталог моих данных: {my_data_dir.getAbsolutePath()}")
# Это создаст 'my_app_data' и 'configs', если они не существуют.

FileSystem.cachedir(*path: str)

Возвращает внешний кэш-каталог приложения. Подобно basedir, он может создавать подкаталоги внутри кэш-каталога.

  • *path (str): Необязательные компоненты пути для добавления к кэш-каталогу.

Пример использования:

# Получить кэш-каталог
cache_dir = CactusUtils.FileSystem.cachedir()
print(f"Кэш-каталог: {cache_dir.getAbsolutePath()}")

# Получить и создать временный подкаталог кэша
temp_cache_dir = CactusUtils.FileSystem.cachedir("temp_images")
print(f"Временный каталог изображений: {temp_cache_dir.getAbsolutePath()}")

FileSystem.tempdir()

Возвращает специальный временный каталог внутри кэш-каталога (cactuslib_temp_files). Этот каталог создается, если он не существует.

Пример использования:

temp_dir = CactusUtils.FileSystem.tempdir()
print(f"Временный каталог CactusLib: {temp_dir.getAbsolutePath()}")

FileSystem.get_file_content(file_path, mode: str = "rb")

Считывает содержимое файла.

  • file_path: Путь к файлу.

  • mode (str, optional): Режим открытия файла. По умолчанию "rb" (чтение бинарных данных).

Пример использования:

# Предположим, что 'my_file.txt' существует в базовом каталоге
file_path = CactusUtils.FileSystem.basedir("my_file.txt").getAbsolutePath()
# Сначала запишем некоторое содержимое в файл для демонстрации
CactusUtils.FileSystem.write_file(file_path, "Привет, мир!", mode="w")

content_bytes = CactusUtils.FileSystem.get_file_content(file_path)
print(f"Содержимое (байты): {content_bytes}")
content_str = CactusUtils.FileSystem.get_file_content(file_path, mode="r")
print(f"Содержимое (строка): {content_str}")

FileSystem.get_temp_file_content(filename: str, mode: str = "rb", delete_after: int = 0)

Считывает содержимое файла, расположенного во временном каталоге. При необходимости удаляет файл после указанной задержки.

  • filename (str): Имя файла во временном каталоге.

  • mode (str, optional): Режим открытия файла. По умолчанию "rb".

  • delete_after (int, optional): Количество секунд, по истечении которых файл будет удален. Если 0, файл не удаляется автоматически. По умолчанию 0.

Пример использования:

temp_file_name = "test_temp.txt"
temp_file_path = CactusUtils.FileSystem.write_temp_file(temp_file_name, "Временные данные!", mode="w")
print(f"Путь к временному файлу: {temp_file_path}")

# Считать содержимое без удаления
content = CactusUtils.FileSystem.get_temp_file_content(temp_file_name, mode="r")
print(f"Содержимое из временного файла: {content}")

# Считать содержимое и удалить через 5 секунд
# CactusUtils.FileSystem.write_temp_file("temp_to_delete.txt", "Это будет удалено!", mode="w")
# content_to_delete = CactusUtils.FileSystem.get_temp_file_content("temp_to_delete.txt", mode="r", delete_after=5)
# print(f"Содержимое из временного файла для удаления: {content_to_delete}")

FileSystem.write_file(file_path, content, mode: str = "wb")

Записывает содержимое в указанный файл.

  • file_path: Путь к файлу.

  • content: Содержимое для записи (байты или строка).

  • mode (str, optional): Режим открытия файла. По умолчанию "wb" (запись бинарных данных).

Пример использования:

output_file = CactusUtils.FileSystem.basedir("output.txt").getAbsolutePath()
CactusUtils.FileSystem.write_file(output_file, "Это некоторый текст.", mode="w")
print(f"Содержимое записано в: {output_file}")

binary_data = b"\x01\x02\x03\x04"
binary_file = CactusUtils.FileSystem.cachedir("binary_data.bin").getAbsolutePath()
CactusUtils.FileSystem.write_file(binary_file, binary_data)
print(f"Бинарные данные записаны в: {binary_file}")

FileSystem.write_temp_file(filename: str, content, mode="wb", delete_after: int = 0)

Записывает содержимое в файл во временном каталоге. При необходимости удаляет файл после указанной задержки.

  • filename (str): Имя файла во временном каталоге.

  • content: Содержимое для записи.

  • mode (str, optional): Режим открытия файла. По умолчанию "wb".

  • delete_after (int, optional): Количество секунд, по истечении которых файл будет удален. Если 0, файл не удаляется автоматически. По умолчанию 0.

Пример использования:

temp_report_name = "report.csv"
temp_report_content = "Имя,Возраст\nИван,30\nМария,25"
path_to_report = CactusUtils.FileSystem.write_temp_file(temp_report_name, temp_report_content, mode="w")
print(f"Отчет записан во временный файл: {path_to_report}")

# Записать временное изображение, которое будет удалено через 10 секунд
# CactusUtils.FileSystem.write_temp_file("image.jpg", b"фиктивные_данные_изображения", delete_after=10)

FileSystem.delete_file_after(file_path, seconds: int = 0)

Удаляет файл после указанной задержки. Если seconds равно 0, файл удаляется немедленно.

  • file_path: Путь к файлу для удаления.

  • seconds (int, optional): Задержка в секундах перед удалением файла. По умолчанию 0.

Пример использования:

file_to_delete = CactusUtils.FileSystem.basedir("old_log.txt").getAbsolutePath()
CactusUtils.FileSystem.write_file(file_to_delete, "Этот лог будет удален.")

# Удалить немедленно
CactusUtils.FileSystem.delete_file_after(file_to_delete)
print(f"Файл удален немедленно: {file_to_delete}")

# Создать еще один файл и запланировать его удаление
file_to_delete_later = CactusUtils.FileSystem.basedir("temp_doc.txt").getAbsolutePath()
CactusUtils.FileSystem.write_file(file_to_delete_later, "Этот документ будет удален через 5 секунд.")
# CactusUtils.FileSystem.delete_file_after(file_to_delete_later, 5)
# print(f"Файл запланирован к удалению через 5 секунд: {file_to_delete_later}")

Другие методы класса


compress_and_encode(data: Union[bytes, str], level: int = 7) -> str

Сжимает данные с помощью zlib, а затем кодирует их с помощью base64.

  • data (bytes или str): Данные для сжатия и кодирования.

  • level (int, optional): Уровень сжатия (0-9). По умолчанию 7.

Пример использования:

original_text = "Это пример текста, который будет сжат и закодирован."
encoded_text = CactusUtils.compress_and_encode(original_text)
print(f"Исходная длина: {len(original_text)}")
print(f"Сжатая и закодированная длина: {len(encoded_text)}")
print(f"Закодированные данные: {encoded_text[:50]}...") # Показать часть

decode_and_decompress(encoded_data: Union[bytes, str])

Декодирует данные, закодированные в base64, а затем распаковывает их с помощью zlib.

  • encoded_data (bytes или str): Данные, закодированные в base64 и сжатые.

Пример использования:

original_text = "Еще один фрагмент текста для демонстрации декодирования и декомпрессии."
encoded_text = CactusUtils.compress_and_encode(original_text)
decoded_bytes = CactusUtils.decode_and_decompress(encoded_text)
decoded_text = decoded_bytes.decode('utf-8')
print(f"Декодированный и декомпрессированный текст: {decoded_text}")

pluralization_string(number: int, words: List[str])

Возвращает строку во множественном числе на основе заданного числа и списка форм слова (единственное, двойственное, множественное число). Этот метод разработан для правил русского языка.

  • number (int): Число для определения формы множественного числа.

  • words (list[str]): Список слов, представляющих формы единственного, двойственного и множественного числа.

Пример использования:

print(CactusUtils.pluralization_string(1, ["жизнь", "жизни", "жизней"]))   # Вывод: 1 жизнь
print(CactusUtils.pluralization_string(2, ["жизнь", "жизни", "жизней"]))   # Вывод: 2 жизни
print(CactusUtils.pluralization_string(5, ["жизнь", "жизни", "жизней"]))   # Вывод: 5 жизней
print(CactusUtils.pluralization_string(21, ["рубль", "рубля", "рублей"])) # Вывод: 21 рубль
print(CactusUtils.pluralization_string(22, ["рубль", "рубля", "рублей"])) # Вывод: 22 рубля
print(CactusUtils.pluralization_string(105, ["апельсин", "апельсина", "апельсинов"])) # Вывод: 105 апельсинов

escape_html(text: str)

Экранирует специальные HTML-символы (&, <, >) в строке.

  • text (str): Строка для экранирования.

Пример использования:

html_string = "Это <b>жирный</b> & очень важный текст!"
escaped_string = CactusUtils.escape_html(html_string)
print(f"Оригинал: {html_string}")
print(f"Экранированный: {escaped_string}")
# Вывод: Экранированный: Это &lt;b&gt;жирный&lt;/b&gt; &amp; очень важный текст!

copy_to_clipboard(text: str)

Копирует данный текст в буфер обмена Android и показывает уведомление «Скопировано в буфер обмена».

  • text (str): Текст для копирования.

Пример использования:

# Эта функция взаимодействует с Android-специфическими API.
# Она будет работать только в среде Android, где доступны AndroidUtilities и BulletinHelper.
# CactusUtils.copy_to_clipboard("Привет из CactusUtils!")

log(message: str, level: str = "INFO", __id__: Optional[str] = __id__)

Записывает сообщение в logcat с указанным уровнем и необязательным идентификатором. Символы новой строки заменяются на <CNL>.

  • message (str): Сообщение для записи в лог.

  • level (str, optional): Уровень логирования (например, «DEBUG», «INFO», «WARN», «ERROR» или пользовательский). По умолчанию "INFO".

  • __id__ (str, optional): Идентификатор записи в логе, часто используется для фильтрации.

Пример использования:

CactusUtils.log("Это информационное сообщение.", level="INFO", __id__="МоеПриложение")
CactusUtils.log("Что-то пошло не так!", level="ERROR", __id__="СетеваяСлужба")
CactusUtils.log("Подробная отладочная информация здесь.\nС несколькими строками.", level="DEBUG")

debug(message: str, __id__: Optional[str] = __id__)

Записывает отладочное сообщение в logcat. Это сокращение для CactusUtils.log с level="DEBUG".

  • message (str): Сообщение для записи в лог.

  • __id__ (str, optional): Идентификатор записи в логе.

Пример использования:

CactusUtils.debug("Отладка переменной X: 123", __id__="ОбработчикДанных")

error(message: str, __id__: Optional[str] = __id__)

Записывает сообщение об ошибке в logcat. Это сокращение для CactusUtils.log с level="ERROR".

  • message (str): Сообщение для записи в лог.

  • __id__ (str, optional): Идентификатор записи в логе.

Пример использования:

try:
    1 / 0
except ZeroDivisionError:
    CactusUtils.error("Попытка деления на ноль!", __id__="calculator")

info(message: str, __id__: Optional[str] = __id__)

Записывает информационное сообщение в logcat. Это сокращение для CactusUtils.log с level="INFO".

  • message (str): Сообщение для записи в лог.

  • __id__ (str, optional): Идентификатор записи в логе.

Пример использования:

CactusUtils.info("Приложение успешно запущено.", __id__="ЖизненныйЦиклПриложения")

warn(message: str, __id__: Optional[str] = __id__)

Записывает предупреждающее сообщение в logcat. Это сокращение для CactusUtils.log с level="WARN".

  • message (str): Сообщение для записи в лог.

  • __id__ (str, optional): Идентификатор записи в логе.

Пример использования:

CactusUtils.warn("Файл конфигурации не найден, используются значения по умолчанию.", __id__="ЗагрузчикКонфига")

runtime_exec(cmd: List[str], return_list_lines: bool = False, raise_errors: bool = True) -> Union[List[str], str]

Выполняет команду с помощью Runtime.getRuntime().exec() (эквивалент выполнения команд оболочки в Android/Java).

  • cmd (List[str]): Список строк, представляющих команду и ее аргументы.

  • return_list_lines (bool, optional): Если True, возвращает вывод в виде списка строк. В противном случае возвращает одну строку, где строки соединены символами новой строки. По умолчанию False.

  • raise_errors (bool, optional): Если True, исключения во время выполнения будут повторно возбуждаться. По умолчанию True.

Пример использования:

# Получить основную системную информацию (пример для среды Android)
# result_list = CactusUtils.runtime_exec(["getprop", "ro.build.version.release"], return_list_lines=True)
# print(f"Версия Android: {result_list[0]}")

# result_string = CactusUtils.runtime_exec(["ls", "-la", "/data/data"], return_list_lines=False)
# print(f"Частичный листинг /data/data:\n{result_string[:200]}...")

get_logs(__id__: Optional[str] = None, times: Optional[int] = None, lvl: Optional[str] = None, as_list: bool = False)

Извлекает сообщения logcat, опционально фильтруя их по ID, времени и уровню логирования.

  • __id__ (Optional[str]): ID плагина/компонента для фильтрации логов.

  • times (Optional[int]): Время в секундах, с которого нужно получить логи (например, times=60 получает логи за последние 60 секунд).

  • lvl (Optional[str]): Уровень логирования для фильтрации (например, «INFO», «ERROR»).

  • as_list (bool, optional): Если True, возвращает логи в виде списка строк. В противном случае возвращает одну строку. По умолчанию False.

Пример использования:

# Получить все логи за последние 5 минут как одну строку
# all_recent_logs = CactusUtils.get_logs(times=300)
# print(f"Недавние логи (первые 500 символов):\n{all_recent_logs[:500]}...")

# Получить логи ошибок для конкретного ID за последний час в виде списка
# my_error_logs = CactusUtils.get_logs(__id__="my_plugin", times=3600, lvl="ERROR", as_list=True)
# if my_error_logs:
#     print(f"Логи ошибок МоегоПлагина:\n{'\n'.join(my_error_logs)}")
# else:
#     print("Логи ошибок для МоегоПлагина не найдены.")

Классы Uri и MessageUri (вложенные в CactusUtils)

Это классы данных для построения строковых URI для межплагинного взаимодействия или специфических схем обмена сообщениями.


Uri

Представляет общий URI для команд, связанных с Cactus.

  • plugin_id (str): ID плагина.

  • command (str): Выполняемая команда.

  • kwargs (Dict[str, str]): Именованные аргументы, которые будут URL-кодированы и включены в URI.

Метод класса: create(cls, plugin, cmd: str, **kwargs)

Удобный метод для создания экземпляра Uri.

  • plugin: Объект с атрибутом id (например, экземпляр плагина).

  • cmd (str): Команда.

  • **kwargs: Дополнительные именованные аргументы.

Метод: string()

Возвращает полную строку URI в формате tg://cactus/{plugin_id}/{command}?{url_encoded_kwargs}.

Пример использования:

from urllib.parse import unquote_plus # Для демонстрации декодирования

class MockPlugin:
    def __init__(self, _id):
        self.id = _id

    def func(self):
        # Создать Uri с использованием метода класса
        uri_instance = CactusUtils.Uri.create(self, "open_settings", theme="dark", version="1.0")
        print(f"Сгенерированный URI: {uri_instance}")

        # Создать экземпляр Uri вручную
        another_uri = CactusUtils.Uri(
            plugin_id=self.id,
            command="open_profile",
            kwargs={"user_id": "12345", "tab": "posts"}
        )
        print(f"Другая строка URI: {another_uri.string()}")

MessageUri

Подкласс Uri специально для URI, связанных с сообщениями. Он использует tg://cactusX/ в качестве своей схемы.

Возвращает полную строку URI в формате tg://cactusX/{plugin_id}/{command}?{url_encoded_kwargs}.

Пример использования:

# Создать MessageUri
message_uri = CactusUtils.MessageUri.create(self, "send_message", chat_id=98765, text="Привет мир!")
print(f"Сгенерированный URI сообщения: {message_uri}")