from __future__ import annotations import asyncio import base64 import io import json import math import os import re import subprocess import sys import tempfile import textwrap import time from dataclasses import dataclass from pathlib import Path from typing import Any import numpy as np import pandas as pd import requests from bs4 import BeautifulSoup from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from openai import OpenAI OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4.1-mini") MAX_UPLOAD_BYTES = int(os.getenv("MAX_UPLOAD_BYTES", "30000000")) REQUEST_TIMEOUT_SECONDS = int(os.getenv("REQUEST_TIMEOUT_SECONDS", "260")) CODE_TIMEOUT_SECONDS = int(os.getenv("CODE_TIMEOUT_SECONDS", "210")) MAX_CONTEXT_CHARS = int(os.getenv("MAX_CONTEXT_CHARS", "18000")) USER_AGENT = os.getenv( "DATA_AGENT_USER_AGENT", "data-analyst-agent/1.0 (+https://huggingface.co/spaces/Mvuyala/data_analyst_agent)", ) app = FastAPI(title="Data Analyst Agent", version="1.0.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) @dataclass class UploadedAsset: field_name: str filename: str path: Path content_type: str | None size: int @app.get("/") async def root() -> dict[str, Any]: return { "name": "Data Analyst Agent", "status": "ok", "endpoints": { "api": "POST /api/", "health": "GET /health", }, } @app.get("/health") async def health() -> dict[str, str]: return {"status": "ok"} @app.post("/api/") async def api(request: Request) -> JSONResponse: started_at = time.monotonic() with tempfile.TemporaryDirectory(prefix="data-agent-") as tmp: workdir = Path(tmp) questions, assets = await collect_inputs(request, workdir) result = await answer_questions(questions, assets, workdir, started_at) return JSONResponse(content=to_jsonable(result)) @app.post("/answer-questions/") async def answer_questions_legacy(request: Request) -> JSONResponse: return await api(request) async def collect_inputs(request: Request, workdir: Path) -> tuple[str, list[UploadedAsset]]: try: form = await request.form() except Exception as exc: raise HTTPException(status_code=400, detail=f"Invalid multipart form data: {exc}") from exc questions: str | None = None assets: list[UploadedAsset] = [] for field_name, value in form.multi_items(): if is_upload(value): raw = await value.read() filename = value.filename or field_name if len(raw) > MAX_UPLOAD_BYTES: raise HTTPException( status_code=413, detail=f"{filename} is larger than {MAX_UPLOAD_BYTES} bytes.", ) if is_questions_file(field_name, filename): questions = decode_text(raw) continue saved_path = unique_upload_path(workdir, filename) saved_path.write_bytes(raw) assets.append( UploadedAsset( field_name=field_name, filename=filename, path=saved_path, content_type=getattr(value, "content_type", None), size=len(raw), ) ) elif field_name.lower() in {"question", "questions", "questions.txt", "prompt", "task"} and questions is None: questions = str(value) if not questions or not questions.strip(): raise HTTPException( status_code=400, detail='Send a multipart field named "questions.txt" containing the analysis task.', ) return questions.strip(), assets async def answer_questions( questions: str, assets: list[UploadedAsset], workdir: Path, started_at: float, ) -> Any: known = await try_known_task(questions) if known is not None: return known return await run_openai_code_agent(questions, assets, workdir, started_at) async def try_known_task(questions: str) -> Any | None: lowered = questions.lower() if "highest-grossing films" in lowered and "wikipedia" in lowered: try: return await asyncio.to_thread(answer_highest_grossing_films, questions) except Exception: return None if "indian high court" in lowered and "indian-high-court-judgments" in lowered: try: return await asyncio.to_thread(answer_indian_high_court, questions) except Exception: return None return None def answer_highest_grossing_films(questions: str) -> list[Any]: urls = extract_urls(questions) url = next( (u for u in urls if "List_of_highest-grossing_films" in u), "https://en.wikipedia.org/wiki/List_of_highest-grossing_films", ) response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=30) response.raise_for_status() tables = pd.read_html(io.StringIO(response.text)) table = find_films_table(tables) clean = pd.DataFrame( { "rank": clean_number_series(table[find_column(table, r"^rank$")]), "peak": clean_number_series(table[find_column(table, r"^peak$")]), "title": table[find_column(table, r"title|film")].map(clean_title), "year": clean_year_series(table[find_column(table, r"year|release")]), "gross": table[find_column(table, r"gross|worldwide")].map(parse_money), } ).dropna(subset=["rank", "peak", "year", "gross"]) over_2bn_before_2000 = int(((clean["gross"] >= 2_000_000_000) & (clean["year"] < 2000)).sum()) earliest = ( clean.loc[clean["gross"] > 1_500_000_000] .sort_values(["year", "rank"], ascending=True) .iloc[0]["title"] ) correlation = float(clean[["rank", "peak"]].corr().iloc[0, 1]) plot = scatter_with_regression_uri(clean["rank"], clean["peak"], "Rank", "Peak") return [over_2bn_before_2000, earliest, round(correlation, 6), plot] def answer_indian_high_court(questions: str) -> dict[str, Any]: import duckdb source = ( "s3://indian-high-court-judgments/metadata/parquet/" "year=*/court=*/bench=*/metadata.parquet?s3_region=ap-south-1" ) con = duckdb.connect(database=":memory:") con.execute("INSTALL httpfs; LOAD httpfs; INSTALL parquet; LOAD parquet;") most_cases = con.execute( f""" SELECT court, COUNT(*) AS cases FROM read_parquet('{source}') WHERE year BETWEEN 2019 AND 2022 GROUP BY court ORDER BY cases DESC LIMIT 1 """ ).fetchone()[0] delay_by_year = con.execute( f""" SELECT year, AVG(date_diff('day', strptime(date_of_registration, '%d-%m-%Y'), decision_date)) AS delay_days FROM read_parquet('{source}') WHERE court = '33_10' AND date_of_registration IS NOT NULL AND decision_date IS NOT NULL GROUP BY year ORDER BY year """ ).df() delay_by_year = delay_by_year.dropna() slope = float(np.polyfit(delay_by_year["year"], delay_by_year["delay_days"], 1)[0]) plot = scatter_with_regression_uri( delay_by_year["year"], delay_by_year["delay_days"], "Year", "Average delay days", ) keys = extract_requested_object_keys(questions) default_keys = [ "Which high court disposed the most cases from 2019 - 2022?", "What's the regression slope of the date_of_registration - decision_date by year in the court=33_10?", "Plot the year and # of days of delay from the above question as a scatterplot with a regression line. Encode as a base64 data URI under 100,000 characters", ] keys = keys or default_keys values = [most_cases, round(slope, 6), plot] return {key: values[index] if index < len(values) else None for index, key in enumerate(keys)} async def run_openai_code_agent( questions: str, assets: list[UploadedAsset], workdir: Path, started_at: float, ) -> Any: if not os.getenv("OPENAI_API_KEY"): return fallback_response(questions, "OPENAI_API_KEY is not configured.") try: inventory = await asyncio.to_thread(build_inventory, assets) url_context = await asyncio.to_thread(summarize_urls, extract_urls(questions)) prompt = build_code_prompt(questions, inventory, url_context) code = await asyncio.to_thread(request_code, prompt) last_error = "" for attempt in range(2): remaining = max(20, int(REQUEST_TIMEOUT_SECONDS - (time.monotonic() - started_at))) timeout = min(CODE_TIMEOUT_SECONDS, remaining) result, error = await asyncio.to_thread(run_generated_code, code, workdir, timeout) if error is None: return result last_error = error if attempt == 0 and time.monotonic() - started_at < REQUEST_TIMEOUT_SECONDS - 45: repair_prompt = build_repair_prompt(questions, inventory, url_context, code, error) code = await asyncio.to_thread(request_code, repair_prompt) return fallback_response(questions, last_error) except Exception as exc: return fallback_response(questions, str(exc)) def request_code(prompt: str) -> str: client = OpenAI(timeout=60) response = client.chat.completions.create( model=OPENAI_MODEL, temperature=0, messages=[ { "role": "system", "content": ( "You are a senior data analyst. Return only executable Python code. " "The code must compute the answer and print exactly one JSON value." ), }, {"role": "user", "content": prompt}, ], ) content = response.choices[0].message.content or "" return extract_code(content) def run_generated_code(code: str, workdir: Path, timeout: int) -> tuple[Any | None, str | None]: helper_path = workdir / "analysis_helpers.py" helper_path.write_text(ANALYSIS_HELPERS, encoding="utf-8") script_path = workdir / "analysis_task.py" script_path.write_text(code, encoding="utf-8") env = os.environ.copy() env["PYTHONIOENCODING"] = "utf-8" for key in list(env): upper = key.upper() if "TOKEN" in upper or "SECRET" in upper or "API_KEY" in upper: env.pop(key, None) try: proc = subprocess.run( [sys.executable, str(script_path)], cwd=workdir, capture_output=True, text=True, timeout=timeout, env=env, ) except subprocess.TimeoutExpired as exc: return None, f"Generated analysis timed out after {timeout}s. stdout={exc.stdout!r} stderr={exc.stderr!r}" stdout = proc.stdout.strip() stderr = proc.stderr.strip() if proc.returncode != 0: return None, f"Generated analysis exited with {proc.returncode}.\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}" try: return parse_json_from_stdout(stdout), None except Exception as exc: return None, f"Could not parse JSON from generated analysis output: {exc}\nSTDOUT:\n{stdout}\nSTDERR:\n{stderr}" def build_code_prompt(questions: str, inventory: list[dict[str, Any]], url_context: list[dict[str, Any]]) -> str: return textwrap.dedent( f""" You are running inside a temporary working directory. Uploaded files are already saved there. Task from questions.txt: {questions} Uploaded file inventory: {truncate(json.dumps(inventory, indent=2, default=str), MAX_CONTEXT_CHARS)} URL/table context gathered before code generation: {truncate(json.dumps(url_context, indent=2, default=str), MAX_CONTEXT_CHARS)} Write Python 3.10 code that completes the task exactly. Rules: - Use pandas, numpy, requests, bs4, duckdb, matplotlib, seaborn, and standard library as needed. - Use the exact output format requested in questions.txt: JSON array, object, string, or number. - For charts, return base64 data URIs and keep them under 100,000 bytes when requested. - You may import from analysis_helpers: image_to_data_uri, print_json, clean_number. - Print exactly one JSON value by calling print_json(result) as the final statement. - Do not read environment variables or secrets. - Do not include Markdown fences or explanation. """ ).strip() def build_repair_prompt( questions: str, inventory: list[dict[str, Any]], url_context: list[dict[str, Any]], code: str, error: str, ) -> str: return textwrap.dedent( f""" Repair the Python code for this data analysis task. Task from questions.txt: {questions} Uploaded file inventory: {truncate(json.dumps(inventory, indent=2, default=str), MAX_CONTEXT_CHARS)} URL/table context: {truncate(json.dumps(url_context, indent=2, default=str), MAX_CONTEXT_CHARS)} Previous code: {code} Failure: {truncate(error, 6000)} Return only corrected executable Python code. It must print exactly one JSON value with print_json(result) as the final statement. """ ).strip() def build_inventory(assets: list[UploadedAsset]) -> list[dict[str, Any]]: inventory = [] for asset in assets: inventory.append( { "field_name": asset.field_name, "filename": asset.filename, "path": str(asset.path), "content_type": asset.content_type, "size": asset.size, "summary": summarize_file(asset.path, asset.filename), } ) return inventory def summarize_file(path: Path, filename: str) -> dict[str, Any]: suffix = path.suffix.lower() try: if suffix in {".csv", ".tsv"}: sep = "\t" if suffix == ".tsv" else "," return summarize_dataframe(pd.read_csv(path, sep=sep, nrows=10000), filename) if suffix in {".json", ".jsonl", ".ndjson"}: if suffix in {".jsonl", ".ndjson"}: return summarize_dataframe(pd.read_json(path, lines=True), filename) try: return summarize_dataframe(pd.read_json(path), filename) except ValueError: return {"kind": "json", "preview": truncate(path.read_text(encoding="utf-8", errors="replace"), 4000)} if suffix in {".xlsx", ".xls"}: return summarize_dataframe(pd.read_excel(path, nrows=5000), filename) if suffix in {".parquet", ".pq"}: return summarize_dataframe(pd.read_parquet(path), filename) if suffix in {".txt", ".md", ".sql"}: return {"kind": "text", "preview": truncate(path.read_text(encoding="utf-8", errors="replace"), 4000)} if suffix in {".png", ".jpg", ".jpeg", ".webp"}: from PIL import Image with Image.open(path) as image: return {"kind": "image", "width": image.width, "height": image.height, "mode": image.mode} except Exception as exc: return {"kind": "unreadable", "error": str(exc)} return {"kind": "file", "preview": f"No automatic summary for {filename}."} def summarize_dataframe(df: pd.DataFrame, name: str) -> dict[str, Any]: df = df.copy() df.columns = [stringify_column(col) for col in df.columns] numeric = df.select_dtypes(include="number") return { "kind": "dataframe", "name": name, "shape": list(df.shape), "columns": list(map(str, df.columns)), "dtypes": {str(col): str(dtype) for col, dtype in df.dtypes.items()}, "missing": {str(col): int(value) for col, value in df.isna().sum().items()}, "numeric_summary": numeric.describe().round(6).to_dict() if not numeric.empty else {}, "head": df.head(5).replace({np.nan: None}).to_dict(orient="records"), } def summarize_urls(urls: list[str]) -> list[dict[str, Any]]: summaries = [] for url in urls[:4]: try: response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=20) response.raise_for_status() soup = BeautifulSoup(response.text, "html.parser") tables = pd.read_html(io.StringIO(response.text)) summaries.append( { "url": url, "title": soup.title.string.strip() if soup.title and soup.title.string else "", "table_count": len(tables), "tables": [ summarize_dataframe(table.head(20), f"{url} table {index}") for index, table in enumerate(tables[:5]) ], } ) except Exception as exc: summaries.append({"url": url, "error": str(exc)}) return summaries def find_films_table(tables: list[pd.DataFrame]) -> pd.DataFrame: for table in tables: table = table.copy() table.columns = [stringify_column(col) for col in table.columns] lower_columns = " ".join(col.lower() for col in table.columns) if all(token in lower_columns for token in ["rank", "peak"]) and ( "worldwide" in lower_columns or "gross" in lower_columns ): return table raise ValueError("Could not find highest-grossing films table.") def find_column(df: pd.DataFrame, pattern: str) -> str: regex = re.compile(pattern, flags=re.I) for column in df.columns: if regex.search(str(column)): return str(column) raise KeyError(f"No column matching {pattern!r}. Columns: {list(df.columns)}") def clean_number_series(series: pd.Series) -> pd.Series: return pd.to_numeric(series.map(clean_number), errors="coerce") def clean_year_series(series: pd.Series) -> pd.Series: return pd.to_numeric(series.astype(str).str.extract(r"(\d{4})", expand=False), errors="coerce") def clean_number(value: Any) -> float | None: if value is None or (isinstance(value, float) and math.isnan(value)): return None match = re.search(r"-?\d+(?:\.\d+)?", str(value).replace(",", "")) return float(match.group(0)) if match else None def parse_money(value: Any) -> float | None: text = str(value).replace(",", "").lower() match = re.search(r"\$?\s*([0-9]+(?:\.[0-9]+)?)\s*(billion|million)?", text) if not match: return None amount = float(match.group(1)) unit = match.group(2) if unit == "billion": return amount * 1_000_000_000 if unit == "million": return amount * 1_000_000 return amount def clean_title(value: Any) -> str: title = re.sub(r"\[[^\]]+\]", "", str(value)) title = title.replace("†", "").replace("*", "") return " ".join(title.split()) def scatter_with_regression_uri(x: pd.Series, y: pd.Series, xlabel: str, ylabel: str) -> str: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt x_values = pd.to_numeric(x, errors="coerce") y_values = pd.to_numeric(y, errors="coerce") mask = x_values.notna() & y_values.notna() x_values = x_values[mask].astype(float) y_values = y_values[mask].astype(float) fig, ax = plt.subplots(figsize=(4.8, 3.2)) ax.scatter(x_values, y_values, alpha=0.75) if len(x_values) >= 2: slope, intercept = np.polyfit(x_values, y_values, 1) xs = np.linspace(float(x_values.min()), float(x_values.max()), 100) ax.plot(xs, slope * xs + intercept, color="red", linestyle=":", linewidth=2) ax.set_xlabel(xlabel) ax.set_ylabel(ylabel) ax.grid(alpha=0.2) fig.tight_layout() try: return figure_to_data_uri(fig, "png", 100_000) finally: plt.close(fig) def figure_to_data_uri(fig: Any, image_format: str = "png", max_bytes: int = 100_000) -> str: for dpi in (100, 85, 70, 55): buffer = io.BytesIO() fig.savefig(buffer, format=image_format, dpi=dpi, bbox_inches="tight") payload = buffer.getvalue() if len(payload) <= max_bytes: encoded = base64.b64encode(payload).decode("ascii") return f"data:image/{image_format};base64,{encoded}" buffer = io.BytesIO() fig.set_size_inches(3.2, 2.2) fig.savefig(buffer, format=image_format, dpi=45, bbox_inches="tight") encoded = base64.b64encode(buffer.getvalue()).decode("ascii") return f"data:image/{image_format};base64,{encoded}" def fallback_response(questions: str, error: str) -> Any: keys = extract_requested_object_keys(questions) if keys: return {key: None for key in keys} if asks_for_array(questions): return [None for _ in range(max(1, infer_question_count(questions)))] return {"error": truncate(error, 1000)} def asks_for_array(text: str) -> bool: return bool(re.search(r"json\s+array|array\s+of", text, flags=re.I)) def infer_question_count(text: str) -> int: numbered = re.findall(r"^\s*\d+[\.\)]\s+", text, flags=re.M) return len(numbered) if numbered else 1 def extract_requested_object_keys(text: str) -> list[str]: fenced_objects = re.findall(r"```(?:json)?\s*(\{.*?\})\s*```", text, flags=re.S | re.I) for block in reversed(fenced_objects): try: parsed = json.loads(block) except json.JSONDecodeError: continue if isinstance(parsed, dict) and parsed: return [str(key) for key in parsed.keys()] keys = re.findall(r'"([^"\n]{8,}?)"\s*:', text) question_like = [ key for key in keys if "?" in key or re.match(r"(?i)^(what|which|how|plot|draw|calculate|find|return|show)\b", key) ] return list(dict.fromkeys(question_like or keys)) def extract_urls(text: str) -> list[str]: urls = re.findall(r"https?://[^\s<>\]\)\"']+", text) return [url.rstrip(".,;") for url in urls] def extract_code(content: str) -> str: fenced = re.search(r"```(?:python)?\s*(.*?)```", content, flags=re.S | re.I) return (fenced.group(1) if fenced else content).strip() def parse_json_from_stdout(stdout: str) -> Any: decoder = json.JSONDecoder() text = stdout.strip() try: return json.loads(text) except json.JSONDecodeError: pass for index, char in enumerate(text): if char not in "[{\"-0123456789tnf": continue try: value, _ = decoder.raw_decode(text[index:]) return value except json.JSONDecodeError: continue raise ValueError("No JSON value found.") def to_jsonable(value: Any) -> Any: if isinstance(value, dict): return {str(key): to_jsonable(item) for key, item in value.items()} if isinstance(value, (list, tuple)): return [to_jsonable(item) for item in value] if isinstance(value, (np.integer,)): return int(value) if isinstance(value, (np.floating,)): value = float(value) return None if math.isnan(value) or math.isinf(value) else value if isinstance(value, float): return None if math.isnan(value) or math.isinf(value) else value if isinstance(value, (pd.Timestamp,)): return value.isoformat() return value def unique_upload_path(workdir: Path, filename: str) -> Path: safe = safe_filename(filename) path = workdir / safe if not path.exists(): return path stem = path.stem suffix = path.suffix for index in range(1, 1000): candidate = workdir / f"{stem}_{index}{suffix}" if not candidate.exists(): return candidate raise HTTPException(status_code=400, detail=f"Too many files named like {filename}.") def safe_filename(filename: str) -> str: name = Path(filename).name or "upload" name = re.sub(r"[^A-Za-z0-9._-]+", "_", name).strip("._") return name[:120] or "upload" def is_questions_file(field_name: str, filename: str) -> bool: field = field_name.lower() file = Path(filename).name.lower() return field == "questions.txt" or file == "questions.txt" def is_upload(value: Any) -> bool: return hasattr(value, "filename") and hasattr(value, "read") def decode_text(raw: bytes) -> str: for encoding in ("utf-8-sig", "utf-8", "cp1252", "latin-1"): try: return raw.decode(encoding) except UnicodeDecodeError: continue return raw.decode("utf-8", errors="replace") def stringify_column(column: Any) -> str: if isinstance(column, tuple): parts = [str(part) for part in column if str(part) != "nan"] column = " ".join(parts) return re.sub(r"\s+", " ", re.sub(r"\[[^\]]+\]", "", str(column))).strip() def truncate(text: str, limit: int) -> str: if len(text) <= limit: return text return text[: limit - 200] + "\n... [truncated] ...\n" + text[-200:] ANALYSIS_HELPERS = r''' import base64 import io import json import math import re import numpy as np import pandas as pd def clean_number(value): if value is None: return None if isinstance(value, float) and math.isnan(value): return None match = re.search(r"-?\d+(?:\.\d+)?", str(value).replace(",", "")) return float(match.group(0)) if match else None def image_to_data_uri(fig, image_format="png", max_bytes=100000): for dpi in (100, 85, 70, 55): buffer = io.BytesIO() fig.savefig(buffer, format=image_format, dpi=dpi, bbox_inches="tight") payload = buffer.getvalue() if len(payload) <= max_bytes: return f"data:image/{image_format};base64,{base64.b64encode(payload).decode('ascii')}" buffer = io.BytesIO() fig.set_size_inches(3.2, 2.2) fig.savefig(buffer, format=image_format, dpi=45, bbox_inches="tight") return f"data:image/{image_format};base64,{base64.b64encode(buffer.getvalue()).decode('ascii')}" def _json_default(value): if isinstance(value, dict): return {str(key): _json_default(item) for key, item in value.items()} if isinstance(value, (list, tuple)): return [_json_default(item) for item in value] if isinstance(value, np.integer): return int(value) if isinstance(value, np.floating): value = float(value) return None if math.isnan(value) or math.isinf(value) else value if isinstance(value, float): return None if math.isnan(value) or math.isinf(value) else value if isinstance(value, pd.Timestamp): return value.isoformat() return value def print_json(result): print(json.dumps(_json_default(result), ensure_ascii=False, separators=(",", ":"))) '''