Upload 6 files

backend/media_analysis/data_export.py

@@ -1,26 +1,37 @@
-"""Stateless CSV export for model-produced structured records.
+"""Generic, safe extraction parser and downloadable CSV assembler.
 
-Only export parsed machine-readable records, not numbers scraped from prose.
-The API returns base64 content for Spyder to save locally; no public files,
-filesystem persistence, or unauthenticated download route needed.
+A standardized transport envelope contains source-defined columns + compact row arrays.
+There are NO hardcoded environmental engineering or chart fields.
 """
 from __future__ import annotations
 import base64
 import csv
 import io
 import json
+import math
 import re
 
+MAX_ROWS = 1000
+MAX_COLUMNS = 100
+MAX_CELL_CHARS = 10000
+PROVENANCE = ["source_paper_id", "source_title", "source_doi", "source_paper_key", "source_media_label", "source_page"]
 
-# Only accepted tabular fields become output columns; other fields may be kept
-# for backward compatibility with earlier full-record responses.
-TABLE_COLUMNS = ["parameter", "unit", "minimum", "maximum", "mean", "std",
-                 "standard_range", "class_label", "classification",
-                 "percent_samples", "references"]
-CHART_COLUMNS = ["point_id", "x", "y", "unit_x", "unit_y", "estimated"]
 
+def _primitive(value):
+    return value is None or isinstance(value, (str, bool, int)) or (isinstance(value, float) and math.isfinite(value))
 
-def decode_structured_answer(answer: str) -> tuple[str, list[dict], str | None]:
+
+def _valid_columns(columns):
+    return (isinstance(columns, list) and 1 <= len(columns) <= MAX_COLUMNS
+            and all(isinstance(x, str) and x.strip() and len(x) <= 160 for x in columns)
+            and len(set(columns)) == len(columns))
+
+
+def decode_structured_answer(answer: str):
+    """Return (short answer, records, warning), for backward-compatible callers.
+
+    New output is generic columns/rows. Legacy records are supported as a fallback.
+    """
     s = (answer or "").strip()
     if s.startswith("```"):
         s = re.sub(r"^```(?:json)?\s*", "", s, flags=re.I)
@@ -28,101 +39,76 @@ def decode_structured_answer(answer: str) -> tuple[str, list[dict], str | None]:
     try:
         obj = json.loads(s)
     except (ValueError, TypeError):
-        return answer, [], "Model output was not complete JSON; CSV not generated"
+        return answer, [], "Incomplete or invalid model JSON; no CSV generated"
     if not isinstance(obj, dict):
-        return answer, [], "Model output must be a JSON object"
-    readable = str(obj.get("answer") or "")
-    if isinstance(obj.get("columns"), list) and isinstance(obj.get("rows"), list):
-        columns = obj["columns"]
-        if columns not in (TABLE_COLUMNS, CHART_COLUMNS):
-            return readable, [], "Unsupported or incorrectly ordered data columns; CSV not generated"
-        data_rows = obj["rows"]
-        if len(data_rows) > 250 or any(not isinstance(r, list) or len(r) != len(columns) for r in data_rows):
-            return readable, [], "Invalid row length or record count; CSV not generated"
-        if any(any(not isinstance(v, (str, int, float, bool, type(None))) for v in row) for row in data_rows):
-            return readable, [], "Nested values in data rows; CSV not generated"
-        return readable, [dict(zip(columns, r)) for r in data_rows], None
-    # Legacy full-record schema remains supported for existing clients.
-    rows = obj.get("records")
-    if not isinstance(rows, list):
-        return readable, [], "No structured records or compact rows; CSV not generated"
-    if len(rows) > 250 or any(not isinstance(r,dict) for r in rows):
-        return readable, [], "Invalid legacy record structure"
-    return readable, rows, None
-
-
-NUMERIC_FIELDS = {"minimum", "maximum", "mean", "std", "value", "x", "y", "percent_value"}
-
-
-def normalize_records(records: list[dict]) -> tuple[list[dict], list[str]]:
-    """Normalize unambiguous numeric fields, retain originals when text formatting differs.
-
-    Never infer a missing sign, range, classification, or null value.
+        return answer, [], "Expected a JSON object"
+    readable = obj.get("answer") if isinstance(obj.get("answer"), str) else ""
+    columns, rows = obj.get("columns"), obj.get("rows")
+    if columns is not None or rows is not None:
+        if not _valid_columns(columns):
+            return readable, [], "Invalid or duplicated column names"
+        if not isinstance(rows, list) or len(rows) > MAX_ROWS:
+            return readable, [], "Invalid row count"
+        for row in rows:
+            if (not isinstance(row, list) or len(row) != len(columns)
+                    or any(not _primitive(v) or (isinstance(v, str) and len(v) > MAX_CELL_CHARS) for v in row)):
+                return readable, [], "Mismatched columns/rows or unsupported cell type"
+        return readable, [dict(zip(columns, row)) for row in rows], None
+    legacy = obj.get("records")
+    if isinstance(legacy, list) and len(legacy) <= MAX_ROWS and all(isinstance(r, dict) for r in legacy):
+        if all(all(isinstance(k, str) and len(k) <= 160 and _primitive(v) for k,v in r.items()) for r in legacy):
+            return readable, legacy, None
+    return readable, [], "No valid columns/rows in model output"
+
+
+def normalize_records(records):
+    """Source-preserving general normalization. Never infer merged-cell content.
+
+    Keep all source-defined column values unchanged; report no table-specific fields.
+    """
+    return [dict(rec) for rec in records], []
+
+
+def _clean_csv_cell(value):
+    if value is None:
+        return ""
+    if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t", "\r", "\n"):
+        return "'" + value
+    return value
+
+
+def csv_attachment(records, sources):
+    """Export arbitrary rectangular records to base64-encoded CSV for the authenticated API.
+
+    Preserve source column order from model's `columns` list (dict insertion order).
+    For several source items, source ID columns are populated only if provided
+    explicitly by the model; never silently ascribe rows to the first source.
     """
