deploy: auto-deploy 17:08:05

app/api/v1/csv_analysis.py

@@ -0,0 +1,315 @@
+from __future__ import annotations
+
+import json
+from typing import Annotated, Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
+from pydantic import BaseModel, Field, ValidationError
+
+from app.api.deps import require_auth
+from app.config import get_settings
+from app.services.chat_service import chat_completion
+from app.services.csv_analysis_service import (
+    analyze_csv_dataset,
+    create_csv_chart,
+    execute_csv_chat_blocks,
+    get_dataset_info,
+)
+from app.services.prompts import get_csv_system_prompt
+from app.utils.json_utils import extract_json_blocks
+
+
+class _AnalyzeBlock(BaseModel):
+    description: str = ""
+    python_code: str = ""
+
+
+class _VisualizationBlock(BaseModel):
+    description: str = ""
+    python_code: str = ""
+
+
+class _AIResponse(BaseModel):
+    analyze: List[_AnalyzeBlock] = []
+    visualization: List[_VisualizationBlock] = []
+    message: str = ""
+
+router = APIRouter()
+_settings = get_settings()
+_MAX_UPLOAD_BYTES = _settings.max_upload_bytes
+
+
+@router.post(
+    "/csv/info",
+    summary="Get metadata for up to 10 CSV files (upload or URL)",
+)
+async def get_csv_info(
+    files: Annotated[Optional[List[UploadFile]], File(description="CSV files to inspect (max 10 total with URLs)")] = None,
+    urls: Annotated[Optional[str], Form(description="JSON array of file URLs (max 10 total with files)")] = None,
+    token: str = Depends(require_auth),
+):
+    parsed_urls: List[str] = []
+    if urls:
+        try:
+            parsed_urls = json.loads(urls)
+            if not isinstance(parsed_urls, list) or not all(isinstance(u, str) for u in parsed_urls):
+                raise ValueError("urls must be a JSON array of strings")
+        except (json.JSONDecodeError, ValueError) as exc:
+            raise HTTPException(status_code=400, detail=str(exc))
+
+    file_count = len(files) if files else 0
+    url_count = len(parsed_urls)
+    total = file_count + url_count
+
+    if total == 0:
+        raise HTTPException(status_code=400, detail="Provide at least one file or URL")
+    if total > 10:
+        raise HTTPException(status_code=400, detail=f"Maximum 10 sources allowed (got {total})")
+
+    results: List[dict] = []
+
+    if files:
+        for f in files:
+            try:
+                data = await f.read()
+            except Exception as exc:
+                results.append({"source": getattr(f, "filename", "unknown"), "success": False, "error": f"Read error: {exc}"})
+                continue
+
+            if len(data) > _MAX_UPLOAD_BYTES:
+                results.append({"source": f.filename or "unknown", "success": False, "error": f"File exceeds {_settings.max_upload_mb} MB limit"})
+                continue
+
+            if not data:
+                results.append({"source": f.filename or "unknown", "success": False, "error": "Empty file"})
+                continue
+
+            try:
+                meta = await get_dataset_info(data)
+                meta["source"] = f.filename or "upload"
+                results.append(meta)
+            except Exception as exc:
+                results.append({"source": f.filename or "upload", "success": False, "error": str(exc)})
+
+    for url in parsed_urls:
+        if not url.startswith(("http://", "https://")):
+            results.append({"source": url, "success": False, "error": "Only http/https URLs are supported"})
+            continue
+
+        try:
+            meta = await get_dataset_info(url)
+            meta["source"] = url
+            results.append(meta)
+        except Exception as exc:
+            results.append({"source": url, "success": False, "error": str(exc)})
+
+    return {
+        "success": True,
+        "total": total,
+        "succeeded": sum(1 for r in results if r.get("success")),
+        "failed": sum(1 for r in results if not r.get("success")),
+        "results": results,
+    }
+
+
+# @router.post(
+#     "/csv/analyze",
+#     summary="Execute Python analysis code against a CSV file (upload or URL)",
+# )
+# async def analyze_csv(
+#     file: Annotated[Optional[UploadFile], File(description="CSV file to analyze")] = None,
+#     url: Annotated[Optional[str], Form(description="URL to a CSV file")] = None,
+#     code: str = Form(..., description="Python code to execute (df pre-loaded with CSV data)"),
+#     token: str = Depends(require_auth),
+# ):
+#     if not file and not url:
+#         raise HTTPException(status_code=400, detail="Provide either a file or a URL")
+
+#     if file and url:
+#         raise HTTPException(status_code=400, detail="Provide either a file or a URL, not both")
+
+#     if file:
+#         data = await file.read()
+#         if len(data) > _MAX_UPLOAD_BYTES:
+#             raise HTTPException(status_code=413, detail=f"File exceeds {_settings.max_upload_mb} MB limit")
+#         if not data:
+#             raise HTTPException(status_code=400, detail="Empty file")
+#         result = await analyze_csv_dataset(data, code)
+#     else:
+#         result = await analyze_csv_dataset(url, code)
+
+#     return result
+
+
+# @router.post(
+#     "/csv/chart",
+#     summary="Generate a chart from CSV data and return as base64 PNG (upload or URL)",
+# )
+# async def chart_csv(
+#     file: Annotated[Optional[UploadFile], File(description="CSV file for chart generation")] = None,
+#     url: Annotated[Optional[str], Form(description="URL to a CSV file")] = None,
+#     code: str = Form(..., description="Python chart code (df pre-loaded, use matplotlib/seaborn)"),
+#     token: str = Depends(require_auth),
+# ):
+#     if not file and not url:
+#         raise HTTPException(status_code=400, detail="Provide either a file or a URL")
+
+#     if file and url:
+#         raise HTTPException(status_code=400, detail="Provide either a file or a URL, not both")
+
+#     if file:
+#         data = await file.read()
+#         if len(data) > _MAX_UPLOAD_BYTES:
+#             raise HTTPException(status_code=413, detail=f"File exceeds {_settings.max_upload_mb} MB limit")
+#         if not data:
+#             raise HTTPException(status_code=400, detail="Empty file")
+#         result = await create_csv_chart(data, code)
+#     else:
+#         result = await create_csv_chart(url, code)
+
+#     return result
+
+
+@router.post(
+    "/csv/chat",
+    summary="Chat with AI about a CSV file — returns analysis + chart code results",
+)
+async def csv_chat(
+    request: Request,
+    file: Annotated[Optional[UploadFile], File(description="CSV file to analyze")] = None,
+    url: Annotated[Optional[str], Form(description="URL to a CSV file")] = None,
+    query: str = Form(..., description="Natural language query about the CSV data"),
+    token: str = Depends(require_auth),
+):
+    if not file and not url:
+        raise HTTPException(status_code=400, detail="Provide either a file or a URL")
+    if file and url:
+        raise HTTPException(status_code=400, detail="Provide either a file or a URL, not both")
+
+    if file:
+        data = await file.read()
+        if len(data) > _MAX_UPLOAD_BYTES:
+            raise HTTPException(status_code=413, detail=f"File exceeds {_settings.max_upload_mb} MB limit")
+        if not data:
+            raise HTTPException(status_code=400, detail="Empty file")
+        source: Any = data
+    else:
+        source = url
+
+    metadata = await get_dataset_info(source)
+    system_prompt = get_csv_system_prompt(metadata)
+
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": query},
+    ]
+
+    redis = getattr(request.app.state, "redis", None)
+    scripts = getattr(request.app.state, "scripts", None)
+
+    try:
+        ai_response = await chat_completion(
+            messages=messages,
+            response_format={"type": "json_object"},
+            max_tokens=12000,
+            redis=redis,
+            scripts=scripts,
+        )
+    except RuntimeError as e:
+        raise HTTPException(status_code=502, detail=str(e))
+
+    parsed = ai_response.get("parsed")
+    if not parsed:
+        choices = ai_response.get("choices", [])
+        content = choices[0].get("message", {}).get("content", "") if choices else ""
+        blocks = extract_json_blocks(content)
+        if blocks:
+            parsed = blocks[0]
+        else:
+            try:
+                parsed = json.loads(content)
+            except (json.JSONDecodeError, TypeError):
+                pass
+
+    if not isinstance(parsed, dict):
+        return {
+            "success": False,
+            "message": None,
+            "analyze": [],
+            "visualizations": [],
+            "error": "AI response was not valid JSON",
+        }
+
+    try:
+        ai_data = _AIResponse(**parsed)
+    except ValidationError as exc:
+        return {
+            "success": False,
+            "message": None,
+            "analyze": [],
+            "visualizations": [],
+            "error": f"AI response failed schema validation: {exc}",
+        }
+
+    message_text = ai_data.message
+    has_content = bool(message_text.strip()) if message_text else False
+
+    analyze_blocks_raw = [b.model_dump() for b in ai_data.analyze]
+    viz_blocks_raw = [b.model_dump() for b in ai_data.visualization]
+
+    exec_result = await execute_csv_chat_blocks(
+        source=source,
+        analyze_blocks=analyze_blocks_raw,
+        viz_blocks=viz_blocks_raw,
+    )
+
+    if not exec_result["success"]:
+        return {
+            "success": False,
+            "message": ai_data.message if has_content else None,
+            "analyze": [],
+            "visualizations": [],
+            "error": exec_result.get("error", "Code execution failed"),
+        }
+
+    results = exec_result.get("results", {})
+    raw_analyze = results.get("analyze", [])
+    raw_visualizations = results.get("visualization", [])
+
+    analyze_results: List[Dict[str, Any]] = []
+    for i, block in enumerate(ai_data.analyze):
+        raw = raw_analyze[i] if i < len(raw_analyze) else {}
+        code = block.python_code.strip()
+        if not code:
+            continue
+        analyze_results.append({
+            "description": block.description,
+            "code": code,
+            "success": raw.get("success", False),
+            "output": raw.get("output", ""),
+            "error": raw.get("error"),
+            "execution_time_ms": exec_result["execution_time_ms"],
+        })
+
+    viz_results: List[Dict[str, Any]] = []
+    for i, block in enumerate(ai_data.visualization):
+        raw = raw_visualizations[i] if i < len(raw_visualizations) else {}
+        code = block.python_code.strip()
+        if not code:
+            continue
+        viz_results.append({
+            "description": block.description,
+            "code": code,
+            "success": raw.get("success", False),
+            "image_base64": raw.get("image_base64"),
+            "error": raw.get("error"),
+            "execution_time_ms": exec_result["execution_time_ms"],
+        })
+
+    return {
+        "success": True,
+        "message": ai_data.message if has_content else None,
+        "analyze": analyze_results,
+        "visualizations": viz_results,
+        "error": None,
+    }

