# meta developer: @limokanews
# requires: whoosh cryptography
from collections import Counter, defaultdict
import shutil
from whoosh.index import create_in, open_dir
from whoosh.fields import Schema, TEXT, ID
from whoosh.qparser import QueryParser, OrGroup
from whoosh.query import FuzzyTerm, Wildcard
from whoosh.writing import LockError
import aiohttp
import random
import logging
import os
import html
import json
import re
import asyncio
from typing import Iterable, Union, List, Dict, Any, Optional
import hashlib
from telethon.types import Message
from telethon.errors.rpcerrorlist import WebpageMediaEmptyError
from telethon import TelegramClient
from telethon.errors.rpcerrorlist import YouBlockedUserError
from telethon import functions
try:
from aiogram.utils.exceptions import BadRequest
except ImportError:
from aiogram.exceptions import TelegramBadRequest as BadRequest
from .. import utils, loader
from ..types import InlineCall
logger = logging.getLogger("Limoka")
__version__ = (1, 0, 0)
WEIGHTS = {
"inline.token_obtainment": 15,
"main": 10,
"inline": 7,
"translations": 5,
"security": 3,
}
DEFAULT_WEIGHT = 1
BASE_DIR = os.getcwd()
def _get_lang_value(data: Dict[str, Any], lang: str) -> str:
if not isinstance(data, dict):
return str(data) if data else ""
return data.get(lang, data.get("default", data.get("en", "")))
class Search:
def __init__(self, query, ix):
self.schema = Schema(
title=TEXT(stored=True), path=ID(stored=True), content=TEXT(stored=True)
)
self.query = query
self.ix = ix
def search_module(self):
with self.ix.searcher() as searcher:
parser = QueryParser("content", self.ix.schema, group=OrGroup.factory(0.8))
query = parser.parse(self.query)
wildcard_query = Wildcard("content", f"*{self.query}*")
fuzzy_query = FuzzyTerm("content", self.query, maxdist=2, prefixlength=1)
for search_query in [query, wildcard_query, fuzzy_query]:
results = searcher.search(search_query)
if results:
return list(set(result["path"] for result in results))
return []
class LimokaAPI:
async def fetch_json(self, base_url, path):
url = f"{base_url}{path}"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return json.loads(await response.text())
@loader.tds
class LimokaLegacy(loader.Module):
"""
Modules are now in one place with easy searching!
For Hikka and FTG Userbots. This module has outdated functionality and is kept for legacy reasons only.
Read https://t.me/limokanews/133 for more information.
"""
strings = {
"name": "Limoka Legacy",
"wait": (
"Just wait\n"
"🔍 A search is underway among {count} modules "
"for the query: {query}\n"
"{fact}"
),
"found_header": (
"🔍 Found module {name} "
"by query: {query}\n\n"
"ℹ️ Description: {description}\n"
"🧑💻 Developer: {username}\n\n"
"🏷 Tags: {tags}\n\n"
),
"found_body": ("{commands}"),
"found_footer": "",
"caption_short": (
"🔍 {safe_name}\n"
"ℹ️ Description: {safe_desc}\n"
"🧑💻 Dev: {dev_username}"
),
"command_template": "{emoji} {prefix}{command} — {description}\n",
"inline_handler_template": "{inline_bot} {command} — {description}\n",
"emojis": {
1: "1️⃣",
2: "2️⃣",
3: "3️⃣",
4: "4️⃣",
5: "5️⃣",
6: "6️⃣",
7: "7️⃣",
8: "8️⃣",
9: "9️⃣",
},
"404": "❌ Not found by query: {query}",
"noargs": "❌ No args",
"?": "🔎 Request too short / not found",
"no_info": "No information",
"facts": [
"🛡 The limoka catalog is carefully moderated!",
"🚀 Limoka performance allows you to search for modules quickly!",
],
"history": (
"🔎 Your search history:\n"
"{history}"
),
"empty_history": "🔎 Your search history is empty!",
"enter_query": "🔍 Enter new search query:",
"global_search": "🔍 Global search for {query} — found {count} modules",
"change_query": "🔍 Change query",
"back": "🔙 Back",
"global_button": "🌍 Results",
"first_page": "This is the first page!",
"last_page": "This is the last page!",
"display_error": "Error displaying module. Please try again.",
"error_occurred": "An error occurred. Please try again.",
"start_search_form": "🔍 Limoka Search\nEnter your query to search for modules:",
"history_cleared": "🧹 Search history cleared!",
"invalid_history_arg": "❌ Invalid argument for history command. Use:\n.lshistory - show history\n.lshistory clear - clear history",
"close": "❌ Close",
"indexing_in_progress": (
"⚠️ Database is busy, "
"try again later. "
"If issue persists, try "
"removing limoka_index in the userbot's root folder. "
"If error persists again, report to developers"
),
"install_btn": "🛠 Install",
"source_btn": "📦 Source",
"installed": "✅ Installed successfully!",
"install_failed": "❌ Installation failed!",
"tags": {
"newbie": "Newbie",
"herokutrusted": "Heroku Trusted",
"hikkatrusted": "Hikka Trusted",
"nonactive": "Non-active repository",
"nonlongermaintained": "Abandoned repository",
}
}
strings_ru = {
"name": "Limoka",
"wait": (
"Подождите\n"
"🔍 Идёт поиск среди {count} модулей по запросу: {query}\n"
"{fact}"
),
"found_header": (
"🔍 Найден модуль {name} "
"по запросу: {query}\n\n"
"ℹ️ Описание: {description}\n"
"🧑💻 Разработчик: {username}\n\n"
"🏷 Теги: {tags}\n\n"
),
"found_body": ("{commands}"),
"found_footer": "",
"caption_short": (
"🔍 {safe_name}\n"
"ℹ️ Описание: {safe_desc}\n"
"🧑💻 Разработчик: {dev_username}"
),
"command_template": "{emoji} {prefix}{command} — {description}\n",
"inline_handler_template": "{inline_bot} {command} — {description}\n",
"emojis": {
1: "1️⃣",
2: "2️⃣",
3: "3️⃣",
4: "4️⃣",
5: "5️⃣",
6: "6️⃣",
7: "7️⃣",
8: "8️⃣",
9: "9️⃣",
},
"404": "❌ Не найдено по запросу: {query}",
"noargs": "❌ Нет аргументов",
"?": "🔎 Запрос слишком короткий / не найден",
"no_info": "Нет информации",
"facts": [
"🛡 Каталог Limoka тщательно модерируется!",
"🚀 Limoka позволяет искать модули с невероятной скоростью!",
],
"history": (
"🔎 История поиска:\n"
"{history}"
),
"empty_history": "🔎 История поиска пуста!",
"enter_query": "🔍 Введите новый поисковый запрос:",
"global_search": "🔍 Глобальный поиск по {query} — найдено {count} модулей",
"change_query": "🔍 Изменить запрос",
"back": "🔙 Назад",
"global_button": "🌍 Результаты",
"first_page": "Это первая страница!",
"last_page": "Это последняя страница!",
"display_error": "Ошибка отображения модуля. Пожалуйста, попробуйте еще раз.",
"error_occurred": "Произошла ошибка. Пожалуйста, попробуйте еще раз.",
"start_search_form": "🔍 Limoka Поиск\nВведите ваш запрос для поиска модулей:",
"history_cleared": "🧹 История поиска очищена!",
"invalid_history_arg": "❌ Неверный аргумент для команды истории. Используйте:\n.lshistory - показать историю\n.lshistory clear - очистить историю",
"close": "❌ Закрыть",
"indexing_in_progress": (
"⚠️ База данных занята, "
"попробуйте снова через несколько секунд. "
"Если ошибка сохраняется, попробуйте "
"удалить limoka_index в корневой папке юзербота. "
"Если ошибка сохраняется снова, сообщите разработчикам"
),
"install_btn": "🛠 Установить",
"source_btn": "📦 Исходный код",
"installed": "✅ Установлено успешно!",
"install_failed": "❌ Установка не удалась!",
"tags": {
"newbie": "Новичок",
"herokutrusted": "Heroku Trusted",
"hikkatrusted": "Hikka Trusted",
"nonactive": "Неактивный репозиторий",
"nonlongermaintained": "Заброшенный репозиторий",
}
}
def __init__(self):
self.api = LimokaAPI()
self.config = loader.ModuleConfig(
loader.ConfigValue(
"limokaurl",
"https://raw.githubusercontent.com/MuRuLOSE/limoka/refs/heads/main/",
lambda: "Mirror (doesn't work): https://raw.githubusercontent.com/MuRuLOSE/limoka-mirror/refs/heads/main/",
validator=loader.validators.String(),
),
loader.ConfigValue(
"external_install_allowed",
True,
lambda: "If enabled, module installation can be handled via external Limoka bot (@limoka_bbot) for better reliability.",
validator=loader.validators.Boolean(),
),
loader.ConfigValue(
"filter_newbies_modules",
False,
lambda: "If enabled, modules from developers with newbies tag will be not shown.",
validator=loader.validators.Boolean(),
),
)
self.name = self.strings["name"]
self._invalid_banners = set()
self._bot_username = "limoka_bbot"
self._base_url = self.config["limokaurl"]
BANNERS = {
"global_search": "https://github.com/MuRuLOSE/hikka-assets/blob/main/Limoka%20-%20Global%20Search.png?raw=true",
"not_found": "https://github.com/MuRuLOSE/hikka-assets/blob/main/Limoka%20-%20Not%20Found.png?raw=true",
"no_banner": "https://github.com/MuRuLOSE/hikka-assets/blob/main/Limoka%20-%20No%20banner.png?raw=true",
}
def _filter_newbies(self, modules: Dict[str, Any]) -> Dict[str, Any]:
try:
if not self.config.get("filter_newbies_modules"):
return modules
except Exception:
return modules
if not getattr(self, "repositories", None):
return modules
filtered: Dict[str, Any] = {}
for path, info in modules.items():
repo_key = "/".join(path.split("/")[:2]) if "/" in path else path
repo = self.repositories.get(repo_key)
tags = repo.get("tags", []) if repo else []
if "newbie" in tags:
continue
filtered[path] = info
return filtered
def _create_search_session(
self,
query: str = "",
results: Optional[List[str]] = None,
current_index: int = 0,
) -> Dict[str, Any]:
return {
"query": query,
"results": results or [],
"current_index": current_index,
}
async def client_ready(self, client, db):
self.client: TelegramClient = client
self.db = db
self.api = LimokaAPI()
self.schema = Schema(
title=TEXT(stored=True), path=ID(stored=True), content=TEXT(stored=True)
)
os.makedirs("limoka_search", exist_ok=True)
if not os.path.exists("limoka_search/index"):
self.ix = create_in("limoka_search", self.schema)
else:
self.ix = open_dir("limoka_search")
self._history = self.pointer("history", [])
self.modules = (await self.api.fetch_json(self._base_url, "modules.json")).get(
"modules", {}
)
raw = (await self.api.fetch_json(self._base_url, "repositories.json")).get(
"repositories", []
)
self.repositories = {repo["url"]: repo for repo in raw}
try:
self.modules = self._filter_newbies(self.modules)
except Exception:
pass
self._service_bot_id = (await self.client.get_entity(self._bot_username)).id
loop = asyncio.get_running_loop()
self.ix_task = loop.run_in_executor(
None, lambda: asyncio.run(self._update_index())
)
if self.config["external_install_allowed"]:
try:
message = await self.client.get_messages(self._bot_username, limit=1)
if not message:
message = await self.client.send_message(
self._bot_username, "/start"
)
await message.delete()
await self.client(
functions.messages.DeleteHistoryRequest(
peer=self._bot_username,
max_id=0,
just_clear=True,
revoke=True,
)
)
except YouBlockedUserError:
logger.warning(
f"Please unblock {self._bot_username} to enable external installation feature. Or disable external_install_allowed in Limoka settings to get rid of this message."
)
self._userbot_bot_username = (await self.inline.bot.get_me()).username
@loader.loop(interval=3600)
async def _update_modules_loop(self):
self.modules = await self.api.fetch_json(self._base_url, "modules.json")
try:
self.modules = self._filter_newbies(self.modules)
except Exception:
pass
await self._update_index()
async def _update_index(self):
try:
writer = self.ix.writer()
modules_to_index = self._filter_newbies(self.modules)
for module_path, module_data in modules_to_index.items():
writer.add_document(
title=module_data["name"],
path=module_path,
content=module_data["name"]
+ " "
+ (
module_data.get("description")
or ""
+ " "
+ (
(module_data.get("meta").get("developer") or "")
if module_data.get("meta")
else ""
)
),
)
for func in module_data.get("commands", []):
for command, description in func.items():
writer.add_document(
title=module_data["name"],
path=module_path,
content=f"{command} {description}",
)
writer.commit()
except LockError:
folder = os.path.join(BASE_DIR, "limoka_search")
if os.path.commonpath([folder, BASE_DIR]) == BASE_DIR and os.path.exists(
folder
):
shutil.rmtree(folder)
await self._update_index()
else:
logger.error(
(
f"Skipping unsafe rmtree for {folder}. Please, report this to developer. ",
f"Debug info: folder={folder}, base_dir={BASE_DIR}, common_path={os.path.commonpath([folder, BASE_DIR])}, exists={os.path.exists(folder)}",
)
)
async def _validate_url(self, url: str) -> Optional[str]:
if not url or url in self._invalid_banners:
return None
try:
async with aiohttp.ClientSession() as session:
async with session.head(
url, timeout=5, allow_redirects=True
) as response:
if response.status != 200:
self._invalid_banners.add(url)
return None
ct = response.headers.get("Content-Type", "").lower()
if not ct.startswith("image/"):
self._invalid_banners.add(url)
return None
return url
except Exception as e:
if url:
self._invalid_banners.add(url)
return None
def find_userbot(self, keys: Iterable[str]) -> str | None:
scores = defaultdict(int)
for key in keys:
parts = key.split(".")
for i in range(1, len(parts)):
prefix = ".".join(parts[:i])
suffix = ".".join(parts[i:])
weight = WEIGHTS.get(suffix, DEFAULT_WEIGHT)
scores[prefix] += weight
if not scores:
return None
return max(scores, key=scores.get)
@property
def user_lang(self) -> str:
def warn(msg: str):
logger.warning(
f"{msg} Defaulting language to English. "
"If this is unexpected, please report to the module developer."
)
lang = self.db.get(f"{__package__.split('.')[0]}.translations", "lang")
if lang:
return lang
logger.warning(
"Cannot determine language from module translations. "
"Trying fallback method."
)
userbot = self.find_userbot(self.db.keys())
if not userbot:
warn("Cannot determine userbot type.")
return "en"
lang = self.db.get(f"{userbot}.translations", "lang")
if not lang:
warn("Cannot determine language from userbot translations.")
return "en"
return lang
def _get_description(self, desc_map: dict, lang: str) -> str:
desc = _get_lang_value(desc_map, lang) or self.strings["no_info"]
return html.escape(desc)
def _get_emoji(self, index: int) -> str:
emojis = self.strings["emojis"]
if index < 10:
return emojis.get(index, "")
return "".join(emojis.get(int(d), "") for d in str(index))
def generate_commands(self, module_info, lang: str = "en"):
commands = []
for i, cmd in enumerate(module_info.get("new_commands", []), 1):
commands.append(
self.strings["command_template"].format(
prefix=self.get_prefix(),
command=html.escape(cmd.get("name", "")),
emoji=self._get_emoji(i),
description=self._get_description(cmd.get("description", {}), lang),
)
)
for handler in module_info.get("inline_handlers", []):
commands.append(
self.strings["inline_handler_template"].format(
inline_bot=self._userbot_bot_username,
command=html.escape(handler.get("name", "")),
description=self._get_description(
handler.get("description", {}), lang
),
)
)
return commands
def _format_module_content(
self,
module_info: Dict[str, Any],
query: str,
module_path: Optional[str] = None,
lang: str = "en",
) -> tuple:
name = html.escape(module_info.get("name") or self.strings["no_info"])
cls_doc = module_info.get("cls_doc", {})
description = html.escape(
_get_lang_value(cls_doc, lang)
or _get_lang_value(module_info.get("description", ""), lang)
or self.strings["no_info"]
)
dev_username = html.escape(module_info["meta"].get("developer") or "Unknown")
raw_path = (
module_path if module_path is not None else module_info.get("path", "")
)
clean_module_path = (raw_path or "").replace("\\", "/")
commands = self.generate_commands(module_info, lang)
if len(description) > 300:
description = description[:297] + "…"
repo_key = (
"/".join(module_path.split("/")[:2]) if "/" in module_path else module_path
)
tags_list = []
for x in self.repositories:
if x.replace("https://github.com/", "") == repo_key:
tags_list = self.repositories.get(x, {}).get("tags", [])
break
logger.info(tags_list)
tags_text = ", ".join(self.strings["tags"].get(tag, tag) for tag in tags_list)
header = self.strings["found_header"].format(
query=html.escape(query),
name=name,
description=description,
username=dev_username,
tags=tags_text,
)
commands_text = "".join(commands)
if len(commands_text) <= 500:
body_pages = [commands_text] if commands_text else [""]
else:
body_pages = []
current_page = []
current_length = 0
for cmd in commands:
if current_length + len(cmd) > 500:
if current_page:
body_pages.append("".join(current_page))
current_page = []
current_length = 0
current_page.append(cmd)
current_length += len(cmd)
if current_page:
body_pages.append("".join(current_page))
if not body_pages:
body_pages = [""]
footer = self.strings["found_footer"]
return header, body_pages, footer
def _build_module_markup(
self,
session: Dict[str, Any],
body_pages: List[str],
page_body: int,
module_path: str,
) -> list:
result = session["results"]
index = session["current_index"]
source_url = self._get_source_url(module_path)
markup = [
[
{
"text": self.strings["install_btn"],
"callback": self._install_module,
"args": (module_path, session),
},
{
"text": self.strings["source_btn"],
"url": source_url,
},
]
]
if len(body_pages) > 1:
markup.append(
[
{
"text": "◀️" if page_body > 0 else "🚫",
"callback": (
self._previous_body_page
if page_body > 0
else self._inline_void
),
"args": (
(session, module_path, page_body) if page_body > 0 else ()
),
},
{
"text": f"Body {page_body + 1}/{len(body_pages)}",
"callback": self._inline_void,
},
{
"text": "▶️" if page_body + 1 < len(body_pages) else "🚫",
"callback": (
self._next_body_page
if page_body + 1 < len(body_pages)
else self._inline_void
),
"args": (
(session, module_path, page_body)
if page_body + 1 < len(body_pages)
else ()
),
},
]
)
page = index + 1
markup.append(
[
{
"text": "⏪" if index > 0 else "🚫",
"callback": self._previous_page if index > 0 else self._inline_void,
"args": (session,) if index > 0 else (),
},
{"text": f"{page}/{len(result)}", "callback": self._inline_void},
{
"text": "⏩" if index + 1 < len(result) else "🚫",
"callback": (
self._next_page
if index + 1 < len(result)
else self._inline_void
),
"args": (session,) if index + 1 < len(result) else (),
},
]
)
markup.append(
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": self._enter_query,
},
]
)
markup.append(
[
{
"text": self.strings["global_button"],
"callback": self._show_global_results,
"args": (session,),
},
]
)
markup.append(
[{"text": self.strings.get("close", "❌ Close"), "action": "close"}]
)
return markup
async def _safe_display(
self,
message_or_call: Union[Message, InlineCall],
text: str,
markup: list,
photo: Optional[Any] = None,
):
try:
if message_or_call is None:
logger.error("message_or_call is None in _safe_display")
return
if isinstance(message_or_call, Message):
if photo is not None:
await self.inline.form(
text=text,
message=message_or_call,
reply_markup=markup,
photo=photo,
)
else:
await self.inline.form(
text=text, message=message_or_call, reply_markup=markup
)
else:
if photo is not None:
await message_or_call.edit(
text=text, reply_markup=markup, photo=photo
)
else:
await message_or_call.edit(text=text, reply_markup=markup)
except (BadRequest, WebpageMediaEmptyError) as e:
logger.exception(f"Error in _safe_display: {e}")
if isinstance(message_or_call, Message):
await utils.answer(message_or_call, self.strings["display_error"])
elif hasattr(message_or_call, "edit"):
await message_or_call.edit(self.strings["display_error"])
async def _display_module(
self,
message_or_call: Union[Message, InlineCall],
module_info: Dict[str, Any],
module_path: str,
session: Dict[str, Any],
page_body: int = 0,
):
try:
query = session["query"]
lang = self.user_lang
module_banner_raw = module_info.get("meta", {}).get("banner")
photo = await self._validate_url(
module_banner_raw
) or await self._validate_url(
"https://raw.githubusercontent.com/MuRuLOSE/hikka-assets/refs/heads/main/Limoka%20-%20No%20banner.png"
)
header, body_pages, footer = self._format_module_content(
module_info,
query,
module_path=module_path,
lang=lang,
)
current_body = body_pages[min(page_body, len(body_pages) - 1)]
full_message = header + current_body + footer
markup = self._build_module_markup(
session, body_pages, page_body, module_path
)
await self._safe_display(message_or_call, full_message, markup, photo)
except Exception as e:
logger.exception(f"Error in _display_module: {e}")
if isinstance(message_or_call, Message):
await utils.answer(message_or_call, self.strings["error_occurred"])
elif hasattr(message_or_call, "edit"):
await message_or_call.edit(self.strings["error_occurred"])
async def _previous_body_page(
self,
call: InlineCall,
session: Dict[str, Any],
module_path: str,
page_body: int,
):
module_info = self.modules[module_path]
new_page_body = max(page_body - 1, 0)
await self._display_module(
call, module_info, module_path, session, page_body=new_page_body
)
async def _next_body_page(
self,
call: InlineCall,
session: Dict[str, Any],
module_path: str,
page_body: int,
):
module_info = self.modules[module_path]
query = session["query"]
header, body_pages, footer = self._format_module_content(
module_info,
query,
module_path=module_path,
lang=self.user_lang,
)
new_page_body = min(page_body + 1, len(body_pages) - 1)
await self._display_module(
call, module_info, module_path, session, page_body=new_page_body
)
async def _install_module(
self, call: InlineCall, module_path: str, session: Dict[str, Any]
):
module_url = self.config["limokaurl"] + module_path
loader_mod = self.lookup("loader")
if not loader_mod:
await call.answer(
self.strings.get("watcher_loader_missing", "Loader not found")
)
return
status = await loader_mod.download_and_install(module_url, None)
if getattr(loader_mod, "fully_loaded", False):
loader_mod.update_modules_in_db()
if status:
await call.answer(self.strings["installed"])
else:
await call.answer(self.strings["install_failed"])
def _get_source_url(self, module_path: str) -> str:
repo_key = (
"/".join(module_path.split("/")[:2]) if "/" in module_path else module_path
)
repo_url = "https://github.com/" + repo_key
repo = self.repositories.get(repo_url, {})
branch = repo.get("branch", "main")
path_in_repo = (
"/".join(module_path.split("/")[2:])
if len(module_path.split("/")) > 2
else module_path
)
return f"{repo_url}/blob/{branch}/{path_in_repo}"
async def _show_results(self, call: InlineCall, session: Dict[str, Any]):
query = session["query"]
searcher = Search(query.lower(), self.ix)
try:
result = searcher.search_module()
except Exception:
await call.edit(self.strings["?"], reply_markup=[])
return
if not result:
markup = [
[{"text": self.strings.get("close", "❌ Close"), "action": "close"}]
]
photo = await self._validate_url(
"https://raw.githubusercontent.com/MuRuLOSE/hikka-assets/main/Limoka%20-%20Not%20Found.png"
)
await self._safe_display(
call, self.strings["404"].format(query=query), markup, photo
)
return
module_path = result[0]
module_info = self.modules[module_path]
display_session = self._create_search_session(
query=query,
results=result,
current_index=0,
)
await self._display_module(call, module_info, module_path, display_session, 0)
async def _enter_query_handler(
self, call_or_query, query: Optional[str] = None, *args, **kwargs
):
call = None
if query is None and isinstance(call_or_query, str):
query = call_or_query
for a in args:
if hasattr(a, "edit") or isinstance(a, Message):
call = a
break
else:
call = call_or_query
if call is None:
logger.error("_enter_query_handler: missing call/context")
return
if not query:
await call.edit(
self.strings["?"],
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": self._enter_query,
}
]
],
)
return
if len(query) <= 1:
await call.edit(
self.strings["?"],
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": self._enter_query,
}
]
],
)
return
searcher = Search(query.lower(), self.ix)
try:
result = searcher.search_module()
except Exception:
await call.edit(
self.strings["?"],
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": self._enter_query,
}
]
],
)
return
if not result:
await call.edit(
self.strings["404"].format(query=query),
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": self._enter_query,
}
],
[
{
"text": self.strings.get("close", "❌ Close"),
"action": "close",
}
],
],
)
return
module_path = result[0]
module_info = self.modules[module_path]
# Create session for displaying module
display_session = self._create_search_session(
query=query,
results=result,
current_index=0,
)
await self._display_module(call, module_info, module_path, display_session, 0)
async def _enter_query(self, call: InlineCall, query: Optional[str] = None):
markup = [
[
{
"text": "✍️ " + self.strings["enter_query"],
"input": self.strings["enter_query"],
"handler": self._enter_query_handler,
}
],
[
{
"text": self.strings["back_to_results"],
"callback": self._show_results,
"args": (
self._create_search_session(
query=query or "",
),
),
}
],
[
{
"text": self.strings.get("close", "❌ Close"),
"action": "close",
}
],
]
await call.edit(self.strings["enter_query"], reply_markup=markup)
async def _show_global_results(self, call: InlineCall, session: Dict[str, Any]):
query = session["query"]
searcher = Search(query.lower(), self.ix)
try:
result = searcher.search_module()
except Exception:
await call.edit(self.strings["?"], reply_markup=[])
return
if not result:
await call.edit(
self.strings["404"].format(query=query),
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": self._enter_query,
}
]
],
)
return
text = self.strings["global_search"].format(
query=html.escape(query), count=len(result)
)
buttons = []
for i, path in enumerate(result[:15]):
info = self.modules.get(path)
if not info:
continue
name = info.get("name", "Unknown")
global_session = self._create_search_session(
query=query,
results=result,
current_index=i,
)
buttons.append(
[
{
"text": f"{i+1}. {name}",
"callback": self._display_module_from_global,
"args": (path, global_session),
}
]
)
buttons.append(
[{"text": self.strings["change_query"], "callback": self._enter_query}]
)
await call.edit(text=text[:4096], reply_markup=buttons)
async def _display_module_from_global(
self, call: InlineCall, module_path: str, session: Dict[str, Any]
):
module_info = self.modules[module_path]
await self._display_module(call, module_info, module_path, session, 0)
async def _next_page(self, call: InlineCall, session: Dict[str, Any]):
result = session["results"]
index = session["current_index"]
if index + 1 >= len(result):
await call.answer(self.strings["last_page"])
return
index += 1
module_path = result[index]
module_info = self.modules[module_path]
new_session = session.copy()
new_session["current_index"] = index
await self._display_module(call, module_info, module_path, new_session, 0)
async def _previous_page(self, call: InlineCall, session: Dict[str, Any]):
result = session["results"]
index = session["current_index"]
if index - 1 < 0:
await call.answer(self.strings["first_page"])
return
index -= 1
module_path = result[index]
module_info = self.modules[module_path]
new_session = session.copy()
new_session["current_index"] = index
await self._display_module(call, module_info, module_path, new_session, 0)
async def _inline_void(self, call: InlineCall):
await call.answer()
@loader.command(ru_doc="[запрос / ничего] — Поиск модулей")
async def limokacmd(self, message: Message):
"""[query / nothing] - Search modules"""
args = utils.get_args_raw(message)
lock_path = os.path.join(BASE_DIR, "limoka_search", "index.lock")
if os.path.exists(lock_path):
await utils.answer(message, self.strings["indexing_in_progress"])
return
if not args:
markup = [
[
{
"text": "✍️ " + self.strings["enter_query"],
"input": self.strings["enter_query"],
"handler": self._enter_query_handler,
}
],
[
{
"text": self.strings["global_button"],
"callback": self._show_global_form,
"args": (message,),
}
],
]
markup.append(
[{"text": self.strings.get("close", "❌ Close"), "action": "close"}]
)
await self.inline.form(
text=self.strings["start_search_form"],
message=message,
reply_markup=markup,
# photo=self._get_banner_for_state("global_search"),
)
return
history = self.get("history", [])
if len(history) >= 10:
history = history[-9:]
history.append(args)
self.set("history", history)
await utils.answer(
message,
self.strings["wait"].format(
count=len(self.modules),
fact=random.choice(self.strings["facts"]),
query=args,
),
)
searcher = Search(args.lower(), self.ix)
try:
result = searcher.search_module()
except Exception:
return await utils.answer(message, self.strings["?"])
if not result:
return await utils.answer(message, self.strings["404"].format(query=args))
module_path = result[0]
module_info = self.modules[module_path]
# Create session for displaying module
display_session = self._create_search_session(
query=args,
results=result,
current_index=0,
)
await self._display_module(
message, module_info, module_path, display_session, 0
)
async def _show_global_form(self, call: InlineCall, message: Message):
markup = [
[
{
"text": "✍️ " + self.strings["enter_query"],
"input": self.strings["enter_query"],
"handler": self._global_search_handler,
"args": (message,),
}
],
[
{
"text": "🔙 " + self.strings["back"],
"callback": self._inline_void,
}
],
[
{
"text": self.strings.get("close", "❌ Close"),
"action": "close",
}
],
]
await call.edit(self.strings["global_search_form"], reply_markup=markup)
async def _global_search_handler(
self, call: InlineCall, query: str, message: Message, *args, **kwargs
):
global_session = self._create_search_session(
query=query,
results=[],
current_index=0,
) # idk what is that crap but it works lol
if len(query) <= 1:
await call.edit(
self.strings["?"],
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": lambda c: self._show_global_form(c, message),
}
],
[
{
"text": self.strings.get("close", "❌ Close"),
"action": "close",
}
],
],
)
return
searcher = Search(query.lower(), self.ix)
try:
result = searcher.search_module()
except Exception:
await call.edit(
self.strings["?"],
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": lambda c: self._show_global_form(c, message),
}
],
[
{
"text": self.strings.get("close", "❌ Close"),
"action": "close",
}
],
],
)
return
if not result:
await call.edit(
self.strings["404"].format(query=query),
reply_markup=[
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": lambda c: self._show_global_form(c, message),
}
],
[
{
"text": self.strings.get("close", "❌ Close"),
"action": "close",
}
],
],
)
return
text = self.strings["global_search"].format(
query=html.escape(query), count=len(result)
)
buttons = []
for i, path in enumerate(result[:15]):
info = self.modules.get(path)
if not info:
continue
name = info.get("name", "Unknown")
buttons.append(
[
{
"text": f"{i+1}. {name}",
"callback": self._display_module_from_global,
"args": (path, global_session),
}
]
)
buttons.append(
[
{
"text": "🔄 " + self.strings["change_query"],
"callback": lambda c: self._show_global_form(c, message),
}
]
)
buttons.append(
[{"text": self.strings.get("close", "❌ Close"), "action": "close"}]
)
await call.edit(text=text[:4096], reply_markup=buttons)
@loader.command(ru_doc="[clear] — Показать или очистить историю поиска")
async def lshistorycmd(self, message: Message):
"""[clear] - Show or clear search history"""
args = utils.get_args_raw(message).strip().lower()
if args == "clear":
self.set("history", [])
await utils.answer(message, self.strings["history_cleared"])
return
if args:
await utils.answer(message, self.strings["invalid_history_arg"])
return
history = self.get("history", [])
if not history:
await utils.answer(message, self.strings["empty_history"])
return
formatted_history = [
f"{i+1}. {utils.escape_html(h)}"
for i, h in enumerate(history[-10:])
]
await utils.answer(
message,
self.strings["history"].format(history="\n".join(formatted_history)),
)
@loader.watcher(from_dl=False)
async def secure_install_watcher(self, message: Message):
if not message.text:
return
if not hasattr(message, "from_id") or not message.from_id:
return
sender_id = None
if hasattr(message.from_id, "user_id"):
sender_id = message.from_id.user_id
elif hasattr(message.from_id, "channel_id"):
sender_id = message.from_id.channel_id
if sender_id != self._service_bot_id:
# logger.debug("Message not from official bot, ignoring")
return
if not self.config["external_install_allowed"]:
return
try:
clean_text = (
getattr(message, "raw_text", None)
or getattr(message, "message", None)
or message.text
or ""
)
if message.entities:
from html import unescape
clean_text = unescape(clean_text)
clean_text = re.sub(r"<[^>]+>", "", clean_text)
match = re.search(r"#limoka:([^\s\"'<>]+)", clean_text)
if not match:
logger.debug(
"No #limoka tag found in cleaned text; leaving original message intact"
)
return
tag_content = match.group(1)
parts = tag_content.split(":", 1)
if len(parts) != 2:
logger.error("Invalid tag format after cleaning")
await utils.answer(message, self.strings["watcher_invalid_format"])
return
module_path, signature_hex = parts
module_path = re.sub(r"[<>\"']", "", module_path).strip()
if module_path.startswith("href="):
module_path = module_path[5:].strip('"').strip("'")
if module_path not in self.modules:
found = False
for db_path in self.modules.keys():
if module_path in db_path or db_path in module_path:
module_path = db_path
found = True
break
if not found:
logger.warning(f"Module not found after cleanup: {module_path}")
await utils.answer(
message,
self.strings["watcher_module_not_found"].format(
path=html.escape(module_path)
),
)
return
try:
import base64
from cryptography.hazmat.primitives.asymmetric import ed25519
PUB_KEY_B64 = (
"MCowBQYDK2VwAyEA1ltSnqtf3pGBuctuAYqHivCXsaRtKOVxavai7yin7ZE="
)
der_bytes = base64.b64decode(PUB_KEY_B64)
raw_pubkey = der_bytes[-32:]
module_url = self.config["limokaurl"] + module_path
async with aiohttp.ClientSession() as session:
async with session.get(module_url, timeout=10) as resp:
if resp.status != 200:
logger.error(
f"Failed to fetch module for verification: {module_url} (HTTP {resp.status})"
)
await utils.answer(
message, self.strings["watcher_loader_missing"]
)
return
module_bytes = await resp.read()
sha256 = hashlib.sha256(module_bytes).hexdigest()
public_key = ed25519.Ed25519PublicKey.from_public_bytes(
raw_pubkey
)
signature = bytes.fromhex(signature_hex)
signed_payload = f"{module_path}|{sha256}".encode()
public_key.verify(signature, signed_payload)
except Exception as e:
logger.error(f"Signature verification failed for {module_path}: {e}")
await utils.answer(message, self.strings["watcher_signature_invalid"])
return
loader_mod = self.lookup("loader")
if not loader_mod:
logger.error("Loader module not found")
await utils.answer(message, self.strings["watcher_loader_missing"])
return
module_url = self.config["limokaurl"] + module_path
status = await loader_mod.download_and_install(module_url, None)
if getattr(loader_mod, "fully_loaded", False):
loader_mod.update_modules_in_db()
try:
await message.delete()
except Exception as e:
logger.error(f"Failed to delete message: {e}")
if status:
try:
bot_peer = await self.client.get_entity(self._service_bot_id)
await self.client.send_message(
bot_peer, f"#limoka:sucsess:{message.id}"
)
except Exception as e:
logger.error(f"Failed to send success confirmation: {e}")
else:
logger.error(f"Installation failed with status: {status}")
try:
bot_peer = await self.client.get_entity(self._service_bot_id)
await self.client.send_message(
bot_peer, f"#limoka:failed:{message.id}"
)
except Exception as e:
logger.error(f"Failed to send failure notification: {e}")
except Exception as e:
logger.exception(f"CRITICAL ERROR in secure_install_watcher: {e}")
try:
await utils.answer(
message, self.strings["watcher_critical"].format(error=str(e)[:100])
)
await asyncio.sleep(5)
await message.delete()
except Exception:
pass