#!/usr/bin/env python3
"""
XAU/USD static dashboard publisher.

Standalone daemon that scans XAU/USD at candle closes and atomically
writes `snapshot.json` + `news.json` (plus a copy of
`static/dashboard.html`) into an Apache-served directory. Apache does
all web serving — this process never opens a port, so a failure here
can only ever mean a stale page.

Independence rules (owner requirement — the telegram bot must be
immune to anything on the web side):
  * Runs as its own process; never imported by, and never imports,
    `telegram_bot` / `oanda_daemon` (telegram_bot resolves credentials
    at import). The confirmation map and boundary math are therefore
    duplicated here as local constants.
  * The only shared-code dependency beyond the standard signal path is
    the additive `NewsEngine.get_week_events` accessor, which no
    production signal path calls.

Scheduling: every M15/H1/H4 boundary is also an M5 boundary (all
intervals are multiples of 5, anchored to UTC midnight), so one
sequential loop sleeps to the next M5 boundary (+10 s buffer) and scans
whichever timeframes just closed. Due-classification uses the RAW wake
time — flooring 12:00:10 to the 5-minute grid yields the 12:00 boundary
that just closed. Do NOT subtract the buffer first: 12:00:10 − 15 s
floors to 11:55 and silently skips every H1/H4 scan (v1 bug, 2026-08-02).

Scans run sequentially M5→M15→H1→H4 through one shared data provider so
later timeframes hit the warm boundary cache — that ordering IS the
duplicate-fetch dedup (providers have no in-flight dedup; Oanda has no
throttle). Keep it sequential.

Run:  python dashboard_publisher.py [--once] [--out DIR]
Out dir precedence: --out → DASHBOARD_OUT_DIR env → /var/www/html/gold-dashboard
"""

import argparse
import json
import logging
import math
import numbers
import os
import shutil
import sys
import time
from datetime import datetime, timezone
from logging.handlers import TimedRotatingFileHandler
from typing import Dict, List, Optional

from config import PlatformConfig, Timeframe, XAUUSD
from main import report_to_dict
from news_engine import NewsEngine
from pair_signals import produce_signal
from providers import make_data_provider
import providers.base as _providers_base

logger = logging.getLogger("dashboard")

DASHBOARD_TFS = [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]

# Local copy of the bot's confirmation map (see independence rules above).
CONFIRMATIONS = {
    Timeframe.M5:  [Timeframe.M15, Timeframe.H1],
    Timeframe.M15: [Timeframe.M5, Timeframe.H1],
    Timeframe.H1:  [Timeframe.M15, Timeframe.H4],
    Timeframe.H4:  [Timeframe.H1, Timeframe.D1],
}

TF_INTERVAL_MINUTES = {
    Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.H1: 60, Timeframe.H4: 240,
}

DEFAULT_OUT_DIR = "/var/www/html/gold-dashboard"
SCAN_BUFFER_SECONDS = 10
SNAPSHOT_FILE = "snapshot.json"
NEWS_FILE = "news.json"
PAGE_FILE = "dashboard.html"


# ══════════════════════════════════════════════════════════════════
#  Pure layer (no network, fully covered by test_all.py)
# ══════════════════════════════════════════════════════════════════

def _safe_num(v):
    """Coerce to a JSON-safe float; NaN/±inf become None."""
    try:
        f = float(v)
    except (TypeError, ValueError):
        return v
    if math.isnan(f) or math.isinf(f):
        return None
    return f


