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}")
# Вывод: Экранированный: Это <b>жирный</b> & очень важный текст!
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}")