From 8fdd2dd3c478075104689a91d30a454c19ab11e6 Mon Sep 17 00:00:00 2001 From: Capybara-z Date: Sat, 18 Oct 2025 00:21:14 +0300 Subject: [PATCH] Add router icon / multi-format media support --- handlers/keys/key_view.py | 17 ++++- handlers/keys/subscriptions.py | 1 + handlers/utils.py | 133 ++++++++++++++++++++++++--------- 3 files changed, 112 insertions(+), 39 deletions(-) diff --git a/handlers/keys/key_view.py b/handlers/keys/key_view.py index 7c2dec55..7d63eac9 100644 --- a/handlers/keys/key_view.py +++ b/handlers/keys/key_view.py @@ -96,7 +96,7 @@ async def process_callback_or_message_view_keys(callback_query_or_message: Messa await render_key_info(target_message, session, key_name, image_path) return - inline_keyboard, response_message = build_keys_response(records) + inline_keyboard, response_message = await build_keys_response(records, session) image_path = os.path.join("img", "pic_keys.jpg") await edit_or_send_message( @@ -110,7 +110,7 @@ async def process_callback_or_message_view_keys(callback_query_or_message: Messa await target_message.answer(text=error_message) -def build_keys_response(records): +async def build_keys_response(records, session): """ Формирует сообщение и клавиатуру для устройств с указанием срока действия подписки. """ @@ -133,7 +133,18 @@ def build_keys_response(records): else: formatted_date_full = "без срока действия" - key_button = InlineKeyboardButton(text=f"🔑 {key_display}", callback_data=f"view_key|{email}") + is_vless = False + if hasattr(record, 'tariff_id') and record.tariff_id: + try: + tariff = await get_tariff_by_id(session, record.tariff_id) + if tariff and tariff.get("vless"): + is_vless = True + except: + pass + + icon = "📶" if is_vless else "🔑" + + key_button = InlineKeyboardButton(text=f"{icon} {key_display}", callback_data=f"view_key|{email}") rename_button = InlineKeyboardButton(text=ALIAS, callback_data=f"rename_key|{client_id}") builder.row(key_button, rename_button) diff --git a/handlers/keys/subscriptions.py b/handlers/keys/subscriptions.py index 821224bf..6350e31c 100644 --- a/handlers/keys/subscriptions.py +++ b/handlers/keys/subscriptions.py @@ -203,6 +203,7 @@ def prepare_headers( "announce": "base64:" + base64.b64encode(announce_str.encode("utf-8")).decode("utf-8"), "profile-web-page-url": f"https://t.me/{USERNAME_BOT}", "subscription-userinfo": subscription_userinfo, + #"routing": "happ://routing/onadd/...", } elif "Hiddify" in user_agent: parts = subscription_info.split(" - ")[0].split(": ") diff --git a/handlers/utils.py b/handlers/utils.py index c3201115..b2fc60da 100644 --- a/handlers/utils.py +++ b/handlers/utils.py @@ -15,6 +15,8 @@ from aiogram.types import ( BufferedInputFile, InlineKeyboardMarkup, InputMediaPhoto, + InputMediaVideo, + InputMediaAnimation, Message, ) from sqlalchemy import func, select @@ -197,6 +199,24 @@ def format_hours(hours: int) -> str: return f"{hours} {get_plural_form(hours, 'час', 'часа', 'часов')}" +def get_media_type(media_path: str) -> str: + if not media_path: + return 'photo' + + ext = os.path.splitext(media_path.lower())[1] + + if ext in ['.jpg', '.jpeg', '.png', '.webp']: + return 'photo' + + if ext in ['.mp4', '.mov', '.avi']: + return 'video' + + if ext == '.gif': + return 'animation' + + return 'photo' + + async def edit_or_send_message( target_message: Message, text: str, @@ -212,47 +232,88 @@ async def edit_or_send_message( edit_or_send_message.lock = asyncio.Lock() edit_or_send_message.max = 256 - if media_path and os.path.isfile(media_path): - async with edit_or_send_message.lock: - cached_id = edit_or_send_message.cache.get(media_path) + def find_media_file(original_path: str) -> str | None: + if not original_path: + return None + + if os.path.isfile(original_path): + return original_path + + base_name = os.path.splitext(original_path)[0] + supported_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.webp', '.mp4', '.mov', '.avi'] + + for ext in supported_extensions: + fallback_path = base_name + ext + if os.path.isfile(fallback_path): + return fallback_path + + return None + + if media_path: + actual_media_path = find_media_file(media_path) + if actual_media_path: + media_type = get_media_type(actual_media_path) + + async with edit_or_send_message.lock: + cached_id = edit_or_send_message.cache.get(actual_media_path) + if cached_id: + edit_or_send_message.cache.move_to_end(actual_media_path) + if cached_id: - edit_or_send_message.cache.move_to_end(media_path) - if cached_id: - try: - await target_message.edit_media(InputMediaPhoto(media=cached_id, caption=text), reply_markup=reply_markup) - return - except Exception: try: - await target_message.answer_photo( - photo=cached_id, - caption=text, - reply_markup=reply_markup, - disable_web_page_preview=disable_web_page_preview, - ) + if media_type == 'photo': + await target_message.edit_media(InputMediaPhoto(media=cached_id, caption=text), reply_markup=reply_markup) + elif media_type == 'video': + await target_message.edit_media(InputMediaVideo(media=cached_id, caption=text), reply_markup=reply_markup) + elif media_type == 'animation': + await target_message.edit_media(InputMediaAnimation(media=cached_id, caption=text), reply_markup=reply_markup) return except Exception: - pass + try: + if media_type == 'photo': + await target_message.answer_photo(photo=cached_id, caption=text, reply_markup=reply_markup, disable_web_page_preview=disable_web_page_preview) + elif media_type == 'video': + await target_message.answer_video(video=cached_id, caption=text, reply_markup=reply_markup, disable_web_page_preview=disable_web_page_preview) + elif media_type == 'animation': + await target_message.answer_animation(animation=cached_id, caption=text, reply_markup=reply_markup, disable_web_page_preview=disable_web_page_preview) + return + except Exception: + pass - async with aiofiles.open(media_path, "rb") as f: - data = await f.read() - upload = BufferedInputFile(data, filename=os.path.basename(media_path)) - try: - msg = await target_message.edit_media(InputMediaPhoto(media=upload, caption=text), reply_markup=reply_markup) - except Exception: - msg = await target_message.answer_photo( - photo=upload, - caption=text, - reply_markup=reply_markup, - disable_web_page_preview=disable_web_page_preview, - ) - if getattr(msg, "photo", None): - fid = msg.photo[-1].file_id - async with edit_or_send_message.lock: - if media_path not in edit_or_send_message.cache: - edit_or_send_message.cache[media_path] = fid - if len(edit_or_send_message.cache) > edit_or_send_message.max: - edit_or_send_message.cache.popitem(last=False) - return + async with aiofiles.open(actual_media_path, "rb") as f: + data = await f.read() + upload = BufferedInputFile(data, filename=os.path.basename(actual_media_path)) + + try: + if media_type == 'photo': + msg = await target_message.edit_media(InputMediaPhoto(media=upload, caption=text), reply_markup=reply_markup) + elif media_type == 'video': + msg = await target_message.edit_media(InputMediaVideo(media=upload, caption=text), reply_markup=reply_markup) + elif media_type == 'animation': + msg = await target_message.edit_media(InputMediaAnimation(media=upload, caption=text), reply_markup=reply_markup) + except Exception: + if media_type == 'photo': + msg = await target_message.answer_photo(photo=upload, caption=text, reply_markup=reply_markup, disable_web_page_preview=disable_web_page_preview) + elif media_type == 'video': + msg = await target_message.answer_video(video=upload, caption=text, reply_markup=reply_markup, disable_web_page_preview=disable_web_page_preview) + elif media_type == 'animation': + msg = await target_message.answer_animation(animation=upload, caption=text, reply_markup=reply_markup, disable_web_page_preview=disable_web_page_preview) + + file_id = None + if hasattr(msg, 'photo') and msg.photo: + file_id = msg.photo[-1].file_id + elif hasattr(msg, 'video') and msg.video: + file_id = msg.video.file_id + elif hasattr(msg, 'animation') and msg.animation: + file_id = msg.animation.file_id + + if file_id: + async with edit_or_send_message.lock: + if actual_media_path not in edit_or_send_message.cache: + edit_or_send_message.cache[actual_media_path] = file_id + if len(edit_or_send_message.cache) > edit_or_send_message.max: + edit_or_send_message.cache.popitem(last=False) + return if not force_text and target_message.caption is not None: try: