"""
Bloomberg Terminal-style Web Dashboard, FastAPI backend
Run: python server.py
Then open: http://localhost:8000
"""

import io
import json
import logging
import os
import sys
import threading
import time
import hashlib
import hmac
from datetime import datetime, timezone
from pathlib import Path

# UTF-8 safety on Windows
if sys.platform == "win32":
    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")

from dotenv import load_dotenv
load_dotenv()

import uvicorn
import secrets
import requests
from fastapi import FastAPI, Request, Header
from fastapi.responses import HTMLResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles

import mailer
import ai_assistant
from analyzers import psychology
from collectors import google_trends as google_trends_mod

from collectors import (
    RSSCollector, RedditCollector, CoinGeckoCollector, FearGreedCollector,
    NewsDataCollector, OnChainCollector, dedupe_articles, cluster_consensus,
)
from collectors.dedupe import _normalize_title
from analyzers import (
    NarrativeDetector, EntityExtractor, TrendScorer, EnsembleAnalyzer,
    SentimentAggregator, SpikeDetector, Alerter, Scorecard,
)
from analyzers.velocity import compute_velocity, top_accelerating
from analyzers.meta_model import MetaModel, features_from_item
from analyzers.relevance import RelevanceClassifier
from analyzers.liquidation import LiquidationHeatmap
from analyzers.magnet_tracker import MagnetTracker
from analyzers.forecast import MarketImpactForecast
from collectors.economic_calendar import EconomicCalendar, fmp_configured
from analyzers.market_reaction import MarketReaction
from analyzers.embedding_narratives import EmbeddingNarratives
from monitoring import monitor
from storage import Database, EmailsDB
import persistence
from config import DB_PATH

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)

# Persistent analyzers (loaded once, reused across cycles)
_meta_model = MetaModel()
_embed_narr = EmbeddingNarratives()

# Docs/schema disabled: no need to advertise the API surface (incl. the gated
# /terminal and admin routes) on a pre-launch public site.
app = FastAPI(title="Crypto Narrative Terminal",
              docs_url=None, redoc_url=None, openapi_url=None)
# Ensure the static dir exists (git doesn't track empty folders, so it may be
# absent on a fresh clone / cloud deploy).
Path("static").mkdir(exist_ok=True)
app.mount("/static", StaticFiles(directory="static"), name="static")

_HTML = Path("templates/index.html").read_text(encoding="utf-8")
_ADMIN_HTML = Path("templates/admin.html").read_text(encoding="utf-8")
_TERMS_HTML = Path("templates/terms.html").read_text(encoding="utf-8")
_JOIN_HTML = Path("templates/join.html").read_text(encoding="utf-8")

# SEO content layer (Articles / FAQ / About) — data-driven, see content/site_content.py
from content import site_content

# Live execution layer (optional). Reads the published signal ONLY; it never
# takes part in computing it. Completely inert unless TRADE_ENABLED=true.
from trading import executor as trader

# Restore the last DB backup (if HF persistence is configured) BEFORE opening it
persistence.restore_db(DB_PATH)
db = Database()
alerter = Alerter()

# Subscribers & email settings live in their own tiny DB so they can be backed
# up to HF immediately on every signup (see persistence.backup_emails).
EMAILS_DB_PATH = str(Path(DB_PATH).resolve().parent / "emails.db")
persistence.restore_emails(EMAILS_DB_PATH)
edb = EmailsDB(EMAILS_DB_PATH)
# Trade state lives beside the DB so a restart can't double-open a position.
trader.init(DB_PATH)
# One-time: pull any existing subscribers from the old combined DB so they
# aren't stranded after the split (no-op once emails.db has them).
try:
    edb.migrate_from(db)
except Exception as _exc:
    logger.warning(f"[EmailsDB] migrate_from skipped: {_exc}")

# Admin auth, simple token issued on password login (set ADMIN_PASSWORD in .env)
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin")
_admin_tokens: set[str] = set()

# Pre-launch preview gate: the real terminal + full data are hidden behind a
# password so the public only sees the waitlist. Defaults to the admin password.
PREVIEW_PASSWORD = os.getenv("PREVIEW_PASSWORD") or ADMIN_PASSWORD
_PREVIEW_TOKEN = secrets.token_urlsafe(24)   # rotates on restart (re-login needed)


def _check_admin(token: str | None) -> bool:
    return bool(token and token in _admin_tokens)


def _persist_now():
    """Back up the small emails DB immediately (in the background) after a
    signup/confirm so a Space rebuild can never lose a subscriber. No-op if
    persistence isn't configured."""
    if not persistence.persistence_configured():
        return
    threading.Thread(
        target=lambda: persistence.backup_emails(edb, min_interval=0), daemon=True
    ).start()


def _has_preview(request: Request) -> bool:
    return request.cookies.get("crx_preview") == _PREVIEW_TOKEN


# ---------------------------------------------------------------------------
# User accounts (email + password). scrypt hashing (stdlib), in-memory sessions.
# Login does NOT unlock the terminal yet (that stays preview-gated), this is the
# SaaS account foundation; flip _login_gates_terminal later when plans/payments
# are wired.
# ---------------------------------------------------------------------------
_sessions: dict[str, str] = {}          # token -> email
_login_fails: dict[str, list] = {}      # email -> [timestamps] (basic rate limit)


def _hash_pw(password: str, salt: bytes | None = None) -> tuple[str, str]:
    salt = salt or os.urandom(16)
    dk = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=16384, r=8, p=1, dklen=32)
    return dk.hex(), salt.hex()


def _verify_pw(password: str, salt_hex: str, hash_hex: str) -> bool:
    try:
        dk, _ = _hash_pw(password, bytes.fromhex(salt_hex))
        return hmac.compare_digest(dk, hash_hex)
    except Exception:
        return False


def _rate_limited(email: str) -> bool:
    now = time.time()
    fails = [t for t in _login_fails.get(email, []) if now - t < 900]  # 15 min window
    _login_fails[email] = fails
    return len(fails) >= 6


def _record_fail(email: str):
    _login_fails.setdefault(email, []).append(time.time())


def _current_user(request: Request) -> str | None:
    return _sessions.get(request.cookies.get("crx_session", ""))


def _has_access(request: Request) -> bool:
    """Terminal access = a logged-in account OR the owner preview cookie."""
    return _current_user(request) is not None or _has_preview(request)

# ---------------------------------------------------------------------------
# In-memory cache, updated by background thread
# ---------------------------------------------------------------------------
_cache: dict = {}
_cache_lock = threading.Lock()
_last_update: str = ""
_is_fetching: bool = False


def _maybe_train_meta_model() -> None:
    """#2 Train the meta-model on stories that now have a 24h reaction label."""
    try:
        labeled = db.labeled_reactions(5000)
        samples = []
        for r in labeled:
            lab = r.get("label_24h")
            if lab not in ("bullish", "bearish"):
                continue
            art = db.get_article(r["url"])
            if not art:
                continue
            samples.append((features_from_item(art), 1 if lab == "bullish" else 0))
        if samples:
            _meta_model.train(samples)   # no-ops if below the min sample threshold
    except Exception as exc:
        logger.debug(f"[MetaModel] training skipped: {exc}")


