main.py

import os, asyncio, io
import httpx
import pandas as pd
from fastapi import FastAPI, Request, HTTPException, UploadFile, File, Form, Body
from fastapi.responses import JSONResponse, Response, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from slowapi import Limiter
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware

# --- Secrets set in the PUBLIC Space ---
UPSTREAM_URL = os.environ.get("hf_url", "").rstrip("/")             # url to access Private Space
HF_TOKEN     = os.environ.get("hf_token")                           # HF access token to access Private Space
DEMO_FORWARD_URL   = os.getenv("DEMO_FORWARD_URL", "").rstrip("/")  # url to acces demo space 
DATFID_DEMO_TOKEN  = os.getenv("DATFID_DEMO_TOKEN", "")             # token to access demo space
# This Space's public URL (used to ping self while waiting so HF does not put this Space to sleep). Override with SELF_URL env if different.
SELF_URL = os.getenv("SELF_URL", "https://datfid-org-datfid-master.hf.space").rstrip("/")
# Admin-only endpoints: require this secret (send as X-Admin-Pass header). Secret name in HF: Admin_pass
Admin_pass = os.getenv("Admin_pass", "")

if not HF_TOKEN:
    raise RuntimeError("Missing secret 'hf_token' in public Space.")
if not UPSTREAM_URL.startswith("https://"):
    raise RuntimeError("Missing/invalid secret 'hf_url' in public Space.")

app = FastAPI(title="DATFID Public Proxy", docs_url="/docs", redoc_url=None)

# Prefer X-Forwarded-For (HF sits behind a proxy)
def client_ip(request: Request):
    xff = request.headers.get("x-forwarded-for")
    return xff.split(",")[0].strip() if xff else (request.client.host or "0.0.0.0")

limiter = Limiter(key_func=client_ip)  # or get_remote_address
GLOBAL_LIMIT = limiter.limit("100/10minute", key_func=lambda: "global:any") # global limit

app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)

@app.exception_handler(RateLimitExceeded)
async def ratelimit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(status_code=429, content={"detail": "Too many requests, slow down."})

SDK_MAX_BODY_BYTES = int(os.getenv("SDK_MAX_BODY_BYTES", "25000000"))  # 25MB default for demo
SDK_MAX_BODY_BYTES_extended = int(os.getenv("SDK_MAX_BODY_BYTES_extended", "125000000")) # 125MB default for prod 

# How long to wait for upstream (API) response. modelforecast_ind on many individuals can take 60–90 min; set UPSTREAM_TIMEOUT high (e.g. 5400) or leave default.
UPSTREAM_TIMEOUT = float(os.getenv("UPSTREAM_TIMEOUT", "900"))  # 15 min default so long forecast_ind runs don't hit ReadTimeout
# While waiting for a response: every PING_INTERVAL seconds we ping the backend (secure_ping) and, if SELF_URL is set, we also ping ourselves (keep-alive) so this Space stays awake
PING_INTERVAL = float(os.getenv("PING_INTERVAL", "270"))

# Early global body-size guard (runs before routes)
@app.middleware("http")
async def limit_body_size(request: Request, call_next):
    cl = request.headers.get("content-length")

    # demo routes use the smaller cap, others use extended
    path = request.url.path or ""
    cap = SDK_MAX_BODY_BYTES if "-demo" in path else SDK_MAX_BODY_BYTES_extended

    if cl and int(cl) > cap:
        return JSONResponse(
            {"detail": f"Payload too large (> {cap} bytes)"},
            status_code=413,
        )
    return await call_next(request)

# CORS for browser
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://datfid.com",
        "https://www.datfid.com"
    ],
    # Optional: allow Vercel preview domains
    # allow_origin_regex=r"^https:\/\/.*\.vercel\.app$",
    allow_methods=["POST", "OPTIONS"],
    allow_headers=["Content-Type"],   # no Authorization header needed from browser
    allow_credentials=False,
    max_age=86400,
)

# to ensure we don’t leak hop-by-hop headers
def _filter_resp_headers(headers: dict) -> dict:
    # pass through useful headers but strip hop-by-hop
    allowed = {"content-disposition"}
    return {k: v for k, v in headers.items() if k.lower() in allowed}

async def _ping_upstream_loop(ping_url: str, headers: dict, interval: float, self_url: str = ""):
    """Every `interval` seconds: if self_url set, GET self_url/keep-alive (keep this Space awake), then GET ping_url (backend). Stops when cancelled."""
    if interval <= 0:
        return
    while True:
        await asyncio.sleep(interval)
        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                if self_url:
                    try:
                        await client.get(f"{self_url}/keep-alive")
                    except Exception:
                        pass
                await client.get(ping_url, headers=headers)
        except asyncio.CancelledError:
            break
        except Exception:
            pass

