# file: battlewords/game_storage.py """ BattleWords-specific storage wrapper for HuggingFace storage operations. This module provides high-level functions for saving and loading BattleWords games using the shared storage module from battlewords.modules. """ __version__ = "0.1.3" import json import tempfile import os from datetime import datetime, timezone from typing import Dict, Any, List, Optional, Tuple import logging from urllib.parse import unquote from battlewords.modules import ( upload_files_to_repo, gen_full_url, HF_REPO_ID, SHORTENER_JSON_FILE, SPACE_NAME ) from battlewords.modules.storage import _get_json_from_repo from battlewords.local_storage import save_json_to_file from battlewords.word_loader import compute_word_difficulties # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) def generate_uid() -> str: """ Generate a unique identifier for a game. Format: YYYYMMDDTHHMMSSZ-RANDOM Example: 20250123T153045Z-A7B9C2 Returns: str: Unique game identifier """ import random import string timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") random_suffix = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6)) return f"{timestamp}-{random_suffix}" def serialize_game_settings( word_list: List[str], username: str, score: int, time_seconds: int, game_mode: str, grid_size: int = 12, spacer: int = 1, may_overlap: bool = False, wordlist_source: Optional[str] = None, challenge_id: Optional[str] = None ) -> Dict[str, Any]: """ Serialize game settings into a JSON-compatible dictionary. Creates initial structure with one user's result. Each user has their own uid and word_list. Args: word_list: List of words used in THIS user's game username: Player's name score: Final score achieved time_seconds: Time taken to complete (in seconds) game_mode: Game mode ("classic" or "too_easy") grid_size: Grid size (default: 12) spacer: Word spacing configuration (0-2, default: 1) may_overlap: Whether words can overlap (default: False) wordlist_source: Source file name (e.g., "classic.txt") challenge_id: Optional challenge ID (generated if not provided) Returns: dict: Serialized game settings with users array """ if challenge_id is None: challenge_id = generate_uid() # Try compute difficulty using the source file; optional difficulty_value: Optional[float] = None try: if wordlist_source: words_dir = os.path.join(os.path.dirname(__file__), "words") wordlist_path = os.path.join(words_dir, wordlist_source) if os.path.exists(wordlist_path): total_diff, _ = compute_word_difficulties(wordlist_path, words_array=word_list) difficulty_value = float(total_diff) except Exception as _e: # optional field, swallow errors difficulty_value = None # Build user result with desired ordering: uid, username, word_list, word_list_difficulty, score, time, timestamp user_result = { "uid": generate_uid(), "username": username, "word_list": word_list, } if difficulty_value is not None: user_result["word_list_difficulty"] = difficulty_value user_result["score"] = score user_result["time"] = time_seconds user_result["timestamp"] = datetime.now(timezone.utc).isoformat() settings = { "challenge_id": challenge_id, "game_mode": game_mode, "grid_size": grid_size, "puzzle_options": { "spacer": spacer, "may_overlap": may_overlap }, "users": [user_result], "created_at": datetime.now(timezone.utc).isoformat(), "version": __version__ } if wordlist_source: settings["wordlist_source"] = wordlist_source return settings def add_user_result_to_game( sid: str, username: str, word_list: List[str], score: int, time_seconds: int, repo_id: Optional[str] = None ) -> bool: """ Add a user's result to an existing shared challenge. Each user gets their own uid and word_list. Args: sid: Short ID of the existing challenge username: Player's name word_list: List of words THIS user played score: Score achieved time_seconds: Time taken (seconds) repo_id: HF repository ID (uses HF_REPO_ID from env if None) Returns: bool: True if successfully added, False otherwise """ if repo_id is None: repo_id = HF_REPO_ID logger.info(f"➕ Adding user result to challenge {sid}") try: # Load existing game settings settings = load_game_from_sid(sid, repo_id) if not settings: logger.error(f"❌ Challenge not found: {sid}") return False # Compute optional difficulty using the saved wordlist_source if available difficulty_value: Optional[float] = None try: wordlist_source = settings.get("wordlist_source") if wordlist_source: words_dir = os.path.join(os.path.dirname(__file__), "words") wordlist_path = os.path.join(words_dir, wordlist_source) if os.path.exists(wordlist_path): total_diff, _ = compute_word_difficulties(wordlist_path, words_array=word_list) difficulty_value = float(total_diff) except Exception: difficulty_value = None # Create new user result with ordering and optional difficulty user_result = { "uid": generate_uid(), "username": username, "word_list": word_list, } if difficulty_value is not None: user_result["word_list_difficulty"] = difficulty_value user_result["score"] = score user_result["time"] = time_seconds user_result["timestamp"] = datetime.now(timezone.utc).isoformat() # Add to users array if "users" not in settings: settings["users"] = [] settings["users"].append(user_result) logger.info(f"👥 Now {len(settings['users'])} users in game") # Get the file path from the sid status, full_url = gen_full_url( short_url=sid, repo_id=repo_id, json_file=SHORTENER_JSON_FILE ) if status != "success_retrieved_full" or not full_url: logger.error(f"❌ Could not resolve sid: {sid}") return False # Extract challenge_id from URL url_parts = full_url.split("/resolve/main/") if len(url_parts) != 2: logger.error(f"❌ Invalid URL format: {full_url}") return False file_path = url_parts[1] # e.g., "games/{challenge_id}/settings.json" challenge_id = file_path.split("/")[1] # Extract challenge_id folder_name = f"games/{challenge_id}" # Save updated settings back to HF try: with tempfile.TemporaryDirectory() as tmpdir: settings_path = save_json_to_file(settings, tmpdir, "settings.json") logger.info(f"📤 Updating {folder_name}/settings.json") response = upload_files_to_repo( files=[settings_path], repo_id=repo_id, folder_name=folder_name, repo_type="dataset" ) logger.info(f"✅ User result added for {username}") return True except Exception as e: logger.error(f"❌ Failed to upload updated settings: {e}") return False except Exception as e: logger.error(f"❌ Failed to add user result: {e}") return False def save_game_to_hf( word_list: List[str], username: str, score: int, time_seconds: int, game_mode: str, grid_size: int = 12, spacer: int = 1, may_overlap: bool = False, repo_id: Optional[str] = None, wordlist_source: Optional[str] = None ) -> Tuple[str, Optional[str], Optional[str]]: """ Save game settings to HuggingFace repository and generate shareable URL. Creates a new game entry with the first user's result. This function: 1. Generates a unique UID for the game 2. Serializes game settings to JSON with first user 3. Uploads settings.json to HF repo under games/{uid}/ 4. Creates a shortened URL (sid) for sharing 5. Returns the full URL and short ID Args: word_list: List of words used in the game username: Player's name score: Final score achieved time_seconds: Time taken to complete (in seconds) game_mode: Game mode ("classic" or "too_easy") grid_size: Grid size (default: 12) spacer: Word spacing configuration (0-2, default: 1) may_overlap: Whether words can overlap (default: False) repo_id: HF repository ID (uses HF_REPO_ID from env if None) wordlist_source: Source wordlist file name (e.g., "classic.txt") Returns: tuple: (challenge_id, full_url, sid) where: - challenge_id: Unique challenge identifier - full_url: Full URL to settings.json - sid: Shortened ID for sharing (8 characters) Raises: Exception: If upload or URL shortening fails Example: >>> uid, full_url, sid = save_game_to_hf( ... word_list=["WORD", "TEST", "GAME", "PLAY", "SCORE", "FINAL"], ... username="Alice", ... score=42, ... time_seconds=180, ... game_mode="classic", ... wordlist_source="classic.txt" ... ) >>> print(f"Share: https://{SPACE_NAME}/?game_id={sid}") """ if repo_id is None: repo_id = HF_REPO_ID logger.info(f"💾 Saving game to HuggingFace repo: {repo_id}") # Generate challenge ID and serialize settings challenge_id = generate_uid() settings = serialize_game_settings( word_list=word_list, username=username, score=score, time_seconds=time_seconds, game_mode=game_mode, grid_size=grid_size, spacer=spacer, may_overlap=may_overlap, challenge_id=challenge_id, wordlist_source=wordlist_source ) logger.debug(f"🆔 Generated Challenge ID: {challenge_id}") # Write settings to a temp directory using a fixed filename 'settings.json' folder_name = f"games/{challenge_id}" try: with tempfile.TemporaryDirectory() as tmpdir: settings_path = save_json_to_file(settings, tmpdir, "settings.json") logger.info(f"📤 Uploading to {folder_name}/settings.json") # Upload to HF repo under games/{uid}/settings.json response = upload_files_to_repo( files=[settings_path], repo_id=repo_id, folder_name=folder_name, repo_type="dataset" ) # Construct full URL to settings.json full_url = f"https://huggingface.co/datasets/{repo_id}/resolve/main/{folder_name}/settings.json" logger.info(f"✅ Uploaded: {full_url}") # Generate short URL logger.info("🔗 Creating short URL...") status, sid = gen_full_url( full_url=full_url, repo_id=repo_id, json_file=SHORTENER_JSON_FILE ) if status in ["created_short", "success_retrieved_short", "exists_match"]: logger.info(f"✅ Short ID created: {sid}") share_url = f"https://{SPACE_NAME}/?game_id={sid}" logger.info(f"🎮 Share URL: {share_url}") return challenge_id, full_url, sid else: logger.warning(f"⚠️ URL shortening failed: {status}") return challenge_id, full_url, None except Exception as e: logger.error(f"❌ Failed to save game: {e}") raise def load_game_from_sid( sid: str, repo_id: Optional[str] = None ) -> Optional[Dict[str, Any]]: """ Load game settings from a short ID (sid). If settings.json cannot be found, return None and allow normal game loading. Args: sid: Short ID (8 characters) from shareable URL repo_id: HF repository ID (uses HF_REPO_ID from env if None) Returns: dict | None: Game settings or None if not found dict: Challenge settings containing: - challenge_id: Unique challenge identifier - wordlist_source: Source wordlist file (e.g., "classic.txt") - game_mode: Game mode - grid_size: Grid size - puzzle_options: Puzzle configuration (spacer, may_overlap) - users: Array of user results, each with: - uid: Unique user game identifier - username: Player name - word_list: Words THIS user played - score: Score achieved - time: Time taken (seconds) - timestamp: When result was recorded - created_at: When challenge was created - version: Storage version Returns None if sid not found or download fails Example: >>> settings = load_game_from_sid("abc12345") >>> if settings: ... print(f"Challenge ID: {settings['challenge_id']}") ... print(f"Wordlist: {settings['wordlist_source']}") ... for user in settings['users']: ... print(f"{user['username']}: {user['score']} pts") """ if repo_id is None: repo_id = HF_REPO_ID logger.info(f"🔍 Loading game from sid: {sid}") try: # Resolve sid to full URL status, full_url = gen_full_url( short_url=sid, repo_id=repo_id, json_file=SHORTENER_JSON_FILE ) if status != "success_retrieved_full" or not full_url: logger.warning(f"⚠️ Could not resolve sid: {sid} (status: {status})") return None logger.info(f"✅ Resolved to: {full_url}") # Extract the file path from the full URL # URL format: https://huggingface.co/datasets/{repo_id}/resolve/main/{path} # We need just the path part: games/{uid}/settings.json try: url_parts = full_url.split("/resolve/main/") if len(url_parts) != 2: logger.error(f"❌ Invalid URL format: {full_url}") return None file_path = url_parts[1] logger.info(f"📥 Downloading {file_path} using authenticated API...") settings = _get_json_from_repo(repo_id, file_path, repo_type="dataset") if not settings: logger.error(f"❌ settings.json not found for sid: {sid}. Loading normal game.") return None logger.info(f"✅ Loaded challenge: {settings.get('challenge_id', 'unknown')}") users = settings.get('users', []) logger.debug(f"Users in challenge: {len(users)}") return settings except Exception as e: logger.error(f"❌ Failed to parse URL or download: {e}") return None except Exception as e: logger.error(f"❌ Unexpected error loading game: {e}") return None def get_shareable_url(sid: str, base_url: str = None) -> str: """ Generate a shareable URL from a short ID. If running locally, use localhost. Otherwise, use HuggingFace Space domain. Additionally, if an "iframe_host" query parameter is present in the current Streamlit request, it takes precedence and will be used as the base URL. Args: sid: Short ID (8 characters) base_url: Optional override for the base URL (for testing or custom deployments) Returns: str: Full shareable URL Example: >>> url = get_shareable_url("abc12345") >>> print(url) https://surn-battlewords.hf.space/?game_id=abc12345 """ import os from battlewords.modules.constants import SPACE_NAME # 0) If not explicitly provided, try to read iframe_host from Streamlit query params if base_url is None: try: import streamlit as st # local import to avoid hard dependency params = getattr(st, "query_params", None) if params is None and hasattr(st, "experimental_get_query_params"): params = st.experimental_get_query_params() if params and "iframe_host" in params: raw_host = params.get("iframe_host") # st.query_params may return str or list[str] if isinstance(raw_host, (list, tuple)): raw_host = raw_host[0] if raw_host else None if raw_host: decoded = unquote(str(raw_host)) if decoded: base_url = decoded except Exception: # Ignore any errors here and fall back to defaults below pass # 1) If base_url is provided (either parameter or iframe_host), use it directly if base_url: sep = '&' if '?' in base_url else '?' return f"{base_url}{sep}game_id={sid}" if os.environ.get("IS_LOCAL", "true").lower() == "true": # 2) Check for local development (common Streamlit env vars) port = os.environ.get("PORT") or os.environ.get("STREAMLIT_SERVER_PORT") or "8501" host = os.environ.get("HOST") or os.environ.get("STREAMLIT_SERVER_ADDRESS") or "localhost" if host in ("localhost", "127.0.0.1") or os.environ.get("IS_LOCAL", "").lower() == "true": return f"http://{host}:{port}/?game_id={sid}" # 3) Otherwise, build HuggingFace Space URL from SPACE_NAME space = (SPACE_NAME or "surn/battlewords").lower().replace("/", "-") return f"https://{space}.hf.space/?game_id={sid}" if __name__ == "__main__": # Example usage print("BattleWords Game Storage Module") print(f"Version: {__version__}") print(f"Target Repository: {HF_REPO_ID}") print(f"Space Name: {SPACE_NAME}") # Example: Save a game print("\n--- Example: Save Game ---") try: challenge_id, full_url, sid = save_game_to_hf( word_list=["WORD", "TEST", "GAME", "PLAY", "SCORE", "FINAL"], username="Alice", score=42, time_seconds=180, game_mode="classic" ) print(f"Challenge ID: {challenge_id}") print(f"Full URL: {full_url}") print(f"Short ID: {sid}") print(f"Share: {get_shareable_url(sid)}") except Exception as e: print(f"Error: {e}") # Example: Load a game print("\n--- Example: Load Game ---") if sid: settings = load_game_from_sid(sid) if settings: print(f"Loaded Challenge: {settings['challenge_id']}") print(f"Wordlist Source: {settings.get('wordlist_source', 'N/A')}") users = settings.get('users', []) print(f"Users: {len(users)}") for user in users: print(f" - {user['username']}: {user['score']} pts in {user['time']}s")