def _sanitize(obj):
    """Recursively make a payload safe for json.dumps(allow_nan=False):
    NaN/inf → None, numpy scalars → native ints/floats."""
    if isinstance(obj, bool):
        return obj
    if isinstance(obj, dict):
        return {k: _sanitize(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_sanitize(v) for v in obj]
    if isinstance(obj, numbers.Integral):
        return int(obj)
    if isinstance(obj, numbers.Real):
        return _safe_num(obj)
    return obj


def analysis_to_dict(analysis) -> Optional[dict]:
    """Serialize a TechnicalAnalysis. Iterates analysis.indicators only —
    never vars()/asdict(), which would drag dynamic pandas attrs
    (_rsi_series, _macd_histogram, …) into the payload. ATR (weight 0)
    is surfaced separately as context, not in the directional grid."""
    if analysis is None:
        return None
    indicators = {}
    atr_context = None
    for name, ind in analysis.indicators.items():
        if name.startswith("_"):
            continue
        entry = {
            "signal": ind.signal.name,
            "value": _safe_num(ind.value),
            "confidence": _safe_num(ind.confidence),
            "weight": _safe_num(ind.weight),
            "description": ind.description,
        }
        if name == "atr":
            atr_context = entry
        elif ind.weight != 0:
            indicators[name] = entry
    return {
        "timeframe": analysis.timeframe,
        "timestamp": str(analysis.timestamp),
        "composite_signal": analysis.composite_signal.name,
        "composite_score": _safe_num(analysis.composite_score),
        "confidence": _safe_num(analysis.confidence),
        "market_regime": analysis.market_regime,
        "volatility_regime": analysis.volatility_regime,
        "pattern_signals": list(analysis.pattern_signals),
        "key_levels": {k: _safe_num(v) for k, v in analysis.key_levels.items()},
        "indicators": indicators,
        "atr_context": atr_context,
    }


def report_to_webapp_dict(report) -> dict:
    """main.report_to_dict + the per-TF detail the dashboard needs."""
    d = report_to_dict(report)
    d["primary_signal"] = report.primary_signal.name
    d["primary_score"] = _safe_num(report.primary_score)
    d["mtf_scores"] = {k: _safe_num(v) for k, v in report.mtf_scores.items()}
    d["exhaustion_count"] = report.exhaustion_count
    d["primary_analysis"] = analysis_to_dict(report.primary_analysis)
    d["confirmation_analyses"] = [
        analysis_to_dict(a) for a in report.confirmation_analyses
    ]
    return _sanitize(d)


def event_to_dict(ev) -> dict:
    """EconomicEvent → dict. surprise/minutes_until/is_high_impact are
    properties, so asdict() would skip them — added explicitly.
    actual/forecast/previous stay strings as published ('228K', '0.3%')."""
    return {
        "time": ev.time.astimezone(timezone.utc).isoformat(),
        "currency": ev.currency,
        "impact": ev.impact,
        "event": ev.event,
        "forecast": ev.forecast,
        "previous": ev.previous,
        "actual": ev.actual,
        "surprise": ev.surprise,
        "minutes_until": _safe_num(ev.minutes_until),
        "is_high_impact": ev.is_high_impact,
    }


def headline_to_dict(h) -> dict:
    return {
        "headline": h.headline,
        "source": h.source,
        "timestamp": h.timestamp.astimezone(timezone.utc).isoformat()
                     if h.timestamp else None,
        "currencies": list(h.currencies_mentioned),
        "score": _safe_num(h.sentiment_score),
        "keywords": list(h.keywords),
        "score_by_currency": {
            k: _safe_num(v) for k, v in h.sentiment_by_currency.items()
        },
    }


def build_news_payload(context, week_events, now: datetime) -> dict:
    """news.json: full calendar week + scored headlines + LLM summary."""
    headlines = sorted(
        context.news_headlines,
        key=lambda h: h.timestamp or now, reverse=True,
    )
    return _sanitize({
        "generated_at": now.isoformat(),
        "llm_summary": context.llm_summary,
        "sentiment": dict(context.news_sentiment),
        "high_impact_warning": context.high_impact_warning,
        "warning_message": context.warning_message,
        "calendar_available": context.calendar_available,
        "events": [event_to_dict(e) for e in week_events],
        "headlines": [headline_to_dict(h) for h in headlines],
    })


def due_timeframes(wake_dt: datetime) -> List[Timeframe]:
    """Timeframes whose candle just closed at the boundary this wake
    belongs to. Takes the RAW wake time (boundary + buffer): flooring to
    the 5-minute grid yields the boundary that just passed. Never
    subtract the buffer first — see module docstring (v1 bug)."""
    minutes = wake_dt.hour * 60 + wake_dt.minute
    floored = minutes - (minutes % 5)
    return [tf for tf in DASHBOARD_TFS if floored % TF_INTERVAL_MINUTES[tf] == 0]


def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
    # Same math as telegram_bot.py / oanda_daemon.py — kept local so the
    # dashboard never imports bot modules (independence rules above).
    now = datetime.now(timezone.utc)
    total_minutes = now.hour * 60 + now.minute
    elapsed_minutes = total_minutes % interval_minutes
    elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
    interval_seconds = interval_minutes * 60
    remaining = interval_seconds - elapsed_seconds + buffer
    if remaining <= buffer:
        remaining += interval_seconds
    return remaining


def _empty_tf_entry() -> dict:
    return {"status": "warming", "scanned_at": None, "scan_seconds": None,
            "error": None, "error_at": None, "report": None}


def build_snapshot_payload(snapshots: Dict[str, dict], now: datetime) -> dict:
    """snapshot.json: one entry per dashboard TF. Timeframes not yet
    scanned appear as status="warming" so the page can render a
    placeholder instead of a hole. server_time_epoch lets the client
    compute its clock offset once so countdowns don't drift."""
    timeframes = {}
    for tf in DASHBOARD_TFS:
        interval = TF_INTERVAL_MINUTES[tf]
        minutes = now.hour * 60 + now.minute
        secs_into = (minutes % interval) * 60 + now.second + now.microsecond / 1e6
        next_close = now.timestamp() + (interval * 60 - secs_into)
        entry = _empty_tf_entry()
        entry.update(snapshots.get(tf.value) or {})
        entry["next_close_epoch"] = next_close
        entry["next_scan_epoch"] = next_close + SCAN_BUFFER_SECONDS
        timeframes[tf.value] = entry
    return _sanitize({
        "symbol": XAUUSD,
        "server_time_epoch": now.timestamp(),
        "market_closed": _providers_base._is_forex_weekend(now),
        "timeframes": timeframes,
    })


def publish(obj: dict, path: str) -> None:
    """Atomic JSON write: temp file + os.replace, so Apache never serves
    a torn file. allow_nan=False — a NaN slipping through would render
    the page blank (JSON.parse rejects bare NaN)."""
    data = json.dumps(obj, allow_nan=False, separators=(",", ":"))
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as fh:
        fh.write(data)
    os.replace(tmp, path)


# ══════════════════════════════════════════════════════════════════
#  Runtime layer (constructed only from main(); never at import)
# ══════════════════════════════════════════════════════════════════

def _setup_logging(level: str = "INFO") -> None:
    log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs")
    os.makedirs(log_dir, exist_ok=True)
    log_file = os.path.join(log_dir, "dashboard.log")

    fmt = logging.Formatter(
        "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S")

    file_handler = TimedRotatingFileHandler(
        log_file, when="h", interval=3, backupCount=240, utc=True)
    file_handler.setFormatter(fmt)

    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setFormatter(fmt)

    logging.basicConfig(
        level=getattr(logging, level.upper(), logging.INFO),
        handlers=[file_handler, console_handler],
        force=True)


class DashboardPublisher:
    def __init__(self, out_dir: str):
        self.out_dir = out_dir
        self.snapshots: Dict[str, dict] = {}
        self.news_payload: Optional[dict] = None
        self._dp = None
        self._news_engine: Optional[NewsEngine] = None

    def _provider(self):
        if self._dp is None:
            self._dp = make_data_provider(PlatformConfig())
        return self._dp

    def _news(self) -> NewsEngine:
        if self._news_engine is None:
            self._news_engine = NewsEngine(PlatformConfig().news)
        return self._news_engine

    def install_static(self) -> None:
        """Copy the repo's dashboard.html into the out dir (atomic), so
        git pull + restart also updates the page."""
        src = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                           "static", PAGE_FILE)
        if not os.path.exists(src):
            logger.warning(f"{src} missing — page not installed")
            return
        dst = os.path.join(self.out_dir, PAGE_FILE)
        shutil.copyfile(src, dst + ".tmp")
        os.replace(dst + ".tmp", dst)

    def scan_tf(self, tf: Timeframe) -> None:
        started = time.time()
        try:
            report = produce_signal(XAUUSD, tf, self._provider(),
                                    CONFIRMATIONS[tf])
            self.snapshots[tf.value] = {
                "status": "ok",
                "scanned_at": datetime.now(timezone.utc).isoformat(),
                "scan_seconds": round(time.time() - started, 2),
                "error": None,
                "error_at": None,
                "report": report_to_webapp_dict(report),
            }
            logger.info(f"{tf.value}: {report.final_signal.name} "
                        f"q={report.signal_quality or '-'} "
                        f"({time.time() - started:.1f}s)")
        except Exception as e:
            # Keep the last good report; the page shows it with an
            # error badge instead of going blank.
            logger.exception(f"{tf.value} scan failed")
            entry = self.snapshots.get(tf.value) or _empty_tf_entry()
            entry["status"] = "error"
            entry["error"] = f"{type(e).__name__}: {e}"
            entry["error_at"] = datetime.now(timezone.utc).isoformat()
            self.snapshots[tf.value] = entry

    def refresh_news(self) -> None:
        try:
            ne = self._news()
            context = ne.get_fundamental_context()
            week = ne.get_week_events()
            self.news_payload = build_news_payload(
                context, week, datetime.now(timezone.utc))
        except Exception:
            logger.exception("news refresh failed — keeping last payload")

    def publish_all(self) -> None:
        now = datetime.now(timezone.utc)
        publish(build_snapshot_payload(self.snapshots, now),
                os.path.join(self.out_dir, SNAPSHOT_FILE))
        if self.news_payload is not None:
            publish(self.news_payload, os.path.join(self.out_dir, NEWS_FILE))

    def run_cycle(self, tfs: List[Timeframe]) -> None:
        for tf in tfs:  # sequential — see module docstring (cache dedup)
            self.scan_tf(tf)
        self.refresh_news()
        self.publish_all()

    def run_forever(self) -> None:
        self.run_cycle(DASHBOARD_TFS)  # populate the page at boot
        while True:
            delay = seconds_until_boundary(5, buffer=SCAN_BUFFER_SECONDS)
            logger.info(f"next M5 boundary in {delay:.0f}s")
            time.sleep(delay)
            try:
                due = due_timeframes(datetime.now(timezone.utc))
                logger.info(f"boundary wake — scanning {[t.value for t in due]}")
                self.run_cycle(due)
            except Exception:
                logger.exception("cycle failed")


def main(argv=None) -> None:
    ap = argparse.ArgumentParser(description="XAU/USD static dashboard publisher")
    ap.add_argument("--once", action="store_true",
                    help="one scan+publish cycle, then exit")
    ap.add_argument("--out",
                    default=os.environ.get("DASHBOARD_OUT_DIR", DEFAULT_OUT_DIR),
                    help="output directory served by Apache "
                         f"(default: $DASHBOARD_OUT_DIR or {DEFAULT_OUT_DIR})")
    args = ap.parse_args(argv)

    _setup_logging()
    os.makedirs(args.out, exist_ok=True)
    pub = DashboardPublisher(args.out)
    pub.install_static()
    if args.once:
        pub.run_cycle(DASHBOARD_TFS)
        print(f"Published to {args.out}")
        return
    pub.run_forever()


if __name__ == "__main__":
    main()