# handlers/user/shop_fragment.py from __future__ import annotations import asyncio import inspect import logging import os import sqlite3 from dataclasses import dataclass from decimal import Decimal, InvalidOperation from html import escape from typing import Any, Final, Optional from aiogram import F, Router from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError from aiogram.fsm.context import FSMContext from aiogram.types import ( CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message, ) from aiogram.utils.keyboard import InlineKeyboardBuilder from core.config import settings from core.ui_emojis import ce, emoji_id, premiumize_html from keyboards.admin_kb import get_owner_order_keyboard from services.market_app_api import ( MarketAppAPIError, MarketAppAuthenticationError, MarketAppClient, MarketAppConfig, MarketAppConflictError, MarketAppConnectionError, MarketAppNotFoundError, MarketAppRateLimitError, MarketAppTimeoutError, MarketAppValidationError, ) from services.pricing_service import ( PricingConfig, PricingService, PricingUnavailableError, ) from states.order import OrderStates try: from utils.finance import finance_manager except Exception: finance_manager = None logger = logging.getLogger(__name__) router = Router( name="legacy_premium_shop_owner_approval" ) # ============================================================ # Configuration # ============================================================ def _env_positive_int( name: str, default: int, ) -> int: raw = os.getenv( name, "", ).strip() if not raw: return int(default) try: value = int(raw) except ValueError: logger.warning( "Invalid %s ignored.", name, ) return int(default) if value <= 0: logger.warning( "%s must be positive; fallback used.", name, ) return int(default) return value def get_database_path() -> str: value = ( getattr( settings, "DB_PATH", None, ) or getattr( settings, "DATABASE_PATH", None, ) or "matrix_bot.db" ) return str(value) DB_PATH: Final[str] = get_database_path() # ============================================================ # Product Catalog # ============================================================ PRODUCTS_DATA: Final[dict[str, dict[str, Any]]] = { "1": { "package_id": "prem_1m", "name": "اشتراک ۱ ماهه پرمیوم تلگرام", "price_toman": _env_positive_int( "PREMIUM_PRICE_1M_TOMAN", 99_000, ), "duration": "۱ ماه", "months": 1, "badge": "پکیج اقتصادی", }, "2": { "package_id": "prem_3m", "name": "اشتراک ۳ ماهه پرمیوم تلگرام", "price_toman": _env_positive_int( "PREMIUM_PRICE_3M_TOMAN", 250_000, ), "duration": "۳ ماه", "months": 3, "badge": "پکیج محبوب", }, "3": { "package_id": "prem_12m", "name": "اشتراک ۱ ساله پرمیوم تلگرام", "price_toman": _env_positive_int( "PREMIUM_PRICE_12M_TOMAN", 890_000, ), "duration": "۱۲ ماه", "months": 12, "badge": "پکیج سالانه", }, } PACKAGE_BY_ID: Final[dict[str, dict[str, Any]]] = { item["package_id"]: { **item, "legacy_id": legacy_id, } for legacy_id, item in PRODUCTS_DATA.items() } # ============================================================ # Order Status # ============================================================ STATUS_PENDING_OWNER: Final[str] = "pending_owner_approval" STATUS_OWNER_PROCESSING: Final[str] = "owner_processing" STATUS_REJECTED: Final[str] = "rejected_by_owner" STATUS_PRICE_CHANGED: Final[str] = "price_changed" STATUS_INSUFFICIENT: Final[str] = "insufficient_balance" STATUS_PROVIDER_PROCESSING: Final[str] = "provider_processing" STATUS_PROVIDER_UNKNOWN: Final[str] = "provider_status_unknown" STATUS_COMPLETED: Final[str] = "completed" STATUS_FAILED_REFUNDED: Final[str] = "failed_refunded" STATUS_FAILED_REFUND_REQUIRED: Final[str] = "failed_refund_required" # ============================================================ # Data Models # ============================================================ @dataclass(slots=True) class StoredOrder: id: int user_id: int product_id: str target: str price: int status: str gateway: str market_order_id: Optional[str] = None owner_id: Optional[int] = None provider_required_price: Optional[int] = None @dataclass(slots=True) class WalletReserveResult: success: bool reason: str balance_before: int = 0 balance_after: int = 0 # ============================================================ # SQLite # ============================================================ def _connect() -> sqlite3.Connection: conn = sqlite3.connect( DB_PATH, timeout=30, check_same_thread=False, ) conn.row_factory = sqlite3.Row conn.execute( "PRAGMA busy_timeout = 30000" ) conn.execute( "PRAGMA foreign_keys = ON" ) return conn def _table_exists( conn: sqlite3.Connection, table_name: str, ) -> bool: row = conn.execute( """ SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1 """, (table_name,), ).fetchone() return row is not None def _column_names( conn: sqlite3.Connection, table_name: str, ) -> set[str]: if not _table_exists( conn, table_name, ): return set() return { str(row["name"]) for row in conn.execute( f"PRAGMA table_info({table_name})" ).fetchall() } def _wallet_column( conn: sqlite3.Connection, ) -> Optional[str]: columns = _column_names( conn, "users", ) if "balance_toman" in columns: return "balance_toman" if "balance" in columns: return "balance" return None async def ensure_shop_schema() -> None: await asyncio.to_thread( _ensure_shop_schema_sync ) def _ensure_shop_schema_sync() -> None: """ Migration فقط ستون‌های کمکی Owner Approval را اضافه می‌کند. جدول orders موجود پروژه حذف/rename نمی‌شود. """ conn = _connect() try: if not _table_exists( conn, "users", ): raise RuntimeError( "users table is missing." ) if not _table_exists( conn, "orders", ): conn.execute( """ CREATE TABLE orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, product_id TEXT NOT NULL, target TEXT, price INTEGER DEFAULT 0, status TEXT DEFAULT 'pending', gateway TEXT DEFAULT 'wallet', market_order_id TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """ ) columns = _column_names( conn, "orders", ) additions = { "owner_id": "INTEGER", "provider_required_price": "INTEGER", "provider_error": "TEXT", "provider_started_at": "TIMESTAMP", "completed_at": "TIMESTAMP", "refunded_at": "TIMESTAMP", } for column, definition in additions.items(): if column not in columns: conn.execute( ( "ALTER TABLE orders " f"ADD COLUMN {column} {definition}" ) ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_orders_status ON orders(status) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_orders_user_status ON orders(user_id, status) """ ) conn.commit() finally: conn.close() # ============================================================ # Order Repository # ============================================================ def _row_to_order( row: sqlite3.Row, ) -> StoredOrder: keys = set( row.keys() ) return StoredOrder( id=int( row["id"] ), user_id=int( row["user_id"] ), product_id=str( row["product_id"] ), target=str( row["target"] or "" ), price=int( row["price"] or 0 ), status=str( row["status"] or "" ), gateway=str( row["gateway"] or "wallet" ), market_order_id=( str( row["market_order_id"] ) if ( "market_order_id" in keys and row["market_order_id"] ) else None ), owner_id=( int( row["owner_id"] ) if ( "owner_id" in keys and row["owner_id"] is not None ) else None ), provider_required_price=( int( row["provider_required_price"] ) if ( "provider_required_price" in keys and row["provider_required_price"] is not None ) else None ), ) async def create_pending_owner_order( *, user_id: int, product_id: str, target: str, price: int, ) -> int: await ensure_shop_schema() return await asyncio.to_thread( _create_pending_owner_order_sync, int(user_id), str(product_id), str(target), int(price), ) def _create_pending_owner_order_sync( user_id: int, product_id: str, target: str, price: int, ) -> int: conn = _connect() try: cursor = conn.execute( """ INSERT INTO orders ( user_id, product_id, target, price, status, gateway, market_order_id ) VALUES (?, ?, ?, ?, ?, 'wallet', NULL) """, ( user_id, product_id, target, price, STATUS_PENDING_OWNER, ), ) conn.commit() return int( cursor.lastrowid ) finally: conn.close() async def get_order( order_id: int, ) -> Optional[StoredOrder]: await ensure_shop_schema() return await asyncio.to_thread( _get_order_sync, int(order_id), ) def _get_order_sync( order_id: int, ) -> Optional[StoredOrder]: conn = _connect() try: row = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, (order_id,), ).fetchone() if not row: return None return _row_to_order( row ) finally: conn.close() async def claim_order_for_owner( order_id: int, owner_id: int, ) -> Optional[StoredOrder]: """ Atomic claim. فقط اولین کلیک مالک از pending_owner_approval عبور می‌کند. """ await ensure_shop_schema() return await asyncio.to_thread( _claim_order_for_owner_sync, int(order_id), int(owner_id), ) def _claim_order_for_owner_sync( order_id: int, owner_id: int, ) -> Optional[StoredOrder]: conn = _connect() try: conn.execute( "BEGIN IMMEDIATE" ) cursor = conn.execute( """ UPDATE orders SET status = ?, owner_id = ?, provider_error = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_OWNER_PROCESSING, owner_id, order_id, STATUS_PENDING_OWNER, ), ) if cursor.rowcount != 1: conn.rollback() return None row = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, (order_id,), ).fetchone() conn.commit() return ( _row_to_order(row) if row else None ) except Exception: conn.rollback() raise finally: conn.close() async def reject_pending_order( order_id: int, owner_id: int, ) -> bool: await ensure_shop_schema() return await asyncio.to_thread( _reject_pending_order_sync, int(order_id), int(owner_id), ) def _reject_pending_order_sync( order_id: int, owner_id: int, ) -> bool: conn = _connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, owner_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_REJECTED, owner_id, order_id, STATUS_PENDING_OWNER, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def restore_order_to_pending( order_id: int, *, error: str, ) -> bool: """ فقط قبل از Debit/Provider buy استفاده شود. """ return await asyncio.to_thread( _restore_order_to_pending_sync, int(order_id), str(error)[:1000], ) def _restore_order_to_pending_sync( order_id: int, error: str, ) -> bool: conn = _connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PENDING_OWNER, error, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def mark_price_changed( order_id: int, required_price: int, ) -> bool: return await asyncio.to_thread( _mark_price_changed_sync, int(order_id), int(required_price), ) def _mark_price_changed_sync( order_id: int, required_price: int, ) -> bool: conn = _connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, provider_required_price = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PRICE_CHANGED, required_price, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def reserve_wallet_for_provider( order_id: int, ) -> WalletReserveResult: """ Debit + provider_processing در یک Transaction. """ return await asyncio.to_thread( _reserve_wallet_for_provider_sync, int(order_id), ) def _reserve_wallet_for_provider_sync( order_id: int, ) -> WalletReserveResult: conn = _connect() try: conn.execute( "BEGIN IMMEDIATE" ) order = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, (order_id,), ).fetchone() if not order: conn.rollback() return WalletReserveResult( success=False, reason="order_not_found", ) if str( order["status"] ) != STATUS_OWNER_PROCESSING: conn.rollback() return WalletReserveResult( success=False, reason="invalid_order_status", ) balance_column = _wallet_column( conn ) if not balance_column: conn.rollback() return WalletReserveResult( success=False, reason="wallet_column_missing", ) user_id = int( order["user_id"] ) price = int( order["price"] or 0 ) user = conn.execute( f""" SELECT {balance_column} AS balance FROM users WHERE user_id = ? LIMIT 1 """, (user_id,), ).fetchone() if not user: conn.rollback() return WalletReserveResult( success=False, reason="user_not_found", ) balance_before = int( user["balance"] or 0 ) if balance_before < price: conn.execute( """ UPDATE orders SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_INSUFFICIENT, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return WalletReserveResult( success=False, reason="insufficient_balance", balance_before=balance_before, balance_after=balance_before, ) balance_after = ( balance_before - price ) conn.execute( f""" UPDATE users SET {balance_column} = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( balance_after, user_id, ), ) if _table_exists( conn, "wallet_transactions", ): columns = _column_names( conn, "wallet_transactions", ) required = { "user_id", "type", "amount", "balance_before", "balance_after", } if required.issubset( columns ): insert_columns = [ "user_id", "type", "amount", "balance_before", "balance_after", ] values: list[Any] = [ user_id, "order_reserve", -price, balance_before, balance_after, ] if "description" in columns: insert_columns.append( "description" ) values.append( f"رزرو موجودی سفارش #{order_id}" ) if "reference_id" in columns: insert_columns.append( "reference_id" ) values.append( f"order:{order_id}:reserve" ) placeholders = ", ".join( "?" for _ in insert_columns ) conn.execute( ( "INSERT INTO wallet_transactions (" + ", ".join(insert_columns) + ") VALUES (" + placeholders + ")" ), tuple(values), ) conn.execute( """ UPDATE orders SET status = ?, provider_started_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PROVIDER_PROCESSING, order_id, STATUS_OWNER_PROCESSING, ), ) conn.commit() return WalletReserveResult( success=True, reason="reserved", balance_before=balance_before, balance_after=balance_after, ) except Exception: conn.rollback() logger.exception( "Wallet reserve failed | order=%s", order_id, ) return WalletReserveResult( success=False, reason="database_error", ) finally: conn.close() async def refund_provider_failure( order_id: int, *, error: str, ) -> bool: """ Refund فقط برای failure قطعی. اگر status دیگر provider_processing نباشد، Refund دوباره انجام نمی‌شود. """ return await asyncio.to_thread( _refund_provider_failure_sync, int(order_id), str(error)[:1000], ) def _refund_provider_failure_sync( order_id: int, error: str, ) -> bool: conn = _connect() try: conn.execute( "BEGIN IMMEDIATE" ) order = conn.execute( """ SELECT * FROM orders WHERE id = ? LIMIT 1 """, (order_id,), ).fetchone() if not order: conn.rollback() return False if str( order["status"] ) != STATUS_PROVIDER_PROCESSING: conn.rollback() return False balance_column = _wallet_column( conn ) if not balance_column: conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_FAILED_REFUND_REQUIRED, error, order_id, STATUS_PROVIDER_PROCESSING, ), ) conn.commit() return False user_id = int( order["user_id"] ) price = int( order["price"] or 0 ) user = conn.execute( f""" SELECT {balance_column} AS balance FROM users WHERE user_id = ? LIMIT 1 """, (user_id,), ).fetchone() if not user: conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? """, ( STATUS_FAILED_REFUND_REQUIRED, error, order_id, ), ) conn.commit() return False before = int( user["balance"] or 0 ) after = before + price conn.execute( f""" UPDATE users SET {balance_column} = ?, updated_at = CURRENT_TIMESTAMP WHERE user_id = ? """, ( after, user_id, ), ) if _table_exists( conn, "wallet_transactions", ): columns = _column_names( conn, "wallet_transactions", ) required = { "user_id", "type", "amount", "balance_before", "balance_after", } if required.issubset( columns ): insert_columns = [ "user_id", "type", "amount", "balance_before", "balance_after", ] values: list[Any] = [ user_id, "order_refund", price, before, after, ] if "description" in columns: insert_columns.append( "description" ) values.append( f"بازگشت وجه سفارش #{order_id}" ) if "reference_id" in columns: insert_columns.append( "reference_id" ) values.append( f"order:{order_id}:refund" ) placeholders = ", ".join( "?" for _ in insert_columns ) conn.execute( ( "INSERT INTO wallet_transactions (" + ", ".join(insert_columns) + ") VALUES (" + placeholders + ")" ), tuple(values), ) conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, refunded_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_FAILED_REFUNDED, error, order_id, STATUS_PROVIDER_PROCESSING, ), ) conn.commit() return True except Exception: conn.rollback() logger.exception( "Refund transaction failed | order=%s", order_id, ) return False finally: conn.close() async def mark_provider_unknown( order_id: int, *, error: str, ) -> bool: return await asyncio.to_thread( _mark_provider_unknown_sync, int(order_id), str(error)[:1000], ) def _mark_provider_unknown_sync( order_id: int, error: str, ) -> bool: conn = _connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, provider_error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_PROVIDER_UNKNOWN, error, order_id, STATUS_PROVIDER_PROCESSING, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def complete_order( order_id: int, *, market_order_id: str, ) -> bool: return await asyncio.to_thread( _complete_order_sync, int(order_id), str(market_order_id), ) def _complete_order_sync( order_id: int, market_order_id: str, ) -> bool: conn = _connect() try: cursor = conn.execute( """ UPDATE orders SET status = ?, market_order_id = ?, completed_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = ? """, ( STATUS_COMPLETED, market_order_id, order_id, STATUS_PROVIDER_PROCESSING, ), ) conn.commit() return ( cursor.rowcount == 1 ) finally: conn.close() async def get_user_balance( user_id: int, ) -> int: await ensure_shop_schema() return await asyncio.to_thread( _get_user_balance_sync, int(user_id), ) def _get_user_balance_sync( user_id: int, ) -> int: conn = _connect() try: balance_column = _wallet_column( conn ) if not balance_column: return 0 row = conn.execute( f""" SELECT {balance_column} AS balance FROM users WHERE user_id = ? LIMIT 1 """, (user_id,), ).fetchone() if not row: return 0 return int( row["balance"] or 0 ) finally: conn.close() # ============================================================ # Admin Access # ============================================================ def get_admin_ids() -> frozenset[int]: method = getattr( settings, "get_admin_list", None, ) if callable(method): try: return frozenset( int(user_id) for user_id in method() if int(user_id) > 0 ) except Exception: logger.exception( "Could not read ADMIN_IDS." ) raw = str( getattr( settings, "ADMIN_IDS", "", ) or "" ) result: set[int] = set() for item in raw.split(","): item = item.strip() if not item: continue try: value = int(item) except ValueError: continue if value > 0: result.add(value) return frozenset(result) def is_admin( user_id: int, ) -> bool: return ( int(user_id) in get_admin_ids() ) # ============================================================ # Pricing # ============================================================ async def _call_maybe_async( func, *args, **kwargs, ): if inspect.iscoroutinefunction( func ): return await func( *args, **kwargs, ) result = await asyncio.to_thread( func, *args, **kwargs, ) if inspect.isawaitable( result ): return await result return result async def get_usdt_toman_rate() -> Optional[Decimal]: """ نرخ تبدیل فقط از سیستم مالی پروژه خوانده می‌شود. اگر نرخ معتبر وجود نداشته باشد، خرید Owner Approval متوقف می‌شود؛ نرخ ثابت مخفی یا قدیمی استفاده نمی‌کنیم. """ if finance_manager is None: return None method = getattr( finance_manager, "get_rate", None, ) if not callable(method): return None try: raw = await _call_maybe_async( method, "usdt", ) rate = Decimal( str(raw) ) if ( not rate.is_finite() or rate <= 0 ): return None return rate except ( InvalidOperation, TypeError, ValueError, ): return None except Exception: logger.exception( "Could not read USDT/Toman rate." ) return None async def get_required_provider_sale_price( *, market_app: MarketAppClient, months: int, ) -> tuple[int, dict[str, Any]]: """ Provider فقط بعد از Owner Approval تماس گرفته می‌شود. این Price Check یک عملیات read-only است و Retry آن مشکلی از جنس duplicate purchase ایجاد نمی‌کند. """ usdt_rate = await get_usdt_toman_rate() if usdt_rate is None: raise PricingUnavailableError( "USDT/Toman rate is not configured." ) config = PricingConfig( usdt_irr_rate=usdt_rate, markup_percent=getattr( settings, "PRICING_MARKUP_PERCENT", Decimal("0"), ), minimum_markup_toman=Decimal("0"), price_round_to=int( getattr( settings, "PRICE_ROUND_TO_TOMAN", 1000, ) or 0 ), minimum_sale_price_toman=int( getattr( settings, "MIN_ORDER_PRICE_TOMAN", 1000, ) or 0 ), refresh_before_purchase=True, ) pricing = PricingService( market_app, config=config, ) quote = await pricing.get_premium_price( int(months) ) price = int( quote.get( "price_toman", 0, ) or 0 ) if price <= 0: raise PricingUnavailableError( "Provider returned invalid premium price." ) return ( price, quote, ) # ============================================================ # One-shot Mutating Client # ============================================================ def build_one_shot_market_client( shared_client: MarketAppClient, ) -> MarketAppClient: """ خرید POST نباید Retry خودکار داشته باشد. MarketAppClient فعلی retries را به صورت عمومی برای request استفاده می‌کند؛ بنابراین برای buy یک Client کوتاه‌عمر با retries=1 ساخته می‌شود و بلافاصله بسته می‌شود. """ config = getattr( shared_client, "config", None, ) if config is None: raise RuntimeError( "MarketApp shared client has no config." ) token = str( getattr( config, "token", "", ) or "" ).strip() if not token: raise RuntimeError( "MarketApp token is not configured." ) one_shot_config = MarketAppConfig( token=token, base_url=str( getattr( config, "base_url", "https://api.marketapp.org/v1", ) ), timeout=float( getattr( config, "timeout", 30.0, ) ), connect_timeout=float( getattr( config, "connect_timeout", 10.0, ) ), retries=1, retry_delay=float( getattr( config, "retry_delay", 1.0, ) ), max_retry_delay=float( getattr( config, "max_retry_delay", 15.0, ) ), proxy=getattr( config, "proxy", None, ), verify_ssl=bool( getattr( config, "verify_ssl", True, ) ), ) return MarketAppClient( one_shot_config ) # ============================================================ # Validation # ============================================================ def normalize_username( raw: str, ) -> Optional[str]: """ Premium MarketApp را فقط با Telegram username ارسال می‌کنیم. """ value = str( raw or "" ).strip() if value.startswith("@"): value = value[1:] if ( len(value) < 5 or len(value) > 32 ): return None if not value[0].isalpha(): return None if not all( char.isalnum() or char == "_" for char in value ): return None return value # ============================================================ # Styled Buttons # ============================================================ def _button( *, text: str, callback_data: str, style: str = "primary", icon_key: str = "sparkles", ) -> InlineKeyboardButton: if style not in { "primary", "success", "danger", }: raise ValueError( "Invalid button style." ) if len( callback_data.encode( "utf-8" ) ) > 64: raise ValueError( "callback_data exceeds 64 bytes." ) return InlineKeyboardButton( text=text, callback_data=callback_data, style=style, icon_custom_emoji_id=( emoji_id( icon_key ) ), ) def shop_main_keyboard() -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() builder.row( _button( text="خرید اشتراک پرمیوم تلگرام", callback_data="shop_cat_vip", style="success", icon_key="premium", ) ) builder.row( _button( text="کیف پول من", callback_data="wallet_menu", style="primary", icon_key="wallet", ), _button( text="حساب کاربری", callback_data="user_profile", style="primary", icon_key="user", ), ) builder.row( _button( text="منوی اصلی ربات", callback_data="main_menu", style="danger", icon_key="back", ) ) return builder.as_markup() def vip_products_keyboard() -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() for legacy_id, product in PRODUCTS_DATA.items(): builder.row( _button( text=( f"{product['duration']} ─ " f"{product['price_toman']:,} تومان" ), callback_data=( f"shop_prod_{legacy_id}" ), style="primary", icon_key="premium", ) ) builder.row( _button( text="بازگشت به فروشگاه", callback_data="shop_main_menu", style="primary", icon_key="back", ) ) return builder.as_markup() def target_keyboard() -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() builder.row( _button( text="انصراف", callback_data="shop_cat_vip", style="danger", icon_key="error", ) ) return builder.as_markup() def invoice_keyboard() -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() builder.row( _button( text="ارسال درخواست برای تایید مالک", callback_data="market_finalize_purchase", style="success", icon_key="success", ) ) builder.row( _button( text="کیف پول", callback_data="wallet_menu", style="primary", icon_key="wallet", ), _button( text="فروشگاه", callback_data="shop_cat_vip", style="primary", icon_key="shop", ), ) builder.row( _button( text="لغو سفارش", callback_data="shop_main_menu", style="danger", icon_key="error", ) ) return builder.as_markup() def success_keyboard() -> InlineKeyboardMarkup: builder = InlineKeyboardBuilder() builder.row( _button( text="بازگشت به فروشگاه", callback_data="shop_main_menu", style="primary", icon_key="shop", ) ) builder.row( _button( text="منوی اصلی", callback_data="main_menu", style="primary", icon_key="back", ) ) return builder.as_markup() # ============================================================ # Telegram Helpers # ============================================================ async def safe_answer( callback: CallbackQuery, text: str | None = None, *, show_alert: bool = False, ) -> bool: try: await callback.answer( text=text, show_alert=show_alert, ) return True except TelegramBadRequest: return False except Exception: logger.exception( "Could not answer shop callback." ) return False async def safe_edit( callback: CallbackQuery, text: str, keyboard: InlineKeyboardMarkup | None = None, ) -> bool: if not callback.message: return False rendered = premiumize_html( text ) try: await callback.message.edit_text( rendered, reply_markup=keyboard, parse_mode="HTML", ) return True except TelegramBadRequest as exc: error = str( exc ).lower() if ( "message is not modified" in error ): return True if ( "message can't be edited" in error or "message to edit not found" in error ): try: await callback.message.answer( rendered, reply_markup=keyboard, parse_mode="HTML", ) return True except Exception: logger.exception( "Could not send replacement shop message." ) return False logger.warning( "Shop edit failed: %s", exc, ) return False except Exception: logger.exception( "Unexpected shop edit error." ) return False async def send_html( message: Message, text: str, *, reply_markup: InlineKeyboardMarkup | None = None, ) -> None: await message.answer( premiumize_html( text ), reply_markup=reply_markup, parse_mode="HTML", ) async def notify_user( bot, user_id: int, text: str, keyboard: InlineKeyboardMarkup | None = None, ) -> bool: try: await bot.send_message( chat_id=user_id, text=premiumize_html( text ), reply_markup=keyboard, parse_mode="HTML", ) return True except TelegramForbiddenError: logger.warning( "User blocked bot | user=%s", user_id, ) return False except Exception: logger.exception( "Could not notify user | user=%s", user_id, ) return False async def notify_admins_new_order( *, bot, order: StoredOrder, product: dict[str, Any], buyer_name: str, buyer_username: Optional[str], ) -> int: sent = 0 username_text = ( f"@{buyer_username}" if buyer_username else "ثبت نشده" ) text = ( f"{ce('package')} " "درخواست جدید خرید پرمیوم\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('list')} سفارش: " f"#{order.id}\n" f"{ce('user')} خریدار: " f"{escape(buyer_name)}\n" f"{ce('chat')} یوزرنیم خریدار: " f"{escape(username_text)}\n" f"{ce('pin')} User ID: " f"{order.user_id}\n\n" f"{ce('premium')} محصول:\n" f"{escape(product['name'])}\n\n" f"{ce('target')} مقصد:\n" f"@{escape(order.target)}\n\n" f"{ce('money')} مبلغ ثبت‌شده:\n" f"{order.price:,} تومان\n\n" f"{ce('warning')} " "تا قبل از تایید شما هیچ مبلغی کسر نشده و " "هیچ درخواست خریدی به MarketApp ارسال نشده است." ) for admin_id in get_admin_ids(): try: await bot.send_message( chat_id=admin_id, text=premiumize_html( text ), reply_markup=( get_owner_order_keyboard( order.id ) ), parse_mode="HTML", ) sent += 1 except Exception: logger.exception( "Could not notify owner | " "admin=%s order=%s", admin_id, order.id, ) return sent # ============================================================ # MarketApp Result Helpers # ============================================================ def is_provider_success( result: Any, ) -> bool: if not isinstance( result, dict, ): return False if "_success" in result: return bool( result.get( "_success" ) ) if "success" in result: return bool( result.get( "success" ) ) status = str( result.get( "status", "" ) or "" ).strip().lower() return status in { "success", "completed", "ok", } def extract_market_order_id( result: dict[str, Any], ) -> str: data = result.get( "data", {} ) if not isinstance( data, dict, ): data = {} value = ( result.get( "order_id" ) or result.get( "id" ) or data.get( "order_id" ) or data.get( "id" ) ) if not value: return "UNKNOWN" return str(value) def provider_error_text( result: Any, ) -> str: if not isinstance( result, dict, ): return ( "پاسخ سرویس‌دهنده معتبر نبود." ) value = ( result.get( "error" ) or result.get( "message" ) or result.get( "detail" ) or "خرید توسط سرویس‌دهنده رد شد." ) return str(value)[:1000] # ============================================================ # Shop Text # ============================================================ SHOP_MAIN_TEXT: Final[str] = ( f"{ce('shop')} " "فروشگاه خدمات تلگرام\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('premium')} " "از این بخش می‌توانید درخواست خرید Premium ثبت کنید.\n\n" f"{ce('warning')} " "فرآیند خرید:\n" "۱. انتخاب پکیج و مقصد\n" "۲. ثبت درخواست خرید\n" "۳. بررسی و تایید مالک\n" "۴. بررسی مجدد قیمت و موجودی\n" "۵. کسر موجودی و ارسال به MarketApp\n\n" "تا قبل از تایید مالک، هیچ مبلغی از کیف پول شما " "کسر نمی‌شود." ) # ============================================================ # Main Shop # ============================================================ @router.callback_query( F.data == "shop_main_menu" ) async def shop_main_menu_handler( callback: CallbackQuery, state: FSMContext, ) -> None: await state.clear() await safe_edit( callback, SHOP_MAIN_TEXT, shop_main_keyboard(), ) await safe_answer( callback ) # ============================================================ # Premium Category # ============================================================ @router.callback_query( F.data == "shop_cat_vip" ) async def shop_category_products_handler( callback: CallbackQuery, state: FSMContext, ) -> None: await state.clear() text = ( f"{ce('premium')} " "بسته‌های Premium تلگرام\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('list')} " "پکیج موردنظر خود را انتخاب کنید.\n\n" f"{ce('warning')} " "قیمت نمایش‌داده‌شده قیمت ثبت درخواست است. " "پس از تایید مالک، قیمت سرویس‌دهنده دوباره بررسی می‌شود؛ " "اگر قیمت لازم افزایش یافته باشد خرید متوقف می‌شود." ) await safe_edit( callback, text, vip_products_keyboard(), ) await safe_answer( callback ) # ============================================================ # Product Detail # ============================================================ @router.callback_query( F.data.startswith( "shop_prod_" ) ) async def shop_product_detail_handler( callback: CallbackQuery, state: FSMContext, ) -> None: legacy_id = str( callback.data or "" ).removeprefix( "shop_prod_" ) product = PRODUCTS_DATA.get( legacy_id ) if not product: await safe_answer( callback, "❌ محصول مورد نظر پیدا نشد.", show_alert=True, ) return await state.clear() await state.update_data( selected_product_id=( product[ "package_id" ] ), product_name=( product[ "name" ] ), price=int( product[ "price_toman" ] ), months=int( product[ "months" ] ), ) text = ( f"{ce('premium')} " "جزئیات محصول\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('package')} عنوان:\n" f"{escape(product['name'])}\n\n" f"{ce('history')} مدت:\n" f"{escape(product['duration'])}\n\n" f"{ce('money')} قیمت ثبت درخواست:\n" f"{int(product['price_toman']):,} تومان\n\n" f"{ce('warning')} " "خرید مستقیم نیست؛ درخواست ابتدا برای تایید مالک ارسال می‌شود." ) builder = InlineKeyboardBuilder() builder.row( _button( text="ادامه و وارد کردن مقصد", callback_data=( f"confirm_market_buy_{legacy_id}" ), style="success", icon_key="success", ) ) builder.row( _button( text="بازگشت", callback_data="shop_cat_vip", style="primary", icon_key="back", ) ) await safe_edit( callback, text, builder.as_markup(), ) await safe_answer( callback ) # ============================================================ # Confirm Product # ============================================================ @router.callback_query( F.data.startswith( "confirm_market_buy_" ) ) async def confirm_purchase_handler( callback: CallbackQuery, state: FSMContext, ) -> None: legacy_id = str( callback.data or "" ).removeprefix( "confirm_market_buy_" ) product = PRODUCTS_DATA.get( legacy_id ) if not product: await safe_answer( callback, "❌ محصول پیدا نشد.", show_alert=True, ) return user_id = int( callback.from_user.id ) balance = await get_user_balance( user_id ) price = int( product[ "price_toman" ] ) if balance < price: shortage = ( price - balance ) text = ( f"{ce('error')} " "موجودی کافی نیست\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('wallet')} موجودی:\n" f"{balance:,} تومان\n\n" f"{ce('money')} مبلغ سفارش:\n" f"{price:,} تومان\n\n" f"{ce('warning')} کسری:\n" f"{shortage:,} تومان" ) builder = InlineKeyboardBuilder() builder.row( _button( text="شارژ کیف پول", callback_data="charge_fiat", style="success", icon_key="wallet", ) ) builder.row( _button( text="بازگشت به فروشگاه", callback_data="shop_cat_vip", style="primary", icon_key="back", ) ) await safe_edit( callback, text, builder.as_markup(), ) await safe_answer( callback ) return await state.update_data( selected_product_id=( product[ "package_id" ] ), product_name=( product[ "name" ] ), price=price, months=int( product[ "months" ] ), ) await state.set_state( OrderStates.waiting_for_target_account ) text = ( f"{ce('target')} " "حساب مقصد Premium\n" "━━━━━━━━━━━━━━━━━━━━\n\n" "یوزرنیم تلگرام مقصد را ارسال کنید.\n\n" "مثال:\n" "@username\n" "username\n\n" f"{ce('warning')} " "Telegram ID عددی در این مسیر پذیرفته نمی‌شود." ) await safe_edit( callback, text, target_keyboard(), ) await safe_answer( callback ) # ============================================================ # Target # ============================================================ @router.message( OrderStates.waiting_for_target_account, F.text, ) async def process_target_id_handler( message: Message, state: FSMContext, ) -> None: target = normalize_username( message.text or "" ) if not target: await send_html( message, ( f"{ce('error')} " "یوزرنیم مقصد معتبر نیست.\n\n" "یوزرنیم باید بین ۵ تا ۳۲ کاراکتر باشد و فقط " "از حروف انگلیسی، عدد و _ تشکیل شود." ), reply_markup=target_keyboard(), ) return data = await state.get_data() package_id = str( data.get( "selected_product_id", "" ) or "" ) product = PACKAGE_BY_ID.get( package_id ) if not product: await state.clear() await send_html( message, ( f"{ce('error')} " "اطلاعات محصول منقضی شده است. " "لطفاً دوباره از فروشگاه شروع کنید." ), reply_markup=success_keyboard(), ) return price = int( data.get( "price", product[ "price_toman" ], ) or 0 ) current_balance = await get_user_balance( message.from_user.id ) await state.update_data( target_account=target ) await state.set_state( OrderStates.waiting_for_request_confirmation ) text = ( f"{ce('list')} " "پیش‌فاکتور درخواست خرید\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('premium')} سرویس:\n" f"{escape(product['name'])}\n\n" f"{ce('target')} مقصد:\n" f"@{escape(target)}\n\n" f"{ce('money')} مبلغ ثبت درخواست:\n" f"{price:,} تومان\n\n" f"{ce('wallet')} موجودی فعلی:\n" f"{current_balance:,} تومان\n\n" f"{ce('warning')} " "با زدن دکمه زیر هیچ پولی کسر نمی‌شود.\n" "درخواست فقط برای تایید مالک ثبت خواهد شد." ) await send_html( message, text, reply_markup=invoice_keyboard(), ) # ============================================================ # Submit Owner Approval Request # ============================================================ @router.callback_query( F.data == "market_finalize_purchase" ) async def finalize_purchase_handler( callback: CallbackQuery, state: FSMContext, ) -> None: """ نام callback قدیمی حفظ شده، اما دیگر خرید MarketApp نیست. فقط: 1) validate 2) persist pending_owner_approval 3) notify owner No debit. No provider call. """ user_id = int( callback.from_user.id ) current_state = await state.get_state() allowed_states = { OrderStates.waiting_for_request_confirmation.state, OrderStates.waiting_for_payment_confirmation.state, } if ( current_state and current_state not in allowed_states ): await safe_answer( callback, "❌ نشست سفارش معتبر نیست.", show_alert=True, ) return data = await state.get_data() package_id = str( data.get( "selected_product_id", "" ) or "" ) target = normalize_username( str( data.get( "target_account", "" ) or "" ) ) product = PACKAGE_BY_ID.get( package_id ) if ( not product or not target ): await state.clear() await safe_answer( callback, "❌ اطلاعات سفارش منقضی شده است.", show_alert=True, ) return quoted_price = int( data.get( "price", product[ "price_toman" ], ) or 0 ) if quoted_price <= 0: await state.clear() await safe_answer( callback, "❌ مبلغ سفارش معتبر نیست.", show_alert=True, ) return balance = await get_user_balance( user_id ) if balance < quoted_price: await state.clear() await safe_edit( callback, ( f"{ce('error')} " "موجودی شما برای ثبت این درخواست کافی نیست.\n\n" f"موجودی: {balance:,} تومان\n" f"مبلغ: {quoted_price:,} تومان" ), success_keyboard(), ) await safe_answer( callback ) return try: order_id = await create_pending_owner_order( user_id=user_id, product_id=package_id, target=target, price=quoted_price, ) order = await get_order( order_id ) if not order: raise RuntimeError( "Created order could not be reloaded." ) # FSM دیگر منبع حقیقت نیست. await state.clear() sent = await notify_admins_new_order( bot=callback.bot, order=order, product=product, buyer_name=( callback.from_user.full_name or "بدون نام" ), buyer_username=( callback.from_user.username ), ) text = ( f"{ce('success')} " "درخواست خرید ثبت شد\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('list')} شماره سفارش:\n" f"#{order_id}\n\n" f"{ce('premium')} محصول:\n" f"{escape(product['name'])}\n\n" f"{ce('target')} مقصد:\n" f"@{escape(target)}\n\n" f"{ce('money')} مبلغ ثبت‌شده:\n" f"{quoted_price:,} تومان\n\n" f"{ce('loading')} وضعیت:\n" "در انتظار تایید مالک\n\n" f"{ce('warning')} " "تا این لحظه هیچ مبلغی از کیف پول شما کسر نشده است." ) if sent == 0: text += ( "\n\n" f"{ce('warning')} " "اعلان مستقیم برای مالک ارسال نشد، اما سفارش " "در دیتابیس ثبت شده است." ) await safe_edit( callback, text, success_keyboard(), ) await safe_answer( callback, "✅ درخواست برای تایید مالک ثبت شد.", show_alert=True, ) logger.info( "Owner approval order created | " "order=%s user=%s package=%s", order_id, user_id, package_id, ) except Exception: logger.exception( "Could not create owner approval order | user=%s", user_id, ) await state.clear() await safe_answer( callback, "❌ ثبت درخواست خرید انجام نشد.", show_alert=True, ) # ============================================================ # Owner Reject # ============================================================ @router.callback_query( F.data.startswith( "owner_reject_order_" ) ) async def owner_reject_order_handler( callback: CallbackQuery, ) -> None: owner_id = int( callback.from_user.id ) if not is_admin( owner_id ): await safe_answer( callback, "⛔ دسترسی مدیریت ندارید.", show_alert=True, ) return try: order_id = int( str( callback.data ).removeprefix( "owner_reject_order_" ) ) except ( TypeError, ValueError, ): await safe_answer( callback, "❌ شناسه سفارش معتبر نیست.", show_alert=True, ) return order = await get_order( order_id ) if not order: await safe_answer( callback, "❌ سفارش پیدا نشد.", show_alert=True, ) return success = await reject_pending_order( order_id, owner_id, ) if not success: fresh = await get_order( order_id ) await safe_answer( callback, ( "⚠️ این سفارش قبلاً پردازش شده است." + ( f"\nوضعیت: {fresh.status}" if fresh else "" ) ), show_alert=True, ) return await safe_edit( callback, ( f"{ce('error')} " "سفارش توسط مالک رد شد.\n\n" f"سفارش: #{order_id}\n" "هیچ مبلغی از کاربر کسر نشده است." ), None, ) await notify_user( callback.bot, order.user_id, ( f"{ce('error')} " "درخواست خرید شما توسط مالک رد شد.\n\n" f"شماره سفارش: #{order_id}\n\n" "هیچ مبلغی از کیف پول شما کسر نشده است." ), success_keyboard(), ) await safe_answer( callback, "سفارش رد شد.", show_alert=True, ) # ============================================================ # Owner Approve # ============================================================ @router.callback_query( F.data.startswith( "owner_approve_order_" ) ) async def owner_approve_order_handler( callback: CallbackQuery, market_app: MarketAppClient | None = None, ) -> None: owner_id = int( callback.from_user.id ) if not is_admin( owner_id ): await safe_answer( callback, "⛔ دسترسی مدیریت ندارید.", show_alert=True, ) return try: order_id = int( str( callback.data ).removeprefix( "owner_approve_order_" ) ) except ( TypeError, ValueError, ): await safe_answer( callback, "❌ شناسه سفارش معتبر نیست.", show_alert=True, ) return # -------------------------------------------------------- # Atomic claim # -------------------------------------------------------- order = await claim_order_for_owner( order_id, owner_id, ) if not order: fresh = await get_order( order_id ) await safe_answer( callback, ( "⚠️ این سفارش قبلاً توسط ادمین دیگری " "یا همین ادمین پردازش شده است." + ( f"\nوضعیت: {fresh.status}" if fresh else "" ) ), show_alert=True, ) return product = PACKAGE_BY_ID.get( order.product_id ) if not product: await restore_order_to_pending( order_id, error="Unknown product_id.", ) await safe_answer( callback, "❌ محصول سفارش معتبر نیست.", show_alert=True, ) return if market_app is None: await restore_order_to_pending( order_id, error="MarketApp client is unavailable.", ) await safe_answer( callback, "❌ MarketApp در Dispatcher فعال نیست.", show_alert=True, ) return await safe_answer( callback, "در حال بررسی نهایی قیمت و موجودی..." ) # -------------------------------------------------------- # Provider read-only price refresh # -------------------------------------------------------- try: required_price, quote = ( await get_required_provider_sale_price( market_app=market_app, months=int( product[ "months" ] ), ) ) except Exception as exc: logger.exception( "Owner price refresh failed | order=%s", order_id, ) await restore_order_to_pending( order_id, error=( "Price refresh failed: " + str(exc) ), ) await notify_user( callback.bot, order.user_id, ( f"{ce('warning')} " "بررسی سفارش هنوز تکمیل نشده است.\n\n" f"شماره سفارش: #{order_id}\n" "به دلیل خطای موقت قیمت، هیچ مبلغی کسر نشد." ), ) await safe_edit( callback, ( f"{ce('warning')} " "قیمت MarketApp قابل بررسی نبود.\n\n" "سفارش دوباره به حالت انتظار مالک برگشت و " "هیچ مبلغی کسر نشد." ), get_owner_order_keyboard( order_id ), ) return # قیمت موردنیاز افزایش پیدا کرده؛ quoted price را تغییر نمی‌دهیم. if required_price > order.price: await mark_price_changed( order_id, required_price, ) await notify_user( callback.bot, order.user_id, ( f"{ce('warning')} " "قیمت سرویس تغییر کرده است.\n\n" f"سفارش: #{order_id}\n" f"قیمت ثبت‌شده: {order.price:,} تومان\n" f"قیمت لازم فعلی: {required_price:,} تومان\n\n" "برای جلوگیری از کسر مبلغ متفاوت، این سفارش خرید نشد." ), success_keyboard(), ) await safe_edit( callback, ( f"{ce('warning')} " "خرید متوقف شد؛ قیمت افزایش یافته است.\n\n" f"ثبت‌شده: {order.price:,} تومان\n" f"قیمت لازم: {required_price:,} تومان\n\n" "هیچ مبلغی کسر نشده و MarketApp buy فراخوانی نشد." ), None, ) return # -------------------------------------------------------- # Atomic wallet reserve # -------------------------------------------------------- reservation = await reserve_wallet_for_provider( order_id ) if not reservation.success: if ( reservation.reason == "insufficient_balance" ): await notify_user( callback.bot, order.user_id, ( f"{ce('error')} " "موجودی شما برای سفارش کافی نیست.\n\n" f"سفارش: #{order_id}\n" f"مبلغ: {order.price:,} تومان\n" f"موجودی: {reservation.balance_before:,} تومان\n\n" "هیچ خریدی از MarketApp انجام نشد." ), success_keyboard(), ) await safe_edit( callback, ( f"{ce('warning')} " "موجودی کاربر کافی نیست.\n\n" "سفارش به MarketApp ارسال نشد." ), None, ) return await restore_order_to_pending( order_id, error=( "Wallet reserve failed: " + reservation.reason ), ) await safe_edit( callback, ( f"{ce('error')} " "رزرو موجودی انجام نشد.\n\n" "MarketApp buy اجرا نشد." ), get_owner_order_keyboard( order_id ), ) return # -------------------------------------------------------- # Exactly-one automatic buy attempt # -------------------------------------------------------- one_shot_client: Optional[ MarketAppClient ] = None try: one_shot_client = ( build_one_shot_market_client( market_app ) ) result = await one_shot_client.buy_premium( username=order.target, months=int( product[ "months" ] ), ) except ( MarketAppTimeoutError, MarketAppConnectionError, ) as exc: # Ambiguous: # درخواست ممکن است به Provider رسیده باشد. await mark_provider_unknown( order_id, error=( f"{type(exc).__name__}: {exc}" ), ) await notify_user( callback.bot, order.user_id, ( f"{ce('warning')} " "وضعیت سفارش نیازمند بررسی است.\n\n" f"شماره سفارش: #{order_id}\n\n" "پاسخ قطعی از MarketApp دریافت نشد. " "برای جلوگیری از خرید دوباره یا Refund اشتباه، " "سفارش به صورت خودکار تکرار یا برگشت داده نشد." ), success_keyboard(), ) await safe_edit( callback, ( f"{ce('warning')} " "provider_status_unknown\n\n" f"سفارش: #{order_id}\n" "Timeout/Connection بعد از شروع درخواست خرید رخ داد.\n\n" "خرید را دوباره اجرا نکنید و Refund خودکار نزنید." ), None, ) logger.critical( "AMBIGUOUS MARKETAPP PURCHASE | " "order=%s user=%s", order_id, order.user_id, ) return except MarketAppAPIError as exc: # Generic APIError در Client فعلی می‌تواند HTTP 5xx باشد. # بعد از شروع buy، آن را ambiguous در نظر می‌گیریم. await mark_provider_unknown( order_id, error=( f"{type(exc).__name__}: {exc}" ), ) await notify_user( callback.bot, order.user_id, ( f"{ce('warning')} " "نتیجه MarketApp قطعی نیست.\n\n" f"شماره سفارش: #{order_id}\n" "سفارش برای بررسی دستی نگه داشته شد." ), success_keyboard(), ) await safe_edit( callback, ( f"{ce('warning')} " "پاسخ MarketApp نامشخص است.\n\n" f"سفارش: #{order_id}\n\n" "Refund یا Retry خودکار انجام نشد." ), None, ) return except Exception as exc: # Exception ناشناخته پس از شروع buy نیز ambiguous است. await mark_provider_unknown( order_id, error=( f"Unexpected: {type(exc).__name__}: {exc}" ), ) logger.exception( "Unexpected provider exception | order=%s", order_id, ) await safe_edit( callback, ( f"{ce('warning')} " "وضعیت سفارش نامشخص شد.\n\n" "Retry/Refund خودکار انجام نشد." ), None, ) return finally: if one_shot_client is not None: try: await one_shot_client.close() except Exception: logger.exception( "Could not close one-shot MarketApp client." ) # -------------------------------------------------------- # Definite provider response # -------------------------------------------------------- if not is_provider_success( result ): error_text = provider_error_text( result ) refunded = await refund_provider_failure( order_id, error=error_text, ) await notify_user( callback.bot, order.user_id, ( f"{ce('error')} " "خرید Premium توسط سرویس‌دهنده انجام نشد.\n\n" f"سفارش: #{order_id}\n\n" + ( f"{ce('success')} مبلغ سفارش به کیف پول شما برگشت داده شد." if refunded else ( f"{ce('warning')} بازگشت وجه نیازمند بررسی مالی ادمین است." ) ) ), success_keyboard(), ) await safe_edit( callback, ( f"{ce('error')} " "MarketApp خرید را ناموفق اعلام کرد.\n\n" f"سفارش: #{order_id}\n" f"Refund: {'completed' if refunded else 'manual_required'}" ), None, ) return market_order_id = extract_market_order_id( result ) completed = await complete_order( order_id, market_order_id=market_order_id, ) if not completed: # Provider success قطعی است، پس Refund ممنوع. await mark_provider_unknown( order_id, error=( "Provider succeeded but local completion update failed. " f"market_order_id={market_order_id}" ), ) logger.critical( "Provider success but DB completion failed | " "order=%s market_order=%s", order_id, market_order_id, ) await notify_user( callback.bot, order.user_id, ( f"{ce('warning')} " "سرویس تحویل شده ولی ثبت نهایی نیازمند بررسی است.\n\n" f"سفارش داخلی: #{order_id}\n" f"MarketApp: {escape(market_order_id)}" ), ) await safe_edit( callback, ( f"{ce('warning')} " "MarketApp موفق، ثبت محلی ناموفق.\n\n" f"Order: #{order_id}\n" f"MarketApp: {escape(market_order_id)}\n\n" "Refund نکنید مگر پس از بررسی دستی." ), None, ) return await notify_user( callback.bot, order.user_id, ( f"{ce('success')} " "خرید Premium با موفقیت انجام شد.\n" "━━━━━━━━━━━━━━━━━━━━\n\n" f"{ce('list')} سفارش: " f"#{order_id}\n" f"{ce('premium')} محصول: " f"{escape(product['name'])}\n" f"{ce('target')} مقصد: " f"@{escape(order.target)}\n" f"{ce('money')} مبلغ: " f"{order.price:,} تومان\n" f"{ce('success')} MarketApp ID: " f"{escape(market_order_id)}" ), success_keyboard(), ) await safe_edit( callback, ( f"{ce('success')} " "سفارش با موفقیت تکمیل شد.\n\n" f"Order: #{order_id}\n" f"MarketApp: {escape(market_order_id)}\n" f"User: {order.user_id}" ), None, ) await safe_answer( callback, "✅ سفارش تکمیل شد.", show_alert=True, ) # ============================================================ # Legacy FSM Compatibility # ============================================================ @router.callback_query( OrderStates.waiting_for_payment_confirmation, F.data == "market_finalize_purchase", ) async def legacy_payment_confirmation_bridge( callback: CallbackQuery, state: FSMContext, ) -> None: """ برای کاربری که قبل از Deploy در State قدیمی بوده است. """ await finalize_purchase_handler( callback, state, ) # ============================================================ # Export # ============================================================ __all__ = [ "router", "PRODUCTS_DATA", "SHOP_MAIN_TEXT", "shop_main_keyboard", "vip_products_keyboard", "target_keyboard", "invoice_keyboard", "success_keyboard", "ensure_shop_schema", "create_pending_owner_order", "claim_order_for_owner", "reject_pending_order", "reserve_wallet_for_provider", "refund_provider_failure", "mark_provider_unknown", "complete_order", ]