def _start_ping_task(ping_url: str, headers: dict, self_url: str = ""):
    """Start background ping task: every PING_INTERVAL seconds ping backend and, if self_url set, also ping self."""
    if PING_INTERVAL <= 0:
        return None
    return asyncio.create_task(_ping_upstream_loop(ping_url, headers, PING_INTERVAL, self_url))

async def _cancel_ping_task(task: asyncio.Task | None):
    if task is None:
        return
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

def _extract_user_token(req: Request) -> str | None:
    """
    Read user's DATFID token from Authorization: Bearer <dt+...>.
    We do NOT verify it here; the private Space does that.
    """
    auth = req.headers.get("authorization", "")
    if not auth.lower().startswith("bearer "):
        return None
    return auth.split(" ", 1)[1].strip()

# Chat endpoints: fetch data from URL, parse to table, forward to API
DATA_URL_FETCH_TIMEOUT = 60.0
DATA_URL_MAX_BYTES = 20 * 1024 * 1024  # 20MB

async def _fetch_url_to_records(data_url: str) -> list:
    """Fetch data_url (https only), parse CSV/Excel/JSON to list of dicts. Raises HTTPException on error."""
    if not data_url.strip().lower().startswith("https://"):
        raise HTTPException(status_code=400, detail="data_url must be an HTTPS URL.")
    async with httpx.AsyncClient(timeout=DATA_URL_FETCH_TIMEOUT, follow_redirects=True) as client:
        r = await client.get(data_url)
        r.raise_for_status()
        raw = r.content
        content_type = (r.headers.get("content-type") or "").lower()
    if len(raw) > DATA_URL_MAX_BYTES:
        raise HTTPException(status_code=413, detail=f"Data at URL exceeds {DATA_URL_MAX_BYTES // (1024*1024)}MB limit.")
    path_lower = data_url.split("?")[0].lower()
    try:
        if "json" in content_type or path_lower.endswith(".json"):
            df = pd.read_json(io.BytesIO(raw))
        elif "spreadsheet" in content_type or "excel" in content_type or path_lower.endswith((".xlsx", ".xls")):
            df = pd.read_excel(io.BytesIO(raw))
        else:
            # CSV or default
            df = pd.read_csv(io.BytesIO(raw))
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Could not parse data from URL: {str(e)[:200]}")
    for col in df.columns:
        if pd.api.types.is_datetime64_any_dtype(df[col]):
            df[col] = df[col].astype(str)
    return df.to_dict(orient="records")


async def _fetch_url_to_dataframe(data_url: str) -> pd.DataFrame:
    """Fetch data_url (https only), parse CSV/Excel/JSON to DataFrame. Raises HTTPException on error."""
    if not data_url.strip().lower().startswith("https://"):
        raise HTTPException(status_code=400, detail="data_url must be an HTTPS URL.")
    async with httpx.AsyncClient(timeout=DATA_URL_FETCH_TIMEOUT, follow_redirects=True) as client:
        r = await client.get(data_url)
        r.raise_for_status()
        raw = r.content
        content_type = (r.headers.get("content-type") or "").lower()
    if len(raw) > DATA_URL_MAX_BYTES:
        raise HTTPException(status_code=413, detail=f"Data at URL exceeds {DATA_URL_MAX_BYTES // (1024*1024)}MB limit.")
    path_lower = data_url.split("?")[0].lower()
    try:
        if "json" in content_type or path_lower.endswith(".json"):
            df = pd.read_json(io.BytesIO(raw))
        elif "spreadsheet" in content_type or "excel" in content_type or path_lower.endswith((".xlsx", ".xls")):
            df = pd.read_excel(io.BytesIO(raw))
        else:
            df = pd.read_csv(io.BytesIO(raw))
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Could not parse data from URL: {str(e)[:200]}")
    for col in df.columns:
        if pd.api.types.is_datetime64_any_dtype(df[col]):
            df[col] = df[col].astype(str)
    return df


