""" Kalino API service for virtual numbers (شماره مجازی). Base URL: https://api.ozvinoo.xyz/ The provider strictly uses the V1 /web/{token}/... format for EVERYTHING: GET /web/{token}/applications → app list (all services) GET /web/{token}/get-prices/{service_id} → countries with base prices GET /web/{token}/getNumber/{service_id}/{country} → buy a virtual number GET /web/{token}/getCode/{request_id} → fetch the SMS verification code GET /web/{token}/get-balance → fetch panel balance The API token lives in ``config.KALINO_API_KEY``. Dynamic Profit Margin: Final_Price = Base_Price + (Base_Price × (Margin / 100)) Margin is fetched live from the ``settings`` table (key ``account_profit_margin``) so admin changes apply immediately. Falls back to config.ACCOUNT_PROFIT_MARGIN_PERCENT (default 30%) if unset. """ import json import logging import time from typing import Optional import aiohttp from config import config from utils.emojis import get_plain_emoji logger = logging.getLogger(__name__) TOKEN = config.KALINO_API_KEY # ─── Cache (5-minute TTL) ──────────────────────────────────────── _cache: dict = {} _cache_ts: float = 0.0 _CACHE_TTL: float = 300.0 # 5 minutes # ─── Persian labels for known application codes ─────────────────── # Every emoji comes from the shared list in utils/emojis.py. _APP_LABELS = { "tg": f"{get_plain_emoji('diamond')} تلگرام پنل اختصاصی", "wa": f"{get_plain_emoji('asterisk')} واتساپ", "ig": f"{get_plain_emoji('rocket')} اینستاگرام", "change": f"{get_plain_emoji('moai')} چنج نامبر", "google": f"{get_plain_emoji('search')} گوگل", "fb": f"{get_plain_emoji('blue_book')} فیسبوک", "imo": f"{get_plain_emoji('mobile')} ایمو", "tiktok": f"{get_plain_emoji('music_note')} تیک تاک", "x": f"{get_plain_emoji('cross')} ایکس (توییتر)", "wechat": f"{get_plain_emoji('comment')} وی چت", "tinder": f"{get_plain_emoji('fire')} تیندر", "likee": f"{get_plain_emoji('sparkling_heart')} لایکی", "yahoo": f"{get_plain_emoji('web')} یاهو", "netflix": f"{get_plain_emoji('film')} نتفلیکس", "paypal": f"{get_plain_emoji('parking')} پی پال", "steam": f"{get_plain_emoji('joystick')} استیم", "microsoft": f"{get_plain_emoji('laptop')} مایکروسافت", "line": f"{get_plain_emoji('clover')} لاین", "uber": f"{get_plain_emoji('car')} اوبر", "alibaba": f"{get_plain_emoji('cloud')} علی بابا", "amazon": f"{get_plain_emoji('box')} آمازون", "discord": f"{get_plain_emoji('video_game')} دیسکورد", "apple": f"{get_plain_emoji('apple')} اپل", "ebay": f"{get_plain_emoji('cart')} ای بی", } def _is_cache_valid() -> bool: return (time.time() - _cache_ts) < _CACHE_TTL def _parse_in_stock(count) -> bool: """Parse the ``count`` field into a boolean stock flag. Supports both numeric counts (0 → out of stock) and Persian/English presence words returned by some providers. """ if count is None: return True s = str(count).strip().replace(",", "").replace("،", "") if not s: return True try: return int(float(s)) > 0 except ValueError: low = s.lower() return ("موجود" in s) or ("available" in low) or (s not in ("", "0", "ناموجود", "unavailable")) def get_app_label(code: str, fallback: str = "") -> str: """Return the Persian display label (with emoji) for an app code.""" label = _APP_LABELS.get(str(code).lower()) if label: return label return fallback or (str(code) or f"{get_plain_emoji('small_blue_diamond')} سرویس") # ══════════════════════════════════════════════════════════════════════ # VIRTUAL NUMBERS (V1 /web/{token}/ API — "شماره مجازی") # ══════════════════════════════════════════════════════════════════════ def _extract_balance(data) -> Optional[int]: """Dig the panel balance out of ANY Kalino response shape. Supports: {"balance": 1234} {"data": {"balance": 1234}} {"success": true, "data": {"balance": "1234"}} 1234 (bare number) "1234" (bare numeric string) Returns an int, or None when the payload contains no balance field. """ if data is None: return None if isinstance(data, dict): for key in ("balance", "remaining_balance", "money"): if key in data: raw = data[key] if isinstance(raw, dict): return _extract_balance(raw) try: return int(float(str(raw).replace(",", "").replace("،", ""))) except (TypeError, ValueError): return None if isinstance(data.get("data"), dict): return _extract_balance(data["data"]) return None if isinstance(data, (int, float)): return int(data) if isinstance(data, str): try: return int(float(data.replace(",", "").replace("،", ""))) except (TypeError, ValueError): return None return None async def get_panel_balance() -> int: """Fetch the Kalino panel balance (API money — separate from the DB wallet). Returns the balance as an integer Toman amount, or 0 if the request fails / the token is rejected. Parses every documented Kalino shape (flat ``balance`` or nested ``data.balance``). """ url = f"https://api.ozvinoo.xyz/web/{TOKEN}/get-balance" try: timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url) as resp: raw = await resp.text() logger.debug(f"RAW KALINO BALANCE [{resp.status}]: {raw[:500]}") data = json.loads(raw) parsed = _extract_balance(data) if parsed is None: logger.warning("Could not parse Kalino balance: %s", raw[:300]) return 0 return parsed except Exception as e: logger.error(f"Error fetching Kalino balance: {e}") return 0 async def get_applications() -> Optional[list[dict]]: """Fetch all available services (applications) from the Kalino API. Returns: List of dicts: {"id", "code", "title", "name", ...} or ``None``. """ try: timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: url = f"https://api.ozvinoo.xyz/web/{TOKEN}/applications" async with session.get(url) as resp: raw = await resp.text() logger.debug(f"RAW APPLICATIONS RESPONSE: {raw[:1000]}") data = json.loads(raw) if isinstance(data, dict): apps = [app for app in data.values() if isinstance(app, dict)] elif isinstance(data, list): apps = [app for app in data if isinstance(app, dict)] else: return None app_list = [] for app in apps: service_id = app.get("id") code = app.get("code", "نامشخص") title = app.get("title") or app.get("name") or code app_list.append({ "service_id": service_id, "code": code, "name": code, "title": title, }) return app_list or None except Exception as exc: logger.error(f"FATAL ERROR in get_applications: {exc}", exc_info=True) async def get_applications_list() -> dict: """Fetch the raw applications dict from the Kalino API. Returns the raw dict keyed by app code (e.g. {"tg": {...}, "wa": {...}}) so the keyboard can map codes to real ``service_id``s dynamically. Returns an empty dict on failure. """ try: timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: url = f"https://api.ozvinoo.xyz/web/{TOKEN}/applications" async with session.get(url) as resp: raw = await resp.text() logger.debug(f"RAW APPLICATIONS LIST: {raw[:1000]}") data = json.loads(raw) if isinstance(data, dict): return data elif isinstance(data, list): # Convert list to dict keyed by code for consistent access return {str(item.get("code", "")): item for item in data if isinstance(item, dict)} return {} except Exception as exc: logger.error(f"Error in get_applications_list: {exc}", exc_info=True) return {} # ─── Dynamic services map (label → service_id) ──────────────────────── _services_map: dict[str, int] = {} _services_map_ts: float = 0.0 _SERVICES_MAP_TTL: float = 300.0 # 5 minutes async def get_apps_display_map() -> dict[str, int]: """Build a live map of display label → numeric service_id. Returns e.g. {"💎 تلگرام پنل اختصاصی": 1, "✳️ واتساپ": 2, ...}. Cached for 5 minutes to avoid hammering the API. """ global _services_map, _services_map_ts now = time.time() if _services_map and (now - _services_map_ts) < _SERVICES_MAP_TTL: return _services_map apps = await get_applications_list() result: dict[str, int] = {} for code, app in apps.items(): if not isinstance(app, dict): continue service_id = app.get("id") if service_id is None: continue fallback = app.get("title") or code label = get_app_label(code, fallback) result[label] = int(service_id) # Telegram always first in the grid ordered: dict[str, int] = {} for label, sid in result.items(): if "تلگرام" in label: ordered[label] = sid for label, sid in result.items(): if "تلگرام" not in label: ordered[label] = sid if ordered: _services_map = ordered _services_map_ts = now logger.info("Fetched %d services from Kalino API", len(ordered)) return _services_map async def warm_services_cache() -> None: """Prefetch the Kalino applications + display map in the background. Runs as an asyncio task at startup. Never raises — any failure just leaves the cache to be populated lazily on the first user tap. """ try: await get_applications_list() await get_apps_display_map() except Exception as exc: logger.error("Kalino services prewarm failed: %s", exc) async def get_telegram_countries(service_id) -> Optional[list[dict]]: """Fetch countries for a given service and apply the live profit margin. ``service_id`` may be a numeric id (e.g. 1) or a provider code (e.g. "tg", "change", "imo", "tg_noreport") — it is injected into the appropriate API URL. For ``tg_noreport`` (non-report Telegram numbers), a special V2 API endpoint is used instead of the standard V1 get-prices path. """ if str(service_id) == "tg_noreport": return await get_tg_noreport_countries() return await get_countries(service_id) async def get_tg_noreport_countries() -> Optional[list[dict]]: """Fetch Telegram non-report countries from the live Telegram price list. The provider exposes the country/stock list through the normal Telegram service (service id 1). The actual non-report purchase is still performed through the V2 ``telegram-numbers/numbers/`` endpoint. """ try: from database.db import get_profit_margin margin = await get_profit_margin( config.ACCOUNT_PROFIT_MARGIN_PERCENT ) / 100 url = f"https://api.ozvinoo.xyz/web/{TOKEN}/get-prices/1" timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url) as resp: raw = await resp.text() logger.debug( f"RAW TG NOREPORT PRICES RESPONSE [{resp.status}]: " f"{raw[:1500]}" ) if resp.status != 200: logger.error( f"TG NOREPORT prices API returned HTTP " f"{resp.status}: {raw[:500]}" ) return None data = json.loads(raw) if not isinstance(data, list): logger.error( "TG NOREPORT prices API returned unexpected type: " f"{type(data).__name__}" ) return None result = [] for item in data: if not isinstance(item, dict): continue price = int(item.get("price", 0)) country = item.get("country", "نامشخص") country_range = str( item.get("range", item.get("country", "1")) ) stock_raw = item.get("count", "") in_stock = _parse_in_stock(stock_raw) result.append({ "country": country, "range": country_range, "service_id": "tg_noreport", "base_price": price, "final_price": int(price + (price * margin)), "in_stock": in_stock, }) logger.info( f"TG NOREPORT countries loaded: {len(result)}" ) return result except Exception as exc: logger.error( f"FATAL ERROR in get_tg_noreport_countries: {exc}", exc_info=True, ) return None async def get_countries(service_id) -> Optional[list[dict]]: """Fetch countries for a service and apply the live profit margin. Every HTTP response is read as RAW TEXT and logged BEFORE parsing, so a non-JSON body is visible in the logs instead of crashing silently. Returns: List of dicts: {"country", "range", "service_id", "base_price", "final_price", "in_stock"} or ``None`` on failure. """ try: # Dynamic profit margin — live from DB, fallback to config. from database.db import get_profit_margin margin = await get_profit_margin(config.ACCOUNT_PROFIT_MARGIN_PERCENT) / 100 timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: url_prices = f"https://api.ozvinoo.xyz/web/{TOKEN}/get-prices/{service_id}" async with session.get(url_prices) as resp: raw_prices = await resp.text() logger.debug(f"RAW PRICES RESPONSE: {raw_prices[:1000]}") countries_data = json.loads(raw_prices) result = [] if isinstance(countries_data, list): for item in countries_data: if not isinstance(item, dict): continue base_price = int(item.get("price", 0)) # Use 'range' field for callback_data (short prefix code) # Falls back to country name if range is missing country_range = str(item.get("range", item.get("country", "1"))) result.append({ "country": item.get("country", "نامشخص"), "range": country_range, "service_id": service_id, "base_price": base_price, "final_price": int(base_price + (base_price * margin)), "in_stock": _parse_in_stock(item.get("count")), }) return result except Exception as exc: logger.error(f"FATAL ERROR in get_countries: {exc}", exc_info=True) return None def _normalize_order(data) -> Optional[dict]: """Normalize ANY Kalino ``getNumber`` response into a stable dict. Returns a dict with the keys: {"request_id", "number", "price", "error_code", "error_msg", "success"} Supported success shapes: {"success": true, "status": 1, "request_id": 123, "number": "...", "price": 0.5} {"request_id": 123, "phone": "..."} {"data": {"id": 123, "phone": "...", "price": 0.5}} Supported error shapes: {"success": false, "error_code": "no_number", "error_msg": "..."} {"error_code": "no_number"} {"status": false} Returns ``None`` for totally unparseable payloads / network failures. """ if data is None: return None if not isinstance(data, dict): return {"success": False, "error_code": "bad_response", "error_msg": f"Unexpected response type: {type(data).__name__}", "request_id": None, "number": None, "price": None} success = True if "success" in data: success = bool(data["success"]) if "status" in data and isinstance(data["status"], bool): success = success and data["status"] request_id = (data.get("request_id") or data.get("id") or data.get("activation_id") or data.get("order_id")) number = data.get("number") or data.get("phone") or data.get("mobile") price = data.get("price") or data.get("cost") or data.get("amount") error_code = data.get("error_code") error_msg = data.get("error_msg") or data.get("error_message") or data.get("message") # Nested payloads (e.g. {"data": {...}}) if isinstance(data.get("data"), dict): nested = data["data"] request_id = request_id or nested.get("request_id") or nested.get("id") \ or nested.get("activation_id") or nested.get("order_id") number = number or nested.get("number") or nested.get("phone") or nested.get("mobile") price = price or nested.get("price") or nested.get("cost") or nested.get("amount") error_code = error_code or nested.get("error_code") error_msg = error_msg or nested.get("error_msg") or nested.get("message") # A number without request_id is NOT a real order (race / failure). if success and (request_id is None or number is None): # Some panels return only one of the two; treat a missing number as a # failed purchase so the fail-safe refund can trigger. if error_code is None: error_code = "no_number" if number is None else "no_request" success = False try: price = float(price) if price is not None else None except (TypeError, ValueError): price = None return { "success": success, "request_id": str(request_id) if request_id is not None else None, "number": str(number) if number is not None else None, "price": price, "error_code": error_code, "error_msg": error_msg, } async def buy_virtual_number(service_id, country: str) -> Optional[dict]: """Buy a virtual number for a service and a country. 'country' must be the exact range/country value returned from the API (e.g. "لهستان 🇵🇱" or its short range code). Returns a NORMALIZED dict (see ``_normalize_order``) — never raises: Success → {"success": True, "request_id": ..., "number": ..., ...} Failure → {"success": False, "error_code": ..., "error_msg": ...} Network → None """ if str(service_id) == "tg_noreport": # Non-report Telegram numbers use the V2 telegram-numbers API. order = await _buy_tg_noreport(country) if order: return order # Fall through to the standard V1 endpoint as a last resort. url = f"https://api.ozvinoo.xyz/web/{TOKEN}/getNumber/{service_id}/{country}" try: timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url) as resp: raw = await resp.text() logger.debug(f"RAW KALINO getNumber [{resp.status}]: {raw[:600]}") data = json.loads(raw) return _normalize_order(data) except Exception as exc: logger.error("Kalino getNumber failed: %s", exc) return None async def _buy_tg_noreport(country: str) -> Optional[dict]: """Buy a non-report Telegram number via the V2 API (best effort).""" try: url = "https://api.ozvinoo.xyz/telegram-numbers/numbers/" headers = { "Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json", } payload = {"none_report": True, "country": country} timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(url, headers=headers, json=payload) as resp: raw = await resp.text() logger.debug( f"RAW KALINO tg_noreport buy [{resp.status}]: {raw[:600]}" ) if resp.status != 200: return None data = json.loads(raw) return _normalize_order(data) except Exception as exc: logger.error("Kalino tg_noreport buy failed: %s", exc) return None def _extract_code(data) -> tuple[Optional[str], bool]: """Extract the SMS code from a Kalino ``getCode`` response. Returns ``(code, waiting)``: • ("12345", False) → code received • (None, True) → still waiting / pending • (None, False) → terminal failure (no more code coming) """ if data is None or not isinstance(data, dict): return None, False code = (data.get("code") or data.get("sms_code") or data.get("sms") or data.get("verification_code")) if isinstance(data.get("data"), dict): nested_code = (data["data"].get("code") or data["data"].get("sms_code") or data["data"].get("sms") or data["data"].get("verification_code")) code = code or nested_code if code is not None and str(code).strip() not in ("", "0"): return str(code).strip(), False # Still-waiting signals documented by the Kalino panel. error_code = str(data.get("error_code") or "").lower() error_msg = str(data.get("error_msg") or data.get("message") or "").lower() status = data.get("status") status_str = str(status).lower() if status is not None else "" waiting_signals = ("wait", "pending", "not_ready", "no_code_yet", "in_progress", "processing", "نرسیده") for signal in waiting_signals: if signal in error_code or signal in error_msg or signal in status_str: return None, True if data.get("success") is False: wait_str = ("wait" in error_msg) or ("طول" in error_msg) \ or ("ندار" in error_msg) or ("در انتظار" in error_msg) return None, wait_str # Neither code nor a terminal error — assume still waiting to be safe. return None, True async def get_number_code(request_id) -> Optional[dict]: """Fetch the SMS verification code for an ordered number. Returns a NORMALIZED dict: {"code": "12345", "waiting": False, "status": "...", "raw": {...}} {"code": None, "waiting": True, "status": "...", "raw": {...}} → ``None`` only on connection errors. """ url = f"https://api.ozvinoo.xyz/web/{TOKEN}/getCode/{request_id}" try: timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url) as resp: raw = await resp.text() logger.debug(f"RAW KALINO getCode [{resp.status}]: {raw[:600]}") data = json.loads(raw) code, waiting = _extract_code(data) return { "code": code, "waiting": waiting, "status": data.get("status") if isinstance(data, dict) else None, "raw": data if isinstance(data, dict) else {}, } except Exception as exc: logger.error("Kalino getCode failed: %s", exc) return None def _parse_cancel(data) -> Optional[dict]: """Parse a Kalino cancel response into a minimal success dict.""" if data is None: return None if not isinstance(data, dict): return {"success": False, "error_code": "bad_response", "error_msg": ""} success = True if "success" in data: success = bool(data["success"]) if isinstance(data.get("status"), bool): success = success and bool(data["status"]) if isinstance(data.get("status"), (int, float)): success = success and int(data["status"]) >= 1 return { "success": success, "error_code": data.get("error_code"), "error_msg": data.get("error_msg") or data.get("message") or "", } async def cancel_virtual_number(request_id) -> Optional[dict]: """Cancel an ordered number on the Kalino panel. Tries the standard panel cancel endpoints in order and tolerates failure — the caller ALWAYS refunds the local wallet regardless, so a cancelled order never leaves the user paying for an unusable number. Returns a parsed dict with ``success`` (bool), or ``None`` on total failure. NEVER raises. """ from config import config as _cfg base = (_cfg.KALINO_BASE_URL or "https://api.ozvinoo.xyz/").rstrip("/") candidates = [ f"{base}/web/{TOKEN}/cancelOrder/{request_id}", f"{base}/web/{TOKEN}/cancelNumber/{request_id}", f"{base}/web/{TOKEN}/cancel/{request_id}", ] timeout = aiohttp.ClientTimeout(total=15) try: async with aiohttp.ClientSession(timeout=timeout) as session: for url in candidates: try: async with session.get(url) as resp: raw = await resp.text() if resp.status >= 500: continue logger.debug(f"KALINO cancel {url} [{resp.status}]: {raw[:300]}") try: data = json.loads(raw) except Exception: continue if isinstance(data, dict): return _normalize_order(data) except Exception as exc: logger.warning("Kalino cancel attempt failed (%s): %s", url, exc) continue # Last resort: some panels expose cancel as a POST. async with aiohttp.ClientSession(timeout=timeout) as session: url = f"{base}/web/{TOKEN}/cancelOrder/{request_id}" try: async with session.post(url) as resp: raw = await resp.text() data = json.loads(raw) if isinstance(data, dict): return _normalize_order(data) except Exception as exc: logger.warning("Kalino cancel POST failed: %s", exc) except Exception as exc: logger.error("Kalino cancel_virtual_number failed: %s", exc) return None # ══════════════════════════════════════════════════════════════════════ # CACHE INVALIDATION # ══════════════════════════════════════════════════════════════════════ def invalidate_cache() -> None: """Clear all cached API responses.""" global _cache, _cache_ts _cache.clear() _cache_ts = 0.0 logger.info("Kalino API cache invalidated.")