Spaces:
Sleeping
Sleeping
File size: 3,185 Bytes
da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 da1806a 36bd0e2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | import json
import logging
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
import pandas as pd
import numpy as np
import os
app = FastAPI()
# ---------------------------------------------------------
# Logging
# ---------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------
# CORS (optional but useful for frontend)
# ---------------------------------------------------------
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# ---------------------------------------------------------
# Lazy-loaded dataset
# ---------------------------------------------------------
DATASET_PATH = "transcripts.csv" # change if needed
df_cache = None
def load_dataset():
global df_cache
if df_cache is None:
logger.info("Loading dataset...")
if not os.path.exists(DATASET_PATH):
raise FileNotFoundError(f"Dataset not found: {DATASET_PATH}")
df_cache = pd.read_csv(DATASET_PATH)
logger.info(f"Dataset loaded with {len(df_cache)} rows.")
return df_cache
# ---------------------------------------------------------
# JSON-safe conversion
# ---------------------------------------------------------
def to_json_safe(obj):
if isinstance(obj, (np.integer, np.int64)):
return int(obj)
if isinstance(obj, (np.floating, np.float64)):
return float(obj)
if isinstance(obj, (np.ndarray, list)):
return [to_json_safe(x) for x in obj]
if isinstance(obj, dict):
return {k: to_json_safe(v) for k, v in obj.items()}
return obj
# ---------------------------------------------------------
# Serve index.html at "/"
# ---------------------------------------------------------
@app.get("/", response_class=HTMLResponse)
def serve_index():
if not os.path.exists("index.html"):
return "<h1>index.html not found</h1>"
with open("index.html", "r") as f:
return f.read()
# ---------------------------------------------------------
# List all tickers
# ---------------------------------------------------------
@app.get("/tickers")
def get_tickers():
df = load_dataset()
tickers = sorted(df["symbol"].unique().tolist())
return {"tickers": tickers}
# ---------------------------------------------------------
# Get transcript for a symbol
# ---------------------------------------------------------
@app.get("/transcript/{symbol}")
def get_transcript(symbol: str):
df = load_dataset()
symbol = symbol.upper()
logger.info(f"Fetching transcript for: {symbol}")
subset = df[df["symbol"] == symbol]
if subset.empty:
raise HTTPException(status_code=404, detail=f"No transcript found for {symbol}")
# Convert each row to JSON-safe dict
records = subset.to_dict(orient="records")
safe_records = [to_json_safe(r) for r in records]
return {"symbol": symbol, "records": safe_records}
|