Fix PDF instructions lint error

agent/assistant.py

@@ -47,9 +47,10 @@
     "When generating PDF reports:\n"
     "- IMPORTANT: When asked to create a PDF report, create it immediately with the information provided\n"
     "- Generate reports even with minimal information - do not ask for clarification\n"
-    "- The 'data_json' parameter should be a JSON string with data to include\n"
+    "- Build dictionary per pdf_schema.json with title, optional insights and list of sections.\n"
+    "  Each section has type 'paragraph', 'table', or 'chart'; charts require 'chart_type' (bar, pie, line) and data.\n"
+    "- Pass this dictionary as a JSON string via the 'data_json' parameter\n"
     "- Always include the generated PDF file path in your response\n"
-    '- Example format: {"title": "Report Title", "data": "Your Data"}\n'
     '- To call the PDF tool, use: {"name": "pdf", "arguments": {"data_json": "JSON string here"}}\n\n'
     "When working with CSV files:\n"
     "- If a user has uploaded a CSV file, it will be available in the uploads directory\n"

app.py

@@ -4,14 +4,19 @@
 import os
 import uuid
 import shutil
+from pathlib import Path
+from jsonschema import validate, ValidationError
 from tools.sql_tool import run_sql
 from tools.csv_tool import summarise_csv
 from tools.pdf_tool import create_pdf
 from tools.default_paths import DATA_DIR, UPLOADS_DIR
-
-# assistant
 from agent import answer, _check_ollama_available, session_manager
 