app/api/v1/router.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from fastapi import APIRouter
 
-from app.api.v1 import auth, batch, chat, code_executor, convert, database, embeddings, qr_generator, reconcile, scraper, semantic_router, sql_validator, system, token_counter, token_generator, url_shortener, vector_stores, web_search, webhook_socket
+from app.api.v1 import auth, batch, chat, code_executor, convert, csv_analysis, database, embeddings, qr_generator, reconcile, scraper, semantic_router, sql_validator, system, token_counter, token_generator, url_shortener, vector_stores, web_search, webhook_socket
 from app.api.verify import router as verify_router
 
 api_v1_router = APIRouter()
@@ -24,5 +24,6 @@ api_v1_router.include_router(token_generator.router, tags=["Token Generator"])
 api_v1_router.include_router(chat.router, tags=["Chat"])
 api_v1_router.include_router(vector_stores.router, tags=["Vector Stores"])
 api_v1_router.include_router(webhook_socket.router, tags=["Webhook / Socket"])
+api_v1_router.include_router(csv_analysis.router, tags=["CSV Analysis"])
 api_v1_router.include_router(qr_generator.router, tags=["QR Generator"])
 api_v1_router.include_router(url_shortener.router, tags=["URL Shortener"])

app/services/csv_analysis_service.py