-    result, warnings = [], []
-    for index, original in enumerate(records, 1):
-        item = dict(original)
-        for key in NUMERIC_FIELDS & item.keys():
-            value = item[key]
-            if isinstance(value, str):
-                cleaned = value.strip()
-                normalized = cleaned.replace(",", "")
-                if re.fullmatch(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)", normalized):
-                    item[key + "_printed"] = value
-                    item[key] = float(normalized) if "." in normalized else int(normalized)
-                elif cleaned:
-                    warnings.append(f"Row {index}: {key} is not an unambiguous number: {value!r}")
-        # Separate class designations that were embedded in a standard range.
-        # This is a purely textual rearrangement: no thresholds or signs are inferred.
-        rng = item.get("standard_range")
-        if isinstance(rng, str):
-            m = re.fullmatch(r"\s*(.*?)\s+(Class\s+(?:I|II|III|IV|V))\s*", rng, re.I)
-            if m and m.group(1).strip():
-                extracted_range = m.group(1).strip()
-                extracted_class = "Class " + m.group(2).split()[-1].upper()
-                existing_class = item.get("class_label")
-                if not existing_class:
-                    item["class_label"] = extracted_class
-                    item["standard_range"] = extracted_range
-                elif re.sub(r"\s+", " ", str(existing_class)).strip().lower() == extracted_class.lower():
-                    item["standard_range"] = extracted_range
-                else:
-                    warnings.append(f"Row {index}: conflicting class_label and standard_range; preserved original")
-        result.append(item)
-    return result, warnings
-
-
-def csv_attachment(records: list[dict], sources: list[dict]) -> dict | None:
     if not records:
         return None
