Spaces:
Running
Running
| """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": "", | |
| } | |
| 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, | |
| ) | |