from __future__ import annotations import asyncio import logging import time import uuid from dataclasses import dataclass, field from decimal import Decimal, InvalidOperation, ROUND_DOWN from typing import Any, Dict, Optional, Tuple import aiohttp logger = logging.getLogger(__name__) # ============================================================ # Exceptions # ============================================================ class MarketAppAPIError(Exception): """Base MarketApp exception.""" class MarketAppTimeoutError(MarketAppAPIError): """Request timeout.""" class MarketAppConnectionError(MarketAppAPIError): """Network connection error.""" class MarketAppRateLimitError(MarketAppAPIError): """HTTP 429.""" def __init__( self, message: str, retry_after: Optional[float] = None, ): super().__init__(message) self.retry_after = retry_after class MarketAppAuthenticationError(MarketAppAPIError): """Authentication/authorization error.""" class MarketAppValidationError(MarketAppAPIError): """Invalid local input.""" class MarketAppNotFoundError(MarketAppAPIError): """HTTP 404.""" class MarketAppConflictError(MarketAppAPIError): """HTTP 409.""" class MarketAppPriceError(MarketAppAPIError): """Invalid/missing price.""" # ============================================================ # Configuration # ============================================================ @dataclass(slots=True) class MarketAppConfig: base_url: str = "https://api.marketapp.ws/v1" token: str = "" timeout: float = 30.0 connect_timeout: float = 10.0 max_connections: int = 100 retries: int = 3 retry_delay: float = 1.5 max_retry_delay: float = 15.0 proxy: Optional[str] = None proxy_auth: Optional[aiohttp.BasicAuth] = None rate_limit_calls: int = 10 rate_limit_period: float = 1.0 custom_headers: Dict[str, str] = field( default_factory=dict ) enable_cache: bool = True cache_ttl: float = 15.0 cache_max_items: int = 500 usdt_rate_url: str = ( "https://api.binance.com/api/v3/ticker/price" ) usdt_rate_cache_ttl: float = 5.0 usdt_decimals: int = 8 user_agent: str = ( "MatrixBot/MarketAppClient/7.0" ) verify_ssl: bool = True # ============================================================ # Rate Limiter # ============================================================ class RateLimiter: def __init__( self, calls: int, period: float, ): if calls <= 0: raise ValueError( "calls must be greater than zero" ) if period <= 0: raise ValueError( "period must be greater than zero" ) self.capacity = float(calls) self.tokens = float(calls) self.rate = ( float(calls) / float(period) ) self.updated = time.monotonic() self.lock = asyncio.Lock() async def acquire(self) -> None: async with self.lock: while True: now = time.monotonic() elapsed = ( now - self.updated ) self.updated = now self.tokens = min( self.capacity, self.tokens + ( elapsed * self.rate ), ) if self.tokens >= 1: self.tokens -= 1 return wait_time = ( 1.0 - self.tokens ) / self.rate await asyncio.sleep( wait_time ) # ============================================================ # Cache # ============================================================ class MarketAppCache: def __init__( self, max_items: int = 500, default_ttl: float = 15.0, ): self.max_items = max( 1, int(max_items), ) self.default_ttl = max( 0.0, float(default_ttl), ) self.data: Dict[ str, Tuple[float, Any], ] = {} def get( self, key: str, ) -> Optional[Any]: item = self.data.get(key) if item is None: return None created_at, value = item if ( self.default_ttl > 0 and time.monotonic() - created_at > self.default_ttl ): self.data.pop( key, None, ) return None return value def set( self, key: str, value: Any, ) -> None: self.data.pop( key, None, ) while ( len(self.data) >= self.max_items ): try: oldest = next( iter(self.data) ) self.data.pop( oldest, None, ) except StopIteration: break self.data[key] = ( time.monotonic(), value, ) def clear(self) -> None: self.data.clear() def __len__(self) -> int: return len(self.data) # ============================================================ # Client # ============================================================ class MarketAppClient: def __init__( self, config: Optional[ MarketAppConfig ] = None, token: Optional[str] = None, ): if config is None: config = MarketAppConfig( token=token or "" ) elif token and not config.token: config.token = token self.config = config self._session: Optional[ aiohttp.ClientSession ] = None self._session_lock = asyncio.Lock() self._limiter = RateLimiter( calls=self.config.rate_limit_calls, period=self.config.rate_limit_period, ) self._cache = MarketAppCache( max_items=self.config.cache_max_items, default_ttl=self.config.cache_ttl, ) self._external_rate_cache = MarketAppCache( max_items=10, default_ttl=self.config.usdt_rate_cache_ttl, ) logger.info( "MarketApp initialized: %s", self.config.base_url, ) # ======================================================== # Session # ======================================================== async def _get_session( self, ) -> aiohttp.ClientSession: if ( self._session and not self._session.closed ): return self._session async with self._session_lock: if ( self._session and not self._session.closed ): return self._session timeout = aiohttp.ClientTimeout( total=self.config.timeout, connect=self.config.connect_timeout, ) connector = aiohttp.TCPConnector( limit=self.config.max_connections, enable_cleanup_closed=True, ttl_dns_cache=300, ssl=self.config.verify_ssl, ) headers = { "Content-Type": "application/json", "Accept": "application/json", "Accept-Language": "en-US,en;q=0.9", "User-Agent": self.config.user_agent, } if self.config.token: headers["Authorization"] = ( self.config.token ) if self.config.custom_headers: headers.update( self.config.custom_headers ) self._session = aiohttp.ClientSession( connector=connector, timeout=timeout, headers=headers, ) return self._session async def close(self) -> None: if ( self._session and not self._session.closed ): await self._session.close() self._session = None async def __aenter__(self): await self._get_session() return self async def __aexit__( self, exc_type, exc, traceback, ): await self.close() # ======================================================== # URL # ======================================================== def _build_url( self, endpoint: str, ) -> str: return ( f"{self.config.base_url.rstrip('/')}/" f"{endpoint.lstrip('/')}" ) # ======================================================== # Parse response # ======================================================== async def _parse_response( self, response: aiohttp.ClientResponse, ) -> Any: content_type = ( response.headers .get("Content-Type", "") .lower() ) if "json" in content_type: try: return await response.json( content_type=None ) except Exception: text = await response.text() return { "raw": text } text = await response.text() if not text: return {} return { "raw": text } # ======================================================== # Normalize response # ======================================================== @staticmethod def _normalize_result( result: Any, status: int, request_id: str, latency_ms: float, ) -> Dict[str, Any]: if isinstance(result, dict): output = dict(result) else: output = { "data": result } output["_request_id"] = request_id output["_latency_ms"] = round( latency_ms, 2, ) output["_status"] = status output["_success"] = ( 200 <= status < 400 ) if "data" not in output: output["data"] = { key: value for key, value in output.items() if not key.startswith("_") } return output # ======================================================== # Error message # ======================================================== @staticmethod def _extract_error_message( payload: Any, default: str, ) -> str: if isinstance(payload, dict): for key in ( "message", "error", "detail", "description", ): value = payload.get(key) if ( isinstance(value, str) and value.strip() ): return value.strip() errors = payload.get("errors") if isinstance(errors, list): return "; ".join( str(x) for x in errors ) if isinstance(errors, dict): return str(errors) return default # ======================================================== # Request # ======================================================== async def request( self, endpoint: str, payload: Optional[ Dict[str, Any] ] = None, method: str = "POST", use_cache: bool = False, timeout: Optional[float] = None, attempts_override: Optional[int] = None, ) -> Dict[str, Any]: method = method.upper() if not endpoint: raise MarketAppValidationError( "Endpoint cannot be empty." ) url = self._build_url(endpoint) cache_key = ( f"{method}:{url}:" f"{repr(payload)}" ) if ( use_cache and method == "GET" and self.config.enable_cache ): cached = self._cache.get( cache_key ) if cached is not None: result = dict(cached) result["_cache"] = True return result attempts = max( 1, int( attempts_override if attempts_override is not None else self.config.retries ), ) last_exception: Optional[ Exception ] = None for attempt in range( 1, attempts + 1, ): await self._limiter.acquire() session = await self._get_session() request_id = uuid.uuid4().hex[:12] started = time.monotonic() try: kwargs: Dict[str, Any] = {} if payload is not None: if method in { "GET", "DELETE", }: kwargs["params"] = payload else: kwargs["json"] = payload if self.config.proxy: kwargs["proxy"] = ( self.config.proxy ) if self.config.proxy_auth: kwargs["proxy_auth"] = ( self.config.proxy_auth ) if timeout is not None: kwargs["timeout"] = ( aiohttp.ClientTimeout( total=timeout, connect=self.config.connect_timeout, ) ) async with session.request( method, url, **kwargs, ) as response: elapsed = ( time.monotonic() - started ) * 1000 status = response.status if status == 401: raise MarketAppAuthenticationError( "Invalid MarketApp API token." ) if status == 403: raise MarketAppAuthenticationError( "MarketApp API access forbidden." ) if status == 404: raise MarketAppNotFoundError( f"Not found: {endpoint}" ) if status == 409: raise MarketAppConflictError( f"Conflict: {endpoint}" ) body = ( await self._parse_response( response ) ) if status == 429: raw = ( response.headers.get( "Retry-After" ) ) retry_after = None if raw: try: retry_after = float( raw ) except ValueError: pass if retry_after is None: retry_after = min( self.config.max_retry_delay, self.config.retry_delay * ( 2 ** ( attempt - 1 ) ), ) raise MarketAppRateLimitError( "MarketApp rate limit.", retry_after, ) if status >= 500: raise MarketAppAPIError( self._extract_error_message( body, f"Server error {status}", ) ) result = ( self._normalize_result( body, status, request_id, elapsed, ) ) if 400 <= status < 500: result["error"] = ( self._extract_error_message( body, f"HTTP {status}", ) ) return result if ( method == "GET" and use_cache and self.config.enable_cache and result["_success"] ): self._cache.set( cache_key, result, ) return result except ( MarketAppAuthenticationError, MarketAppValidationError, MarketAppNotFoundError, MarketAppConflictError, ): raise except MarketAppRateLimitError as exc: last_exception = exc if attempt >= attempts: break await asyncio.sleep( min( exc.retry_after or self.config.retry_delay, self.config.max_retry_delay, ) ) except asyncio.TimeoutError as exc: last_exception = ( MarketAppTimeoutError( str(exc) ) ) if attempt >= attempts: break await asyncio.sleep( min( self.config.max_retry_delay, self.config.retry_delay * ( 2 ** ( attempt - 1 ) ), ) ) except aiohttp.ClientError as exc: last_exception = ( MarketAppConnectionError( str(exc) ) ) if attempt >= attempts: break await asyncio.sleep( min( self.config.max_retry_delay, self.config.retry_delay * ( 2 ** ( attempt - 1 ) ), ) ) except MarketAppAPIError as exc: last_exception = exc if attempt >= attempts: break await asyncio.sleep( min( self.config.max_retry_delay, self.config.retry_delay * ( 2 ** ( attempt - 1 ) ), ) ) except Exception as exc: raise MarketAppAPIError( f"Unexpected API error: {exc}" ) from exc raise ( last_exception or MarketAppAPIError( "Request failed." ) ) # ======================================================== # Decimal # ======================================================== @staticmethod def decimal( value: Any, ) -> Decimal: try: if isinstance(value, Decimal): return value if isinstance(value, bool): raise ValueError return Decimal( str(value).strip() ) except ( InvalidOperation, ValueError, TypeError, ) as exc: raise MarketAppPriceError( f"Invalid numeric value: {value}" ) from exc # ======================================================== # Recursive value search # ======================================================== @classmethod def _find_value( cls, data: Any, keys: tuple[str, ...], ) -> Optional[Decimal]: if isinstance(data, dict): # اولویت با کلیدهای مشخص قیمت for key in keys: if key not in data: continue value = data[key] if value is None: continue try: return cls.decimal(value) except MarketAppPriceError: continue # جستجوی recursive for value in data.values(): found = cls._find_value( value, keys, ) if found is not None: return found elif isinstance(data, list): for item in data: found = cls._find_value( item, keys, ) if found is not None: return found return None # ======================================================== # Currency extraction # ======================================================== @staticmethod def _normalize_currency( value: Any, ) -> Optional[str]: if value is None: return None text = ( str(value) .strip() .upper() ) aliases = { "TON": "TON", "TONCOIN": "TON", "USDT": "USDT", "TETHER": "USDT", "USD": "USD", "$": "USD", "IRR": "IRR", "RIAL": "IRR", "TMN": "TMM", "TMM": "TMM", "TOMAN": "TMM", } return aliases.get(text) @classmethod def extract_marketapp_currency( cls, response: Any, ) -> Optional[str]: if not isinstance( response, (dict, list), ): return None currency_keys = ( "currency", "currency_code", "price_currency", "amount_currency", "unit", "unit_currency", "quote_currency", ) def find_currency( data: Any, ) -> Optional[str]: if isinstance(data, dict): for key in currency_keys: if key in data: value = ( cls._normalize_currency( data[key] ) ) if value: return value for value in data.values(): found = find_currency( value ) if found: return found elif isinstance(data, list): for item in data: found = find_currency( item ) if found: return found return None return find_currency(response) # ======================================================== # Stars # ======================================================== async def get_stars_price( self, amount: int, ) -> Dict[str, Any]: if isinstance(amount, bool): raise MarketAppValidationError( "Stars amount must be integer." ) try: amount = int(amount) except ( TypeError, ValueError, ) as exc: raise MarketAppValidationError( "Stars amount must be integer." ) from exc if amount < 50: raise MarketAppValidationError( "Minimum is 50 Stars." ) if amount > 10_000: raise MarketAppValidationError( "Maximum is 10000 Stars." ) return await self.request( "fragment/stars/price/", payload={ "amount": amount }, method="POST", ) # ======================================================== # Premium # ======================================================== async def get_premium_price( self, months: int, ) -> Dict[str, Any]: if isinstance(months, bool): raise MarketAppValidationError( "Months must be integer." ) try: months = int(months) except ( TypeError, ValueError, ) as exc: raise MarketAppValidationError( "Months must be integer." ) from exc if months not in { 1, 3, 6, 12, }: raise MarketAppValidationError( "Months must be 1, 3, 6 or 12." ) return await self.request( "fragment/premium/price/", payload={ "months": months }, method="POST", ) # ======================================================== # TON / USDT # ======================================================== async def get_ton_usdt_rate( self, force_refresh: bool = False, ) -> Decimal: cache_key = "TONUSDT_RATE" if ( self.config.enable_cache and not force_refresh ): cached = ( self._external_rate_cache.get( cache_key ) ) if cached is not None: return self.decimal( cached ) session = await self._get_session() params = { "symbol": "TONUSDT" } try: async with session.get( self.config.usdt_rate_url, params=params, timeout=aiohttp.ClientTimeout( total=self.config.timeout, connect=self.config.connect_timeout, ), ) as response: if response.status != 200: text = await response.text() raise MarketAppPriceError( "Cannot get TON/USDT rate: " f"HTTP {response.status} " f"{text}" ) data = await response.json() except asyncio.TimeoutError as exc: raise MarketAppTimeoutError( "TON/USDT price request timeout." ) from exc except aiohttp.ClientError as exc: raise MarketAppConnectionError( f"TON/USDT network error: {exc}" ) from exc if not isinstance(data, dict): raise MarketAppPriceError( "Invalid TON/USDT response." ) price = data.get("price") if price is None: raise MarketAppPriceError( "TON/USDT response has no price." ) rate = self.decimal(price) if rate <= 0: raise MarketAppPriceError( "Invalid TON/USDT rate." ) if self.config.enable_cache: self._external_rate_cache.set( cache_key, rate, ) return rate # ======================================================== # Extract price # ======================================================== @classmethod def extract_marketapp_price( cls, response: Any, ) -> Decimal: keys = ( "price_ton", "amount_ton", "total_ton", "cost_ton", "price", "amount", "total", "cost", "value", ) value = cls._find_value( response, keys, ) if value is None: raise MarketAppPriceError( "Could not find price in " "MarketApp response." ) if value < 0: raise MarketAppPriceError( "MarketApp returned negative price." ) return value # ======================================================== # Extract price + currency # ======================================================== @classmethod def extract_marketapp_price_with_currency( cls, response: Any, ) -> Tuple[ Decimal, str, ]: price = cls.extract_marketapp_price( response ) currency = ( cls.extract_marketapp_currency( response ) ) # ---------------------------------------------------- # MarketApp Fragment endpoints historically return TON. # اگر API currency نفرستاد، TON را به عنوان پیش‌فرض # در نظر می‌گیریم. # ---------------------------------------------------- if currency is None: currency = "TON" return ( price, currency, ) # ======================================================== # TON -> USDT # ======================================================== def ton_to_usdt( self, ton_amount: Decimal, ton_usdt_rate: Decimal, ) -> Decimal: if ton_amount < 0: raise MarketAppPriceError( "TON amount cannot be negative." ) if ton_usdt_rate <= 0: raise MarketAppPriceError( "TON/USDT rate must be positive." ) result = ( ton_amount * ton_usdt_rate ) quantum = ( Decimal("1") / ( Decimal("10") ** self.config.usdt_decimals ) ) return result.quantize( quantum, rounding=ROUND_DOWN, ) # ======================================================== # Stars price USDT # ======================================================== async def get_stars_price_usdt( self, amount: int, ) -> Dict[str, Any]: response = await self.get_stars_price( amount ) ( marketapp_price, currency, ) = self.extract_marketapp_price_with_currency( response ) # ---------------------------------------------------- # اگر API مستقیماً USDT داده باشد # ---------------------------------------------------- if currency == "USDT": usdt_price = ( marketapp_price ) ton_usdt = None # ---------------------------------------------------- # USD تقریباً برابر USDT # ---------------------------------------------------- elif currency == "USD": usdt_price = ( marketapp_price ) ton_usdt = None # ---------------------------------------------------- # TON -> USDT # ---------------------------------------------------- elif currency == "TON": ton_usdt = ( await self.get_ton_usdt_rate() ) usdt_price = ( self.ton_to_usdt( marketapp_price, ton_usdt, ) ) else: raise MarketAppPriceError( f"Unsupported MarketApp currency: " f"{currency}" ) return { "stars": amount, "marketapp_price": str( marketapp_price ), "marketapp_price_ton": ( str(marketapp_price) if currency == "TON" else None ), "source_currency": currency, "ton_usdt": ( str(ton_usdt) if ton_usdt is not None else None ), "price_usdt": str( usdt_price ), "currency": "USDT", "marketapp_response": response, } # ======================================================== # Premium price USDT # ======================================================== async def get_premium_price_usdt( self, months: int, ) -> Dict[str, Any]: response = await self.get_premium_price( months ) ( marketapp_price, currency, ) = self.extract_marketapp_price_with_currency( response ) if currency in { "USDT", "USD", }: usdt_price = ( marketapp_price ) ton_usdt = None elif currency == "TON": ton_usdt = ( await self.get_ton_usdt_rate() ) usdt_price = ( self.ton_to_usdt( marketapp_price, ton_usdt, ) ) else: raise MarketAppPriceError( f"Unsupported MarketApp currency: " f"{currency}" ) return { "months": months, "marketapp_price": str( marketapp_price ), "marketapp_price_ton": ( str(marketapp_price) if currency == "TON" else None ), "source_currency": currency, "ton_usdt": ( str(ton_usdt) if ton_usdt is not None else None ), "price_usdt": str( usdt_price ), "currency": "USDT", "marketapp_response": response, } # ======================================================== # Username # ======================================================== @staticmethod def validate_username( username: str, ) -> str: if not isinstance( username, str, ): raise MarketAppValidationError( "Username must be string." ) username = ( username .strip() .lstrip("@") .strip() ) if not username: raise MarketAppValidationError( "Username cannot be empty." ) if len(username) < 4: raise MarketAppValidationError( "Username is too short." ) if len(username) > 32: raise MarketAppValidationError( "Username is too long." ) allowed = ( "abcdefghijklmnopqrstuvwxyz" "ABCDEFGHIJKLMNOPQRSTUVWXYZ" "0123456789_" ) if any( c not in allowed for c in username ): raise MarketAppValidationError( "Invalid username." ) return username # ======================================================== # Recipient lookup # ======================================================== async def find_stars_recipient( self, username: str, ) -> Dict[str, Any]: username = self.validate_username( username ) return await self.request( "fragment/stars/recipient/", payload={ "username": username, }, method="POST", ) async def find_premium_recipient( self, username: str, ) -> Dict[str, Any]: username = self.validate_username( username ) return await self.request( "fragment/premium/recipient/", payload={ "username": username, }, method="POST", ) async def find_recipient( self, username: str, ) -> Dict[str, Any]: """ Backward-compatible alias used by old PremiumService. """ return await self.find_premium_recipient( username ) # ======================================================== # Buy Stars # ======================================================== async def buy_stars( self, username: str, amount: int, ) -> Dict[str, Any]: username = self.validate_username( username ) try: amount = int(amount) except ( TypeError, ValueError, ) as exc: raise MarketAppValidationError( "Stars amount must be integer." ) from exc if amount < 50: raise MarketAppValidationError( "Minimum is 50 Stars." ) if amount > 10_000: raise MarketAppValidationError( "Maximum is 10000 Stars." ) # IMPORTANT: # A purchase is a mutating operation. Never blindly retry it after # timeout/connection failure because the first request may have # reached MarketApp. return await self.request( "fragment/stars/buy/", payload={ "username": username, "amount": amount, }, method="POST", attempts_override=1, ) # ======================================================== # Buy Premium # ======================================================== async def buy_premium( self, username: str, months: int, ) -> Dict[str, Any]: username = self.validate_username( username ) if months not in { 1, 3, 6, 12, }: raise MarketAppValidationError( "Months must be 1, 3, 6 or 12." ) # IMPORTANT: # A purchase is a mutating operation. Never blindly retry it after # timeout/connection failure because the first request may have # reached MarketApp. return await self.request( "fragment/premium/buy/", payload={ "username": username, "months": months, }, method="POST", attempts_override=1, ) # ======================================================== # Orders # ======================================================== async def get_order_status( self, order_id: str, ) -> Dict[str, Any]: order_id = str( order_id ).strip() if not order_id: raise MarketAppValidationError( "Order ID cannot be empty." ) return await self.request( f"fragment/orders/{order_id}", method="GET", use_cache=False, ) # ======================================================== # TON rate endpoint # ======================================================== async def get_ton_rate( self, ) -> Dict[str, Any]: return await self.request( "ton/rate", method="GET", use_cache=True, ) # ======================================================== # Ping # ======================================================== async def ping(self) -> bool: try: result = await self.request( "ping", method="GET", ) return bool( result.get("_success") ) except Exception: return False # ======================================================== # Cache # ======================================================== def clear_cache(self) -> None: self._cache.clear() self._external_rate_cache.clear() def cache_size(self) -> int: return len(self._cache) # ============================================================ # Factory # ============================================================ async def create_market_app_client( token: str, ) -> MarketAppClient: return MarketAppClient( MarketAppConfig( token=token ) ) async def get_market_app_client( token: Optional[str] = None, ) -> MarketAppClient: return MarketAppClient( token=token ) # ============================================================ # Direct helpers # ============================================================ async def get_stars_price_usdt_direct( amount: int, token: str, ) -> Dict[str, Any]: client = MarketAppClient( token=token ) try: return await client.get_stars_price_usdt( amount ) finally: await client.close() async def get_premium_price_usdt_direct( months: int, token: str, ) -> Dict[str, Any]: client = MarketAppClient( token=token ) try: return await client.get_premium_price_usdt( months ) finally: await client.close() async def buy_stars_direct( username: str, amount: int, token: str, ) -> Dict[str, Any]: client = MarketAppClient( token=token ) try: return await client.buy_stars( username, amount, ) finally: await client.close() async def buy_premium_direct( username: str, months: int, token: str, ) -> Dict[str, Any]: client = MarketAppClient( token=token ) try: return await client.buy_premium( username, months, ) finally: await client.close()