-    reserved = {"paper_id", "paper_title", "doi", "paper_key", "media_label", "page", "extraction_method", "estimated", "source_quality"}
-    allowed = set()
-    for rec in records:
-        for key, value in rec.items():
-            if isinstance(key,str) and len(key) <= 70 and isinstance(value,(str,int,float,bool,type(None))):
-                allowed.add(key)
-    if not allowed:
-        return None
-    preferred = TABLE_COLUMNS + CHART_COLUMNS
-    fields = [x for x in preferred if x in allowed] + sorted(allowed - reserved - set(preferred)) + [
-        "paper_id", "paper_title", "doi", "paper_key", "media_label", "page", "extraction_method", "estimated", "source_quality"]
-    output = io.StringIO(newline="")
-    writer = csv.DictWriter(output, fieldnames=fields, extrasaction="ignore")
+    if len(records) > MAX_ROWS:
+        raise ValueError("Too many rows")
+    columns = list(dict.fromkeys(k for r in records for k in r.keys()))
+    if not _valid_columns(columns):
+        raise ValueError("Invalid CSV columns")
+    for r in records:
+        if any(not _primitive(v) for v in r.values()):
+            raise ValueError("CSV data must contain only scalar values")
+    extras = PROVENANCE if len(sources) == 1 else []
+    names = columns + [x for x in extras if x not in columns]
+    stream = io.StringIO(newline="")
+    writer = csv.DictWriter(stream, fieldnames=names, extrasaction="ignore")
     writer.writeheader()
-    for rec in records:
-        row = {k:v for k,v in rec.items() if k in fields and isinstance(v,(str,int,float,bool,type(None)))}
-        if len(sources) == 1:
-            src = sources[0]
-            for k, source_key in [("paper_id","paper_id"),("paper_title","paper_title"),("doi","doi"),("paper_key","paper_key"),("media_label","label"),("page","page")]:
-                row[k] = src.get(source_key) or ""
-        # Never silently claim exactness for unverified positioned text.
-        row.setdefault("extraction_method", "printed_table_transcription" if "parameter" in rec else "model_assisted")
-        row.setdefault("estimated", False if "parameter" in rec else "")
-        row.setdefault("source_quality", "model_extracted_unverified")
-        writer.writerow({k:("'"+v if isinstance(v,str) and v[:1] in ("=","+","-","@") else v)
-                         for k,v in row.items()})
-    data=output.getvalue().encode("utf-8-sig")
-    if len(data) > 1_000_000:
-        raise ValueError("Generated CSV exceeds 1MB response limit")
+    src = sources[0] if len(sources) == 1 else {}
+    provenance = dict(zip(PROVENANCE, [src.get(k) for k in ("paper_id","paper_title","doi","paper_key","label","page")]))
+    for record in records:
+        output = dict(record)
+        for name in extras:
+            if name not in output:
+                output[name] = provenance[name]
+        writer.writerow({key:_clean_csv_cell(output.get(key)) for key in names})
+    data=stream.getvalue().encode("utf-8-sig")
+    if len(data)>2_000_000:
+        raise ValueError("CSV exceeds 2MB response limit")
     return {"filename":"waterscope_media_extracted.csv", "mime_type":"text/csv", "encoding":"base64",
-            "content_base64":base64.b64encode(data).decode("ascii"), "row_count":len(records)}
+            "content_base64":base64.b64encode(data).decode("ascii"), "row_count":len(records), "columns":names}

backend/media_analysis/figure_analysis.py

@@ -153,7 +153,10 @@ def analyze_media_question(question: str, selected_papers: list[dict], *, enable
                            structured_output=export_csv)
     if export_csv:
         from .data_export import decode_structured_answer, csv_attachment, normalize_records
