"""Daily cloud refresh run inside the Space: snapshot sources that keep no history (or move daily) and need no
API key, written to the PRIVATE dataset (never to the Space). Schedule: 06:30 Asia/Tokyo (21:30 UTC) by
default; if the Space slept through the slot it catches up on wake when the last run is >20 h old.

Output in the dataset (HF_DATA_REPO):
  warehouse/snapshots/<source>/<YYYY-MM-DD>.parquet   dated snapshots (append-only; never overwritten by other days)
  warehouse/snapshots/_status/latest.json              last run time + per-source result
Env: CLOUD_REFRESH=off disables; CLOUD_REFRESH_UTC=HH:MM overrides the slot.
"""
from __future__ import annotations

import datetime as dt
import io
import json
import os
import re
import tempfile
import threading
import time
import traceback

import pandas as pd
import requests

UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36"}
NASDAQ_H = {**UA, "Accept": "application/json, text/plain, */*", "Origin": "https://www.nasdaq.com", "Referer": "https://www.nasdaq.com/"}
FRED_IDS = ["DGS10", "DFII10", "T10YIE", "DGS2", "DFF", "SOFR", "BAMLH0A0HYM2", "BAMLH0A1HYBB", "BAMLH0A3HYC", "BAMLC0A4CBBB",
            "BAMLC0A0CM", "BAMLC0A4CBBBEY", "DBAA", "DAAA", "T10Y3M", "T10Y2Y", "DCOILBRENTEU", "DCOILWTICO", "VIXCLS", "NFCI",
            "WALCL", "RRPONTSYD", "ICSA", "DTWEXBGS"]
_LOCK = threading.Lock()


def _repo():
    return os.getenv("HF_DATA_REPO", "").strip()


def _api():
    from huggingface_hub import HfApi
    return HfApi(token=os.getenv("HF_WRITE_TOKEN") or os.getenv("HF_TOKEN"))


def _get(url, headers=UA, timeout=30, tries=2, params=None):
    """Browser-impersonating GET (FRED, SSGA and others reset plain script connections), falling back to requests."""
    last = None
    for i in range(tries):
        try:
            from curl_cffi import requests as cr
            r = cr.get(url, params=params, impersonate="chrome", timeout=timeout)
            if r.status_code == 200 and r.content:
                return r
            last = f"HTTP {r.status_code}"
        except Exception as e:                                   # curl_cffi missing or connection failure
            last = type(e).__name__
            try:
                r = requests.get(url, params=params, headers=headers, timeout=timeout)
                if r.status_code == 200 and r.content:
                    return r
                last = f"HTTP {r.status_code}"
            except requests.RequestException as e2:
                last = type(e2).__name__
        time.sleep(3 * (i + 1))
    raise RuntimeError(f"fetch failed ({last}): {url}")


# ---------------------------------------------------------------- sources
def spy_holdings():
    raw = _get("https://www.ssga.com/library-content/products/fund-data/etfs/us/holdings-daily-us-en-spy.xlsx").content
    x = pd.read_excel(io.BytesIO(raw), header=None)
    hdr = next(i for i in range(10) if (x.iloc[i].astype(str).str.strip() == "Ticker").any())
    df = x.iloc[hdr + 1:].copy(); df.columns = [str(c).strip() for c in x.iloc[hdr]]
    return df.dropna(subset=["Ticker"]).astype("string")


def fred():
    frames, fails = [], 0
    for sid in FRED_IDS:
        if fails >= 3:
            break
        try:
            d = pd.read_csv(io.BytesIO(_get(f"https://fred.stlouisfed.org/graph/fredgraph.csv?id={sid}").content))
            d.columns = ["period", "value"]; d["value"] = pd.to_numeric(d["value"], errors="coerce")
            d["series_id"] = sid; frames.append(d.dropna().tail(400))     # last ~400 obs keeps ICE spreads past FRED's window
            fails = 0
        except Exception:
            fails += 1
            continue
        time.sleep(0.5)
    return pd.concat(frames, ignore_index=True)


