""" UEFN Store — Installer Browse the store, manage your library, and install assets directly into UEFN. No dependencies required — runs with Python standard library only. Get your API key at: https://www.uefnstore.com/dashboard/settings """ import os import json import zipfile import tempfile import threading import webbrowser import http.server import socketserver try: import unreal IN_UEFN = True except ImportError: IN_UEFN = False import urllib.request import urllib.error # ────────────────────────────────────────────── # Configuration # ────────────────────────────────────────────── API_BASE = "https://www.uefnstore.com/api/script" STORE_URL = "https://www.uefnstore.com" CONFIG_FILE = os.path.join(os.path.expanduser("~"), ".uefnstore_config") PORT = 48521 # ────────────────────────────────────────────── # API Client # ────────────────────────────────────────────── class APIClient: def __init__(self): self.api_key = None self._load_config() def _load_config(self): if os.path.exists(CONFIG_FILE): try: with open(CONFIG_FILE, "r") as f: data = json.load(f) self.api_key = data.get("api_key") except (json.JSONDecodeError, IOError): self.api_key = None def _save_config(self): with open(CONFIG_FILE, "w") as f: json.dump({"api_key": self.api_key}, f) def _clear_config(self): self.api_key = None if os.path.exists(CONFIG_FILE): os.remove(CONFIG_FILE) def set_api_key(self, key): self.api_key = key.strip() self._save_config() def _request(self, method, endpoint, data=None): url = "{}/{}".format(API_BASE, endpoint) headers = {"Content-Type": "application/json"} if self.api_key: headers["X-API-Key"] = self.api_key body = json.dumps(data).encode() if data else None req = urllib.request.Request(url, data=body, headers=headers, method=method) class _RedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, hdrs, newurl): return urllib.request.Request( newurl, data=req.data, headers=dict(req.header_items()), method=req.get_method() ) opener = urllib.request.build_opener(_RedirectHandler) try: resp = opener.open(req, timeout=30) return json.loads(resp.read().decode()), None except urllib.error.HTTPError as e: try: err = json.loads(e.read().decode()) return None, err.get("error", "HTTP {}".format(e.code)) except Exception: return None, "HTTP {}".format(e.code) except urllib.error.URLError as e: return None, "Connection error: {}".format(e.reason) except Exception as e: return None, str(e) def get_purchases(self): data, err = self._request("GET", "purchases") if err: return None, err return data.get("purchases", []), None def get_download_url(self, product_id, version_id): data, err = self._request("POST", "download", {"product_id": product_id, "version_id": version_id}) if err: return None, err return data, None def logout(self): self._clear_config() @property def is_connected(self): return self.api_key is not None and self.api_key.startswith("uefn_") # ────────────────────────────────────────────── # Installer Logic # ────────────────────────────────────────────── def get_uefn_content_dir(): if IN_UEFN: project = unreal.ValkyrieProjectLibrary.get_main_project() return unreal.ValkyrieProjectLibrary.get_project_content_dir(project) return None def install_product(download_url, product_title, content_dir): slug = ( product_title.lower() .replace(" ", "-") .encode("ascii", "ignore") .decode() ) slug = "".join(c if c.isalnum() or c == "-" else "" for c in slug).strip("-") folder_name = slug or "product" dest_folder = os.path.join(content_dir, folder_name) tmp_path = os.path.join(tempfile.gettempdir(), "uefn_{}.zip".format(slug)) urllib.request.urlretrieve(download_url, tmp_path) if tmp_path.lower().endswith(".zip"): with zipfile.ZipFile(tmp_path, "r") as zf: zf.extractall(dest_folder) else: raise RuntimeError("Only .zip is supported for auto-extraction.") os.remove(tmp_path) if IN_UEFN: unreal.log("[UEFN Store] Installed '{}' to {}".format(product_title, dest_folder)) return dest_folder # ────────────────────────────────────────────── # API layer (used by the HTTP handler) # ────────────────────────────────────────────── class InstallerAPI: def __init__(self, content_dir=None): self.client = APIClient() self.content_dir = content_dir def get_status(self): project_name = None if self.content_dir: project_name = os.path.basename(os.path.dirname(self.content_dir)) result = { "connected": self.client.is_connected, "uefn_project": project_name, } if self.client.is_connected: result["api_key"] = self.client.api_key return result def connect(self, api_key): if not api_key or not api_key.startswith("uefn_"): return {"success": False, "error": "API key must start with uefn_"} self.client.set_api_key(api_key) purchases, err = self.client.get_purchases() if err: self.client.logout() return {"success": False, "error": err} return {"success": True, **self.get_status()} def disconnect(self): self.client.logout() return {"success": True} def get_purchases(self): if not self.client.is_connected: return {"error": "Not connected"} purchases, err = self.client.get_purchases() if err: return {"error": err} return {"purchases": purchases} def do_install(self, product_id, version_id, title): if not self.client.is_connected: return {"success": False, "error": "Not connected"} dl_data, err = self.client.get_download_url(product_id, version_id) if err: return {"success": False, "error": err} target_dir = self.content_dir if not target_dir: target_dir = os.path.join(os.path.expanduser("~"), "UEFN_Downloads") os.makedirs(target_dir, exist_ok=True) try: dest = install_product(dl_data["url"], title, target_dir) return {"success": True, "folder": os.path.basename(dest)} except Exception as e: return {"success": False, "error": str(e)} # ────────────────────────────────────────────── # HTML UI # ────────────────────────────────────────────── HTML_PAGE = r""" UEFN Store
Install purchased assets directly into your UEFN projects
Get your key from uefnstore.com/dashboard/settings
Your API key is stored locally and never shared.
UEFN Store
Browse only — to purchase, open in your browser using the ↗ button above

