external_api.py

"""Adapter between the Gradio app and the external stock-analysis project.

The external project is not hard-coded. Set ``EXTERNAL_API_URL`` to its HTTP
endpoint when it is ready. The app sends a small JSON request, keeps the
upstream body, and passes JSON through unchanged. Non-JSON bodies are wrapped
so the Gradio JSON output always has a stable shape.

This module is the only place to change when the external request or response
contract changes.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any

import requests

DEFAULT_TIMEOUT_SECONDS = 120.0

_settings: dict[str, Any] = {
    "url": "",
    "timeout": DEFAULT_TIMEOUT_SECONDS,
    "model_id": "",
}


@dataclass(frozen=True)
class ExternalResult:
    """Normalized result returned by :func:`query_external`."""

    payload: Any
    raw: str
    http_status: int | None = None
    content_type: str | None = None
    is_json: bool = False
    error: str | None = None


def configure(
    *,
    url: str | None = None,
    timeout: float | None = None,
    model_id: str | None = None,
) -> dict[str, Any]:
    """Apply settings from ``app.py``. All values are optional."""
    if url is not None:
        _settings["url"] = str(url).strip()
    if timeout is not None:
        _settings["timeout"] = float(timeout)
    if model_id is not None:
        _settings["model_id"] = str(model_id).strip()
    return settings()


def settings() -> dict[str, Any]:
    """Return a copy of the current settings, without any credentials."""
    return {
        "configured": is_configured(),
        "timeout": float(_settings["timeout"]),
        "model_id": _resolved_model_id(),
    }


def is_configured() -> bool:
    return bool(_resolved_url())


def model_id() -> str:
    return _resolved_model_id()


def _resolved_url() -> str:
    return str(_settings["url"] or os.getenv("EXTERNAL_API_URL", "")).strip()


def _resolved_model_id() -> str:
    return str(_settings["model_id"] or os.getenv("STOCK_MODEL_ID", "")).strip()


def _token() -> str | None:
    return os.getenv("HF_TOKEN") or None


def _redact(text: str, token: str | None) -> str:
    if token:
        text = text.replace(token, "hf_***")
    return text


def _headers(token: str | None) -> dict[str, str]:
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json",
        "User-Agent": "StockAnalysis-Space/1.0",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"
    return headers


def _not_configured(ticker: str) -> ExternalResult:
    message = "Set EXTERNAL_API_URL to the stock-analysis project endpoint."
    return ExternalResult(
        payload={
            "status": "not_configured",
            "ticker": ticker,
            "message": message,
        },
        raw="",
        error=message,
    )


def query_external(
    ticker: str,
    horizon: str,
    risk_profile: str,
    question: str,
) -> ExternalResult:
    """Send one analysis request to the external project.

    The request body is intentionally small and easy to map onto another
    service. If ``STOCK_MODEL_ID`` is set, it is included as ``model`` so the
    external project can select the uploaded custom model.
    """
    url = _resolved_url()
    if not url:
        return _not_configured(ticker)

    token = _token()
    payload: dict[str, Any] = {
        "ticker": ticker,
        "horizon": horizon,
        "risk_profile": risk_profile,
        "question": question,
    }
    resolved_model = _resolved_model_id()
    if resolved_model:
        payload["model"] = resolved_model

    try:
        response = requests.post(
            url,
            json=payload,
            headers=_headers(token),
            timeout=float(_settings["timeout"]),
        )
    except requests.RequestException as exc:
        message = _redact(f"{type(exc).__name__}: {exc}", token)
        return ExternalResult(
            payload={
                "status": "error",
                "ticker": ticker,
                "error": message,
            },
            raw="",
            error=message,
        )

    raw = response.text or ""
    content_type = response.headers.get("content-type", "").split(";")[0].strip() or None
    try:
        parsed = response.json()
    except ValueError:
        parsed = None

    if not response.ok:
        message = _redact(f"HTTP {response.status_code}: {raw[:500]}", token)
        return ExternalResult(
            payload={
                "status": "error",
                "ticker": ticker,
                "http_status": response.status_code,
                "content_type": content_type,
                "error": message,
                "raw": raw,
            },
            raw=raw,
            http_status=response.status_code,
            content_type=content_type,
            error=message,
        )

    if parsed is not None:
        return ExternalResult(
            payload=parsed,
            raw=raw,
            http_status=response.status_code,
            content_type=content_type,
            is_json=True,
        )

    return ExternalResult(
        payload={
            "status": "ok",
            "ticker": ticker,
            "http_status": response.status_code,
            "content_type": content_type,
            "raw": raw,
        },
        raw=raw,
        http_status=response.status_code,
        content_type=content_type,
        is_json=False,
    )