-        answer, records, warning = decode_structured_answer(result.get("answer", ""))
+        if result.get("response_status") == "incomplete" or result.get("incomplete_reason"):
+            answer, records, warning = "", [], "Model output incomplete; CSV was not generated"
+        else:
+            answer, records, warning = decode_structured_answer(result.get("answer", ""))
         result["answer"] = answer if not warning else (
             "Structured extraction did not complete; no downloadable CSV was generated. "
             "Please retry or narrow the request."
@@ -161,7 +164,14 @@ def analyze_media_question(question: str, selected_papers: list[dict], *, enable
         records, normalization_warnings = normalize_records(records)
         result["structured_data"] = records
         result["normalization_warnings"] = normalization_warnings
-        result["csv_attachment"] = csv_attachment(records, result.get("sources") or [])
+        sources = result.get("sources") or []
+        if len(sources) > 1 and records:
+            names = {key.lower() for rec in records for key in rec}
+            if not any(name in names for name in ("source", "source_id", "paper_id", "paper_title", "source_media_label", "media_label", "source_label")):
+                warning = warning or "Multiple sources require a source identifier column; CSV not generated"
+                records = []
+                result["structured_data"] = []
+        result["csv_attachment"] = csv_attachment(records, sources)
         if warning:
             result["export_warning"] = warning
     return {"status": "ok", "vision_inspected": evidence_mode != "json", "evidence_mode": evidence_mode, "analysis_mode": analysis_mode, **result}

backend/media_analysis/vision_client.py

@@ -116,26 +116,27 @@ def analyze_images(question: str, evidence: list[dict], *, model: str | None = N
     )
     if structured_output:
         instructions += (
-            " For downloadable data extraction return ONLY one compact valid JSON object with "
-            "keys 'answer', 'columns', and 'rows'. 'answer' MUST be one short sentence, "
-            "NOT a reproduction of the table. 'columns' MUST be exactly "
-            "['parameter','unit','minimum','maximum','mean','std','standard_range',"
-            "'class_label','classification','percent_samples','references']. "
-            "'rows' MUST be an array of positional arrays following those columns, with "
-            "one array per ACTUAL classification/data subrow in source order. "
-            "Preserve ALL requested cells, parameters, subrows, units, percentage values, "
-            "inequalities and reference text. For stats/units/reference spanning multiple "
-            "subrows, give the value only on its printed first row and null thereafter. "
-            "Use null for genuinely blank/unreadable cells; never infer a missing percentage. "
-            "Numbers must be JSON numbers with no thousands separators, preserving signs. "
-            "Class I/II/III goes in class_label, not standard_range. "
-            "Print-table transcriptions are NOT visual estimates. Do not add repeated "
-            "paper IDs, sources, confidence descriptions, provenance, CSV text, or any "
-            "other fields to rows: the server adds metadata and generates the CSV. "
-            "Do NOT include Markdown fences or a duplicate CSV in the answer. "
-            "If source is an image plot rather than a table, use columns "
-            "['point_id','x','y','unit_x','unit_y','estimated'] and data rows; "
-            "numerical visual estimates must have estimated=true."
+            " Return ONLY a JSON object with keys 'answer', 'columns', 'rows'. "
+            "'answer' is one concise sentence. 'columns' is an array of the ORIGINAL "
+            "source data column names appropriate to this figure/table; choose them "
+            "from the actual source, NEVER from a fixed schema. 'rows' is a compact "
+            "array of arrays with exactly one value per column. Include every requested "
+            "data row and every classification/continuation row in source order. "
+            "For complete extraction, keep ALL source data, units, groupings, "
+            "multilevel headers and references where present. When a group/merged cell "
+            "spans several rows, fill its displayed text in each affected data row "
+            "to make the CSV self-contained, only if group membership is unambiguous; "
+            "otherwise use null and flag uncertainty in the short answer. "
+            "Statistics that are printed only in the group's first row may remain "
+            "null in subsequent rows to distinguish them from truly repeated data. "
+            "Never invent numbers; preserve signs, inequalities, units and printed "
+            "precision. For plotted values, state in the answer they are approximate "
+            "and add an 'estimated' column when providing numerical estimates. "
+            "Use JSON numbers for numerical cells where possible; use strings for "
+            "ranges and expressions. No per-row metadata, confidence essays, markdown, "
+            "or embedded CSV. The server adds provenance and prepares the download. "
+            "Use at most 100 columns and 1000 rows; if the media is too large, "
+            "explain limitations instead of silently omitting data."
         )
     content: list[dict] = [{"type": "input_text", "text": f"Question: {question.strip()}"}]
     used = []