Files
limoka/fiksofficial/python-modules/deviceinfo.py
2025-11-21 01:04:46 +00:00

503 lines
26 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# ______ ___ ___ _ _
# ____ | ___ \ | \/ | | | | |
# / __ \| |_/ / _| . . | ___ __| |_ _| | ___
# / / _` | __/ | | | |\/| |/ _ \ / _` | | | | |/ _ \
# | | (_| | | | |_| | | | | (_) | (_| | |_| | | __/
# \ \__,_\_| \__, \_| |_/\___/ \__,_|\__,_|_|\___|
# \____/ __/ |
# |___/
# На модуль распространяется лицензия "GNU General Public License v3.0"
# https://github.com/all-licenses/GNU-General-Public-License-v3.0
# meta developer: @pymodule
# requires: aiohttp cachetools
import asyncio
import logging
from typing import List, Dict, Any
import aiohttp
from cachetools import TTLCache
from .. import loader, utils
from ..inline.types import InlineMessage
logger = logging.getLogger(__name__)
@loader.tds
class DeviceInfo(loader.Module):
"""A module for obtaining information about smartphones"""
strings_ru = {
"name": "DeviceInfo",
"_cls_doc": "Модуль для получения информации о смартфонах",
"searching": "🔍 Ищу устройства по запросу: <b>{}</b>...",
"no_query": "❌ Укажи название устройства! Пример: <code>.di iPhone 15</code>",
"no_results": "📭 Устройства не найдены для запросу: <b>{}</b>",
"device_list": "📱 Найдено {} устройств по запросу <b>{}</b>:",
"device_info": "📱 <b>{}</b>\n\n{}",
"error": "❌ Ошибка: {}. Попробуй позже или проверь API.",
"network": "📡 <b>Сеть</b>: {}\n",
"launched": "📅 <b>Дата выпуска</b>:\n Анонс: {}\n Статус: {}\n",
"body": "📏 <b>Корпус</b>:\n Размеры: {}\n Вес: {}\n SIM: {}\n Прочее: {}\n",
"display": "🖥️ <b>Дисплей</b>:\n Тип: {}\n Размер: {}\n Разрешение: {}\n Защита: {}\n",
"platform": "⚙️ <b>Платформа</b>:\n ОС: {}\n Чипсет: {}\n CPU: {}\n GPU: {}\n",
"memory": "💾 <b>Память</b>:\n Карта памяти: {}\n Внутренняя: {}\n Прочее: {}\n",
"main_camera": "📷 <b>Основная камера</b>:\n Модули: {}\n Функции: {}\n Видео: {}\n",
"selfie_camera": "🤳 <b>Фронтальная камера</b>:\n Модули: {}\n Функции: {}\n Видео: {}\n",
"sound": "🔊 <b>Звук</b>:\n Динамик: {}\n Аудиоразъём: {}\n Прочее: {}\n",
"comms": "🌐 <b>Связь</b>:\n Wi-Fi: {}\n Bluetooth: {}\n GPS: {}\n NFC: {}\n Инфракрасный порт: {}\n Радио: {}\n USB: {}\n",
"sensors": "🛠️ <b>Датчики</b>: {}\n",
"battery": "🔋 <b>Батарея</b>:\n Тип: {}\n Зарядка: {}\n",
"misc": "🎨 <b>Разное</b>:\n Цвета: {}\n Модели: {}\n",
"show_body": "📏 Корпус",
"show_memory": "💾 Память",
"show_cameras": "📷 Камеры",
"show_sound": "🔊 Звук",
"show_comms": "🌐 Связь",
"show_sensors": "🛠️ Датчики",
"show_misc": "🎨 Разное",
"next_photo": "▶️ След. фото",
"prev_photo": "◀️ Пред. фото",
"back": "🔙 Назад",
"back_to_device": "🔙 К устройству",
"config_saved": "✅ Конфигурация сохранена!",
"retrying": "🔄 Повторяю запрос... (попытка {}/{} )"
}
strings = {
"name": "DeviceInfo",
"searching": "🔍 Searching devices for: <b>{}</b>...",
"no_query": "❌ Specify a device name! Example: <code>.di iPhone 15</code>",
"no_results": "📭 No devices found for query: <b>{}</b>",
"device_list": "📱 Found {} devices for query <b>{}</b>:",
"device_info": "📱 <b>{}</b>\n\n{}",
"error": "❌ Error: {}. Try again later or check the API.",
"network": "📡 <b>Network</b>: {}\n",
"launched": "📅 <b>Launch</b>:\n Announced: {}\n Status: {}\n",
"body": "📏 <b>Body</b>:\n Dimensions: {}\n Weight: {}\n SIM: {}\n Other: {}\n",
"display": "🖥️ <b>Display</b>:\n Type: {}\n Size: {}\n Resolution: {}\n Protection: {}\n",
"platform": "⚙️ <b>Platform</b>:\n OS: {}\n Chipset: {}\n CPU: {}\n GPU: {}\n",
"memory": "💾 <b>Memory</b>:\n Card slot: {}\n Internal: {}\n Other: {}\n",
"main_camera": "📷 <b>Main Camera</b>:\n Modules: {}\n Features: {}\n Video: {}\n",
"selfie_camera": "🤳 <b>Selfie Camera</b>:\n Modules: {}\n Features: {}\n Video: {}\n",
"sound": "🔊 <b>Sound</b>:\n Loudspeaker: {}\n Audio Jack: {}\n Other: {}\n",
"comms": "🌐 <b>Comms</b>:\n Wi-Fi: {}\n Bluetooth: {}\n GPS: {}\n NFC: {}\n Infrared: {}\n Radio: {}\n USB: {}\n",
"sensors": "🛠️ <b>Sensors</b>: {}\n",
"battery": "🔋 <b>Battery</b>:\n Type: {}\n Charging: {}\n",
"misc": "🎨 <b>Misc</b>:\n Colors: {}\n Models: {}\n",
"show_body": "📏 Body",
"show_memory": "💾 Memory",
"show_cameras": "📷 Cameras",
"show_sound": "🔊 Sound",
"show_comms": "🌐 Comms",
"show_sensors": "🛠️ Sensors",
"show_misc": "🎨 Misc",
"next_photo": "▶️ Next Photo",
"prev_photo": "◀️ Prev Photo",
"back": "🔙 Back",
"back_to_device": "🔙 To Device",
"config_saved": "✅ Configuration saved!",
"retrying": "🔄 Retrying request... (attempt {}/{})"
}
def __init__(self):
self.config = loader.ModuleConfig(
loader.ConfigValue(
"api_base_url",
"https://mobilespecs.fiksofficial.fun",
lambda: "API Url",
validator=loader.validators.String()
),
loader.ConfigValue(
"max_results",
20,
lambda: "Maximum search results to display",
validator=loader.validators.Integer(minimum=1, maximum=50)
),
loader.ConfigValue(
"timeout",
10,
lambda: "Timeout for API requests (seconds)",
validator=loader.validators.Integer(minimum=5, maximum=30)
),
loader.ConfigValue(
"retry_attempts",
3,
lambda: "Number of retry attempts for API requests",
validator=loader.validators.Integer(minimum=1, maximum=5)
)
)
self.cache = TTLCache(maxsize=100, ttl=300)
self.session: aiohttp.ClientSession = None
async def client_ready(self, client, db):
"""Initialize aiohttp session on client ready"""
self.session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.config["timeout"])
)
logger.info("DeviceInfo: aiohttp session initialized")
self.client = client
async def on_unload(self):
"""Close aiohttp session on module unload"""
if self.session:
await self.session.close()
logger.info("DeviceInfo: aiohttp session closed")
async def _resolve_entity(self, call: InlineMessage, message_id: int, chat_id: int = None):
"""Resolve Telegram entity to Message or int"""
if hasattr(call, "message") and call.message:
logger.debug("DeviceInfo: Using call.message")
return call.message
if chat_id:
logger.warning(f"DeviceInfo: call.message is None, falling back to chat_id {chat_id}")
return chat_id
logger.warning(f"DeviceInfo: call.message and chat_id are None, falling back to message_id {message_id}")
return message_id
async def _fetch_json(self, endpoint: str, params: Dict[str, Any] = None) -> Any:
"""Fetch JSON from API with retry and caching"""
cache_key = f"{endpoint}_{params}" if params else endpoint
if cache_key in self.cache:
logger.debug(f"Cache hit for {cache_key}")
return self.cache[cache_key]
url = f"{self.config['api_base_url']}/gsm/{endpoint}"
params_clean = {k: v for k, v in (params or {}).items() if k != "message"}
for attempt in range(1, self.config["retry_attempts"] + 1):
try:
async with self.session.get(url, params=params_clean or None) as resp:
if resp.status != 200:
error_text = await resp.text()
logger.error(f"DeviceInfo: HTTP {resp.status} for {url}: {error_text}")
raise aiohttp.ClientError(f"HTTP {resp.status}: {error_text}")
content_type = resp.headers.get("Content-Type", "")
if "application/json" not in content_type:
error_text = await resp.text()
logger.error(f"DeviceInfo: Invalid content-type {content_type} for {url}: {error_text}")
raise ValueError(f"Invalid content-type: {content_type}")
data = await resp.json()
if data is None:
error_text = await resp.text()
logger.error(f"DeviceInfo: API returned None for {url}: {error_text}")
if endpoint.startswith("search"):
data = []
else:
data = {}
if not isinstance(data, (list, dict)):
logger.error(f"DeviceInfo: Unexpected API response type for {url}: {type(data)}")
raise ValueError(f"Unexpected API response type: {type(data)}")
self.cache[cache_key] = data
logger.debug(f"Cache set for {cache_key}")
return data
except (aiohttp.ClientError, asyncio.TimeoutError, ValueError) as e:
logger.warning(f"DeviceInfo: Request failed for {endpoint} (attempt {attempt}): {e}")
if attempt < self.config["retry_attempts"]:
if params and "message" in params:
await utils.answer(params["message"], self.strings["retrying"].format(attempt, self.config["retry_attempts"]))
await asyncio.sleep(2 * attempt)
else:
logger.error(f"DeviceInfo: API failed after {self.config['retry_attempts']} attempts: {e}")
if endpoint.startswith("search"):
return []
return {}
def _format_essential_info(self, device: Dict[str, Any]) -> str:
"""Format essential device info (name, network, launch, display, platform, battery)"""
info_text = ""
if "network" in device:
info_text += self.strings["network"].format(device.get("network", "N/A"))
if "launced" in device:
info_text += self.strings["launched"].format(
device["launced"].get("announced", "N/A"),
device["launced"].get("status", "N/A")
)
if "display" in device:
info_text += self.strings["display"].format(
device["display"].get("type", "N/A"),
device["display"].get("size", "N/A"),
device["display"].get("resolution", "N/A"),
device["display"].get("protection", "N/A")
)
if "platform" in device:
info_text += self.strings["platform"].format(
device["platform"].get("os", "N/A"),
device["platform"].get("chipset", "N/A"),
device["platform"].get("cpu", "N/A"),
device["platform"].get("gpu", "N/A")
)
if "battery" in device:
info_text += self.strings["battery"].format(
device["battery"].get("battType", "N/A"),
device["battery"].get("charging", "N/A")
)
return info_text
def _format_section(self, section: str, device: Dict[str, Any]) -> str:
"""Format a specific section of device info"""
if section == "body" and "body" in device:
return self.strings["body"].format(
device["body"].get("dimension", "N/A"),
device["body"].get("weight", "N/A"),
device["body"].get("sim", "N/A"),
device["body"].get("other", "N/A")
)
if section == "memory" and "memory" in device:
memory = {item.get("label", ""): item.get("value", "N/A") for item in device.get("memory", [])}
return self.strings["memory"].format(
memory.get("card", "N/A"),
memory.get("internal", "N/A"),
memory.get("otherMemory", "N/A")
)
if section == "cameras":
cam_text = ""
if "mainCamera" in device:
cam_text += self.strings["main_camera"].format(
device["mainCamera"].get("mainModules", "N/A"),
device["mainCamera"].get("mainFeatures", "N/A"),
device["mainCamera"].get("mainVideo", "N/A")
)
if "selfieCamera" in device:
cam_text += self.strings["selfie_camera"].format(
device["selfieCamera"].get("selfieModules", "N/A"),
device["selfieCamera"].get("selfieFeatures", "N/A"),
device["selfieCamera"].get("selfieVideo", "N/A")
)
return cam_text
if section == "sound" and "sound" in device:
return self.strings["sound"].format(
device["sound"].get("loudSpeaker", "N/A"),
device["sound"].get("audioJack", "N/A"),
device["sound"].get("otherSound", "N/A")
)
if section == "comms" and "comms" in device:
return self.strings["comms"].format(
device["comms"].get("wlan", "N/A"),
device["comms"].get("bluetooth", "N/A"),
device["comms"].get("gps", "N/A"),
device["comms"].get("nfc", "N/A"),
device["comms"].get("infrared", "N/A"),
device["comms"].get("radio", "N/A"),
device["comms"].get("usb", "N/A")
)
if section == "sensors" and "sensors" in device:
return self.strings["sensors"].format(device.get("sensors", "N/A"))
if section == "misc" and "misc" in device:
return self.strings["misc"].format(
device["misc"].get("colors", "N/A"),
device["misc"].get("models", "N/A")
)
return "N/A"
@loader.command(ru_doc="(.di) <название устройства> - Получить информацию о смартфоне", alias="di")
async def deviceinfo(self, message):
"""(.di) <device name> - Get smartphone info by name"""
query = utils.get_args_raw(message).strip()
if not query:
await utils.answer(message, self.strings["no_query"])
return
await utils.answer(message, self.strings["searching"].format(query))
try:
devices = await self._fetch_json("search", {"q": query, "message": message})
if not devices:
await utils.answer(message, self.strings["no_results"].format(query))
return
devices = devices[:self.config["max_results"]]
button_rows = [[{"text": device["name"], "callback": self.show_device_info, "args": [device["id"], query, message.id, message.chat_id, 0, None]}] for device in devices]
await self.inline.list(
message=message,
strings=[self.strings["device_list"].format(len(devices), query)],
custom_buttons=button_rows,
ttl=60,
force_me=True,
manual_security=True,
silent=True
)
except Exception as e:
logger.error(f"DeviceInfo: Failed to fetch search results: {e}")
await utils.answer(message, self.strings["error"].format(str(e)))
async def show_device_info(self, call: InlineMessage, device_id: str, query: str, message_id: int, chat_id: int, photo_idx: int, prev_call_id: str = None):
"""Handle device selection and show essential info with buttons for details"""
message = await self._resolve_entity(call, message_id, chat_id)
try:
device = await self._fetch_json(f"info/{device_id}", {"message": message})
if not device:
raise ValueError("No device info returned")
images_data = await self._fetch_json(f"images/{device_id}", {"message": message})
images = images_data.get("images", []) if isinstance(images_data, dict) else []
info_text = self._format_essential_info(device)
full_text = self.strings["device_info"].format(device.get("name", "N/A"), info_text)
# Truncate for media caption (Telegram limit: 1024 chars)
caption = full_text[:1024] + ("..." if len(full_text) > 1024 else "") if images else full_text
logger.debug(f"DeviceInfo: Caption length: {len(caption)} characters, photo_idx: {photo_idx}")
# Buttons for additional sections and photo navigation
buttons = [
[
{"text": self.strings["show_body"], "callback": self.show_section, "args": ["body", device_id, query, message_id, chat_id, photo_idx]},
{"text": self.strings["show_memory"], "callback": self.show_section, "args": ["memory", device_id, query, message_id, chat_id, photo_idx]},
],
[
{"text": self.strings["show_cameras"], "callback": self.show_section, "args": ["cameras", device_id, query, message_id, chat_id, photo_idx]},
{"text": self.strings["show_sound"], "callback": self.show_section, "args": ["sound", device_id, query, message_id, chat_id, photo_idx]},
],
[
{"text": self.strings["show_comms"], "callback": self.show_section, "args": ["comms", device_id, query, message_id, chat_id, photo_idx]},
{"text": self.strings["show_sensors"], "callback": self.show_section, "args": ["sensors", device_id, query, message_id, chat_id, photo_idx]},
],
[
{"text": self.strings["show_misc"], "callback": self.show_section, "args": ["misc", device_id, query, message_id, chat_id, photo_idx]},
],
[
{"text": self.strings["prev_photo"], "callback": self.show_device_info, "args": [device_id, query, message_id, chat_id, max(0, photo_idx - 1), call.id]} if photo_idx > 0 else None,
{"text": self.strings["next_photo"], "callback": self.show_device_info, "args": [device_id, query, message_id, chat_id, min(len(images) - 1, photo_idx + 1), call.id]} if images and photo_idx < len(images) - 1 else None,
{"text": self.strings["back"], "callback": self.back_to_search, "args": [query, message_id, chat_id]},
]
]
# Filter out None buttons
buttons = [[btn for btn in row if btn] for row in buttons if any(row)]
# Always edit the message (for device selection or photo navigation)
logger.debug(f"DeviceInfo: Editing message for device_id: {device_id}, photo_idx: {photo_idx}, call_id: {call.id}")
await call.edit(
text=caption,
reply_markup=buttons,
photo=images[photo_idx] if images else None,
disable_web_page_preview=True
)
except Exception as e:
logger.error(f"DeviceInfo: Failed to show device info: {e}")
await call.edit(
text=self.strings["error"].format(str(e)),
reply_markup=[],
disable_web_page_preview=True
)
async def show_section(self, call: InlineMessage, section: str, device_id: str, query: str, message_id: int, chat_id: int, photo_idx: int):
"""Show a specific section of device info"""
message = await self._resolve_entity(call, message_id, chat_id)
try:
device = await self._fetch_json(f"info/{device_id}", {"message": message})
if not device:
raise ValueError("No device info returned")
section_text = self._format_section(section, device)
full_text = self.strings["device_info"].format(device.get("name", "N/A"), section_text)
# Truncate for Telegram message limit (4000 chars)
full_text = full_text[:4000] + ("..." if len(full_text) > 4000 else "")
# Buttons for returning to device info
buttons = [
[{"text": self.strings["back_to_device"], "callback": self.show_device_info, "args": [device_id, query, message_id, chat_id, photo_idx, call.id]}]
]
# Try to edit the message
try:
logger.debug(f"DeviceInfo: Editing message for section: {section}, call_id: {call.id}")
await call.edit(
text=full_text,
reply_markup=buttons,
photo=None, # Sections don't include photos to avoid media/text mismatch
disable_web_page_preview=True
)
except Exception as edit_error:
logger.warning(f"DeviceInfo: Failed to edit message for section {section}: {edit_error}")
# Fallback to new inline form if edit fails
await self.inline.form(
text=full_text,
message=message,
reply_markup=buttons,
ttl=300,
force_me=True,
silent=True
)
except Exception as e:
logger.error(f"DeviceInfo: Failed to show section {section}: {e}")
try:
await call.edit(
text=self.strings["error"].format(str(e)),
reply_markup=[],
disable_web_page_preview=True
)
except Exception as edit_error:
logger.warning(f"DeviceInfo: Failed to edit error message: {edit_error}")
await self.inline.form(
text=self.strings["error"].format(str(e)),
message=message,
silent=True
)
async def back_to_search(self, call: InlineMessage, query: str, message_id: int, chat_id: int):
"""Handle 'Back' button to return to search results"""
message = await self._resolve_entity(call, message_id, chat_id)
try:
devices = await self._fetch_json("search", {"q": query, "message": message})
logger.debug(f"DeviceInfo: Fetched {len(devices)} devices for query: {query}")
if not devices:
logger.warning(f"DeviceInfo: No devices found for query: {query}")
try:
await call.edit(
text=self.strings["no_results"].format(query),
reply_markup=[],
photo=None,
disable_web_page_preview=True
)
except Exception as edit_error:
logger.warning(f"DeviceInfo: Failed to edit no_results message: {edit_error}")
await self.inline.form(
text=self.strings["no_results"].format(query),
message=message,
silent=True
)
return
devices = devices[:self.config["max_results"]]
button_rows = [[{"text": device["name"], "callback": self.show_device_info, "args": [device["id"], query, message_id, chat_id, 0, None]}] for device in devices]
list_text = self.strings["device_list"].format(len(devices), query)
# Try to edit the message
try:
logger.debug(f"DeviceInfo: Editing message for back_to_search, query: {query}, call_id: {call.id}")
await call.edit(
text=list_text,
reply_markup=button_rows,
photo=None,
disable_web_page_preview=True
)
except Exception as edit_error:
logger.warning(f"DeviceInfo: Failed to edit back_to_search message: {edit_error}")
await self.inline.list(
message=message_id,
strings=[list_text],
custom_buttons=button_rows,
ttl=60,
force_me=True,
manual_security=True,
silent=True
)
except Exception as e:
logger.error(f"DeviceInfo: Failed to return to search: {e}")
try:
await call.edit(
text=self.strings["error"].format(str(e)),
reply_markup=[],
photo=None,
disable_web_page_preview=True
)
except Exception as edit_error:
logger.warning(f"DeviceInfo: Failed to edit error message: {edit_error}")
await self.inline.form(
text=self.strings["error"].format(str(e)),
message=message,
silent=True
)