trumpsignal-backend/scripts/preflight.py

#!/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()))