# @pattern Template Method """ KHY-Quant first-run bootstrap. Handles npm install, .env generation, and database seeding for the standalone khy-quant package. """ import os import secrets import subprocess import sys from pathlib import Path MARKER_FILE = ".khy_quant_bootstrapped" SEED_MARKER_FILE = ".khy_quant_seeded" _NPM_TIMEOUT = 600 if os.name == "nt" else 300 def _subprocess_kwargs(): kwargs = {} if os.name == "nt": si = subprocess.STARTUPINFO() si.dwFlags |= subprocess.STARTF_USESHOWWINDOW si.wShowWindow = 0 kwargs["startupinfo"] = si kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW return kwargs def _write_marker(marker: Path, content: str): try: marker.write_text(f"{content}\n", encoding="utf-8") except OSError: pass def _path_exists_safe(p: Path) -> bool: try: return p.exists() except (OSError, ValueError): return False def _node_env(backend_dir: Path) -> dict: env = os.environ.copy() node_modules = str(backend_dir / "node_modules") existing = env.get("NODE_PATH", "") env["NODE_PATH"] = ( f"{node_modules}{os.pathsep}{existing}" if existing else node_modules ) return env def ensure_bootstrapped(backend_dir: Path, node_cmd: str): """Run first-time setup if not already done.""" marker = backend_dir / MARKER_FILE seed_marker = backend_dir / SEED_MARKER_FILE if marker.exists() and seed_marker.exists(): return print() print(" KHY-Quant first-run setup...") print() install_ok = True seed_ok = seed_marker.exists() if not marker.exists(): install_ok = _ensure_npm_install(backend_dir) and install_ok _ensure_env_file(backend_dir) if not seed_ok: seed_ok = _ensure_db_init(backend_dir, node_cmd) _register_app(backend_dir) if install_ok: _write_marker(marker, "bootstrapped") if seed_ok: _write_marker(seed_marker, "seeded") else: print(" [WARN] Database seed not confirmed; will retry on next startup.", file=sys.stderr) print() print(" Setup complete!") print() def _ensure_npm_install(backend_dir: Path) -> bool: node_modules = backend_dir / "node_modules" if _path_exists_safe(node_modules / "express"): print(" [OK] Node.js dependencies already installed") return True print(" Installing Node.js dependencies...") npm_cmd = "npm.cmd" if os.name == "nt" else "npm" npm_args = [npm_cmd, "install", "--no-audit", "--no-fund"] # Configure China mirror if needed lang = os.environ.get("LANG", "") + os.environ.get("LC_ALL", "") tz = os.environ.get("TZ", "") if lang.startswith("zh") or "Shanghai" in tz or "Chongqing" in tz: npmrc = backend_dir / ".npmrc" if not npmrc.exists(): try: npmrc.write_text("registry=https://registry.npmmirror.com\n", encoding="utf-8") except OSError: pass try: result = subprocess.run( npm_args, cwd=str(backend_dir), timeout=_NPM_TIMEOUT, **_subprocess_kwargs(), ) if result.returncode == 0: print(" [OK] Node.js dependencies installed") return True else: print(" [WARN] npm install exited with errors.", file=sys.stderr) return False except FileNotFoundError: print(f" [WARN] '{npm_cmd}' not found.", file=sys.stderr) return False except subprocess.TimeoutExpired: print(f" [WARN] npm install timed out ({_NPM_TIMEOUT // 60} min)", file=sys.stderr) return False def _ensure_env_file(backend_dir: Path): env_file = backend_dir / ".env" if env_file.exists(): print(" [OK] .env configuration exists") return print(" Generating .env configuration...") jwt_secret = secrets.token_hex(32) content = ( "# KHY-Quant Environment Configuration\n" "# Generated by khy-quant bootstrap\n\n" "DB_TYPE=sqlite\n" "PORT=3000\n" "NODE_ENV=development\n" "LOG_LEVEL=info\n\n" f"JWT_SECRET={jwt_secret}\n" "JWT_EXPIRES_IN=7d\n\n" "ENABLE_AKSHARE=true\n" "DEFAULT_DATA_SOURCE=reliable\n" ) env_file.write_text(content) if os.name != "nt": try: os.chmod(str(env_file), 0o600) except OSError: pass print(" [OK] .env generated") def _ensure_db_init(backend_dir: Path, node_cmd: str) -> bool: seed_script = backend_dir / "scripts" / "seed.js" if not seed_script.exists(): return False print(" Initializing database...") try: result = subprocess.run( [node_cmd, str(seed_script)], cwd=str(backend_dir), env=_node_env(backend_dir), capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=120, **_subprocess_kwargs(), ) combined = f"{result.stdout or ''}\n{result.stderr or ''}".lower() if "seed completed" in combined or (result.returncode == 0 and "seed warning" not in combined): print(" [OK] Database initialized") return True print(" [WARN] Database seed did not confirm success", file=sys.stderr) return False except subprocess.TimeoutExpired: print(" [WARN] Database initialization timed out", file=sys.stderr) return False except Exception as e: print(f" [WARN] Database initialization error: {e}", file=sys.stderr) return False def _register_app(backend_dir: Path): import json apps_dir = Path(os.path.expanduser("~")) / ".khyquant" / "apps" apps_dir.mkdir(parents=True, exist_ok=True) manifest_file = apps_dir / "khyquant.json" version = "0.0.0" pkg_json = backend_dir / "package.json" try: if pkg_json.exists(): pkg = json.loads(pkg_json.read_text(encoding="utf-8")) version = pkg.get("version", version) except Exception: pass manifest = { "name": "khyquant", "version": version, "description": "AI-powered quantitative trading terminal", "entry": str(backend_dir / "server.js"), "port": 3000, "source": "pip", "commands": ["khyquant", "khy-quant"], "autoStart": False, "installedAt": __import__("datetime").datetime.now().isoformat(), } try: manifest_file.write_text(json.dumps(manifest, indent=2, ensure_ascii=False)) print(" [OK] khyquant registered in app registry") except OSError as e: print(f" [WARN] Failed to register app manifest: {e}", file=sys.stderr)