@@ -0,0 +1,299 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import shutil
+import subprocess
+import signal
+import sys
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple, Union
+from urllib.parse import unquote, urlparse
+
+import aiohttp
+
+from app.services.code_executor_service import CodeSanitizer
+from app.services.dataset_metadata_service import extract_metadata
+
+logger = logging.getLogger(__name__)
+
+_PYTHON = getattr(sys, "executable", None) or "python3"
+_MAX_OUTPUT_BYTES = 65536
+_MAX_CONCURRENT = 8
+_DOWNLOAD_CHUNK_SIZE = 256 * 1024
+_DOWNLOAD_TIMEOUT = 120
+
+_semaphore = asyncio.Semaphore(_MAX_CONCURRENT)
+
+
+class CSVAnalysisError(Exception):
+    pass
+
+
+async def _download_file(url: str) -> bytes:
+    timeout = aiohttp.ClientTimeout(total=_DOWNLOAD_TIMEOUT)
+    try:
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with session.get(url) as resp:
+                if resp.status != 200:
+                    raise CSVAnalysisError(f"HTTP {resp.status} when fetching {url}")
+                chunks: List[bytes] = []
+                async for chunk in resp.content.iter_chunked(_DOWNLOAD_CHUNK_SIZE):
+                    chunks.append(chunk)
+                return b"".join(chunks)
+    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
+        raise CSVAnalysisError(f"Download failed for {url}: {exc}") from exc
+
+
+async def _resolve_source(source: Union[str, bytes]) -> Tuple[bytes, Optional[str]]:
+    if isinstance(source, str) and source.lower().startswith(("http://", "https://")):
+        data = await _download_file(source)
+        parsed = urlparse(source)
+        filename = unquote(Path(parsed.path).name) if parsed.path else None
+        return data, filename
+    elif isinstance(source, bytes):
+        return source, None
+    else:
+        raise TypeError(f"Unsupported source type: {type(source)}")
+
+
+def _run_subprocess(cmd: List[str], timeout: float, max_output: int) -> Dict[str, Any]:
+    proc = subprocess.Popen(
+        cmd,
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    try:
+        stdout_bytes, stderr_bytes = proc.communicate(timeout=timeout)
+        timed_out = False
+    except subprocess.TimeoutExpired:
+        try:
+            if os.name == "nt":
+                proc.kill()
+            else:
+                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+        except Exception:
+            proc.kill()
+        stdout_bytes, stderr_bytes = proc.communicate()
+        timed_out = True
+
+    return {
+        "stdout": (stdout_bytes.decode("utf-8", errors="replace")[:max_output] if stdout_bytes else ""),
+        "stderr": (stderr_bytes.decode("utf-8", errors="replace")[:max_output] if stderr_bytes else ""),
+        "exit_code": proc.returncode,
+        "timed_out": timed_out,
+    }
+
+
+_CHAT_SCRIPT = """\
+import json, sys, io, base64, traceback
+import pandas as pd, numpy as np
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+import seaborn as sns
+
+df = pd.read_csv(r"{csv_path}")
+
+with open(r"{blocks_path}", "r") as _f:
+    _data = json.load(_f)
+
+_results = {{"analyze": [], "visualization": []}}
+_ns = {{"df": df, "pd": pd, "np": np, "plt": plt, "sns": sns}}
+
+for _b in _data.get("analyze", []):
+    _code = (_b.get("python_code") or "").strip()
+    if not _code:
+        _results["analyze"].append({{"success": True, "output": "", "error": None}})
+        continue
+    _old = sys.stdout
+    sys.stdout = io.StringIO()
+    try:
+        exec(_code, _ns)
+        _results["analyze"].append({{"success": True, "output": sys.stdout.getvalue(), "error": None}})
+    except Exception:
+        _results["analyze"].append({{"success": False, "output": sys.stdout.getvalue(), "error": traceback.format_exc()}})
+    finally:
+        sys.stdout = _old
+
+for _b in _data.get("visualization", []):
+    _code = (_b.get("python_code") or "").strip()
+    if not _code:
+        _results["visualization"].append({{"success": True, "image_base64": "", "error": None}})
+        continue
+    _full = _code + (
+        "\\nfrom io import BytesIO\\nimport base64\\n"
+        "_buf = BytesIO()\\nplt.savefig(_buf, format='png', bbox_inches='tight', dpi=150)\\n"
+        "_buf.seek(0)\\nprint(base64.b64encode(_buf.read()).decode(), end='')\\n"
+        "plt.close('all')\\n"
+    )
+    _old = sys.stdout
+    sys.stdout = io.StringIO()
+    try:
+        exec(_full, _ns)
+        _results["visualization"].append({{"success": True, "image_base64": sys.stdout.getvalue().strip(), "error": None}})
+    except Exception:
+        _results["visualization"].append({{"success": False, "image_base64": None, "error": traceback.format_exc()}})
+    finally:
+        sys.stdout = _old
+    plt.close("all")
+
+print(json.dumps(_results))
+"""
+
+
+async def execute_csv_chat_blocks(
+    source: Union[str, bytes],
+    analyze_blocks: List[Dict[str, Any]],
+    viz_blocks: List[Dict[str, Any]],
+    timeout: int = 60,
+) -> Dict[str, Any]:
+    data, _ = await _resolve_source(source)
+    if not data:
+        return {"success": False, "results": None, "error": "No data provided"}
+
+    async with _semaphore:
+        run_dir = None
+        start = time.monotonic()
+        try:
+            run_dir = Path(tempfile.mkdtemp())
+            csv_path = run_dir / "data.csv"
+            csv_path.write_bytes(data)
+
+            blocks_path = run_dir / "blocks.json"
+            blocks_path.write_text(
+                json.dumps({"analyze": analyze_blocks, "visualization": viz_blocks}),
+                encoding="utf-8",
+            )
+
+            script = _CHAT_SCRIPT.format(
+                csv_path=csv_path.as_posix(),
+                blocks_path=blocks_path.as_posix(),
+            )
+
+            script_path = run_dir / "chat_exec.py"
+            script_path.write_text(script, encoding="utf-8")
+
+            cmd = [_PYTHON, str(script_path)]
+            result = await asyncio.to_thread(_run_subprocess, cmd, timeout, _MAX_OUTPUT_BYTES)
+            elapsed_ms = (time.monotonic() - start) * 1000
+
+            if result["exit_code"] != 0:
+                return {
+                    "success": False,
+                    "results": None,
+                    "error": result["stderr"] or "Subprocess failed",
+                    "execution_time_ms": round(elapsed_ms, 2),
+                }
+
+            parsed = json.loads(result["stdout"])
+            return {
+                "success": True,
+                "results": parsed,
+                "error": None,
+                "execution_time_ms": round(elapsed_ms, 2),
+            }
+        except json.JSONDecodeError as exc:
+            elapsed_ms = (time.monotonic() - start) * 1000
+            return {"success": False, "results": None, "error": f"Failed to parse output: {exc}", "execution_time_ms": round(elapsed_ms, 2)}
+        except FileNotFoundError:
+            elapsed_ms = (time.monotonic() - start) * 1000
+            return {"success": False, "results": None, "error": f"Python runtime ({_PYTHON}) not found", "execution_time_ms": round(elapsed_ms, 2)}
+        except Exception as exc:
+            elapsed_ms = (time.monotonic() - start) * 1000
+            logger.exception("CSV chat execution error")
+            return {"success": False, "results": None, "error": f"Execution error: {exc}", "execution_time_ms": round(elapsed_ms, 2)}
+        finally:
+            if run_dir and run_dir.exists():
+                shutil.rmtree(run_dir, ignore_errors=True)
+
+
+async def get_dataset_info(source: Union[str, bytes, Any]) -> Dict[str, Any]:
+    return await extract_metadata(source)
+
+
+async def analyze_csv_dataset(
+    source: Union[str, bytes],
+    code: str,
+    timeout: int = 30,
+) -> Dict[str, Any]:
+    data, _ = await _resolve_source(source)
+
+    if not data:
+        return {"success": False, "output": "", "error": "No data provided", "execution_time_ms": None}
+
+    sanitized, err = CodeSanitizer.sanitize(code, "python")
+    if not sanitized:
+        return {"success": False, "output": "", "error": err, "execution_time_ms": None}
+
+    async with _semaphore:
+        run_dir = None
+        start = time.monotonic()
+        try:
+            run_dir = Path(tempfile.mkdtemp())
+            csv_path = run_dir / "data.csv"
+            csv_path.write_bytes(data)
+
+            loader = (
+                "import pandas as pd, numpy as np\n"
+                f"df = pd.read_csv(r'{csv_path}')\n"
+            )
+            full_code = loader + code
+
+            code_path = run_dir / "analysis.py"
+            code_path.write_text(full_code, encoding="utf-8")
+
+            cmd = [_PYTHON, str(code_path)]
+            result = await asyncio.to_thread(_run_subprocess, cmd, timeout, _MAX_OUTPUT_BYTES)
+
+            elapsed_ms = (time.monotonic() - start) * 1000
+
+            return {
+                "success": result["exit_code"] == 0 and not result["timed_out"],
+                "output": result["stdout"],
+                "error": result["stderr"] or None,
+                "execution_time_ms": round(elapsed_ms, 2),
+                "timed_out": result["timed_out"],
+            }
+
+        except FileNotFoundError:
+            elapsed_ms = (time.monotonic() - start) * 1000
+            return {"success": False, "output": "", "error": f"Python runtime ({_PYTHON}) not found", "execution_time_ms": round(elapsed_ms, 2)}
+        except Exception as exc:
+            elapsed_ms = (time.monotonic() - start) * 1000
+            logger.exception("CSV analysis execution error")
+            return {"success": False, "output": "", "error": f"Execution error: {exc}", "execution_time_ms": round(elapsed_ms, 2)}
+        finally:
+            if run_dir and run_dir.exists():
+                shutil.rmtree(run_dir, ignore_errors=True)
+
+
+async def create_csv_chart(
+    source: Union[str, bytes],
+    code: str,
+    timeout: int = 30,
+) -> Dict[str, Any]:
+    preamble = (
+        "import matplotlib\nmatplotlib.use('Agg')\n"
+        "import matplotlib.pyplot as plt\nimport seaborn as sns\nimport pandas as pd\n"
+    )
+    postamble = (
+        "\n\nfrom io import BytesIO\nimport base64\n"
+        "_buf = BytesIO()\n"
+        "plt.savefig(_buf, format='png', bbox_inches='tight', dpi=150)\n"
+        "_buf.seek(0)\n"
+        "print(base64.b64encode(_buf.read()).decode(), end='')\n"
+        "plt.close('all')\n"
+    )
+
+    full_code = preamble + code + postamble
+    result = await analyze_csv_dataset(source, full_code, timeout)
+
+    if result["success"]:
+        return {"success": True, "image_base64": result.get("output", "") or "", "error": None, "execution_time_ms": result["execution_time_ms"]}
+    return {"success": False, "image_base64": None, "error": result.get("error") or "Chart generation failed or produced no output", "execution_time_ms": result["execution_time_ms"]}

app/services/prompts/__init__.py

@@ -0,0 +1,3 @@
+from app.services.prompts.csv_system_prompt import get_csv_system_prompt
+
+__all__ = ["get_csv_system_prompt"]

app/services/prompts/csv_system_prompt.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Any, Dict
+
+_JSON_EXAMPLE = (
+    '```json\n'
+    '{\n'
+    '  "analyze": [\n'
+    '    {\n'
+    '      "description": "Short explanation of the math",\n'
+    '      "python_code": "# Clean data first\\ndf[\'col\'] = ...\\n\\n# Perform analysis\\nresult = df.groupby..."\n'
+    '    }\n'
+    '  ],\n'
+    '  "visualization": [\n'
+    '    {\n'
+    '      "description": "Short explanation of the chart",\n'
+    '      "python_code": "# Clean data first\\ndf[\'col\'] = ...\\n\\n# Plot\\nplt.figure(figsize=(12,6))\\nsns.barplot(data=df, ...)"\n'
+    '    }\n'
+    '  ],\n'
+    '  "message": "Fill this ONLY if the user is greeting, asking non-data questions or asking for wrong information."\n'
+    '}\n'
+    '```'
+)
+
+
+def get_csv_system_prompt(metadata: Dict[str, Any]) -> str:
+    shape = metadata.get("shape", {})
+    num_rows = shape.get("rows", "?")
+    num_cols = shape.get("columns", "?")
+    columns = metadata.get("columns", [])
+    dtypes = metadata.get("dtypes", {})
+    sample_data = metadata.get("sample_data", [])
+    numeric_cols = metadata.get("numeric_columns", [])
+    categorical_cols = metadata.get("categorical_columns", [])
+
+    columns_str = ", ".join(columns)
+    dtypes_str = json.dumps(dtypes)
+    sample_str = json.dumps(sample_data[:1], indent=2) if sample_data else "[]"
+    numeric_str = ", ".join(numeric_cols) if numeric_cols else "None"
+    categorical_str = ", ".join(categorical_cols) if categorical_cols else "None"
+
+    info_block = (
+        f"CSV Info:\n"
+        f"- Shape: {num_rows} rows x {num_cols} cols\n"
+        f"- Columns: {columns_str}\n"
+        f"- Sample Data: {sample_str}\n"
+        f"- Data Types: {dtypes_str}\n"
+        f"- Numeric Columns: {numeric_str}\n"
+        f"- Categorical Columns: {categorical_str}\n"
+    )
+
+    prompt = f"""\
+You are a Senior Data Analyst AI and CSV analysis assistant. Your goal is to extract actionable insights, perform statistical analysis, answer complex questions, and generate professional visualizations using the provided dataset.
+
+The pandas DataFrame is pre-loaded as 'df' - use this variable.
+
+{info_block}\
+STRICT OPERATIONAL REQUIREMENTS:
+1. NEVER guess, predict, or estimate values yourself. ALWAYS generate executable Python code to calculate precise answers.
+2. USE THE EXISTING 'df' - Do not attempt to reload or recreate the dataframe.
+3. VARIABLE ASSIGNMENT IS MANDATORY: Every result, calculation, filtered subset, or visualization must be assigned to a descriptive, snake_case variable name.
+4. JSON FOR STRUCTURED DATA: For any data structures (Lists, Records, Tables, Dictionaries, etc.), return them as JSON with correct indentation so the UI can parse it.
+5. CLEANLINESS: If the analysis requires handling missing values (NaNs) or data cleaning, perform it on a copy (e.g., 'cleaned_df') before analyzing.
+
+ANALYSIS GUIDELINES:
+- Descriptive Statistics: Use .describe(), .value_counts(), and .nunique().
+- Relationships: Calculate correlations using .corr() or group data using .groupby().
+- Filtering: Always store filtered results in a specific variable (e.g., 'high_value_customers = ...').
+- Aggregation: When grouping, reset indices (.reset_index()) to keep results in a flat, readable format.
+- Outliers: Use IQR or Z-score methods when asked to find anomalies.
+
+VISUALIZATION STANDARDS:
+- Use matplotlib/seaborn only.
+- Professional quality: proper sizing, labels, titles.
+- Figure size: (14, 8) for complex charts, (12, 6) for simple charts.
+- Fonts: Clear titles (fontsize=16), labels (fontsize=14).
+- Ticks: Rotate x-labels if needed (45 degree), fontsize=12.
+- Aesthetics: Add annotations/gridlines where helpful; use colorblind-friendly palettes.
+- Final Step: Always include plt.tight_layout() and plt.show().
+- Variable Assignment: Assign figure/axis objects when needed (e.g., fig, ax = plt.subplots...).
+
+VARIABLE ASSIGNMENT RULES:
+1. Every operation must store its result in a variable.
+2. Variable names should be descriptive and snake_case.
+3. For DataFrame operations: result_df = df.operation()
+4. For statistical results: summary_stats = df.describe(include='all')
+5. For filtered data: filtered_data = df[df['column'] > value]
+6. For grouped analysis: revenue_by_region = df.groupby('region')['revenue'].sum().reset_index()
+7. For correlation matrices: correlation_matrix = df.corr(numeric_only=True)
+8. For visualizations: fig, ax = plt.subplots(...)
+
+EXAMPLES:
+
+1. Professional Chart (with variable assignment):
+fig, ax = plt.subplots(figsize=(14, 8))
+sns.barplot(x='category', y='value', data=df, palette='muted', ax=ax)
+ax.set_title('Value by Category', fontsize=16)
+ax.set_xlabel('Category', fontsize=14)
+ax.set_ylabel('Value', fontsize=14)
+ax.set_xticklabels(ax.get_xticklabels(), rotation=45)
+ax.grid(alpha=0.3)
+plt.tight_layout()
+plt.show()
+
+2. Professional Analysis (Clean, Assigned, Modular):
+# Calculate the percentage of missing values per column
+missing_data_report = df.isnull().mean() * 100
+
+# Identify top 5 performing categories by sales
+top_categories_sales = df.groupby('category')['sales'].sum().nlargest(5).reset_index()
+
+# Check for correlation between price and quantity
+price_quantity_corr = df['price'].corr(df['quantity'])
+
+3. Good vs Bad (Assignment Check):
+# GOOD (with variable assignment)
+sample_transactions = df.sample(5)[['id', 'date', 'amount']]
+transaction_stats = df['amount'].describe()
+
+# BAD (no variable assignment)
+df.sample(5)[['id', 'date', 'amount']]  # No variable assigned!
+
+Return complete, executable code that follows these rules.
+Your response should be modular, precise, and favor variable assignment over direct printing.
+
+### 2. STRICT OUTPUT FORMAT
+Return your response ONLY as a JSON object.
+
+- **If the user asks for analysis/charts:** Fill "analyze" and "visualization" arrays with Python code.
+- **If the user greets you or asks a generic question:** Use the "message" field for your response and keep the arrays empty.
+
+{_JSON_EXAMPLE}"""
+
+    return prompt.strip()
+
+
+# if __name__ == "__main__":
+#     import asyncio
+#     import sys
+#     _root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+#     sys.path.insert(0, _root)
+#     from app.services.csv_analysis_service import get_dataset_info
+
+#     url = "https://raw.githubusercontent.com/datasciencedojo/datasets/master/titanic.csv"
+#     metadata = asyncio.run(get_dataset_info(url))
+#     prompt = get_csv_system_prompt(metadata)
+#     print(prompt)
+#     print()
+#     print(f"(length: {len(prompt)} chars)")

pyproject.toml

@@ -27,6 +27,8 @@ dependencies = [
     "pillow>=10.0.0",
     "pypdfium2>=4.30.0",
     "pandas>=2.0.0",
+    "matplotlib>=3.8.0",
+    "seaborn>=0.13.0",
     "spacy>=3.7.0",
     "phonenumbers>=8.13.0",
 ]

requirements.txt

@@ -12,6 +12,8 @@ onnxruntime>=1.18.0
 pillow>=10.0.0
 pypdfium2>=4.30.0
 pandas>=2.0.0
+matplotlib>=3.8.0
+seaborn>=0.13.0
 sentence-transformers==5.6.0
 
 aiohttp>=3.9.0