""" HTG Real-Time Stock & Reseller Intelligence Engine for SL CAT TOPUP Calculates dynamic stock counts for every product based on real-time HTG account balances, wholesale product rates, and auto-failover configurations. """ import json import logging import math import os import time from typing import Any, Dict, List, Optional, Tuple import requests from config import CLIENT_PRICES, HTG_ACCOUNTS_PATH, get_product_price from database import get_setting, set_setting from htg_engine import ( DEFAULT_HEADERS, PRODUCT_CATALOG, check_htg_balance, load_all_accounts, ) logger = logging.getLogger("htg_stock_manager") # Static default fallback rates for HTG Digital (HTG Wholesale Cost in LKR) # Synchronized with live HTG topup.php & levelup.php wholesale rate endpoints DEFAULT_HTG_RATES = { # Memberships "1": 575.0, # Weekly Membership "2": 2785.0, # Monthly Membership "3": 130.0, # Weekly Lite Membership # Diamonds "6": 120.0, # 25 Diamonds "7": 350.0, # 100 Diamonds "8": 1000.0, # 310 Diamonds "9": 1700.0, # 520 Diamonds "10": 3300.0, # 1060 Diamonds "11": 6500.0, # 2180 Diamonds "12": 15750.0, # 5600 Diamonds "13": 33000.0, # 11500 Diamonds # Level Up Pass "86": 90.0, # level up pass 6 "87": 195.0, # level up pass 10 "88": 195.0, # level up pass 15 "89": 195.0, # level up pass 20 "90": 195.0, # level up pass 25 "91": 310.0, # level up pass 30 } # Sub-item definitions for composite bundles BUNDLE_MAP = { "bundle_vip": [("1", 1), ("2", 1)], "bundle_svip": [("1", 4), ("2", 1)], "bundle_vip_max": [("1", 1), ("2", 1), ("3", 1)], "bundle_svip_max": [("1", 4), ("2", 1), ("3", 4)], } # In-memory cache for stock calculations and live rates _STOCK_CACHE = { "rates": {}, "statuses": {}, "rates_last_fetched": 0, "stock_data": None, "stock_last_calculated": 0, "ttl_seconds": 15, # Real-time responsiveness: 15 seconds TTL } def invalidate_stock_cache(): """Forces cache eviction so next request recalculates immediately.""" _STOCK_CACHE["stock_data"] = None _STOCK_CACHE["stock_last_calculated"] = 0 _STOCK_CACHE["rates_last_fetched"] = 0 def fetch_live_htg_rates(force: bool = False) -> Tuple[Dict[str, float], Dict[str, str]]: """ Fetches wholesale rates & item statuses from HTG Digital. Tries levelup.php?action=rates (which contains full catalog including levelup passes), with fallback to DEFAULT_HTG_RATES. """ now = time.time() if not force and _STOCK_CACHE["rates"] and (now - _STOCK_CACHE["rates_last_fetched"] < 120): return _STOCK_CACHE["rates"], _STOCK_CACHE["statuses"] rates = dict(DEFAULT_HTG_RATES) statuses = {} accounts_data = load_all_accounts() active_key = accounts_data.get("active_account", "account_1") cookies = accounts_data.get("accounts", {}).get(active_key, {}).get("cookies", {}) try: session = requests.Session() session.headers.update(DEFAULT_HEADERS) if cookies: session.cookies.update(cookies) url = "https://htgtopup.com/levelup.php?action=rates" resp = session.get(url, timeout=5) if resp.status_code == 200 and resp.text.startswith("{"): data = resp.json() if data.get("status") == "ok" and isinstance(data.get("rates"), dict): for k, v in data["rates"].items(): try: rates[str(k)] = float(v) except (ValueError, TypeError): pass if isinstance(data.get("statuses"), dict): statuses = {str(k): str(v).lower() for k, v in data["statuses"].items()} except Exception as e: logger.warning(f"Could not reach HTG rates API, using verified default rates: {e}") # Compute composite bundle rates dynamically for bundle_id, sub_items in BUNDLE_MAP.items(): total_bundle_cost = 0.0 for sub_id, sub_qty in sub_items: sub_cost = rates.get(sub_id, DEFAULT_HTG_RATES.get(sub_id, 0.0)) total_bundle_cost += sub_cost * sub_qty rates[bundle_id] = round(total_bundle_cost, 2) _STOCK_CACHE["rates"] = rates _STOCK_CACHE["statuses"] = statuses _STOCK_CACHE["rates_last_fetched"] = now return rates, statuses def get_htg_wholesale_cost(product_id: str) -> float: """Returns the wholesale cost (in LKR) charged by HTG for a given product.""" rates, _ = fetch_live_htg_rates() pid = str(product_id) if pid in rates: return rates[pid] if pid in DEFAULT_HTG_RATES: return DEFAULT_HTG_RATES[pid] if pid in BUNDLE_MAP: total = 0.0 for sub_id, qty in BUNDLE_MAP[pid]: total += get_htg_wholesale_cost(sub_id) * qty return round(total, 2) return 0.0 def get_stock_configuration() -> Dict[str, Any]: """Retrieves admin stock engine settings from database.""" mode = get_setting("htg_stock_mode", "max_single") # "max_single" ensures order fits in a single HTG account; "combined", "active_only" safety_reserve = float(get_setting("htg_stock_safety_reserve", "0.0") or 0.0) auto_sync = get_setting("htg_stock_auto_sync", "true").lower() in ("true", "1", "yes") raw_overrides = get_setting("htg_stock_overrides", "{}") try: overrides = json.loads(raw_overrides) if raw_overrides else {} except Exception: overrides = {} return { "mode": mode, "safety_reserve": safety_reserve, "auto_sync": auto_sync, "overrides": overrides, } def save_stock_configuration(mode: str, safety_reserve: float, overrides: Optional[Dict[str, Any]] = None): """Saves stock engine settings and invalidates cache.""" set_setting("htg_stock_mode", str(mode)) set_setting("htg_stock_safety_reserve", str(float(safety_reserve))) if overrides is not None: set_setting("htg_stock_overrides", json.dumps(overrides)) invalidate_stock_cache() def get_pending_reserved_balance(exclude_order_code: Optional[str] = None) -> float: """ Calculates total HTG wholesale balance currently reserved by active 15-minute LankaQR orders, dedicated quantity reservations, and orders in 'processing' queue. Prevents race conditions where two customers purchase simultaneously from the same shared pool. Accepts exclude_order_code to avoid self-reservation deadlocks when paying for an existing order. """ from database import get_connection import datetime reserved_cost = 0.0 now = datetime.datetime.now() cutoff_time = (now - datetime.timedelta(minutes=15)).strftime("%Y-%m-%d %H:%M:%S") try: with get_connection() as conn: c = conn.cursor() if exclude_order_code: c.execute(""" SELECT order_code, product_id, quantity, items_json, status, created_at FROM orders WHERE (((status IN ('payment_submitted', 'pending_payment', 'reserved')) AND created_at >= ?) OR (status = 'processing')) AND order_code != ? """, (cutoff_time, str(exclude_order_code).strip())) else: c.execute(""" SELECT order_code, product_id, quantity, items_json, status, created_at FROM orders WHERE ((status IN ('payment_submitted', 'pending_payment', 'reserved')) AND created_at >= ?) OR (status = 'processing') """, (cutoff_time,)) rows = c.fetchall() for r in rows: items_json = r["items_json"] if items_json: try: cart_items = json.loads(items_json) for itm in cart_items: p_id = itm.get("product_id") or itm.get("id") qty = int(itm.get("quantity") or 1) reserved_cost += get_htg_wholesale_cost(p_id) * qty except Exception: pass else: p_id = r["product_id"] qty = int(r["quantity"] or 1) if p_id and p_id != "wallet_deposit": reserved_cost += get_htg_wholesale_cost(p_id) * qty except Exception as e: logger.warning(f"Error calculating pending reserved balance: {e}") return round(reserved_cost, 2) def get_htg_accounts_balance_summary(exclude_order_code: Optional[str] = None) -> Dict[str, Any]: """ Summarizes balances across all configured HTG accounts. Returns: { "active_account": "account_1", "active_balance": float, "total_combined_balance": float, "max_single_balance": float, "reserved_balance": float, "effective_combined_balance": float, "effective_active_balance": float, "effective_max_single_balance": float, "accounts": list_of_accounts, "auto_failover": bool } """ data = load_all_accounts() active_key = data.get("active_account", "account_1") auto_failover = data.get("auto_failover", True) total_combined = 0.0 max_single = 0.0 active_bal = 0.0 acc_list = [] for acc_id, acc in data.get("accounts", {}).items(): enabled = acc.get("enabled", True) bal = float(acc.get("balance", 0.0) or 0.0) status = acc.get("status", "unconfigured") name = acc.get("name", acc_id) if enabled: total_combined += bal if bal > max_single: max_single = bal if acc_id == active_key: active_bal = bal acc_list.append({ "id": acc_id, "name": name, "balance": bal, "balance_formatted": f"Rs. {bal:,.2f}", "enabled": enabled, "is_active": (acc_id == active_key), "status": status, }) reserved_balance = get_pending_reserved_balance(exclude_order_code=exclude_order_code) effective_combined = max(0.0, total_combined - reserved_balance) effective_active = max(0.0, active_bal - reserved_balance) effective_max_single = max(0.0, max_single - reserved_balance) return { "active_account": active_key, "active_balance": active_bal, "total_combined_balance": total_combined, "max_single_balance": max_single, "reserved_balance": reserved_balance, "effective_combined_balance": effective_combined, "effective_active_balance": effective_active, "effective_max_single_balance": effective_max_single, "accounts": acc_list, "auto_failover": auto_failover, } def calculate_available_stock_for_product( product_id: str, wholesale_cost: float, accounts_summary: Dict[str, Any], config: Dict[str, Any], htg_status: str = "active", ) -> Dict[str, Any]: """ Advanced mathematical calculation of real-time stock for a specific product. Formula: In Combined Mode (with failover): floor(max(0, effective_combined_balance - safety_reserve) / wholesale_cost) In Active-Only Mode: floor(max(0, effective_active_balance - safety_reserve) / wholesale_cost) In Max-Single Mode: floor(max(0, effective_max_single_balance - safety_reserve) / wholesale_cost) """ pid = str(product_id) # Check if item is flagged under maintenance (Level Up Pass products: 86-91) LEVEL_UP_PASS_IDS = {"86", "87", "88", "89", "90", "91"} catalog_prod = next((p for p in PRODUCT_CATALOG if str(p["id"]) == pid), None) overrides = config.get("overrides", {}) if ( pid in LEVEL_UP_PASS_IDS or (catalog_prod and catalog_prod.get("maintenance")) or (pid in overrides and (overrides[pid].get("maintenance") or overrides[pid].get("status") == "maintenance")) ): return { "stock": 0, "status": "maintenance", "badge_text": "🛠️ Maintenance", "badge_sinhala": "🛠️ නඩත්තු වෙමින් පවතී", "badge_class": "badge-maintenance", "badge_color": "#f59e0b", "is_available": False, "is_maintenance": True, "wholesale_cost": wholesale_cost, } if pid in overrides: override_val = overrides[pid] if isinstance(override_val, dict): if override_val.get("force_out_of_stock"): return { "stock": 0, "status": "out_of_stock", "badge_text": "Out of Stock", "badge_class": "badge-out-of-stock", "badge_color": "#ef4444", "is_available": False, "is_manual_override": True, } if "manual_stock" in override_val: man_stock = max(0, int(override_val["manual_stock"])) return { "stock": man_stock, "status": "in_stock" if man_stock > 3 else ("low_stock" if man_stock > 0 else "out_of_stock"), "badge_text": f"{man_stock} In Stock" if man_stock > 0 else "Out of Stock", "badge_class": "badge-in-stock" if man_stock > 3 else "badge-low-stock", "badge_color": "#10b981" if man_stock > 3 else ("#f59e0b" if man_stock > 0 else "#ef4444"), "is_available": man_stock > 0, "is_manual_override": True, } # If HTG marked the item inactive on their store if htg_status in ["inactive", "disabled", "maintenance"]: return { "stock": 0, "status": "inactive", "badge_text": "Temporarily Unavailable", "badge_class": "badge-inactive", "badge_color": "#94a3b8", "is_available": False, "is_inactive": True, } if wholesale_cost <= 0: return { "stock": 0, "status": "out_of_stock", "badge_text": "Unavailable", "badge_class": "badge-out-of-stock", "badge_color": "#ef4444", "is_available": False, } reserve = float(config.get("safety_reserve", 0.0) or 0.0) mode = config.get("mode", "combined") stock = 0 if mode == "combined": avail_bal = max(0.0, accounts_summary.get("effective_combined_balance", 0.0) - reserve) stock = math.floor(avail_bal / wholesale_cost) elif mode == "active_only": avail_bal = max(0.0, accounts_summary.get("effective_active_balance", 0.0) - reserve) stock = math.floor(avail_bal / wholesale_cost) elif mode == "max_single": avail_bal = max(0.0, accounts_summary.get("effective_max_single_balance", 0.0) - reserve) stock = math.floor(avail_bal / wholesale_cost) else: avail_bal = max(0.0, accounts_summary.get("effective_combined_balance", 0.0) - reserve) stock = math.floor(avail_bal / wholesale_cost) stock = max(0, int(stock)) # Determine visual badge and state if stock == 0: status = "out_of_stock" badge_text = "❌ Out of Stock" badge_sinhala = "තොග අවසන්" badge_class = "badge-out-of-stock" badge_color = "#ef4444" is_available = False elif 1 <= stock <= 3: status = "low_stock" badge_text = f"⚡ Only {stock} Left!" badge_sinhala = f"ඉතිරිව ඇත්තේ {stock}ක් පමණි!" badge_class = "badge-low-stock" badge_color = "#f59e0b" is_available = True else: status = "in_stock" badge_text = f"🟢 In Stock ({stock})" badge_sinhala = f"තොග ඇත ({stock})" badge_class = "badge-in-stock" badge_color = "#10b981" is_available = True return { "stock": stock, "status": status, "badge_text": badge_text, "badge_sinhala": badge_sinhala, "badge_class": badge_class, "badge_color": badge_color, "is_available": is_available, "wholesale_cost": wholesale_cost, } def get_all_products_live_stock(force_refresh: bool = False, exclude_order_code: Optional[str] = None) -> Dict[str, Any]: """ Returns full stock and reseller intelligence directory for all catalog products. Uses 15-second TTL cache for high performance. If exclude_order_code is provided, performs dynamic calculation bypassing global cache. """ now = time.time() if ( not force_refresh and not exclude_order_code and _STOCK_CACHE["stock_data"] and (now - _STOCK_CACHE["stock_last_calculated"] < _STOCK_CACHE["ttl_seconds"]) ): return _STOCK_CACHE["stock_data"] rates, statuses = fetch_live_htg_rates(force=force_refresh) accounts_summary = get_htg_accounts_balance_summary(exclude_order_code=exclude_order_code) config = get_stock_configuration() products_stock = {} in_stock_count = 0 low_stock_count = 0 out_of_stock_count = 0 for prod in PRODUCT_CATALOG: pid = str(prod["id"]) cost = get_htg_wholesale_cost(pid) htg_status = statuses.get(pid, "active") stock_info = calculate_available_stock_for_product( product_id=pid, wholesale_cost=cost, accounts_summary=accounts_summary, config=config, htg_status=htg_status, ) selling_price = get_product_price(pid) net_profit = round(selling_price - cost, 2) profit_percent = round((net_profit / cost * 100), 1) if cost > 0 else 0.0 if stock_info["status"] == "in_stock": in_stock_count += 1 elif stock_info["status"] == "low_stock": low_stock_count += 1 else: out_of_stock_count += 1 products_stock[pid] = { "id": pid, "name": prod["name"], "category": prod["category"], "diamonds": prod.get("diamonds", ""), "selling_price": selling_price, "selling_price_formatted": f"Rs. {selling_price:,.2f}", "htg_cost": cost, "htg_cost_formatted": f"Rs. {cost:,.2f}", "net_profit": net_profit, "profit_percent": profit_percent, "stock": stock_info["stock"], "status": stock_info["status"], "badge_text": stock_info["badge_text"], "badge_sinhala": stock_info.get("badge_sinhala", ""), "badge_class": stock_info["badge_class"], "badge_color": stock_info["badge_color"], "is_available": stock_info["is_available"], } result = { "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "config": config, "accounts_summary": accounts_summary, "summary": { "total_items": len(PRODUCT_CATALOG), "in_stock_count": in_stock_count, "low_stock_count": low_stock_count, "out_of_stock_count": out_of_stock_count, "total_balance_lkr": accounts_summary["total_combined_balance"], "effective_balance_lkr": accounts_summary.get("effective_combined_balance", accounts_summary["total_combined_balance"]), "reserved_balance_lkr": accounts_summary.get("reserved_balance", 0.0), }, "products": products_stock, } if not exclude_order_code: _STOCK_CACHE["stock_data"] = result _STOCK_CACHE["stock_last_calculated"] = now return result def validate_order_stock( product_id: Optional[str] = None, quantity: int = 1, items: Optional[List[Dict[str, Any]]] = None, exclude_order_code: Optional[str] = None, ) -> Tuple[bool, str, Dict[str, Any]]: """ Validates if customer order quantity can be fulfilled with currently available stock. Returns: (is_valid: bool, error_message: str, stock_details: dict) """ live_data = get_all_products_live_stock(force_refresh=bool(exclude_order_code), exclude_order_code=exclude_order_code) products_stock = live_data["products"] summary = live_data.get("summary", {}) effective_bal = float(summary.get("effective_balance_lkr", 0.0) or 0.0) in_stock_cnt = int(summary.get("in_stock_count", 0) or 0) total_items = int(summary.get("total_items", 0) or 0) try: from database import get_setting threshold = float(get_setting("low_balance_threshold", "3000.0") or 3000.0) except Exception: threshold = 3000.0 if (effective_bal < threshold) or (in_stock_cnt == 0 and total_items > 0): return ( False, "සමාවන්න! සජීවී Server Delivery Wallet ශේෂය නැවත Recharge වෙමින් පවතී. මෙම නිවේදනය ස්වයංක්‍රීයව ඉවත් වන තුරු (Auto-Clear වන තුරු) කරුණාකර නව Top-Up ඇණවුම් දැමීමෙන් වළකින්න. සර්වර් ශේෂය යථා තත්ත්වයට පත් වූ සැණින් ඇණවුම් භාරගැනීම ස්වයංක්‍රීයව නැවත ආරම්භ වේ.", {"is_low_balance": True, "effective_balance": effective_bal, "threshold": threshold}, ) # 1. Multi-cart items check if items and isinstance(items, list) and len(items) > 0: total_cart_htg_cost = 0.0 for itm in items: pid = str(itm.get("product_id") or itm.get("id") or "").strip() qty = max(1, int(itm.get("quantity") or itm.get("qty") or 1)) stock_entry = products_stock.get(pid) if not stock_entry: continue available = stock_entry["stock"] name = stock_entry["name"] cost = get_htg_wholesale_cost(pid) total_cart_htg_cost += cost * qty if stock_entry.get("status") == "maintenance": return ( False, f"🛠️ සමාවන්න! '{name}' මේ වන විට නඩත්තු වෙමින් පවතී (Under Maintenance). කරුණාකර එය Cart එකෙන් ඉවත් කරන්න.", {"product_id": pid, "status": "maintenance"}, ) if available <= 0: return ( False, f"සමාවන්න! '{name}' සඳහා දැනට සජීවී තොග අවසන්ව ඇත (Out of Stock). කරුණාකර එය Cart එකෙන් ඉවත් කරන්න.", {"product_id": pid, "available_stock": 0, "requested_quantity": qty}, ) if qty > available: return ( False, f"සමාවන්න! '{name}' සඳහා දැනට ඇත්තේ තොග {available}ක් පමණි (ඉල්ලුම් කළ ප්‍රමාණය: {qty}). කරුණාකර ප්‍රමාණය අඩු කරන්න.", {"product_id": pid, "available_stock": available, "requested_quantity": qty}, ) # Shared Pool Check: Ensure combined wholesale cost does not exceed effective balance summary = live_data.get("summary", {}) effective_bal = float(summary.get("effective_balance_lkr", 0.0) or 0.0) if effective_bal > 0 and total_cart_htg_cost > (effective_bal + 0.5): return ( False, f"සමාවන්න! ඔබ තෝරාගත් සියලුම Packages වල එකතුව සඳහා අපගේ සජීවී Server තොග සීමාව ඉක්මවා ඇත. කරුණාකර සමහර Items වල ප්‍රමාණය අඩු කරන්න.", {"total_cart_cost": total_cart_htg_cost, "effective_balance": effective_bal}, ) return True, "", {} # 2. Single item check if product_id: pid = str(product_id).strip() stock_entry = products_stock.get(pid) if not stock_entry: return True, "", {} available = stock_entry["stock"] name = stock_entry["name"] if stock_entry.get("status") == "maintenance": return ( False, f"🛠️ සමාවන්න! '{name}' මේ වන විට නඩත්තු වෙමින් පවතී (Under Maintenance). කරුණාකර වෙනත් Package එකක් තෝරන්න.", {"product_id": pid, "status": "maintenance"}, ) if available <= 0: return ( False, f"සමාවන්න! '{name}' සඳහා දැනට සජීවී තොග අවසන්ව ඇත (Out of Stock). නව තොග පැමිණි වහාම සක්‍රිය වේ.", {"product_id": pid, "available_stock": 0, "requested_quantity": quantity}, ) if quantity > available: return ( False, f"සමාවන්න! '{name}' සඳහා දැනට ඇත්තේ තොග {available}ක් පමණි (ඉල්ලුම් කළ ප්‍රමාණය: {quantity}). කරුණාකර ප්‍රමාණය {available}ක් හෝ ඊට අඩුවෙන් තෝරන්න.", {"product_id": pid, "available_stock": available, "requested_quantity": quantity}, ) return True, "", {} def force_sync_htg_and_refresh_stock() -> Dict[str, Any]: """ Manually forces a live wallet balance query to HTG for all accounts, fetches latest wholesale rates, and recalculates all product stocks. """ load_all_accounts() # Refresh live balance from HTG from htg_engine import refresh_all_accounts_balance try: refresh_all_accounts_balance() except Exception as e: logger.warning(f"Error refreshing HTG balance during stock sync: {e}") invalidate_stock_cache() return get_all_products_live_stock(force_refresh=True)