async def _forward(path: str, method: str = "GET", json_body=None, user_token: str | None = None, success_response_override: dict | None = None):
    """
    Forward request to the PRIVATE Space:
      - 'Authorization: Bearer <HF_TOKEN>' to pass HF private gate
      - 'X-API-Key: <dt+...>' so your private app can validate the user token
    While waiting, every PING_INTERVAL seconds: secure_ping to backend and, if SELF_URL set, ping self (keep-alive).
    """
    url = f"{UPSTREAM_URL}{path}"
    headers = {
        "Authorization": f"Bearer {HF_TOKEN}",
        "Accept": "application/json",
    }
    if user_token:
        headers["X-API-Key"] = user_token

    # While waiting: ping backend (secure_ping) and, if SELF_URL set, ping self so this Space stays awake
    ping_url = f"{UPSTREAM_URL}/secure-ping/"
    ping_headers = {"Authorization": f"Bearer {HF_TOKEN}"}
    if user_token:
        ping_headers["X-API-Key"] = user_token

    timeout = httpx.Timeout(UPSTREAM_TIMEOUT)
    r = None

    async def do_request():
        nonlocal r
        async with httpx.AsyncClient(timeout=timeout) as client:
            r = await client.request(method, url, headers=headers, json=json_body)

    request_task = asyncio.create_task(do_request())
    wait_sec = PING_INTERVAL if PING_INTERVAL > 0 else 0.0
    try:
        while not request_task.done():
            if wait_sec <= 0:
                await request_task
                break
            ping_sleep = asyncio.create_task(asyncio.sleep(wait_sec))
            done, pending = await asyncio.wait(
                {request_task, ping_sleep},
                return_when=asyncio.FIRST_COMPLETED,
                timeout=UPSTREAM_TIMEOUT + 10,
            )
            for t in pending:
                if t is not request_task:
                    t.cancel()
                    try:
                        await t
                    except asyncio.CancelledError:
                        pass
            if request_task in done:
                break
            async with httpx.AsyncClient(timeout=10.0) as c:
                if SELF_URL:
                    try:
                        await c.get(f"{SELF_URL}/keep-alive")
                    except Exception:
                        pass
                try:
                    await c.get(ping_url, headers=ping_headers)
                except Exception:
                    pass
        if not request_task.done():
            request_task.cancel()
            try:
                await request_task
            except asyncio.CancelledError:
                pass
        await request_task
    except asyncio.CancelledError:
        request_task.cancel()
        try:
            await request_task
        except asyncio.CancelledError:
            pass
        raise
    exc = request_task.exception()
    if exc is not None:
        raise exc
    if r is None:
        raise RuntimeError("Upstream request did not complete")

    # When caller wants a minimal response on success (e.g. for chat to avoid response size limits)
    if success_response_override is not None and r.status_code == 200:
        return JSONResponse(status_code=200, content=success_response_override)

    ct = r.headers.get("content-type", "")

    if "application/json" in ct:
        try:
            return JSONResponse(status_code=r.status_code, content=r.json())
        except Exception:
            return JSONResponse(status_code=r.status_code, content={"error": r.text[:500]})
    # Fallback: return short text envelope if non-JSON
    return JSONResponse(status_code=r.status_code, content={"text": r.text[:1000]})


async def _forward_and_get_body(path: str, method: str = "GET", json_body=None, user_token: str | None = None) -> tuple[int, dict | list | None]:
    """Same as _forward but returns (status_code, parsed_json_body) so caller can build a custom response."""
    url = f"{UPSTREAM_URL}{path}"
    headers = {
        "Authorization": f"Bearer {HF_TOKEN}",
        "Accept": "application/json",
    }
    if user_token:
        headers["X-API-Key"] = user_token
    ping_url = f"{UPSTREAM_URL}/secure-ping/"
    ping_headers = {"Authorization": f"Bearer {HF_TOKEN}"}
    if user_token:
        ping_headers["X-API-Key"] = user_token
    timeout = httpx.Timeout(UPSTREAM_TIMEOUT)
    r = None

    async def do_request():
        nonlocal r
        async with httpx.AsyncClient(timeout=timeout) as client:
            r = await client.request(method, url, headers=headers, json=json_body)

    request_task = asyncio.create_task(do_request())
    wait_sec = PING_INTERVAL if PING_INTERVAL > 0 else 0.0
    try:
        while not request_task.done():
            if wait_sec <= 0:
                await request_task
                break
            ping_sleep = asyncio.create_task(asyncio.sleep(wait_sec))
            done, pending = await asyncio.wait(
                {request_task, ping_sleep},
                return_when=asyncio.FIRST_COMPLETED,
                timeout=UPSTREAM_TIMEOUT + 10,
            )
            for t in pending:
                if t is not request_task:
                    t.cancel()
                    try:
                        await t
                    except asyncio.CancelledError:
                        pass
            if request_task in done:
                break
            async with httpx.AsyncClient(timeout=10.0) as c:
                if SELF_URL:
                    try:
                        await c.get(f"{SELF_URL}/keep-alive")
                    except Exception:
                        pass
                try:
                    await c.get(ping_url, headers=ping_headers)
                except Exception:
                    pass
        if not request_task.done():
            request_task.cancel()
            try:
                await request_task
            except asyncio.CancelledError:
                pass
        await request_task
    except asyncio.CancelledError:
        request_task.cancel()
        try:
            await request_task
        except asyncio.CancelledError:
            pass
        raise
    exc = request_task.exception()
    if exc is not None:
        raise exc
    if r is None:
        raise RuntimeError("Upstream request did not complete")

    ct = r.headers.get("content-type", "")
    if "application/json" in ct:
        try:
            return (r.status_code, r.json())
        except Exception:
            return (r.status_code, {"error": r.text[:500]})
    return (r.status_code, {"text": r.text[:1000]})