+# Load PDF schema for validation
+PDF_SCHEMA_PATH = Path("static/pdf_schema.json")
+with open(PDF_SCHEMA_PATH, "r", encoding="utf-8") as _f:
+    PDF_SCHEMA = json.load(_f)
+
 
 def server_status() -> str:
     """
@@ -163,13 +168,16 @@ def create_pdf_wrapper(data_json, out_path=None, include_chart=True):
                     ),
                 }
         else:
-            # Use the data directly
             data = data_json
 
         try:
             # Handle basic data type conversion
             if isinstance(data, dict):
-                # Dictionary - use as is
+                if "sections" in data:
+                    try:
+                        validate(instance=data, schema=PDF_SCHEMA)
+                    except ValidationError as ve:
+                        data = {"error": "Invalid PDF schema", "details": ve.message}
                 pass
             elif isinstance(data, list):
                 # Convert list to simple dictionary with indexed keys
@@ -337,7 +345,7 @@ def respond(message, history, model_choice, csv_file, session_id=None, prev_resu
 
         # Log if we have a previous result object
         if prev_result:
-            print(f"Using previous result object for conversation continuity")
+            print("Using previous result object for conversation continuity")
         else:
             print("No previous result object available")
 

requirements.txt

@@ -85,3 +85,4 @@ tzdata==2025.2
 urllib3==2.4.0
 uvicorn==0.34.2
 websockets==12.0
+jsonschema==4.22.0

static/pdf_schema.json

@@ -0,0 +1,55 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "title": "PDF Report Schema",
+  "type": "object",
+  "description": "Structure for creating rich PDF reports with cover page, insights and various section types.",
+  "properties": {
+    "title": {
+      "type": "string",
+      "description": "Main report title"
+    },
+    "cover": {
+      "type": "object",
+      "description": "Optional cover page settings",
+      "properties": {
+        "logo_path": {
+          "type": "string",
+          "description": "Path to a logo image"
+        }
+      },
+      "additionalProperties": false
+    },
+    "insights": {
+      "type": "array",
+      "description": "List of insight paragraphs to highlight on the first page",
+      "items": {"type": "string"}
+    },
+    "sections": {
+      "type": "array",
+      "description": "Report sections such as paragraphs, tables or charts",
+      "items": {
+        "type": "object",
+        "properties": {
+          "title": {"type": "string"},
+          "type": {"type": "string", "enum": ["paragraph", "table", "chart"]},
+          "text": {"type": "string"},
+          "data": {"type": ["array", "object"]},
+          "chart_spec": {
+            "type": "object",
+            "properties": {
+              "chart_type": {"type": "string", "enum": ["bar", "pie", "line"]},
+              "labels": {"type": "array", "items": {"type": "string"}},
+              "values": {"type": "array", "items": {"type": "number"}}
+            },
+            "required": ["chart_type", "labels", "values"],
+            "additionalProperties": false
+          }
+        },
+        "required": ["title", "type"],
+        "additionalProperties": false
+      }
+    }
+  },
+  "required": ["title", "sections"],
+  "additionalProperties": false
+}

tests/test_pdf_tool.py

@@ -231,3 +231,54 @@ def test_empty_value_handling(tmp_path):
     }
     output_path = create_pdf(data, out_path=tmp_path / "empty_values.pdf")
     assert Path(output_path).exists()
+
+
+def _count_images(pdf_path: Path) -> int:
+    with open(pdf_path, "rb") as f:
+        return f.read().count(b"/Subtype /Image")
+
+
+def test_pdf_with_sections_and_charts(tmp_path):
+    """Generate a PDF using the new schema with multiple chart types."""
+    data = {
+        "title": "Complex Report",
+        "insights": ["Insight one", "Another insight"],
+        "sections": [
+            {
+                "title": "Numbers",
+                "type": "table",
+                "data": [{"a": 1, "b": 2}, {"a": 3, "b": 4}],
+            },
+            {
+                "title": "Bar Chart",
+                "type": "chart",
+                "chart_spec": {
+                    "chart_type": "bar",
+                    "labels": ["A", "B"],
+                    "values": [1, 2],
+                },
+            },
+            {
+                "title": "Pie Chart",
+                "type": "chart",
+                "chart_spec": {
+                    "chart_type": "pie",
+                    "labels": ["X", "Y"],
+                    "values": [3, 7],
+                },
+            },
+            {
+                "title": "Line Chart",
+                "type": "chart",
+                "chart_spec": {
+                    "chart_type": "line",
+                    "labels": [1, 2, 3],
+                    "values": [1, 4, 9],
+                },
+            },
+        ],
+    }
+
+    pdf_path = Path(create_pdf(data, out_path=tmp_path / "complex.pdf"))
+    assert pdf_path.exists()
+    assert _count_images(pdf_path) >= 4

tools/pdf_tool.py

@@ -6,7 +6,7 @@
 
 import datetime as _dt
 from pathlib import Path
-from typing import Dict, List
+from typing import Dict, List, Any
 import tempfile
 import os
 
@@ -112,6 +112,49 @@ def _build_table(data: Dict[str, object]) -> Table:
     return Table(rows, style=TableStyle(styles))
 
 
+def _build_table_from_list(data: List[Any]) -> Table:
+    """Create a table from a list of dicts or rows."""
+    if not data:
+        return Table([["No data"]])
+
+    if all(isinstance(row, dict) for row in data):
+        headers = sorted({k for row in data for k in row.keys()})
+        rows = [headers]
+        for row in data:
+            rows.append([row.get(h, "") for h in headers])
+    else:
+        rows = data
+
+    styles = [
+        ("GRID", (0, 0), (-1, -1), 0.25, colors.grey),
+        ("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey),
+    ]
+    return Table(rows, style=TableStyle(styles))
+
+
+def create_chart(chart_spec: Dict[str, Any]) -> str:
+    """Generate a chart image from a specification and return PNG path."""
+    chart_type = chart_spec.get("chart_type", "bar")
+    labels = chart_spec.get("labels", [])
+    values = chart_spec.get("values", [])
+    fig, ax = _plt.subplots(figsize=(6, 3.5))
+
+    if chart_type == "bar":
+        ax.bar(labels, values, color="#143d8d")
+    elif chart_type == "pie":
+        ax.pie(values, labels=labels, autopct="%1.1f%%")
+    elif chart_type == "line":
+        ax.plot(labels, values, marker="o", color="#143d8d")
+    else:
+        raise ValueError(f"Unsupported chart type: {chart_type}")
+
+    fig.tight_layout()
+    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
+    fig.savefig(tmp.name, bbox_inches="tight")
+    _plt.close(fig)
+    return tmp.name
+
+
 def create_pdf(
     data: Dict[str, object],
     out_path: str | None = None,
@@ -152,55 +195,101 @@ def create_pdf(
     styles = getSampleStyleSheet()
     story: List[object] = []
 
-    # logo (if file exists)
     logo_path = Path(__file__).resolve().parent.parent / "assets" / "logo.png"
-    if logo_path.exists():
-        story.append(
-            Image(
-                str(logo_path),
-                width=80,
-                height=80,
-                kind="proportional",
-                hAlign="CENTER",
+
+    # Determine if new schema is used
+    has_sections = isinstance(data, dict) and "sections" in data
+
+    tmp_pngs: List[str] = []
+
+    if has_sections:
+        if logo_path.exists():
+            story.append(
+                Image(
+                    str(logo_path),
+                    width=80,
+                    height=80,
+                    kind="proportional",
+                    hAlign="CENTER",
+                )
+            )
+            story.append(Spacer(1, 12))
+
+        story.append(Paragraph(data.get("title", "Data Assistant Report"), styles["Title"]))
+        timestamp_text = f"Generated: {_dt.datetime.now().isoformat(timespec='seconds')}"
+        story.append(Paragraph(timestamp_text, styles["Normal"]))
+        story.append(Spacer(1, 12))
+
+        for ins in data.get("insights", []):
+            story.append(Paragraph(ins, styles["Normal"]))
+            story.append(Spacer(1, 6))
+
+        for i, section in enumerate(data.get("sections", []), start=1):
+            story.append(Spacer(1, 12))
+            story.append(Paragraph(section.get("title", f"Section {i}"), styles["Heading2"]))
+            stype = section.get("type")
+            if stype == "paragraph":
+                story.append(Paragraph(section.get("text", ""), styles["Normal"]))
+            elif stype == "table":
+                table_data = section.get("data", {})
+                if isinstance(table_data, dict):
+                    table = _build_table(table_data)
+                else:
+                    table = _build_table_from_list(table_data)
+                story.append(table)
+            elif stype == "chart":
+                chart_spec = section.get("chart_spec", {})
+                png = create_chart(chart_spec)
+                story.append(Image(png, width=400, height=250))
+                tmp_pngs.append(png)
+            else:
+                story.append(Paragraph("Unsupported section type", styles["Italic"]))
+
+        doc.build(story)
+    else:
+        # Fallback to legacy behaviour for plain dicts
+        if logo_path.exists():
+            story.append(
+                Image(
+                    str(logo_path),
+                    width=80,
+                    height=80,
+                    kind="proportional",
+                    hAlign="CENTER",
+                )
             )
-        )
-        story.append(Spacer(1, 12))  # more air below logo
-
-    story.append(Paragraph("Data Assistant Report", styles["Title"]))
-    timestamp_text = f"Generated: {_dt.datetime.now().isoformat(timespec='seconds')}"
-    story.append(Paragraph(timestamp_text, styles["Normal"]))
-    story.append(Spacer(1, 12))
-    story.append(_build_table(data))
-
-    # optional bar chart
-    tmp_png = None
-    if include_chart:
-        numeric_items = {k: v for k, v in data.items() if isinstance(v, (int, float))}
-        if len(numeric_items) >= 3:
-            labels, values = zip(*numeric_items.items())
-            fig, ax = _plt.subplots(figsize=(6, 3.5))
-
-            # filter zero values
-            filtered = [(k, v) for k, v in numeric_items.items() if v]
-            labels, values = zip(*filtered) if filtered else (labels, values)
-
-            ax.bar(range(len(labels)), values, color="#143d8d")  # NeurArk blue
-            ax.set_ylabel("Value (€)")
-            ax.set_xticks(range(len(labels)))
-            ax.set_xticklabels(labels, rotation=45, ha="right")
-            fig.tight_layout()
-            tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
-            fig.savefig(tmp.name, bbox_inches="tight")
-            _plt.close(fig)
-            story.append(Spacer(1, 24))
-            story.append(Image(tmp.name, width=400, height=250))
-            tmp_png = tmp.name
-
-    doc.build(story)
-
-    # clean temporary PNG
-    if tmp_png and os.path.exists(tmp_png):
-        os.unlink(tmp_png)
+            story.append(Spacer(1, 12))
+
+        story.append(Paragraph("Data Assistant Report", styles["Title"]))
+        timestamp_text = f"Generated: {_dt.datetime.now().isoformat(timespec='seconds')}"
+        story.append(Paragraph(timestamp_text, styles["Normal"]))
+        story.append(Spacer(1, 12))
+        story.append(_build_table(data))
+
+        if include_chart:
+            numeric_items = {k: v for k, v in data.items() if isinstance(v, (int, float))}
+            if len(numeric_items) >= 3:
+                labels, values = zip(*numeric_items.items())
+                fig, ax = _plt.subplots(figsize=(6, 3.5))
+                filtered = [(k, v) for k, v in numeric_items.items() if v]
+                labels, values = zip(*filtered) if filtered else (labels, values)
+                ax.bar(range(len(labels)), values, color="#143d8d")
+                ax.set_ylabel("Value (€)")
+                ax.set_xticks(range(len(labels)))
+                ax.set_xticklabels(labels, rotation=45, ha="right")
+                fig.tight_layout()
+                tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".png")
+                fig.savefig(tmp.name, bbox_inches="tight")
+                _plt.close(fig)
+                story.append(Spacer(1, 24))
+                story.append(Image(tmp.name, width=400, height=250))
+                tmp_pngs.append(tmp.name)
+
+        doc.build(story)
+
+    for png in tmp_pngs:
+        if os.path.exists(png):
+            os.unlink(png)
     return str(out_path.resolve())