def cboe_putcall():
    frames = []
    for name in ["totalpc", "equitypc", "indexpc", "vixpc", "spxpc"]:
        txt = _get(f"https://cdn.cboe.com/resources/options/volume_and_call_put_ratios/{name}.csv").text.splitlines()
        hdr = next(i for i, l in enumerate(txt) if l.upper().startswith("DATE"))
        d = pd.read_csv(io.StringIO("\n".join(txt[hdr:]))).tail(30); d["series"] = name
        frames.append(d.astype("string"))
    return pd.concat(frames, ignore_index=True)


def gpu_prices():
    return pd.read_csv(io.BytesIO(_get("https://getdeploying.com/dataset/gpu-prices/weekly.csv").content)).astype("string")


def openrouter_models():
    rows = []
    for x in _get("https://openrouter.ai/api/v1/models").json().get("data", []):
        p = x.get("pricing") or {}
        rows.append({"id": x.get("id"), "name": x.get("name"), "created": x.get("created"), "context_length": x.get("context_length"),
                     "price_prompt": p.get("prompt"), "price_completion": p.get("completion")})
    return pd.DataFrame(rows).astype("string")


def earnings_calendar(days=30):
    rows = []
    for k in range(days):
        d = dt.date.today() + dt.timedelta(days=k)
        if d.weekday() >= 5:
            continue
        try:
            j = _get("https://api.nasdaq.com/api/calendar/earnings", headers=NASDAQ_H, params={"date": d.isoformat()}, tries=1).json()
            rows += [{"date": d.isoformat(), **{kk: str(vv) for kk, vv in r.items()}} for r in ((j.get("data") or {}).get("rows") or [])]
        except Exception:
            pass
        time.sleep(0.5)
    return pd.DataFrame(rows)


def estimates(spy: pd.DataFrame | None):
    if spy is None or spy.empty:
        return pd.DataFrame()
    fails = 0
    tickers = sorted({str(t).strip().replace(".", "-") for t in spy["Ticker"] if re.fullmatch(r"[A-Z][A-Z.\-]{0,6}", str(t).strip())})
    rows = []
    for t in tickers:
        try:
            j = _get(f"https://api.nasdaq.com/api/analyst/{t}/earnings-forecast", headers=NASDAQ_H, tries=1).json().get("data") or {}
            for k in ("quarterlyForecast", "yearlyForecast"):
                rows += [{"ticker": t, "kind": k, **{kk: str(vv) for kk, vv in r.items()}} for r in (j.get(k) or {}).get("rows") or []]
            fails = 0
        except Exception:
            fails += 1
            if fails >= 10:                                       # endpoint blocked: stop instead of 500 timeouts
                break
        time.sleep(0.4)
    return pd.DataFrame(rows)


def treasury_curves():
    y = dt.date.today().year
    frames = []
    for kind, tag in [("daily_treasury_yield_curve", "nominal"), ("daily_treasury_real_yield_curve", "real")]:
        url = ("https://home.treasury.gov/resource-center/data-chart-center/interest-rates/"
               f"daily-treasury-rates.csv/{y}/all?type={kind}&field_tdr_date_value={y}&page&_format=csv")
        d = pd.read_csv(io.BytesIO(_get(url).content)); d["curve"] = tag; frames.append(d.astype("string"))
    return pd.concat(frames, ignore_index=True)


def dram_spot():
    from curl_cffi import requests as cffi
    r = cffi.Session(impersonate="chrome").get("https://www.dramexchange.com/", timeout=60)
    tabs = [t for t in pd.read_html(io.BytesIO(r.content)) if t.shape[1] >= 5 and t.shape[0] >= 2]
    for i, t in enumerate(tabs):
        t.columns = [" ".join(map(str, c)) if isinstance(c, tuple) else str(c) for c in t.columns]
        t["table_no"] = i
    return pd.concat(tabs, ignore_index=True).astype("string") if tabs else pd.DataFrame()