def _round_val(x, ndigits: int = 4):
    """Round numbers to ndigits; leave non-numbers as-is."""
    if isinstance(x, (int, float)) and not isinstance(x, bool):
        return round(float(x), ndigits)
    return x


# Row labels for alpha/beta: Estimate, std.er., t_stat, p_val (backend order = _ROW4)
_FIT_ROW_LABELS = ("Estimate", "std.er.", "t_stat", "p_val")


def _format_fit_table(rows: list, ndigits: int = 4) -> list:
    """
    Transform backend table (list of 4 dicts, keys=columns) into chat-friendly format.
    Row 0: column names -> rounded values (Estimate). Rows 1,2,3: row label -> value(s).
    For single column: [{'col': v0}, {'std.er.': v1}, {'t_stat': v2}, {'p_val': v3}].
    For multi-column: row 0 = {col: v}; rows 1,2,3 = {'std.er.': [v,...]}, etc.
    """
    if not isinstance(rows, list) or len(rows) != 4:
        return rows
    col_names = list(rows[0].keys()) if rows[0] else []
    out = []
    # Row 0: Estimate (column names as keys)
    out.append({k: _round_val(rows[0].get(k)) for k in col_names})
    # Rows 1,2,3: row name as key
    for r in range(1, 4):
        label = _FIT_ROW_LABELS[r]
        vals = [rows[r].get(k) for k in col_names]
        if len(col_names) == 1:
            out.append({label: _round_val(vals[0])})
        else:
            out.append({label: [_round_val(v) for v in vals]})
    return out


def _build_modelfit_chat_minimal(body: dict) -> dict:
    """Extract formula, alpha, beta, and Performance subset (col 2, rows 3,4,5) for chat response."""
    out = {"ok": True, "formula": body.get("formula")}
    alpha = body.get("alpha")
    beta = body.get("beta")
    out["alpha"] = _format_fit_table(alpha) if isinstance(alpha, list) else alpha
    out["beta"] = _format_fit_table(beta) if isinstance(beta, list) else beta
    perf = body.get("Performance")
    if isinstance(perf, list) and len(perf) >= 5:
        keys = list(perf[0].keys()) if perf[0] else []
        col_idx = 1 if len(keys) > 1 else 0
        col2_name = keys[col_idx] if keys else None
        if col2_name is not None:
            row_names = ["R2 overall", "MSE", "MAE"]
            values = [_round_val(perf[i].get(col2_name)) for i in [2, 3, 4]]
            out["Performance_subset"] = {"metric_names": row_names, "values": values}
        else:
            out["Performance_subset"] = None
    else:
        out["Performance_subset"] = None
    return out


async def _forward_stream(path: str, files=None, data=None, user_token: str | None = None, method: str = "POST"):
    url = f"{UPSTREAM_URL}{path}"
    headers = {
        "Authorization": f"Bearer {HF_TOKEN}",
        "Accept": "*/*",
    }
    if user_token:
        headers["X-API-Key"] = user_token

    ping_headers = {"Authorization": f"Bearer {HF_TOKEN}"}
    if user_token:
        ping_headers["X-API-Key"] = user_token
    ping_task = _start_ping_task(f"{UPSTREAM_URL}/secure-ping/", ping_headers, SELF_URL)
    timeout = httpx.Timeout(UPSTREAM_TIMEOUT)
    client = httpx.AsyncClient(timeout=timeout, follow_redirects=True)

    # Don't open the context yet; the iterator must own the context lifetime.
    stream_ctx = client.stream(method, url, headers=headers, files=files, data=data)

    # Mutable holders we can fill once the stream opens
    status_holder = {"code": 200}
    media_type_holder = {"ct": "application/octet-stream"}
    headers_holder = {}

    async def body_iter():
        chunk_queue: asyncio.Queue = asyncio.Queue()
        stream_done = {"done": False, "exc": None}

        async def stream_reader():
            try:
                async with stream_ctx as resp:
                    status_holder["code"] = resp.status_code
                    media_type_holder["ct"] = resp.headers.get("content-type", "application/octet-stream")
                    headers_holder.update(_filter_resp_headers(resp.headers))
                    if resp.status_code >= 400:
                        chunk = await resp.aread()
                        await chunk_queue.put(chunk)
                        await chunk_queue.put(None)
                        return
                    async for chunk in resp.aiter_raw():
                        await chunk_queue.put(chunk)
                    await chunk_queue.put(None)
            except Exception as e:
                stream_done["exc"] = e
                await chunk_queue.put(None)
            finally:
                stream_done["done"] = True

        reader_task = asyncio.create_task(stream_reader())
        try:
            while True:
                chunk = await chunk_queue.get()
                if chunk is None:
                    if stream_done["exc"]:
                        raise stream_done["exc"]
                    break
                yield chunk
        finally:
            reader_task.cancel()
            try:
                await reader_task
            except asyncio.CancelledError:
                pass
            await _cancel_ping_task(ping_task)
            await client.aclose()

    response = StreamingResponse(
        body_iter(),
        status_code=status_holder["code"],
        media_type=media_type_holder["ct"],
        headers=headers_holder,
    )
    return response


