#!/usr/bin/env python3 """ Pre-launch readiness check. Run this RIGHT BEFORE flipping production traffic. Verifies that: * Every required env var is set * KEK / shared secrets have plausible entropy (not "change_me") * DB is reachable + schema is at the latest Alembic head * The Telegram bot token authenticates with the actual @username * AI provider answers with a real model id * No leftover open positions in bot_trades that the bot doesn't know about Exits non-zero on any failure so you can wire it into CI / a deploy gate. DATABASE_URL=... venv/bin/python scripts/preflight.py """ from __future__ import annotations import asyncio import os import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import httpx from sqlalchemy import text from app.config import settings from app.database import AsyncSessionLocal, engine # Vars that MUST be non-empty for production. If you genuinely don't need one # (e.g. running without Telegram), comment it out — don't push junk values. REQUIRED = [ ("database_url", "Where the API stores everything"), ("frontend_url", "Used as the single CORS origin"), ("encryption_key", "KEK for HL API key envelope encryption"), ("ingest_api_key", "Shared secret for /api/signals/ingest"), ] # Required for the listed feature; non-fatal but logged as a warning. OPTIONAL_BUT_RECOMMENDED = [ ("ai_api_key", "Trump signal AI scoring (or anthropic_api_key)"), ("anthropic_api_key", "Required if ai_api_key empty AND you want KOL analysis"), ("telegram_bot_token", "Telegram push alerts (whole feature disabled if empty)"), ("telegram_bot_username","Required for the dashboard's Telegram connect deep link"), ("etherscan_api_key", "KOL on-chain (talks-vs-trades) — Ethereum side"), # NOTE: glassnode_api_key is no longer required — the BTC bottom-reversal # scanner switched to AHR999 + 200-week MA + Pi Cycle Bottom (all derived # from public price candles, no Glassnode call). The setting is kept on # config.Settings for future Glassnode-backed signals but isn't checked. ] # These should NEVER appear unchanged in production. SUSPECT_DEFAULTS = { "change_me_in_production", "your_key_here", "CHANGE_ME", "", } def red(s): return f"\033[31m{s}\033[0m" def green(s): return f"\033[32m{s}\033[0m" def yellow(s): return f"\033[33m{s}\033[0m" async def check_env() -> list[str]: errors = [] print("── env vars ──────────────────────────────────────") for name, why in REQUIRED: v = getattr(settings, name, "") if not v or v in SUSPECT_DEFAULTS: print(red(f" ✗ {name:25s} EMPTY or default — required: {why}")) errors.append(f"{name} empty") else: shown = v if len(v) < 30 else v[:8] + "…" + v[-4:] print(green(f" ✓ {name:25s} = {shown}")) print() print("── recommended (warnings only) ────────────────────") for name, why in OPTIONAL_BUT_RECOMMENDED: v = getattr(settings, name, "") if not v: print(yellow(f" ! {name:25s} empty — {why}")) else: print(green(f" ✓ {name:25s} set")) return errors async def check_db() -> list[str]: errors = [] print() print("── database ──────────────────────────────────────") try: async with AsyncSessionLocal() as db: await db.execute(text("SELECT 1")) print(green(" ✓ DB reachable")) except Exception as exc: print(red(f" ✗ DB unreachable: {exc}")) errors.append("db unreachable") return errors # Schema check — current head try: async with engine.begin() as conn: r = await conn.execute(text("SELECT version_num FROM alembic_version")) row = r.fetchone() current = row[0] if row else None # Read the latest version from alembic/versions/ versions_dir = Path(__file__).resolve().parent.parent / "alembic" / "versions" head_files = sorted(p.stem for p in versions_dir.glob("*.py") if p.stem != "__init__") latest_prefix = max(int(f.split("_")[0]) for f in head_files if f.split("_")[0].isdigit()) if current and current.startswith(f"{latest_prefix:03d}".rstrip("0") or "0"): print(green(f" ✓ alembic at head {current}")) elif current: print(yellow(f" ! alembic at {current} but versions/ has up to {latest_prefix:03d}")) print(yellow(f" → run: alembic upgrade head")) else: print(red(" ✗ alembic_version table empty — schema unmanaged")) errors.append("schema unmanaged") except Exception as exc: print(yellow(f" ! schema version check failed: {exc}")) # Orphan open positions try: async with AsyncSessionLocal() as db: r = await db.execute(text( "SELECT COUNT(*) FROM bot_trades WHERE closed_at IS NULL" )) n = r.scalar() or 0 if n: print(yellow(f" ! {n} open bot_trades rows — verify these match HL state before launch")) else: print(green(" ✓ no orphan open positions")) except Exception as exc: print(yellow(f" ! open-positions check failed: {exc}")) return errors async def check_telegram() -> list[str]: errors = [] print() print("── telegram ──────────────────────────────────────") if not settings.telegram_bot_token: print(yellow(" ! token empty — skipping")) return errors try: async with httpx.AsyncClient(timeout=10) as c: r = await c.get( f"https://api.telegram.org/bot{settings.telegram_bot_token}/getMe" ) if r.status_code != 200: print(red(f" ✗ getMe HTTP {r.status_code}: {r.text[:120]}")) errors.append("telegram auth failed") return errors data = r.json() if not data.get("ok"): print(red(f" ✗ getMe returned ok=false: {data}")) errors.append("telegram auth failed") return errors bot_username = data["result"]["username"] print(green(f" ✓ token authenticates as @{bot_username}")) if settings.telegram_bot_username and bot_username != settings.telegram_bot_username: print(red(f" ✗ TELEGRAM_BOT_USERNAME='{settings.telegram_bot_username}'" f" but token is for @{bot_username}")) errors.append("telegram username mismatch") except Exception as exc: print(red(f" ✗ telegram check failed: {exc}")) errors.append("telegram unreachable") return errors async def check_ai() -> list[str]: errors = [] print() print("── ai provider ───────────────────────────────────") if settings.anthropic_api_key: # Cheap, free models endpoint check. try: async with httpx.AsyncClient(timeout=10) as c: r = await c.get( "https://api.anthropic.com/v1/models", headers={ "x-api-key": settings.anthropic_api_key, "anthropic-version": "2023-06-01", }, ) if r.status_code == 200: print(green(" ✓ anthropic_api_key authenticates")) else: print(red(f" ✗ anthropic /v1/models HTTP {r.status_code}: {r.text[:120]}")) errors.append("anthropic auth failed") except Exception as exc: print(red(f" ✗ anthropic check failed: {exc}")) elif settings.ai_api_key: try: async with httpx.AsyncClient(timeout=10) as c: r = await c.get( f"{settings.ai_base_url.rstrip('/')}/models", headers={"Authorization": f"Bearer {settings.ai_api_key}"}, ) if r.status_code == 200: print(green(f" ✓ ai_api_key authenticates at {settings.ai_base_url}")) else: print(red(f" ✗ /models HTTP {r.status_code}: {r.text[:120]}")) errors.append("ai auth failed") except Exception as exc: print(red(f" ✗ ai check failed: {exc}")) else: print(yellow(" ! no ai provider key set — Trump signals will not be scored")) return errors async def main() -> int: print(f"Preflight check — env: {settings.environment}\n") if settings.environment != "production": print(yellow("WARNING: settings.environment is not 'production'. Some dev-only")) print(yellow(" endpoints (/api/dev/*) and auto-create_all() are enabled.\n")) all_errors: list[str] = [] all_errors += await check_env() all_errors += await check_db() all_errors += await check_telegram() all_errors += await check_ai() await engine.dispose() print() if all_errors: print(red(f"❌ {len(all_errors)} fatal issue(s):")) for e in all_errors: print(red(f" - {e}")) return 1 print(green("✅ All checks passed. Safe to launch.")) return 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))