# ---------------------------------------------------------------- runner
def run_once() -> dict:
    """Fetch every source, upload dated snapshots to the private dataset, write a status file."""
    repo = _repo()
    if not repo:
        return {"error": "HF_DATA_REPO not set"}
    api = _api()
    if not api.dataset_info(repo).private:
        return {"error": "dataset is public; refusing to write"}
    # snapshot dates use the user's calendar (UTC+9, Asia/Tokyo) so they match the local 06:30 run's dates
    today = (dt.datetime.utcnow() + dt.timedelta(hours=float(os.getenv("CLOUD_REFRESH_TZ_HOURS", "9")))).date().isoformat()
    status, spy = {"started_utc": dt.datetime.utcnow().isoformat(timespec="seconds"), "sources": {}}, None
    jobs = [("spy_holdings", spy_holdings), ("fred", fred), ("cboe_putcall", cboe_putcall), ("gpu_prices", gpu_prices),
            ("openrouter_models", openrouter_models), ("earnings_calendar", earnings_calendar), ("treasury_curves", treasury_curves),
            ("dram_spot", dram_spot), ("estimates_nasdaq", lambda: estimates(spy))]
    with tempfile.TemporaryDirectory() as tmp:
        for name, fn in jobs:
            try:
                df = fn()
                if name == "spy_holdings":
                    spy = df
                if df is None or df.empty:
                    status["sources"][name] = "empty"; continue
                df["snapshot_date"] = today
                p = os.path.join(tmp, f"{name}.parquet")
                df.to_parquet(p, index=False)
                api.upload_file(path_or_fileobj=p, path_in_repo=f"warehouse/snapshots/{name}/{today}.parquet", repo_id=repo,
                                repo_type="dataset", commit_message=f"cloud refresh {name} {today}")
                status["sources"][name] = f"ok {len(df)} rows"
            except Exception as e:
                status["sources"][name] = f"error {type(e).__name__}: {str(e)[:120]}"
        status["finished_utc"] = dt.datetime.utcnow().isoformat(timespec="seconds")
        sp = os.path.join(tmp, "latest.json")
        with open(sp, "w") as fh:
            json.dump(status, fh, indent=1)
        try:
            api.upload_file(path_or_fileobj=sp, path_in_repo="warehouse/snapshots/_status/latest.json", repo_id=repo,
                            repo_type="dataset", commit_message=f"cloud refresh status {today}")
        except Exception as e:                                  # e.g. read-only token: report, do not raise
            status["status_upload"] = f"error {type(e).__name__}: {str(e)[:120]}"
    return status


def _last_run_utc():
    try:
        from huggingface_hub import hf_hub_download
        p = hf_hub_download(_repo(), "warehouse/snapshots/_status/latest.json", repo_type="dataset",
                            token=os.getenv("HF_TOKEN"), force_download=True)
        return dt.datetime.fromisoformat(json.load(open(p))["finished_utc"])
    except Exception:
        return None


def _loop():
    """One attempt per scheduled slot. A failed or partial run is not retried until the next slot, so a read-only
    token or a blocked source can never turn into a refetch loop."""
    hh, mm = (int(x) for x in os.getenv("CLOUD_REFRESH_UTC", "21:30").split(":"))
    time.sleep(90)                                         # let the app finish starting
    attempted_slot = None
    while True:
        try:
            now = dt.datetime.utcnow()
            slot = now.replace(hour=hh, minute=mm, second=0, microsecond=0)
            if slot > now:
                slot -= dt.timedelta(days=1)                   # most recent scheduled slot
            if attempted_slot != slot:
                last = _last_run_utc()
                if last is None or last < slot:                # missed or due: run now (covers waking after sleep)
                    attempted_slot = slot
                    with _LOCK:
                        print("[cloud_refresh] running for slot", slot.isoformat(), flush=True)
                        try:
                            print("[cloud_refresh]", json.dumps(run_once())[:800], flush=True)
                        except Exception:
                            traceback.print_exc()
                else:
                    attempted_slot = slot
            nxt = slot + dt.timedelta(days=1)
            time.sleep(max(60, min((nxt - dt.datetime.utcnow()).total_seconds(), 3600)))
        except Exception:
            traceback.print_exc()
            time.sleep(3600)


def start_background():
    if os.getenv("CLOUD_REFRESH", "on").lower() == "off":
        return False
    threading.Thread(target=_loop, name="cloud_refresh", daemon=True).start()
    return True