async def _forward_multipart_json(path: str, files=None, data=None, user_token: str | None = None, method: str = "POST"):
    """POST multipart to upstream and return JSON. While waiting: secure_ping to backend; if SELF_URL set, ping self."""
    url = f"{UPSTREAM_URL}{path}"
    headers = {
        "Authorization": f"Bearer {HF_TOKEN}",
        "Accept": "application/json",
    }
    if user_token:
        headers["X-API-Key"] = user_token
    ping_url = f"{UPSTREAM_URL}/secure-ping/"
    ping_headers = {"Authorization": f"Bearer {HF_TOKEN}"}
    if user_token:
        ping_headers["X-API-Key"] = user_token
    timeout = httpx.Timeout(UPSTREAM_TIMEOUT)
    r = None

    async def do_request():
        nonlocal r
        async with httpx.AsyncClient(timeout=timeout) as client:
            r = await client.request(method, url, headers=headers, files=files, data=data)

    request_task = asyncio.create_task(do_request())
    wait_sec = PING_INTERVAL if PING_INTERVAL > 0 else 0.0
    try:
        while not request_task.done():
            if wait_sec <= 0:
                await request_task
                break
            ping_sleep = asyncio.create_task(asyncio.sleep(wait_sec))
            done, pending = await asyncio.wait(
                {request_task, ping_sleep},
                return_when=asyncio.FIRST_COMPLETED,
                timeout=UPSTREAM_TIMEOUT + 10,
            )
            for t in pending:
                if t is not request_task:
                    t.cancel()
                    try:
                        await t
                    except asyncio.CancelledError:
                        pass
            if request_task in done:
                break
            async with httpx.AsyncClient(timeout=10.0) as c:
                if SELF_URL:
                    try:
                        await c.get(f"{SELF_URL}/keep-alive")
                    except Exception:
                        pass
                try:
                    await c.get(ping_url, headers=ping_headers)
                except Exception:
                    pass
        if not request_task.done():
            request_task.cancel()
            try:
                await request_task
            except asyncio.CancelledError:
                pass
        await request_task
    except asyncio.CancelledError:
        request_task.cancel()
        try:
            await request_task
        except asyncio.CancelledError:
            pass
        raise
    exc = request_task.exception()
    if exc is not None:
        raise exc
    if r is None:
        raise RuntimeError("Upstream request did not complete")
    ct = r.headers.get("content-type", "")
    if "application/json" in ct:
        try:
            return JSONResponse(status_code=r.status_code, content=r.json())
        except Exception:
            return JSONResponse(status_code=r.status_code, content={"error": r.text[:500]})
    return JSONResponse(status_code=r.status_code, content={"text": r.text[:1000]})


