Spaces:
Running on Zero
Running on Zero
| """ | |
| Token Vault & Economy Engine for Nova AI (ChatGPT Astra Edition). | |
| Manages user token balances, deductions, daily claims, API key verification, | |
| and persistent transaction history. | |
| """ | |
| import os | |
| import sys | |
| import time | |
| import json | |
| import secrets | |
| from pathlib import Path | |
| from typing import Dict, Any, List, Optional, Tuple | |
| PROJECT_ROOT = Path(__file__).resolve().parent.parent | |
| DATA_DIR = PROJECT_ROOT / "data" | |
| TOKEN_LEDGER_FILE = DATA_DIR / "user_tokens.json" | |
| # Token Costs per Action | |
| TOKEN_COSTS = { | |
| "chat_query": 2, # Standard conversational chat | |
| "deep_reasoning": 5, # Astra Deep Reasoning (Chain-of-Thought) | |
| "web_search": 3, # Live Web Search & Grounding | |
| "image_generation": 15, # AI Image synthesis | |
| "video_generation": 30, # AI Video synthesis | |
| "website_generation": 20, # Full-Stack Website builder | |
| "prompt_spark": 1 # Astra Spark prompt optimizer | |
| } | |
| DEFAULT_STARTING_BALANCE = 1000 | |
| DAILY_BONUS_AMOUNT = 250 | |
| class TokenVault: | |
| """Thread-safe persistent token manager for Nova AI.""" | |
| def __init__(self, ledger_file: Optional[Path] = None): | |
| self.ledger_file = ledger_file or TOKEN_LEDGER_FILE | |
| self._ensure_storage() | |
| def _ensure_storage(self): | |
| """Ensure storage directory and ledger JSON exist.""" | |
| DATA_DIR.mkdir(parents=True, exist_ok=True) | |
| if not self.ledger_file.exists(): | |
| default_data = { | |
| "users": { | |
| "default_user": { | |
| "balance": DEFAULT_STARTING_BALANCE, | |
| "api_key": f"nova_astra_{secrets.token_hex(16)}", | |
| "tier": "Astra Free Pioneer", | |
| "tier_id": "free", | |
| "last_daily_claim": 0, | |
| "created_at": time.time(), | |
| "transactions": [ | |
| { | |
| "id": f"tx_{secrets.token_hex(6)}", | |
| "timestamp": time.time(), | |
| "date": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| "action": "welcome_bonus", | |
| "amount": DEFAULT_STARTING_BALANCE, | |
| "type": "credit", | |
| "balance_after": DEFAULT_STARTING_BALANCE, | |
| "description": "Welcome to Nova AI! 1,000 Astra Starter Tokens granted." | |
| } | |
| ] | |
| } | |
| } | |
| } | |
| self._save_ledger(default_data) | |
| def _load_ledger(self) -> Dict[str, Any]: | |
| """Load the user ledger from disk.""" | |
| try: | |
| with open(self.ledger_file, "r", encoding="utf-8") as f: | |
| return json.load(f) | |
| except Exception: | |
| return {"users": {}} | |
| def _save_ledger(self, data: Dict[str, Any]): | |
| """Save ledger to disk safely.""" | |
| with open(self.ledger_file, "w", encoding="utf-8") as f: | |
| json.dump(data, f, indent=2, ensure_ascii=False) | |
| def get_user_data(self, user_id: str = "default_user") -> Dict[str, Any]: | |
| """Retrieve user ledger profile, initializing if absent.""" | |
| data = self._load_ledger() | |
| if user_id not in data.get("users", {}): | |
| data.setdefault("users", {})[user_id] = { | |
| "balance": DEFAULT_STARTING_BALANCE, | |
| "api_key": f"nova_astra_{secrets.token_hex(16)}", | |
| "tier": "Astra Free Pioneer", | |
| "tier_id": "free", | |
| "last_daily_claim": 0, | |
| "created_at": time.time(), | |
| "transactions": [ | |
| { | |
| "id": f"tx_{secrets.token_hex(6)}", | |
| "timestamp": time.time(), | |
| "date": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| "action": "welcome_bonus", | |
| "amount": DEFAULT_STARTING_BALANCE, | |
| "type": "credit", | |
| "balance_after": DEFAULT_STARTING_BALANCE, | |
| "description": "Initial Astra starter tokens granted." | |
| } | |
| ] | |
| } | |
| self._save_ledger(data) | |
| return data["users"][user_id] | |
| def get_balance(self, user_id: str = "default_user") -> int: | |
| """Get current token balance.""" | |
| return self.get_user_data(user_id).get("balance", 0) | |
| def get_api_key(self, user_id: str = "default_user") -> str: | |
| """Get API key for mobile client connectivity.""" | |
| return self.get_user_data(user_id).get("api_key", "") | |
| def deduct_tokens(self, action: str, user_id: str = "default_user", custom_cost: Optional[int] = None) -> Tuple[bool, str, int]: | |
| """ | |
| Deduct tokens for an action. | |
| Returns: (success: bool, message: str, remaining_balance: int) | |
| """ | |
| cost = custom_cost if custom_cost is not None else TOKEN_COSTS.get(action, 2) | |
| data = self._load_ledger() | |
| user_info = self.get_user_data(user_id) | |
| current_balance = user_info.get("balance", 0) | |
| if current_balance < cost: | |
| return ( | |
| False, | |
| f"⚠️ Insufficient Astra Tokens! Action '{action}' requires {cost} tokens, but you only have {current_balance} tokens remaining. Claim your daily bonus or refill tokens.", | |
| current_balance | |
| ) | |
| new_balance = current_balance - cost | |
| user_info["balance"] = new_balance | |
| tx = { | |
| "id": f"tx_{secrets.token_hex(6)}", | |
| "timestamp": time.time(), | |
| "date": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| "action": action, | |
| "amount": -cost, | |
| "type": "debit", | |
| "balance_after": new_balance, | |
| "description": f"Used {cost} tokens for {action.replace('_', ' ').title()}" | |
| } | |
| user_info.setdefault("transactions", []).insert(0, tx) | |
| user_info["transactions"] = user_info["transactions"][:100] # Keep last 100 | |
| data["users"][user_id] = user_info | |
| self._save_ledger(data) | |
| return True, f"✓ {cost} tokens deducted for {action}. Balance: {new_balance}", new_balance | |
| def add_tokens(self, amount: int, reason: str = "Token Recharge", user_id: str = "default_user") -> int: | |
| """Add tokens to user account.""" | |
| data = self._load_ledger() | |
| user_info = self.get_user_data(user_id) | |
| current_balance = user_info.get("balance", 0) | |
| new_balance = current_balance + amount | |
| user_info["balance"] = new_balance | |
| tx = { | |
| "id": f"tx_{secrets.token_hex(6)}", | |
| "timestamp": time.time(), | |
| "date": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| "action": "recharge", | |
| "amount": amount, | |
| "type": "credit", | |
| "balance_after": new_balance, | |
| "description": reason | |
| } | |
| user_info.setdefault("transactions", []).insert(0, tx) | |
| user_info["transactions"] = user_info["transactions"][:100] | |
| data["users"][user_id] = user_info | |
| self._save_ledger(data) | |
| return new_balance | |
| def claim_daily_bonus(self, user_id: str = "default_user") -> Tuple[bool, str, int]: | |
| """Claim daily bonus tokens adapted to user's subscription tier.""" | |
| data = self._load_ledger() | |
| user_info = self.get_user_data(user_id) | |
| now = time.time() | |
| last_claim = user_info.get("last_daily_claim", 0) | |
| elapsed = now - last_claim | |
| cooldown = 86400 # 24 hours in seconds | |
| if elapsed < cooldown and last_claim > 0: | |
| remaining_hours = round((cooldown - elapsed) / 3600, 1) | |
| return ( | |
| False, | |
| f"⏳ Daily bonus already claimed! Next bonus available in {remaining_hours} hours.", | |
| user_info.get("balance", 0) | |
| ) | |
| tier_id = user_info.get("tier_id", "free") | |
| bonus_amount = 1000 if tier_id == "pro" else (5000 if tier_id == "enterprise" else DAILY_BONUS_AMOUNT) | |
| user_info["last_daily_claim"] = now | |
| new_balance = user_info.get("balance", 0) + bonus_amount | |
| user_info["balance"] = new_balance | |
| tx = { | |
| "id": f"tx_{secrets.token_hex(6)}", | |
| "timestamp": now, | |
| "date": time.strftime("%Y-%m-%d %H:%M:%S"), | |
| "action": "daily_claim", | |
| "amount": bonus_amount, | |
| "type": "credit", | |
| "balance_after": new_balance, | |
| "description": f"Claimed daily Astra reward (+{bonus_amount:,} tokens for {user_info.get('tier', 'Free')})" | |
| } | |
| user_info.setdefault("transactions", []).insert(0, tx) | |
| user_info["transactions"] = user_info["transactions"][:100] | |
| data["users"][user_id] = user_info | |
| self._save_ledger(data) | |
| return True, f"🎉 Claimed +{bonus_amount:,} Astra Tokens! Current Balance: {new_balance:,}", new_balance | |
| def get_transaction_history(self, user_id: str = "default_user") -> List[Dict[str, Any]]: | |
| """Return transaction records for display.""" | |
| return self.get_user_data(user_id).get("transactions", []) | |
| # Global shared vault instance | |
| token_vault = TokenVault() | |