def _build_chart_series(price_hist: list[dict], sent_hist: list[dict],
                        liq_hist: list[dict], hours: int = 168,
                        max_points: int = 420) -> list[dict]:
    """Align BTC price with the sentiment index (and the strongest liquidation
    magnet) on one timeline so the UI can overlay them.

    Returns [{t, p, i, l}], time, price, index (0-100), liq strength (0-100).
    Downsampled to `max_points` so the payload stays small.
    """
    def _ts(iso):
        try:
            dt = datetime.fromisoformat(iso)
            return (dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt).timestamp()
        except Exception:
            return None

    cutoff = datetime.now(timezone.utc).timestamp() - hours * 3600
    pts = [(t, p["price"]) for p in price_hist
           if (t := _ts(p.get("snapshot_at", ""))) and t >= cutoff and p.get("price")]
    if not pts:
        return []
    pts.sort()

    idx = sorted((t, s.get("index_val")) for s in sent_hist
                 if (t := _ts(s.get("snapshot_at", ""))) and s.get("index_val") is not None)
    liq = sorted((t, s.get("strength")) for s in liq_hist
                 if (t := _ts(s.get("snapshot_at", ""))) and s.get("strength") is not None)

    def _nearest(series, target, tol=1800):
        """Nearest value within tol seconds (series must be sorted by time)."""
        best, best_d = None, tol + 1
        for t, v in series:
            d = abs(t - target)
            if d < best_d:
                best, best_d = v, d
            elif t > target + tol:
                break
        return best if best_d <= tol else None

    step = max(1, -(-len(pts) // max_points))   # ceil -> never exceed max_points
    out = []
    for i in range(0, len(pts), step):
        t, price = pts[i]
        out.append({
            "t": datetime.fromtimestamp(t, tz=timezone.utc).isoformat(),
            "p": round(price, 2),
            "i": _nearest(idx, t),
            "l": _nearest(liq, t),
        })
    return out


def _compute_divergence(sentiment_24h: dict, price_hist: list[dict],
                        hours: int = 24) -> dict:
    """Sentiment-vs-price divergence read (educational).

    Compares how the sentiment index moved vs how BTC moved over the window.
    The thesis: news mood often LEADS price, so a gap between them is a signal.
      - index up while price flat/down  -> bullish divergence (price may catch up)
      - index down while price flat/up   -> bearish divergence (pullback risk)
      - both same direction              -> confirmed / in agreement
    NOT a prediction, framed as an observation.
    """
    ov = (sentiment_24h or {}).get("overall", {}) or {}
    idx_now = ov.get("index")
    idx_delta = sentiment_24h.get("trend_delta_index")   # points vs ~24h ago
    if idx_now is None or idx_delta is None:
        return {}

    def _ts(iso):
        try:
            dt = datetime.fromisoformat(iso)
            return (dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt).timestamp()
        except Exception:
            return None

    pts = sorted((t, p["price"]) for p in price_hist
                 if (t := _ts(p.get("snapshot_at", ""))) and p.get("price"))
    if len(pts) < 2:
        return {}
    price_now = pts[-1][1]
    target = pts[-1][0] - hours * 3600
    then = min(pts, key=lambda x: abs(x[0] - target))
    if abs(then[0] - target) > 8 * 3600 or then[1] == 0:
        return {}
    price_change = (price_now - then[1]) / then[1] * 100

    IDX_TH, PX_TH = 4.0, 1.0
    idx_up, idx_dn = idx_delta >= IDX_TH, idx_delta <= -IDX_TH
    px_up, px_dn = price_change >= PX_TH, price_change <= -PX_TH

    if idx_up and not px_up:
        state, tone = "bullish_divergence", "bull"
        head = "Bullish divergence"
        note = ("Sentiment is rising but price hasn't followed yet. News mood "
                "often leads price, so this can precede a catch-up move up, "
                "not a guarantee.")
    elif idx_dn and not px_dn:
        state, tone = "bearish_divergence", "bear"
        head = "Bearish divergence"
        note = ("Sentiment is cooling while price holds up. Mood tends to lead, "
                "so pullback risk is building, watch closely.")
    elif idx_up and px_up:
        state, tone = "confirmed_bull", "bull"
        head = "Confirmed bullish"
        note = "Sentiment and price are both rising, the move up agrees with the news mood."
    elif idx_dn and px_dn:
        state, tone = "confirmed_bear", "bear"
        head = "Confirmed bearish"
        note = "Sentiment and price are both falling, the decline agrees with the news mood."
    else:
        state, tone = "in_sync", "flat"
        head = "In step"
        note = "Sentiment and price are roughly in step right now, no notable divergence."

    return {
        "state":        state,
        "tone":         tone,
        "headline":     head,
        "note":         note,
        "index_change": round(idx_delta, 1),
        "price_change": round(price_change, 2),
        "window_h":     hours,
    }


def run_pipeline() -> dict:
    global _is_fetching
    _is_fetching = True
    try:
        rss_items      = RSSCollector().fetch_all()
        newsdata_items = NewsDataCollector().fetch_all()
        reddit_items   = RedditCollector().fetch_all()

        cg = CoinGeckoCollector()
        trending_coins = cg.fetch_trending()
        global_market  = cg.fetch_global()
        coin_prices    = cg.fetch_prices()
        trending_ids   = cg.get_trending_ids()
        btc_now = next((c.get("price_usd") for c in coin_prices if c.get("id") == "bitcoin"), None)
        eth_now = next((c.get("price_usd") for c in coin_prices if c.get("id") == "ethereum"), None)

        fg = FearGreedCollector()
        fear_greed_history = fg.fetch()
        fear_greed_current = fg.current()

        db.save_fear_greed(fear_greed_history)
        db.save_prices(coin_prices)

        # #5 On-chain + derivatives signals
        onchain_raw = OnChainCollector().fetch(prev_oi_btc=db.prev_oi_btc())
        if onchain_raw:
            db.save_onchain(onchain_raw)
        onchain = OnChainCollector.context_summary(onchain_raw)

        all_items = rss_items + newsdata_items + reddit_items
        if not all_items:
            monitor.component_failure("collectors", "no items collected")
            return {}

        # Collapse the same story arriving from multiple feeds (keeps the most
        # credible copy), cleaner feed + no double-counting in sentiment.
        before = len(all_items)
        all_items = dedupe_articles(all_items)
        logger.info(f"[Dedupe] {before} -> {len(all_items)} items")

        # Crypto Relevance Filter, classify CRYPTO / MACRO / IRRELEVANT and
        # DISCARD irrelevant noise (sports, entertainment, off-topic) BEFORE
        # any scoring so it never reaches the sentiment index.
        all_items = RelevanceClassifier().classify_batch(all_items)
        all_items, discarded = RelevanceClassifier.filter_relevant(all_items)
        if discarded:
            monitor.component_failure("relevance_filter", f"discarded {discarded} irrelevant")
        logger.info(f"[Relevance] kept {len(all_items)}, discarded {discarded}")
        relevance_counts = {
            "crypto": sum(1 for i in all_items if i.get("relevance") == "crypto"),
            "macro":  sum(1 for i in all_items if i.get("relevance") == "macro"),
            "discarded_irrelevant": discarded,
        }

        sa = EnsembleAnalyzer()
        all_items = sa.analyze_batch(all_items)

        nd = NarrativeDetector()
        all_items = nd.detect_batch(all_items)

        ee = EntityExtractor()
        all_items = ee.extract_batch(all_items)
        coin_mentions = ee.coin_mention_frequency(all_items)

        # #3 Consensus / contradiction across duplicate sources
        consensus = cluster_consensus(rss_items + newsdata_items + reddit_items)
        for it in all_items:
            c = consensus.get(_normalize_title(it.get("title", "")))
            if c:
                it["consensus_strength"] = c["consensus_strength"]
                it["sentiment_variance"] = c["sentiment_variance"]
                it["cluster_size"] = c["cluster_size"]
                if it.get("url"):
                    db.save_consensus(it["url"], c["cluster_size"],
                                      c["sentiment_variance"], c["consensus_strength"])

        # #6 Social attention velocity (acceleration of mentions over time)
        db.save_mentions(coin_mentions, kind="coin")
        velocity_map = compute_velocity(db.mention_series("coin", 400), coin_mentions)
        for it in all_items:
            coins = it.get("entities", {}).get("coins", [])
            vs = max((velocity_map.get(c, {}).get("velocity_score", 0.0) for c in coins),
                     default=0.0)
            it["velocity_score"] = vs
        accelerating = top_accelerating(velocity_map, 8)

        # #2 Meta-model probability of bullish reaction (per item)
        for it in all_items:
            it["bull_prob"] = round(_meta_model.predict_proba(it), 4)

        # #4 Embedding-based narrative discovery (replaces TF-IDF); fallback below
        emerging_narratives = _embed_narr.discover(all_items, top_n=12)
        if not emerging_narratives:
            monitor.model_fallback("embeddings", "using TF-IDF keywords")
            nd.fit_corpus(all_items)
            emerging_narratives = [
                {"label": kw, "size": 0, "terms": [kw], "known_narrative": None}
                for kw, _ in nd.top_emerging_keywords(all_items, top_n=12)
            ]

        scorer = TrendScorer(trending_coin_ids=trending_ids)
        scored_narratives = scorer.score_narratives(all_items)

        # Market-psychology signals (heuristic, from the scored items).
        hype = psychology.hype_signals(all_items)
        retail = psychology.retail_excitement(all_items)
        # Google Trends search-interest (best-effort; may be empty on cloud IPs).
        try:
            _trend_kw = ["bitcoin", "crypto"] + [n.get("name", "") for n in scored_narratives[:3]]
            google_trends = google_trends_mod.fetch_trends([k for k in _trend_kw if k])
        except Exception:
            google_trends = {}

        # Spike detection vs prior snapshots (before saving the new one, so
        # the current run is compared against history, not itself).
        prior_snaps = db.get_snapshots(limit=12)
        spike_alerts = SpikeDetector(prior_snaps).detect(scored_narratives)
        if spike_alerts:
            alerter.push(spike_alerts)   # external channels if configured

        # 24h aggregate sentiment, merge fresh items with cached recent ones
        # so the prior-24h trend comparison has history to work with.
        db.save_articles(all_items)
        db.save_snapshot(scored_narratives)

        # #1 Market reaction labeling: seed new stories, fill matured ones
        mr = MarketReaction()
        mr.seed(db, all_items, btc_now, eth_now)
        mr.fill(db)

        # #2 Retrain meta-model on accumulated 24h-labelled reactions
        _maybe_train_meta_model()

        recent_cached = db.get_recent_articles(hours=48)
        # Reliable "vs ~24h ago" using the real stored snapshot, not stale articles
        prior_snapshot = db.get_index_near(hours_ago=24, tolerance_hours=8)
        # ~30 days of past net scores -> baseline for the RELATIVE index (so the
        # gauge can read bearish when mood sours vs its own recent normal).
        baseline_hist = db.get_sentiment_history(limit=4320)   # ~30d at 10-min
        baseline_nets = [h.get("net_score") for h in baseline_hist
                         if h.get("net_score") is not None]
        sentiment_24h = SentimentAggregator().aggregate(
            recent_cached or all_items, prior_snapshot, baseline_nets)
        # #7 nudge volatility probability with the derivatives liquidation proxy
        if onchain.get("liq_proxy") is not None:
            ov = sentiment_24h.get("overall", {})
            extra = min(0.3, onchain["liq_proxy"] / 40.0)
            ov["vol_prob"] = round(min(1.0, ov.get("vol_prob", 0.0) + extra), 4)
        db.save_sentiment_history(sentiment_24h)
        sentiment_history = db.get_sentiment_history(limit=96)   # recent, for the UI sparkline
        # Accuracy scorecard needs the FULL history (many days), not just the
        # recent window, otherwise it can only ever find ~1 independent call.
        # btc_history (large) is reused by the liquidation heatmap below, which
        # only slices its most recent 300 points.
        # ~45 days is plenty for the track record; loading tens of thousands of
        # rows every cycle was heavy on the small Space and could OOM.
        btc_history = db.get_price_history("bitcoin", limit=7000)
        sc_hist = db.get_sentiment_history(limit=7000)
        scorecard = Scorecard().compute(sc_hist, btc_history)

        # Liquidation heatmap (estimated from OI + leverage tiers vs recent prices).
        # Fall back to the last stored BTC price if CoinGecko was rate-limited.
        liq_price = btc_now or (btc_history[-1]["price"] if btc_history else None)
        liquidation = LiquidationHeatmap().compute(
            liq_price, btc_history, oi_btc=(onchain_raw or {}).get("oi_btc"))
        # Record the strongest-magnet strength for this cycle so the chart can
        # mark high-liquidation moments (this history was never being written).
        db.save_magnet_seed(liquidation)

        # Overlay series: BTC price vs sentiment index (+ liquidation intensity)
        chart_series = _build_chart_series(
            btc_history, sc_hist, db.get_magnet_strength_history(2000))
        divergence = _compute_divergence(sentiment_24h, btc_history)

        # Magnet target tracker: hit-check + expire + register (every cycle),
        # each level tracked for 24h, never re-registered while active.
        _mt = MagnetTracker()
        _mt.update(db, liquidation, liq_price)
        magnet_scorecard = _mt.compute(db)

        # Economic calendar (FMP) + Market Impact Forecast for the next 24-48h
        calendar = EconomicCalendar().fetch(days_ahead=7)
        forecast = MarketImpactForecast().compute(
            sentiment_24h=sentiment_24h, onchain=onchain, liquidation=liquidation,
            calendar=calendar, fear_greed=fear_greed_current)

        # #8 flush system-health events for this cycle
        monitor.flush(db)
        health = db.health_summary(24)
        db.purge_old()

        # Persist the DB to the HF Dataset backup (throttled; no-op if unset).
        # Fully isolated: a backup problem must NEVER discard the cycle's data.
        try:
            persistence.backup_db(db)
        except Exception as exc:
            logger.warning(f"[Persist] backup skipped: {exc}")

        # Build news feed (most recent 60 items with title + source)
        news_feed = sorted(
            [i for i in all_items if i.get("title")],
            key=lambda x: x.get("published", ""),
            reverse=True,
        )[:60]

        result = {
            "narratives":        scored_narratives,
            "global_market":     global_market,
            "fear_greed":        fear_greed_current,
            "fear_greed_history": fear_greed_history,
            "trending_coins":    trending_coins,
            "coin_prices":       coin_prices,
            "coin_mentions":     coin_mentions,
            "emerging_narratives": emerging_narratives,
            "accelerating":      accelerating,
            "onchain":           onchain,
            "liquidation":       liquidation,
            "magnet_scorecard":  magnet_scorecard,
            "calendar":          calendar,
            "forecast":          forecast,
            "hype":              hype,
            "retail_excitement": retail,
            "google_trends":     google_trends,
            "fmp_configured":    fmp_configured(),
            "health":            health,
            "relevance_counts":  relevance_counts,
            "meta_model_trained": _meta_model.trained,
            "article_count":     len(all_items),
            "sentiment_24h":     sentiment_24h,
            "sentiment_history": sentiment_history,
            "spike_alerts":      spike_alerts,
            "scorecard":         scorecard,
            "chart_series":      chart_series,
            "divergence":        divergence,
            "news_feed":         [
                {
                    "title":     n.get("title", "")[:120],
                    "source":    n.get("source", ""),
                    "url":       n.get("url", ""),
                    "published": n.get("published", ""),
                    "sentiment": n.get("sentiment", {}).get("label", "neutral"),
                    "compound":  n.get("sentiment", {}).get("compound", 0),
                    "confidence": n.get("sentiment", {}).get("confidence", 0),
                    "engine":    n.get("sentiment", {}).get("engine", "vader"),
                    "bull_prob": n.get("bull_prob", 0.5),
                    "consensus_strength": n.get("consensus_strength"),
                    "narratives": n.get("narratives", []),
                    "is_macro":  bool(n.get("is_macro")),
                    "relevance": n.get("relevance", "crypto"),
                }
                for n in news_feed
            ],
            "fetch_time": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"),
        }
        return result
    except Exception as exc:
        logger.exception("Pipeline failed")
        return {"error": str(exc)}
    finally:
        _is_fetching = False


REFRESH_MINUTES = int(os.getenv("REFRESH_INTERVAL_MINUTES", "10"))


def refresh_loop():
    while True:
        data = run_pipeline()
        if data:
            with _cache_lock:
                _cache.update(data)
                global _last_update
                _last_update = datetime.now(timezone.utc).strftime("%H:%M:%S")
            # Mirror the (already computed) signal onto the exchange. Wrapped so
            # a trading problem can never affect the analysis pipeline.
            try:
                with _cache_lock:
                    snapshot = dict(_cache)
                trader.on_cycle(snapshot)
            except Exception:
                logger.exception("[Trade] hook failed (analysis unaffected)")
        time.sleep(max(3, REFRESH_MINUTES) * 60)   # default every 10 min (configurable)


def _send_brief_now(slot_label: str = "manual") -> tuple[bool, str]:
    """Build the current brief and email it to all active clients."""
    with _cache_lock:
        data = dict(_cache)
    if not data or not data.get("sentiment_24h"):
        # Cache not warmed yet (e.g. just after a restart). Kick off a fetch
        # so the next attempt succeeds, and tell the user clearly.
        if not _is_fetching:
            threading.Thread(target=lambda: _cache.update(run_pipeline()), daemon=True).start()
        return False, "Data still loading, wait ~2 min after restart, then try again"
    settings = edb.get_email_settings()
    recipients = [c["email"] for c in edb.get_clients(active_only=True)]
    if not recipients:
        edb.log_email(slot_label, 0, "skipped", "no active clients")
        return False, "No active clients"
    html = mailer.build_brief_html(data)
    idx = data["sentiment_24h"]["overall"].get("index", "")
    label = data["sentiment_24h"]["overall"].get("label", "")
    subject = f"{settings.get('subject_prefix','Crypto Narrative Brief')}, {label} ({idx}/100)"
    ok, detail = mailer.send_brief(recipients, subject, html)
    edb.log_email(slot_label, len(recipients), "sent" if ok else "failed", detail)
    return ok, detail


def email_scheduler():
    """Checks every 30s whether a scheduled send time has arrived (UTC)."""
    while True:
        try:
            settings = edb.get_email_settings()
            if settings.get("enabled"):
                now = datetime.now(timezone.utc)
                hhmm = now.strftime("%H:%M")
                today = now.strftime("%Y-%m-%d")
                # ONE brief per day, at send_time_1 only.
                if settings.get("send_time_1") == hhmm:
                    slot_id = f"{today}#1"
                    if settings.get("last_sent_slot") != slot_id:
                        ok, detail = _send_brief_now(slot_id)
                        edb.update_email_settings(last_sent_slot=slot_id)
                        logger.info(f"[Scheduler] slot {slot_id}: {ok} {detail}")
        except Exception as exc:
            logger.warning(f"[Scheduler] error: {exc}")
        time.sleep(30)


# ---------------------------------------------------------------------------
# Routes
# ---------------------------------------------------------------------------

@app.get("/health")
async def health():
    return {"status": "ok"}

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(a: FastAPI):
    threading.Thread(target=refresh_loop, daemon=True).start()
    threading.Thread(target=email_scheduler, daemon=True).start()
    # Timed exit for live trades, independent of the analysis pipeline so a
    # stalled pipeline can never leave a leveraged position open past its hold.
    threading.Thread(target=trader.watchdog_loop, daemon=True).start()
    yield

app.router.lifespan_context = lifespan


@app.api_route("/", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def index():
    # Public front door pre-launch = the waitlist. The real terminal lives at
    # /terminal behind the preview password.
    return HTMLResponse(_JOIN_HTML)


def _preview_gate_page() -> str:
    body = ('<p>This is a private preview. Enter the access password to open the '
            'terminal.</p>'
            '<form onsubmit="g(event)"><input type="password" id="p" placeholder="Password" required>'
            '<button type="submit">Enter</button></form>'
            '<p id="m" style="margin-top:12px;color:#F2555A"></p>'
            '<script>async function g(ev){ev.preventDefault();'
            'var r=await fetch("/api/preview-login",{method:"POST",headers:{"content-type":"application/json"},'
            'body:JSON.stringify({password:document.getElementById("p").value})});'
            'if(r.ok){location.href="/terminal";}else{document.getElementById("m").textContent="Wrong password.";}}</script>')
    return _page("Private preview", "&#128274; Private preview", body)


@app.get("/terminal", response_class=HTMLResponse)
async def terminal(request: Request):
    # Terminal is open to everyone, no login required. Accounts add the daily
    # email brief and the AI analyst, which the UI nudges toward.
    return HTMLResponse(_HTML)


@app.post("/api/preview-login")
async def preview_login(payload: dict):
    if payload.get("password") == PREVIEW_PASSWORD:
        resp = JSONResponse({"ok": True})
        resp.set_cookie("crx_preview", _PREVIEW_TOKEN, httponly=True,
                        samesite="lax", max_age=60 * 60 * 24 * 30)
        return resp
    return JSONResponse({"ok": False, "error": "Wrong password"}, status_code=401)


@app.api_route("/terms", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def terms():
    return HTMLResponse(_TERMS_HTML)


# ---------------------------------------------------------------------------
# SEO content pages — Articles (blog), FAQ, About. Data-driven + structured data.
# ---------------------------------------------------------------------------
@app.api_route("/articles", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def articles_index(request: Request):
    return HTMLResponse(site_content.render_articles_index(_public_base(request)))


@app.api_route("/articles/{slug}", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def article_page(slug: str, request: Request):
    html = site_content.render_article(_public_base(request), slug)
    if html is None:
        return RedirectResponse("/articles", status_code=302)
    return HTMLResponse(html)


@app.api_route("/faq", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def faq_page(request: Request):
    return HTMLResponse(site_content.render_faq(_public_base(request)))


@app.api_route("/about", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def about_page(request: Request):
    return HTMLResponse(site_content.render_about(_public_base(request)))


@app.api_route("/tools", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def tools_page(request: Request):
    return HTMLResponse(site_content.render_tools(_public_base(request)))


# ---------------------------------------------------------------------------
# Auth pages + API (accounts foundation; terminal stays preview-gated for now)
# ---------------------------------------------------------------------------

def _auth_html(mode: str) -> str:
    login = mode == "login"
    title = "Log in" if login else "Create account"
    other = ('<a href="/signup">Create one</a>' if login
             else '<a href="/login">Log in</a>')
    other_label = ("New here? " if login else "Already have an account? ")
    action = "/api/auth/login" if login else "/api/auth/signup"
    google_btn = ''
    if google_configured():
        google_btn = (
            '<a href="/auth/google/login" style="display:flex;align-items:center;justify-content:center;'
            'gap:10px;border:1px solid #23262f;border-radius:12px;padding:12px;color:#EAECEF;'
            'font-weight:600;font-size:14px;text-decoration:none;margin-bottom:14px">'
            '<svg width="18" height="18" viewBox="0 0 48 48"><path fill="#EA4335" d="M24 9.5c3.5 0 6.6 1.2 9 3.6l6.7-6.7C35.9 2.4 30.4 0 24 0 14.6 0 6.4 5.4 2.5 13.3l7.8 6c1.9-5.6 7.1-9.8 13.7-9.8z"/><path fill="#4285F4" d="M46.1 24.5c0-1.6-.1-3.1-.4-4.5H24v9h12.4c-.5 2.9-2.1 5.3-4.6 7l7.1 5.5c4.2-3.9 6.6-9.6 6.6-16.5z"/><path fill="#FBBC05" d="M10.3 28.7c-.5-1.4-.7-2.9-.7-4.7s.3-3.3.7-4.7l-7.8-6C.9 16.5 0 20.1 0 24s.9 7.5 2.5 10.7l7.8-6z"/><path fill="#34A853" d="M24 48c6.5 0 11.9-2.1 15.9-5.8l-7.1-5.5c-2 1.3-4.6 2.1-8.8 2.1-6.6 0-11.8-4.2-13.7-9.8l-7.8 6C6.4 42.6 14.6 48 24 48z"/></svg>'
            'Continue with Google</a>'
            '<div style="display:flex;align-items:center;gap:12px;color:#5b6070;font-size:11px;margin-bottom:14px">'
            '<span style="flex:1;height:1px;background:#23262f"></span>OR<span style="flex:1;height:1px;background:#23262f"></span></div>'
        )
    body = (
        f'<p style="color:#7A8090;font-size:13px;margin:-6px 0 18px">'
        f'{"Welcome back." if login else "Free account, get the daily brief and the AI analyst."}</p>'
        + google_btn +
        '<form onsubmit="return go(event)">'
        '<input type="email" id="email" placeholder="you@email.com" required autocomplete="email">'
        f'<input type="password" id="password" placeholder="Password'
        f'{"" if login else " (min 8 chars)"}" required '
        f'autocomplete="{"current-password" if login else "new-password"}" minlength="8">'
        f'<button type="submit" id="sb">{title}</button>'
        '</form>'
        '<p class="msg" id="m"></p>'
        f'<p style="color:#7A8090;font-size:12.5px;margin-top:18px">{other_label}{other}</p>'
        f'<script>async function go(e){{e.preventDefault();'
        'var m=document.getElementById("m"),b=document.getElementById("sb");'
        'm.textContent="";b.disabled=true;'
        f'var r=await fetch("{action}",{{method:"POST",headers:{{"content-type":"application/json"}},'
        'body:JSON.stringify({email:document.getElementById("email").value,'
        'password:document.getElementById("password").value})});'
        'var d=await r.json();b.disabled=false;'
        'if(d.ok){location.href="/terminal";}else{m.className="msg err";m.textContent=d.error||"Failed";}'
        'return false;}'
        # show any ?err= from a failed Google round-trip
        'var q=new URLSearchParams(location.search),e=q.get("err");'
        'if(e){var mm=document.getElementById("m");mm.className="msg err";'
        'mm.textContent="Sign-in error: "+e.replace("google_","")+(q.get("d")?" ("+q.get("d")+")":"");}'
        '</script>'
    )
    return _page(title + ", Sentiments Analyzer", title, body)


@app.api_route("/login", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def login_page():
    return HTMLResponse(_auth_html("login"))


@app.api_route("/signup", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def signup_page():
    return HTMLResponse(_auth_html("signup"))


@app.get("/account", response_class=HTMLResponse)
async def account_page(request: Request):
    email = _current_user(request)
    if not email:
        return HTMLResponse('<meta http-equiv="refresh" content="0; url=/login">')
    u = edb.get_user(email) or {}
    plan = (u.get("plan") or "free").title()
    body = (
        f'<p style="color:#EAECEF;font-size:14px">Signed in as <b>{email}</b></p>'
        f'<p style="color:#7A8090;font-size:13px;margin:8px 0 20px">Plan: '
        f'<b style="color:#E8A33D">{plan}</b> &nbsp;·&nbsp; full terminal access included, free.</p>'
        '<div style="text-align:left;margin:0 0 20px">'
        '<div style="color:#7A8090;font-size:12px;margin-bottom:8px">EXPERIENCE LEVEL</div>'
        '<div id="lvl-opts" style="display:flex;flex-direction:column;gap:8px">'
        '<button class="lo" data-lvl="1" onclick="setLvl(1)">Beginner</button>'
        '<button class="lo" data-lvl="2" onclick="setLvl(2)">Intermediate</button>'
        '<button class="lo" data-lvl="3" onclick="setLvl(3)">Advanced</button>'
        '</div></div>'
        '<a class="btn" href="/terminal">Open the terminal &rarr;</a>'
        '<p style="margin-top:16px"><a href="#" onclick="lo();return false;" '
        'style="color:#7A8090;font-size:12.5px">Log out</a></p>'
        '<style>.lo{background:rgba(255,255,255,0.03);border:1px solid #23262f;color:#EAECEF;'
        'font-family:inherit;font-size:14px;padding:11px 14px;border-radius:10px;cursor:pointer;text-align:left}'
        '.lo.on{background:rgba(232,163,61,0.14);border-color:rgba(232,163,61,0.5);color:#F5B65A;font-weight:700}</style>'
        '<script>'
        'function mark(){var c=Number(localStorage.getItem("crx_level")||1);'
        'document.querySelectorAll(".lo").forEach(function(b){b.classList.toggle("on",Number(b.dataset.lvl)===c);});}'
        'function setLvl(l){localStorage.setItem("crx_level",String(l));mark();}'
        'mark();'
        'async function lo(){await fetch("/api/auth/logout",{method:"POST"});location.href="/";}</script>'
    )
    return HTMLResponse(_page("Account, Sentiments Analyzer", "&#128100; Your account", body))


@app.post("/api/auth/signup")
async def auth_signup(payload: dict):
    email = (payload.get("email") or "").strip().lower()
    password = payload.get("password") or ""
    if not _valid_email(email):
        return JSONResponse({"ok": False, "error": "Enter a valid email."}, status_code=400)
    if len(password) < 8:
        return JSONResponse({"ok": False, "error": "Password must be at least 8 characters."},
                            status_code=400)
    h, s = _hash_pw(password)
    if not edb.create_user(email, h, s):
        return JSONResponse({"ok": False, "error": "An account with this email already exists."},
                            status_code=409)
    edb.add_client(email, "")   # subscribe the new account to the daily brief
    _persist_now()
    token = secrets.token_urlsafe(24)
    _sessions[token] = email
    resp = JSONResponse({"ok": True})
    resp.set_cookie("crx_session", token, httponly=True, samesite="lax",
                    max_age=60 * 60 * 24 * 30)
    return resp


@app.post("/api/auth/login")
async def auth_login(payload: dict):
    email = (payload.get("email") or "").strip().lower()
    password = payload.get("password") or ""
    if _rate_limited(email):
        return JSONResponse({"ok": False, "error": "Too many attempts. Try again in a few minutes."},
                            status_code=429)
    u = edb.get_user(email)
    if not u or not _verify_pw(password, u.get("pw_salt", ""), u.get("pw_hash", "")):
        _record_fail(email)
        return JSONResponse({"ok": False, "error": "Wrong email or password."}, status_code=401)
    _login_fails.pop(email, None)
    token = secrets.token_urlsafe(24)
    _sessions[token] = email
    resp = JSONResponse({"ok": True})
    resp.set_cookie("crx_session", token, httponly=True, samesite="lax",
                    max_age=60 * 60 * 24 * 30)
    return resp


@app.post("/api/auth/logout")
async def auth_logout(request: Request):
    _sessions.pop(request.cookies.get("crx_session", ""), None)
    resp = JSONResponse({"ok": True})
    resp.delete_cookie("crx_session")
    return resp


# ---------------------------------------------------------------------------
# Google OAuth (one-click sign-in; email comes verified -> no confirm email)
# ---------------------------------------------------------------------------
GOOGLE_CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID", "")
GOOGLE_CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET", "")
_GOOGLE_AUTH = "https://accounts.google.com/o/oauth2/v2/auth"
_GOOGLE_TOKEN = "https://oauth2.googleapis.com/token"
_GOOGLE_USERINFO = "https://openidconnect.googleapis.com/v1/userinfo"


def google_configured() -> bool:
    return bool(GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET)


def _new_session_for(email: str) -> str:
    token = secrets.token_urlsafe(24)
    _sessions[token] = email
    return token


from fastapi.responses import RedirectResponse


@app.get("/auth/google/login")
async def google_login(request: Request):
    if not google_configured():
        return RedirectResponse("/login?err=google_config", status_code=302)
    import urllib.parse
    state = secrets.token_urlsafe(16)
    redirect_uri = _public_base(request) + "/auth/google/callback"
    params = urllib.parse.urlencode({
        "client_id": GOOGLE_CLIENT_ID,
        "redirect_uri": redirect_uri,
        "response_type": "code",
        "scope": "openid email profile",
        "state": state,
        "access_type": "online",
        "prompt": "select_account",
    })
    resp = RedirectResponse(f"{_GOOGLE_AUTH}?{params}", status_code=302)
    # Not httponly so a same-site check still works even if the return nav drops
    # httponly lax edge cases; still secure enough for a short-lived CSRF token.
    resp.set_cookie("crx_oauth_state", state, samesite="lax", max_age=600,
                    secure=True, path="/")
    return resp


@app.get("/auth/google/callback")
async def google_callback(request: Request, code: str = "", state: str = "", error: str = ""):
    def _fail(reason):
        logger.warning(f"[Google] callback fail: {reason}")
        return RedirectResponse(f"/login?err=google_{reason}", status_code=302)

    if error:
        return _fail("denied")
    if not google_configured():
        return _fail("config")
    if not code:
        return _fail("nocode")
    if not state or state != request.cookies.get("crx_oauth_state", ""):
        return _fail("state")
    redirect_uri = _public_base(request) + "/auth/google/callback"
    try:
        tok = requests.post(_GOOGLE_TOKEN, data={
            "code": code, "client_id": GOOGLE_CLIENT_ID,
            "client_secret": GOOGLE_CLIENT_SECRET, "redirect_uri": redirect_uri,
            "grant_type": "authorization_code",
        }, timeout=20)
        if tok.status_code != 200:
            logger.warning(f"[Google] token {tok.status_code}: {tok.text[:200]}")
            return _fail("token")
        tdata = tok.json()
        # Prefer the id_token (JWT from Google over HTTPS -> trusted): decode its
        # payload for the email, avoids a second userinfo network call.
        email, name = "", ""
        id_token = tdata.get("id_token")
        if id_token and id_token.count(".") == 2:
            import base64
            pl = id_token.split(".")[1]
            pl += "=" * (-len(pl) % 4)
            claims = json.loads(base64.urlsafe_b64decode(pl.encode()))
            email = (claims.get("email") or "").strip().lower()
            name = claims.get("name", "")
        if not email:   # fallback to userinfo endpoint
            access = tdata.get("access_token")
            if access:
                info = requests.get(_GOOGLE_USERINFO,
                                    headers={"Authorization": f"Bearer {access}"}, timeout=20).json()
                email = (info.get("email") or "").strip().lower()
                name = info.get("name", "")
        if not email:
            return _fail("email")
    except Exception as exc:
        import urllib.parse
        detail = urllib.parse.quote(f"{type(exc).__name__}: {exc}"[:120])
        logger.warning(f"[Google] callback exception: {exc!r}")
        return RedirectResponse(f"/login?err=google_exception&d={detail}", status_code=302)

    is_new = edb.ensure_oauth_user(email)
    if is_new:
        edb.add_client(email, name or "")             # subscribe to the daily brief
        _persist_now()
    token = _new_session_for(email)
    resp = RedirectResponse("/terminal", status_code=302)
    resp.set_cookie("crx_session", token, httponly=True, samesite="lax",
                    secure=True, max_age=60 * 60 * 24 * 30, path="/")
    resp.delete_cookie("crx_oauth_state")
    return resp


@app.get("/api/auth/me")
async def auth_me(request: Request):
    email = _current_user(request)
    return {"email": email, "google": google_configured()}


# --- SEO: let crawlers index the public waitlist, never the gated product ---
@app.api_route("/robots.txt", methods=["GET", "HEAD"])
async def robots(request: Request):
    base = _public_base(request)
    body = (
        "User-agent: *\n"
        "Allow: /$\n"
        "Allow: /tools\n"
        "Allow: /articles\n"
        "Allow: /faq\n"
        "Allow: /about\n"
        "Allow: /terms\n"
        "Disallow: /terminal\n"
        "Disallow: /admin\n"
        "Disallow: /api/\n"
        "Disallow: /confirm\n"
        "Disallow: /unsubscribe\n"
        f"\nSitemap: {base}/sitemap.xml\n"
    )
    return Response(content=body, media_type="text/plain")


@app.api_route("/sitemap.xml", methods=["GET", "HEAD"])
async def sitemap(request: Request):
    base = _public_base(request)
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    body = (
        '<?xml version="1.0" encoding="UTF-8"?>\n'
        '<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">\n'
        f"  <url><loc>{base}/</loc><lastmod>{today}</lastmod>"
        "<changefreq>daily</changefreq><priority>1.0</priority></url>\n"
        f"  <url><loc>{base}/tools</loc><lastmod>{today}</lastmod>"
        "<changefreq>monthly</changefreq><priority>0.7</priority></url>\n"
        f"  <url><loc>{base}/articles</loc><lastmod>{today}</lastmod>"
        "<changefreq>weekly</changefreq><priority>0.8</priority></url>\n"
        f"  <url><loc>{base}/faq</loc><lastmod>{today}</lastmod>"
        "<changefreq>monthly</changefreq><priority>0.7</priority></url>\n"
        f"  <url><loc>{base}/about</loc><lastmod>{today}</lastmod>"
        "<changefreq>monthly</changefreq><priority>0.5</priority></url>\n"
        + "".join(
            f"  <url><loc>{base}/articles/{s}</loc><lastmod>{today}</lastmod>"
            "<changefreq>monthly</changefreq><priority>0.6</priority></url>\n"
            for s in site_content.article_slugs()
        ) +
        f"  <url><loc>{base}/terms</loc><lastmod>{today}</lastmod>"
        "<changefreq>monthly</changefreq><priority>0.3</priority></url>\n"
        "</urlset>\n"
    )
    return Response(content=body, media_type="application/xml")


# ---------------------------------------------------------------------------
# Public self-serve signup (double opt-in), the SaaS front door
# ---------------------------------------------------------------------------

def _valid_email(e: str) -> bool:
    e = (e or "").strip()
    if len(e) < 5 or len(e) > 254 or " " in e or e.count("@") != 1:
        return False
    local, _, domain = e.partition("@")
    return bool(local) and "." in domain and not domain.startswith(".") \
        and not domain.endswith(".")


def _mask_email(e: str) -> str:
    """Privacy-safe display: 'ar***@gmail.com' (keeps enough for self-recognition)."""
    local, _, domain = (e or "").partition("@")
    if not domain:
        return "someone"
    head = local[:2] if len(local) > 2 else local[:1]
    return f"{head}***@{domain}"


def _page(title: str, heading: str, body: str, accent: str = "#E8A33D") -> str:
    """Minimal themed page for confirm/unsubscribe results."""
    return f"""<!DOCTYPE html><html lang="en"><head><meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0"><title>{title}</title>
<meta name="robots" content="noindex, nofollow">
<link rel="icon" type="image/svg+xml" href="/static/favicon.svg">
<style>@import url('https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700;800&display=swap');
body{{background:#0a0b0f;color:#EAECEF;font-family:'Inter',system-ui,sans-serif;display:flex;
min-height:100vh;align-items:center;justify-content:center;margin:0;padding:20px;-webkit-font-smoothing:antialiased;
background-image:radial-gradient(700px 420px at 50% 0%, rgba(232,163,61,0.07), transparent 60%)}}
.box{{max-width:400px;width:100%;text-align:center;background:rgba(24,27,35,0.6);backdrop-filter:blur(10px);
border:1px solid #23262f;border-radius:18px;padding:38px 32px;box-shadow:0 10px 40px rgba(0,0,0,.35)}}
h1{{color:{accent};font-size:22px;font-weight:800;margin:0 0 14px;letter-spacing:-0.3px}}
p{{color:#7A8090;font-size:14px;line-height:1.6;margin:0 0 20px}}
a{{color:{accent};text-decoration:none}} a:hover{{color:#F5B65A}}
a.btn{{display:inline-block;background:{accent};color:#0a0b0e;text-decoration:none;font-weight:700;
padding:11px 22px;border-radius:11px;font-size:14px}}
input{{background:rgba(0,0,0,0.28);border:1px solid #23262f;border-radius:11px;padding:12px 14px;color:#EAECEF;
font-family:inherit;font-size:14px;width:100%;margin-bottom:11px;box-sizing:border-box;outline:none}}
input:focus{{border-color:rgba(232,163,61,0.5)}}
button{{background:{accent};color:#0a0b0e;border:none;border-radius:11px;padding:12px 22px;
font-family:inherit;font-weight:700;font-size:14px;cursor:pointer;width:100%;transition:.15s}}
button:hover{{background:#F5B65A}} button:disabled{{opacity:.6;cursor:default}}
.msg{{min-height:18px;font-size:13px;margin:2px 0 0}} .msg.err{{color:#F2555A}}</style></head>
<body><div class="box"><h1>{heading}</h1>{body}</div></body></html>"""


# Waitlist cap, never take more sign-ups than we can actually email.
# Brevo free = 300 emails/day, so 300 is the natural ceiling. Override with env.
WAITLIST_CAP = int(os.getenv("WAITLIST_CAP", "300"))


def _public_base(request: Request) -> str:
    """The public-facing base URL for building links. Prefers an explicit env,
    then the proxy's forwarded host (so links use the custom domain, not the
    internal hf.space host), then the request's own base URL."""
    env = os.getenv("PUBLIC_BASE_URL")
    if env:
        return env.rstrip("/")
    xf_host = request.headers.get("x-forwarded-host")
    if xf_host:
        proto = request.headers.get("x-forwarded-proto", "https")
        return f"{proto}://{xf_host}".rstrip("/")
    return str(request.base_url).rstrip("/")


@app.api_route("/join", methods=["GET", "HEAD"], response_class=HTMLResponse)
async def join_page():
    return HTMLResponse(_JOIN_HTML)


@app.post("/api/subscribe")
async def subscribe(payload: dict, request: Request):
    email = (payload.get("email") or "").strip().lower()
    if not _valid_email(email):
        return JSONResponse({"ok": False, "error": "Please enter a valid email."},
                            status_code=400)
    # Enforce the cap on confirmed subscribers (the ones we actually email).
    if edb.subscriber_count(confirmed_only=True) >= WAITLIST_CAP:
        return JSONResponse(
            {"ok": False, "error": "The waitlist is full for now, we've hit capacity. "
                                   "Check back soon as we open more spots."},
            status_code=409)
    token = secrets.token_urlsafe(24)
    ref_code = secrets.token_urlsafe(6)                      # this signup's own code
    referred_by = (payload.get("ref") or "").strip()[:32]    # who referred them
    result = edb.add_pending_subscriber(email, token, source="web",
                                       ref_code=ref_code, referred_by=referred_by)
    if result == "exists":
        return {"ok": True, "message": "You're already on the waitlist, check your inbox."}
    _persist_now()   # save the new pending signup right away
    # Build confirm link from the public base URL (custom domain, not hf.space)
    base = _public_base(request)
    confirm_url = f"{base}/confirm?token={token}"
    subject = "Confirm your spot on the Crypto Narrative waitlist"
    html = f"""<div style="font-family:Arial,sans-serif;max-width:480px;margin:auto;color:#222">
      <h2 style="color:#B4791E">Confirm your waitlist spot</h2>
      <p>Thanks for joining the <b>Crypto Narrative</b> waitlist.
      Click below to confirm your email, we'll notify you the moment early access opens:</p>
      <p><a href="{confirm_url}" style="display:inline-block;background:#E8A33D;color:#000;
      padding:12px 22px;border-radius:8px;text-decoration:none;font-weight:bold">Confirm my email</a></p>
      <p style="color:#888;font-size:12px">If you didn't request this, just ignore this email —
      you won't be added. Not financial advice.</p></div>"""
    ok, detail = mailer.send_transactional(email, subject, html)
    if not ok:
        logger.warning(f"[Subscribe] confirmation email failed for {email}: {detail}")
        return JSONResponse(
            {"ok": False, "error": "Couldn't send the confirmation email. Try again later."},
            status_code=502)
    return {"ok": True, "message": "Almost there! Check your inbox, and your spam or "
                                   "promotions folder, to confirm your spot."}


@app.get("/confirm", response_class=HTMLResponse)
async def confirm(token: str = "", request: Request = None):
    info = edb.confirm_subscriber(token)
    if info:
        _persist_now()   # persist the confirmation + referral credit immediately
        base = _public_base(request)
        ref_link = f"{base}/?ref={info['ref_code']}"
        body = (
            '<p>You\'re on the waitlist. We\'ll email you the moment early access '
            'opens, no spam in between.</p>'
            '<p style="color:#E6E7EB;margin-bottom:8px">Want in sooner? Share your '
            'link, every friend who joins moves you up.</p>'
            f'<input id="rl" readonly value="{ref_link}" onclick="this.select()">'
            '<button onclick="cp()">Copy my link</button>'
            f'<p style="margin-top:14px"><b id="rc" style="color:#26D07C">{info["referrals"]}</b>'
            ' friends referred so far</p>'
            f'<script>const C="{info["ref_code"]}";'
            'function cp(){var i=document.getElementById("rl");i.select();'
            'navigator.clipboard&&navigator.clipboard.writeText(i.value);'
            'var b=event.target;b.textContent="Copied!";setTimeout(()=>b.textContent="Copy my link",1500);}'
            'fetch("/api/ref?code="+C).then(r=>r.json()).then(d=>{if(d.referrals!=null)'
            'document.getElementById("rc").textContent=d.referrals;}).catch(()=>{});</script>')
        return HTMLResponse(_page("You're on the list", "&#10003; You're on the list!", body))
    body = ('<p>This confirmation link is invalid or has expired. '
            'Try subscribing again.</p><a class="btn" href="/join">Back to signup</a>')
    return HTMLResponse(_page("Invalid link", "Link not valid", body, accent="#F2555A"))


@app.get("/api/ref")
async def ref_status(code: str = ""):
    n = edb.referral_count(code)
    return JSONResponse({"referrals": n})


@app.get("/unsubscribe", response_class=HTMLResponse)
async def unsubscribe_page():
    body = ('<p>Enter your email to stop receiving the daily brief.</p>'
            '<form onsubmit="u(event)"><input type="email" id="e" placeholder="you@email.com" required>'
            '<button type="submit">Unsubscribe</button></form>'
            '<p id="m" style="margin-top:14px"></p>'
            '<script>async function u(ev){ev.preventDefault();var e=document.getElementById("e").value;'
            'var r=await fetch("/api/unsubscribe",{method:"POST",headers:{"content-type":"application/json"},'
            'body:JSON.stringify({email:e})});var d=await r.json();'
            'document.getElementById("m").textContent=d.message||d.error||"";}</script>')
    return HTMLResponse(_page("Unsubscribe", "Unsubscribe", body))


@app.post("/api/unsubscribe")
async def unsubscribe(payload: dict):
    email = (payload.get("email") or "").strip().lower()
    if not _valid_email(email):
        return JSONResponse({"ok": False, "error": "Please enter a valid email."}, status_code=400)
    edb.unsubscribe_email(email)
    _persist_now()
    # Always report success (don't reveal whether an email was on the list)
    return {"ok": True, "message": "Done, you've been unsubscribed."}


@app.get("/api/data")
async def get_data(request: Request):
    # Open, the terminal is free for everyone.
    with _cache_lock:
        if not _cache:
            return JSONResponse({"status": "loading", "is_fetching": _is_fetching})
        return JSONResponse({**_cache, "last_update": _last_update, "is_fetching": _is_fetching})


@app.post("/api/ask")
async def ask_ai(payload: dict, request: Request):
    """Premium AI assistant, answers from the terminal's own live data.
    Gated behind the preview cookie (part of the paid terminal)."""
    if not _has_access(request):
        return JSONResponse({"ok": False, "error": "gated"}, status_code=403)
    with _cache_lock:
        snapshot = dict(_cache)
    ok, answer = ai_assistant.ask(payload.get("question", ""), snapshot)
    return {"ok": ok, "answer": answer,
            "configured": ai_assistant.ai_configured()}


@app.get("/api/ask/status")
async def ask_status(request: Request):
    if not _has_access(request):
        return JSONResponse({"configured": False}, status_code=403)
    return {"configured": ai_assistant.ai_configured()}


@app.get("/api/teaser")
async def get_teaser():
    """Public, safe subset for the waitlist page, headline numbers only."""
    with _cache_lock:
        c = _cache or {}
        o = (c.get("sentiment_24h") or {}).get("overall", {}) or {}
        sc = c.get("scorecard") or {}
        try:
            waitlist = edb.subscriber_count(confirmed_only=True)
        except Exception:
            waitlist = None
        try:
            leaderboard = [{"name": _mask_email(r["email"]), "referrals": r["referrals"]}
                           for r in edb.top_referrers(3)]
        except Exception:
            leaderboard = []
        return JSONResponse({
            "index":           o.get("index"),
            "signal_hit_rate": sc.get("signal_hit_rate"),
            "win_rate":        sc.get("win_rate"),
            "narratives":      len(c.get("narratives") or []),
            "waitlist":        waitlist,
            "leaderboard":     leaderboard,
            # Public teaser of the price/index overlay, shows the product works.
            "chart_series":    c.get("chart_series") or [],
            "divergence":      c.get("divergence") or {},
        })


def _public_output(cache: dict) -> dict:
    """The canonical public JSON other platforms consume."""
    s = cache.get("sentiment_24h") or {}
    o = s.get("overall") or {}
    label = (o.get("label") or "").title() or None
    conf = o.get("confidence")
    # short narrative labels (e.g. "ETF Approval" -> "ETF", "AI & Crypto" -> "AI")
    def _short(nm):
        nm = (nm or "").strip()
        for cut in (" (", " / ", " & ", " Approval", " Adoption", " Scaling"):
            if cut in nm:
                nm = nm.split(cut)[0]
        return nm.split()[0] if nm else nm
    narrs = [_short(n.get("name")) for n in (cache.get("narratives") or [])[:3] if n.get("name")]
    def _ascii(t):
        return (t or "").replace("—", "-").replace("→", "->").replace("↓", "down").replace("↑", "up")
    warnings = []
    for a in (cache.get("spike_alerts") or [])[:4]:
        n, m = a.get("narrative"), a.get("message")
        warnings.append(_ascii(f"{n}: {m}" if n and m else (m or f"{n} spike")))
    # add manufactured-hype warnings
    warnings += [_ascii(w) for w in (cache.get("hype") or {}).get("warnings", [])[:3]]
    f = cache.get("forecast") or {}
    direction = (f.get("direction") or o.get("label") or "").title()
    forecast = f"{direction} next 48H" if direction else None
    return {
        "market_sentiment": o.get("index"),
        "trend": label,
        "confidence": round(conf * 100) if isinstance(conf, (int, float)) else None,
        "leading_narratives": narrs,
        "warnings": warnings,
        "forecast": forecast,
    }


@app.api_route("/api/v1/sentiment", methods=["GET", "HEAD"])
async def public_sentiment():
    """Public, versioned API, the canonical market-psychology output for other
    platforms to consume. Open + CORS-enabled."""
    with _cache_lock:
        c = dict(_cache) if _cache else {}
    # ALWAYS the same fixed 6-key schema so every response is consistent;
    # before the first pipeline run the values are just null / empty lists.
    body = _public_output(c)
    return JSONResponse(body, headers={
        "Access-Control-Allow-Origin": "*",
        "Cache-Control": "public, max-age=60",
    })


@app.get("/api/refresh")
async def trigger_refresh(request: Request):
    """Manually trigger a data refresh, gated so the public can't spam it."""
    if not _has_access(request):
        return JSONResponse({"status": "gated"}, status_code=403)
    if not _is_fetching:
        t = threading.Thread(target=lambda: _cache.update(run_pipeline()), daemon=True)
        t.start()
    return {"status": "refreshing"}


# ---------------------------------------------------------------------------
# Admin panel
# ---------------------------------------------------------------------------

@app.get("/admin", response_class=HTMLResponse)
async def admin_page():
    return HTMLResponse(_ADMIN_HTML)


@app.post("/api/admin/login")
async def admin_login(payload: dict):
    if payload.get("password") == ADMIN_PASSWORD:
        token = secrets.token_urlsafe(24)
        _admin_tokens.add(token)
        return {"ok": True, "token": token}
    return JSONResponse({"ok": False, "error": "Wrong password"}, status_code=401)


def _auth_or_401(token: str | None):
    if not _check_admin(token):
        return JSONResponse({"ok": False, "error": "Unauthorized"}, status_code=401)
    return None


# ── live trading (admin only; all money settings come from env/secrets) ──
@app.get("/api/admin/trade/status")
async def admin_trade_status(x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    return {"ok": True, "trade": trader.status()}


@app.get("/api/admin/trade/diagnose")
async def admin_trade_diagnose(x_token: str | None = Header(default=None)):
    """Which MEXC hosts/paths this server can actually reach. Places no orders."""
    err = _auth_or_401(x_token)
    if err: return err
    return {"ok": True, "checks": trader.diagnose()}


@app.post("/api/admin/trade/test")
async def admin_trade_test(x_token: str | None = Header(default=None)):
    """Smallest possible live round-trip so you can verify orders really land."""
    err = _auth_or_401(x_token)
    if err: return err
    ok, msg = trader.test_trade()
    return JSONResponse({"ok": ok, "message": msg}, status_code=200 if ok else 400)


@app.post("/api/admin/trade/close")
async def admin_trade_close(x_token: str | None = Header(default=None)):
    """Panic button: cancel all orders and market-close any open position."""
    err = _auth_or_401(x_token)
    if err: return err
    ok, msg = trader.panic_close()
    return JSONResponse({"ok": ok, "message": msg}, status_code=200 if ok else 400)


@app.get("/api/admin/state")
async def admin_state(x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    return {
        "ok": True,
        "clients":  edb.get_clients(),
        "settings": edb.get_email_settings(),
        "log":      edb.get_email_log(20),
        "smtp_configured": mailer.email_configured(),
        "email_provider": "Brevo (HTTP)" if mailer.brevo_configured() else ("SMTP" if mailer.smtp_configured() else "none"),
        "has_data": bool(_cache.get("sentiment_24h")),
        "scorecard": _cache.get("scorecard", {}),
        "data_span": db.sentiment_span(),
        "index_basis": (_cache.get("sentiment_24h", {}).get("overall", {}) or {}).get("index_basis"),
        "baseline_net": (_cache.get("sentiment_24h", {}).get("overall", {}) or {}).get("baseline_net"),
    }


@app.post("/api/admin/clients")
async def admin_add_client(payload: dict, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    email = (payload.get("email") or "").strip()
    if not email or "@" not in email:
        return JSONResponse({"ok": False, "error": "Invalid email"}, status_code=400)
    ok = edb.add_client(email, payload.get("name", ""))
    return {"ok": ok, "error": None if ok else "Already exists"}


@app.delete("/api/admin/clients/{client_id}")
async def admin_remove_client(client_id: int, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    edb.remove_client(client_id)
    return {"ok": True}


@app.post("/api/admin/clients/{client_id}/toggle")
async def admin_toggle_client(client_id: int, payload: dict, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    edb.set_client_active(client_id, bool(payload.get("active")))
    return {"ok": True}


@app.post("/api/admin/settings")
async def admin_settings(payload: dict, x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    edb.update_email_settings(
        enabled=1 if payload.get("enabled") else 0,
        send_time_1=payload.get("send_time_1", "09:00"),
        send_time_2=payload.get("send_time_2", "21:00"),
        subject_prefix=payload.get("subject_prefix", "Crypto Narrative Brief"),
    )
    return {"ok": True, "settings": edb.get_email_settings()}


@app.post("/api/admin/send-now")
async def admin_send_now(x_token: str | None = Header(default=None)):
    err = _auth_or_401(x_token)
    if err: return err
    ok, detail = _send_brief_now("manual")
    return {"ok": ok, "detail": detail}


# ---------------------------------------------------------------------------
# Cron endpoints, hit by an external free scheduler (GitHub Actions) so
# briefs fire and the host stays awake even on sleepy free tiers.
# Secured by a shared secret (CRON_SECRET in .env).
# ---------------------------------------------------------------------------
CRON_SECRET = os.getenv("CRON_SECRET", "")


@app.get("/api/cron/ping")
async def cron_ping(key: str = ""):
    """Keep-alive + ensure data is fresh. Safe to call frequently."""
    if CRON_SECRET and key != CRON_SECRET:
        return JSONResponse({"ok": False, "error": "bad key"}, status_code=403)
    if not _is_fetching and not _cache:
        threading.Thread(target=lambda: _cache.update(run_pipeline()), daemon=True).start()
    return {"ok": True, "awake": True, "has_data": bool(_cache.get("sentiment_24h"))}


@app.get("/api/cron/send-brief")
async def cron_send_brief(key: str = ""):
    """Send the brief now to all active clients (called at scheduled times).
    Fail-closed: refuses if CRON_SECRET is unset or the key doesn't match, so a
    missing secret can never leave mass-email triggering open to the public."""
    if not CRON_SECRET or key != CRON_SECRET:
        return JSONResponse({"ok": False, "error": "bad key"}, status_code=403)
    ok, detail = _send_brief_now("cron")
    return {"ok": ok, "detail": detail}


if __name__ == "__main__":
    print("\n  Crypto Narrative Terminal")
    print("  Open http://localhost:8000 in your browser\n")
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="warning")