# for demo
async def _forward_demo_stream(path: str, *, files: dict | None, data: dict | None, method: str = "POST"):
    if not DEMO_FORWARD_URL or not HF_TOKEN or not DATFID_DEMO_TOKEN:
        raise HTTPException(status_code=500, detail="Demo not configured.")

    url = DEMO_FORWARD_URL + path
    headers = {
        # HF private-space gate:
        "Authorization": f"Bearer {HF_TOKEN}",
        # App-level demo token (checked by the private API):
        "X-DATFID-Token": DATFID_DEMO_TOKEN,
        "Accept": "*/*",
    }
    ping_headers = {"Authorization": f"Bearer {HF_TOKEN}", "X-DATFID-Token": DATFID_DEMO_TOKEN}
    ping_task = _start_ping_task(f"{DEMO_FORWARD_URL.rstrip('/')}/", ping_headers)

    timeout = httpx.Timeout(120.0)
    client = httpx.AsyncClient(timeout=timeout, follow_redirects=True)
    stream_ctx = client.stream(method, url, headers=headers, files=files, data=data)

    status_holder = {"code": 200}
    media_type_holder = {"ct": "application/octet-stream"}
    headers_holder = {}

    async def body_iter():
        chunk_queue: asyncio.Queue = asyncio.Queue()
        stream_done = {"done": False, "exc": None}

        async def stream_reader():
            try:
                async with stream_ctx as resp:
                    status_holder["code"] = resp.status_code
                    media_type_holder["ct"] = resp.headers.get("content-type", "application/octet-stream")
                    headers_holder.update(_filter_resp_headers(resp.headers))
                    if resp.status_code >= 400:
                        chunk = await resp.aread()
                        await chunk_queue.put(chunk)
                        await chunk_queue.put(None)
                        return
                    async for chunk in resp.aiter_raw():
                        await chunk_queue.put(chunk)
                    await chunk_queue.put(None)
            except Exception as e:
                stream_done["exc"] = e
                await chunk_queue.put(None)
            finally:
                stream_done["done"] = True

        reader_task = asyncio.create_task(stream_reader())
        try:
            while True:
                chunk = await chunk_queue.get()
                if chunk is None:
                    if stream_done["exc"]:
                        raise stream_done["exc"]
                    break
                yield chunk
        finally:
            reader_task.cancel()
            try:
                await reader_task
            except asyncio.CancelledError:
                pass
            await _cancel_ping_task(ping_task)
            await client.aclose()

    response = StreamingResponse(
        body_iter(),
        status_code=status_holder["code"],
        media_type=media_type_holder["ct"],
        headers=headers_holder,
    )
    return response

@app.get("/")
async def root(req: Request):
    # Forward to private root (private gate still needs HF token)
    user_token = _extract_user_token(req)  # optional here
    return await _forward("/", "GET", user_token=user_token)


@app.get("/keep-alive")
async def keep_alive():
    """
    Hit this to keep this Space (datfid_master) and the backend awake.
    - External cron (e.g. every 4 min): UptimeRobot, cron-job.org — keeps both spaces awake when idle.
    - During a long request (>5 min): the client can call this in a background thread every ~4 min
      so HF does not put this Space to sleep while it is waiting for the backend (no response
      sent to the client yet). This endpoint also pings the upstream (backend). No auth required.
    """
    try:
        url = f"{UPSTREAM_URL}/"
        async with httpx.AsyncClient(timeout=10.0) as client:
            r = await client.get(url, headers={"Authorization": f"Bearer {HF_TOKEN}"})
        return JSONResponse(content={"ok": True, "upstream_status": r.status_code})
    except Exception as e:
        return JSONResponse(status_code=502, content={"ok": False, "error": str(e)[:200]})

@app.get("/secure-ping/")
async def secure_ping(req: Request):
    # Require user's DATFID token in Authorization: Bearer <dt+...>
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    return await _forward("/secure-ping/", "GET", user_token=user_token)


def _require_admin_token(request: Request) -> None:
    """Raise 401 if Admin_pass is not set or X-Admin-Pass does not match."""
    if not Admin_pass:
        raise HTTPException(status_code=503, detail="Admin endpoints not configured (set Admin_pass in secrets).")
    token = request.headers.get("X-Admin-Pass", "").strip()
    if not token or token != Admin_pass:
        raise HTTPException(status_code=401, detail="Invalid or missing X-Admin-Pass.")


@app.get("/admin/verify")
async def admin_verify(req: Request):
    """Verify DATFID token (Authorization: Bearer) and admin secret. Returns 200 + {ok: true}."""
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...).")
    _require_admin_token(req)
    return JSONResponse(content={"ok": True})


@app.post("/modelfit/")
async def modelfit(req: Request):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    body = await req.json()
    return await _forward("/modelfit/", "POST", json_body=body, user_token=user_token)

@app.post("/modelforecast/")
async def modelforecast(req: Request):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    body = await req.json()
    return await _forward("/modelforecast/", "POST", json_body=body, user_token=user_token)