Could not load the store in this view.

My Library 0
Loading library...
Checking...
""" # ────────────────────────────────────────────── # HTTP Server # ────────────────────────────────────────────── import time class InstallerHandler(http.server.BaseHTTPRequestHandler): api = None last_heartbeat = 0 def log_message(self, fmt, *args): if IN_UEFN: unreal.log("[UEFN Store] " + (fmt % args)) def _json(self, data, status=200): body = json.dumps(data).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _body(self): length = int(self.headers.get("Content-Length", 0)) return json.loads(self.rfile.read(length).decode()) if length else {} def do_GET(self): if self.path == "/": InstallerHandler.last_heartbeat = time.time() html = HTML_PAGE.replace("STORE_URL_PLACEHOLDER", STORE_URL).encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(html))) self.end_headers() self.wfile.write(html) elif self.path == "/api/heartbeat": InstallerHandler.last_heartbeat = time.time() self._json({"ok": True}) elif self.path == "/api/status": self._json(self.api.get_status()) elif self.path == "/api/purchases": self._json(self.api.get_purchases()) else: self.send_error(404) def do_POST(self): if self.path == "/api/connect": self._json(self.api.connect(self._body().get("api_key", ""))) elif self.path == "/api/disconnect": self._json(self.api.disconnect()) elif self.path == "/api/install": data = self._body() self._json(self.api.do_install( data.get("product_id"), data.get("version_id"), data.get("title", "product") )) elif self.path == "/api/shutdown": self._json({"success": True}) threading.Thread(target=self.server.shutdown, daemon=True).start() else: self.send_error(404) # ────────────────────────────────────────────── # Entry Point # ────────────────────────────────────────────── _server_instance = None def main(content_dir=None): global _server_instance if _server_instance is not None: try: _server_instance.shutdown() _server_instance.server_close() except Exception: pass api = InstallerAPI(content_dir=content_dir) InstallerHandler.api = api socketserver.ThreadingTCPServer.allow_reuse_address = True server = socketserver.ThreadingTCPServer(("127.0.0.1", PORT), InstallerHandler) server.daemon_threads = True _server_instance = server InstallerHandler.last_heartbeat = time.time() def watchdog(): """Shut down the server if no heartbeat received for 10 seconds.""" while True: time.sleep(5) if time.time() - InstallerHandler.last_heartbeat > 10: msg = "[UEFN Store] Browser closed — shutting down." if IN_UEFN: unreal.log(msg) else: print(msg) server.shutdown() break threading.Thread(target=watchdog, daemon=True).start() url = "http://127.0.0.1:{}".format(PORT) msg = "[UEFN Store] Installer running at {}".format(url) if IN_UEFN: unreal.log(msg) else: print(msg) webbrowser.open(url) try: server.serve_forever(poll_interval=0.5) except KeyboardInterrupt: pass finally: try: server.server_close() except Exception: pass if __name__ == "__main__": _content_dir = get_uefn_content_dir() if IN_UEFN: threading.Thread(target=main, args=(_content_dir,), daemon=True).start() else: main(_content_dir)