@app.post("/modelfit_chat/")
async def modelfit_chat(
    req: Request,
    data_url: str = Body(..., embed=True),
):
    """
    Fetch training data from data_url (HTTPS), infer schema from column order, then call /modelfit/ on the API.
    Column order: 1st = id_col, 2nd = time_col, 3rd..second-to-last = current_features, last = y.
    On success returns a small informative JSON: formula, alpha, beta, Performance_subset (col 2, rows 3–5).
    """
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    df = await _fetch_url_to_dataframe(data_url)
    cols = list(df.columns)
    if len(cols) < 3:
        raise HTTPException(
            status_code=400,
            detail="Data must have at least 3 columns (order: id_col, time_col, ...current_features..., y).",
        )
    id_col = cols[0]
    time_col = cols[1]
    y = cols[-1]
    current_features = cols[2:-1]  # empty if exactly 3 columns
    payload = {
        "df": df.to_dict(orient="records"),
        "id_col": id_col,
        "time_col": time_col,
        "y": y,
        "lag_y": None,
        "lagged_features": {},
        "current_features": current_features,
        "filter_by_significance": False,
        "dummy_sqrt": False,
        "dummy_interact": False,
        "dummy_logs": False,
        "dummy_yejo": False,
        "dummy_asinh": False,
        "dummy_logs_interact": False,
        "meanvar_test": False,
        "signif": 0.05,
    }
    status, body = await _forward_and_get_body("/modelfit/", "POST", json_body=payload, user_token=user_token)
    if status != 200 or not isinstance(body, dict):
        return JSONResponse(status_code=status, content=body if isinstance(body, dict) else {"error": str(body)})
    minimal = _build_modelfit_chat_minimal(body)
    return JSONResponse(status_code=200, content=minimal)


@app.post("/modelforecast_chat/")
async def modelforecast_chat(
    req: Request,
    data_url: str = Body(..., embed=True),
):
    """Fetch forecast input data from data_url (HTTPS), then call /modelforecast/ on the API. Returns JSON forecast list."""
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    df_records = await _fetch_url_to_records(data_url)
    # Backend modelforecast expects the raw JSON array as body (single Body(...) param), not {"df_forecast": [...]}
    return await _forward("/modelforecast/", "POST", json_body=df_records, user_token=user_token)


@app.post("/modelfit-file/")
async def modelfit_file(
    req: Request,
    file: UploadFile = File(...),
    id_col: str = Form(...),
    time_col: str = Form(...),
    y: str = Form(...),
    # optional knobs
    lag_y: str = Form(""),
    lagged_features: str = Form(""),          # JSON string or empty
    current_features: str = Form(""),         # "all" | JSON string | ""
    filter_by_significance: str = Form("false"),
    dummy_sqrt: str = Form("false"),
    dummy_interact: str = Form("false"),
    dummy_logs: str = Form("false"),
    dummy_yejo: str = Form("false"),
    dummy_asinh: str = Form("false"),
    dummy_logs_interact: str = Form("false"),
    meanvar_test: str = Form("false"),
    signif: str = Form("0.05"),
):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")

    raw = await file.read()
    if len(raw) > SDK_MAX_BODY_BYTES_extended:
        raise HTTPException(status_code=413, detail="Payload too large.")

    files = {
        "file": (file.filename, raw, file.content_type or "application/octet-stream"),
    }
    data = {
        "id_col": id_col,
        "time_col": time_col,
        "y": y,
        "lag_y": lag_y,
        "lagged_features": lagged_features,
        "current_features": current_features,
        "filter_by_significance": filter_by_significance,
        "dummy_sqrt": dummy_sqrt,
        "dummy_interact": dummy_interact,
        "dummy_logs": dummy_logs,
        "dummy_yejo": dummy_yejo,
        "dummy_asinh": dummy_asinh,
        "dummy_logs_interact": dummy_logs_interact,
        "meanvar_test": meanvar_test,
        "signif": signif,
    }
    return await _forward_stream("/modelfit-file/", files=files, data=data, user_token=user_token, method="POST")

@app.post("/modelforecast-file/")
async def modelforecast_file(
    req: Request,
    df_forecast: UploadFile = File(...),
):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    
    raw = await df_forecast.read()
    if len(raw) > SDK_MAX_BODY_BYTES_extended:
        raise HTTPException(status_code=413, detail="Payload too large.")

    files = {
        "df_forecast": (df_forecast.filename, raw, df_forecast.content_type or "application/octet-stream"),
    }
    return await _forward_stream("/modelforecast-file/", files=files, data=None, user_token=user_token, method="POST")

@app.post("/modelfit-file-demo/")
@GLOBAL_LIMIT
@limiter.limit("10/10minute")  # 10 calls per 10 minutes per IP
async def modelfit_file_demo(
    request: Request,
    file: UploadFile = File(...),
    id_col: str = Form(...),
    time_col: str = Form(...),
    y: str = Form(...),
    # optional knobs
    lag_y: str = Form(""),
    lagged_features: str = Form(""),
    current_features: str = Form(""),
    filter_by_significance: str = Form("false"),
    meanvar_test: str = Form("false"),
    session_id: str = Form(...),
):
    
    # read once, size-guard it, then forward
    raw = await file.read()
    if len(raw) > SDK_MAX_BODY_BYTES:
        raise HTTPException(status_code=413, detail="Payload too large.")
    
    sid = session_id.strip()
    if not sid:
        raise HTTPException(status_code=400, detail="Missing session_id")
    
    files = {
        "file": (file.filename, raw, file.content_type or "application/octet-stream"),
    }
    data = {
        "id_col": id_col,
        "time_col": time_col,
        "y": y,
        "lag_y": lag_y,
        "lagged_features": lagged_features,
        "current_features": current_features,
        "filter_by_significance": filter_by_significance,
        "meanvar_test": meanvar_test,
        "session_id": sid,
    }
    # the path goes to the private demo route
    return await _forward_demo_stream("/modelfit-file-demo/", files=files, data=data, method="POST")

@app.post("/modelforecast-file-demo/")
@GLOBAL_LIMIT
@limiter.limit("10/10minute")
async def modelforecast_file_demo(
    request: Request,
    df_forecast: UploadFile = File(...),
    session_id: str = Form(...),
):
    
    raw = await df_forecast.read()
    if len(raw) > SDK_MAX_BODY_BYTES:
        raise HTTPException(status_code=413, detail="Payload too large.")
    
    sid = session_id.strip()
    if not sid:
        raise HTTPException(status_code=400, detail="Missing session_id")
    
    files = {
        "df_forecast": (df_forecast.filename, raw, df_forecast.content_type or "application/octet-stream"),
    }
    data = {
        "session_id": sid,
    }

    return await _forward_demo_stream("/modelforecast-file-demo/", files=files, data=data, method="POST")

@app.get("/health-demo-proxy")
async def health_demo_proxy():
    # convenience endpoint to test private-space + demo token hop
    try:
        return await _forward_demo_stream("/health-demo", files=None, data=None, method="GET")
    except HTTPException as e:
        # bubble up errors so you can diagnose missing tokens, wrong URL, etc.
        raise e
    
@app.post("/modelfit_ind/")
async def modelfit_ind(req: Request):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    body = await req.json()
    return await _forward("/modelfit_ind/", "POST", json_body=body, user_token=user_token)

@app.post("/modelforecast_ind/")
async def modelforecast_ind(req: Request):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    body = await req.json()
    return await _forward("/modelforecast_ind/", "POST", json_body=body, user_token=user_token)


@app.post("/modelfit-file_ind/")
async def modelfit_file_ind(
    req: Request,
    file: UploadFile = File(...),
    id_col: str = Form(...),
    time_col: str = Form(...),
    y: str = Form(...),
    lag_y: str = Form(""),
    lagged_features: str = Form(""),
    current_features: str = Form(""),
    filter_by_significance: str = Form("false"),
    dummy_sqrt: str = Form("false"),
    dummy_interact: str = Form("false"),
    dummy_logs: str = Form("false"),
    dummy_yejo: str = Form("false"),
    dummy_asinh: str = Form("false"),
    dummy_logs_interact: str = Form("false"),
    meanvar_test: str = Form("false"),
    signif: str = Form("0.05"),
):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    raw = await file.read()
    if len(raw) > SDK_MAX_BODY_BYTES_extended:
        raise HTTPException(status_code=413, detail="Payload too large.")
    files = {
        "file": (file.filename, raw, file.content_type or "application/octet-stream"),
    }
    data = {
        "id_col": id_col,
        "time_col": time_col,
        "y": y,
        "lag_y": lag_y,
        "lagged_features": lagged_features,
        "current_features": current_features,
        "filter_by_significance": filter_by_significance,
        "dummy_sqrt": dummy_sqrt,
        "dummy_interact": dummy_interact,
        "dummy_logs": dummy_logs,
        "dummy_yejo": dummy_yejo,
        "dummy_asinh": dummy_asinh,
        "dummy_logs_interact": dummy_logs_interact,
        "meanvar_test": meanvar_test,
        "signif": signif,
    }
    return await _forward_stream("/modelfit-file_ind/", files=files, data=data, user_token=user_token, method="POST")


@app.post("/modelforecast-file_ind/")
async def modelforecast_file_ind(
    req: Request,
    df_forecast: UploadFile = File(...),
):
    user_token = _extract_user_token(req)
    if not user_token:
        raise HTTPException(status_code=401, detail="Missing Authorization Bearer token (dt+...)")
    raw = await df_forecast.read()
    if len(raw) > SDK_MAX_BODY_BYTES_extended:
        raise HTTPException(status_code=413, detail="Payload too large.")
    files = {
        "df_forecast": (df_forecast.filename, raw, df_forecast.content_type or "application/octet-stream"),
    }
    return await _forward_stream("/modelforecast-file_ind/", files=files, data=None, user_token=user_token, method="POST")