Replace rigid LLM tools with a shared prompt editor

batch_processing/batch_method.py

@@ -10,6 +10,7 @@
 from batch_processing.batch_error_handling import handle_batch_fail
 from file_handling.data_conversion import to_long_df, save_as_csv, join_datasets
 from file_handling.data_import import import_data
+from prompt_editor.engine import load_prompt_editor_batch_results
 from settings import config, secrets_store
 from settings.user_config import get_setting
 from openai import OpenAI
@@ -179,6 +180,11 @@ def get_batch_results(batch_id: str) -> None:
         Results are automatically saved as a DataFrame with columns for
         quote, label, and confidence from the classification response.
     """
+    prompt_editor_df = load_prompt_editor_batch_results(batch_id)
+    if prompt_editor_df is not None:
+        save_as_csv(prompt_editor_df)
+        return
+
     client = get_client()
     status = get_batch_status(batch_id)
 

file_handling/data_import.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import csv
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Iterable, Optional, Sequence, Tuple, List
 
@@ -200,7 +201,7 @@ def rebuild(self, column_labels: list[str]):
 # -----------------------------
 # Public API: Import Wizard
 # -----------------------------
-def import_data(
+def import_tabular_data(
     parent: Optional[tk.Misc] = None,
     title: str = "Import Data",
     filetypes: Sequence[tuple[str, str]] = (
@@ -210,7 +211,7 @@ def import_data(
         ("TSV files", "*.tsv"),
         ("Text files", "*.txt"),
     ),
-) -> Optional[tuple[list[str], str]]:
+) -> Optional[ImportedDataset]:
     """
     Open an "Import Wizard" dialog: select a file, preview it, choose 'has headers',
     pick exactly one column via radio buttons, choose a dataset name, and return that
@@ -447,11 +448,10 @@ def _sel_hdr():
     _initial_blank_preview()  # start 5x5
 
     # -------------- Import handler --------------
-    result_values: Optional[list[str]] = None
-    result_dataset_name: Optional[str] = None
+    result_dataset: Optional[ImportedDataset] = None
 
     def _do_import():
-        nonlocal result_values, result_dataset_name
+        nonlocal result_dataset
         path = file_var.get().strip()
         if not path:
             messagebox.showerror("Error", "No file selected.", parent=dlg)
@@ -464,16 +464,15 @@ def _do_import():
             messagebox.showerror("Load error", str(e), parent=dlg)
             return
 
-        body = rows[1:] if (has_headers.get() and rows) else rows
+        header, body = normalize_tabular_rows(rows, has_headers.get())
         if not body:
             messagebox.showerror("Error", "The file appears to have no data rows.", parent=dlg)
             return
 
         col_idx = selected_col.get()
-        out: list[str] = []
-        for r in body:
-            val = r[col_idx] if col_idx < len(r) else ""
-            out.append("" if val is None else str(val))
+        if col_idx >= len(header):
+            messagebox.showerror("Error", "Selected column is out of range for the imported file.", parent=dlg)
+            return
 
         # Resolve dataset_name
         mode = dataset_name_mode.get()
@@ -493,8 +492,15 @@ def _do_import():
                 messagebox.showerror("Dataset Name error", "Please type a dataset_name in the 'Other' field.", parent=dlg)
                 return
 
-        result_values = out
-        result_dataset_name = dataset_name
+        result_dataset = ImportedDataset(
+            file_path=path,
+            dataset_name=dataset_name,
+            has_headers=has_headers.get(),
+            selected_column_index=col_idx,
+            selected_column_name=header[col_idx],
+            columns=header,
+            rows=rows_to_records(header, body),
+        )
         dlg.destroy()
 
     process_btn.configure(command=_do_import)
@@ -513,6 +519,64 @@ def _do_import():
     owner.wait_window(dlg)
     _safe_destroy(created_root)
 
-    if result_values is None or result_dataset_name is None:
+    return result_dataset
+
+
+def import_data(
+    parent: Optional[tk.Misc] = None,
+    title: str = "Import Data",
+    filetypes: Sequence[tuple[str, str]] = (
+        ("All files", "*.*"),
+        ("Excel files", "*.xlsx *.xls"),
+        ("CSV files", "*.csv"),
+        ("TSV files", "*.tsv"),
+        ("Text files", "*.txt"),
+    ),
+) -> Optional[tuple[list[str], str]]:
+    imported = import_tabular_data(parent=parent, title=title, filetypes=filetypes)
+    if imported is None:
         return None
-    return result_values, result_dataset_name
+    return imported.selected_values, imported.dataset_name
+@dataclass
+class ImportedDataset:
+    file_path: str
+    dataset_name: str
+    has_headers: bool
+    selected_column_index: int
+    selected_column_name: str
+    columns: list[str]
+    rows: list[dict[str, str]]
+
+    @property
+    def selected_values(self) -> list[str]:
+        return [row.get(self.selected_column_name, "") for row in self.rows]
+
+
+def normalize_tabular_rows(rows: list[list[str]], has_headers: bool) -> tuple[list[str], list[list[str]]]:
+    if rows:
+        if has_headers:
+            header = rows[0]
+            body = rows[1:]
+        else:
+            max_cols = max((len(r) for r in rows), default=1)
+            header = [f"Column {i+1}" for i in range(max_cols)]
+            body = rows
+    else:
+        header, body = [f"Column {i+1}" for i in range(5)], []
+
+    normalized_header = [
+        str(h) if (h is not None and str(h).strip() != "") else f"Column {i+1}"
+        for i, h in enumerate(header)
+    ]
+    return normalized_header, body
+
+
+def rows_to_records(columns: Sequence[str], rows: Sequence[Sequence[str]]) -> list[dict[str, str]]:
+    records: list[dict[str, str]] = []
+    for row in rows:
+        record: dict[str, str] = {}
+        for idx, column in enumerate(columns):
+            value = row[idx] if idx < len(row) else ""
+            record[column] = "" if value is None else str(value)
+        records.append(record)
+    return records

live_processing/keyword_extraction_live.py

@@ -1,76 +1,15 @@
-"""
-Live text classification processing using OpenAI API.
+"""Legacy keyword-extraction entry point backed by the prompt editor preset."""
 
-This module provides real-time text classification functionality that processes
-text snippets immediately using OpenAI's API, as opposed to batch processing.
-This is useful for smaller datasets or when immediate results are needed.
-"""
+from __future__ import annotations
 
 from typing import Optional
 import tkinter as tk
-from tkinter import messagebox
-from pydantic import BaseModel, ValidationError, Field, ConfigDict
-from file_handling.data_import import import_data
-from file_handling.data_conversion import save_as_csv, to_long_df
-from settings import config
-from batch_processing.batch_method import get_client
 
-# Progress UI lives in a separate module
-from ui.progress_ui import ProgressController
+from prompt_editor.presets import KEYWORD_EXTRACTION_PRESET_ID
+from prompt_editor.ui import open_prompt_editor
 
 
 def keyword_extraction_pipeline(parent: Optional[tk.Misc] = None):
-    """
-    Prompt for quotes CSVs, extract keywords from each quote,
-    show progress, then save results to CSV.
-    """
-    try:
-        client = get_client()
-    except Exception as e:
-        messagebox.showerror("API Key Required", str(e))
+    if parent is None:
         return
-
-    # Get quotes data
-    from_import = import_data(parent, "Select the quotes data")
-    if from_import is None:
-        return  # user hit Cancel
-    quotes, quotes_nickname = from_import
-
-    class KeywordExtraction(BaseModel):
-        id: int | None = None
-        quote: str
-        keywords: list[str] = Field(..., min_length=1)
-        model_config = ConfigDict()
-
-    total = len(quotes)
-
-    progress = ProgressController.open(parent=parent, total_count=total, title="Processing quotes…")
-
-    results: list[KeywordExtraction] = []
-    try:
-        for idx, q in enumerate(quotes, start=1):
-            try:
-                resp = client.responses.parse(
-                    model=config.model,
-                    input=[{"role": "system", "content": "You are an expert at structured data extraction."},
-                        {"role": "user", "content": f"Extract the keywords from this quote: {q}"}],
-                    text_format=KeywordExtraction,
-                )
-                decision = resp.output_parsed
-                row = KeywordExtraction(
-                    id=idx,
-                    quote=q,
-                    **decision.model_dump(exclude={'id', 'quote'})  # <- prevents duplicate kwargs
-                )
-                results.append(row)
-            except ValidationError as ve:
-                print(f"[VALIDATION ERROR] {str(q)[:60]}... -> {ve}")
-            except Exception as e:
-                print(f"[API ERROR] {str(q)[:60]}... -> {e}")
-            finally:
-                progress.update(idx, message=f"Processed {idx} of {total} quotes")
-    finally:
-        progress.close()
-
-    df = to_long_df(results)
-    save_as_csv(df)
+    open_prompt_editor(parent, preset_id=KEYWORD_EXTRACTION_PRESET_ID, execution_mode="live")

live_processing/multi_label_live.py

@@ -1,88 +1,15 @@
-"""
-Live text classification processing using OpenAI API.
+"""Legacy multi-label entry point backed by the prompt editor preset."""
 
-This module provides real-time text classification functionality that processes
-text snippets immediately using OpenAI's API, as opposed to batch processing.
-This is useful for smaller datasets or when immediate results are needed.
-"""
+from __future__ import annotations
 
-from typing import Optional, List
+from typing import Optional
 import tkinter as tk
-from tkinter import messagebox
-from pydantic import BaseModel, ValidationError, Field, ConfigDict
-from file_handling.data_import import import_data
-from file_handling.data_conversion import make_str_enum, save_as_csv, to_long_df
-from settings import config
-from batch_processing.batch_method import get_client
 
-# Progress UI lives in a separate module
-from ui.progress_ui import ProgressController
+from prompt_editor.presets import MULTI_LABEL_PRESET_ID
+from prompt_editor.ui import open_prompt_editor
 
 
 def multi_label_pipeline(parent: Optional[tk.Misc] = None):
-    """
-    Prompt for labels/quotes CSVs, classify each quote with 1+ labels,
-    show progress, then save results to CSV.
-    """
-    try:
-        client = get_client()
-    except Exception as e:
-        messagebox.showerror("API Key Required", str(e))
+    if parent is None:
         return
-
-    # Get labels data
-    from_import = import_data(parent, "Select the labels data")
-    if from_import is None:
-        return  # user hit Cancel
-    label_values, labels_nickname = from_import
-    labels = make_str_enum("Label", label_values)
-
-    # Get quotes data
-    from_import = import_data(parent, "Select the quotes data")
-    if from_import is None:
-        return  # user hit Cancel
-    quotes, quotes_nickname = from_import
-
-    class LabeledQuoteMulti(BaseModel):
-        id: int | None = None
-        quote: str
-        label: List[labels] = Field(..., min_length=1)
-        model_config = ConfigDict(use_enum_values=True, extra='forbid')
-
-    total = len(quotes)
-
-    progress = ProgressController.open(parent=parent, total_count=total, title="Processing quotes…")
-
-    results: list[LabeledQuoteMulti] = []
-    try:
-        for idx, q in enumerate(quotes, start=1):
-            try:
-                resp = client.responses.parse(
-                    model=config.model,
-                    input=[{
-                        "role": "user",
-                        "content": (
-                            "Label this quote with labels from the allowed set only. "
-                            f"Allowed: {', '.join(labels)}\nQuote: {q}"
-                        ),
-                    }],
-                    text_format=LabeledQuoteMulti,
-                )
-                decision = resp.output_parsed
-                row = LabeledQuoteMulti(
-                    id=idx,
-                    quote=q,
-                    **decision.model_dump(exclude={'id', 'quote'})  # <- prevents duplicate kwargs
-                )
-                results.append(row)
-            except ValidationError as ve:
-                print(f"[VALIDATION ERROR] {str(q)[:60]}... -> {ve}")
-            except Exception as e:
-                print(f"[API ERROR] {str(q)[:60]}... -> {e}")
-            finally:
-                progress.update(idx, message=f"Processed {idx} of {total} quotes")
-    finally:
-        progress.close()
-
-    df = to_long_df(results)
-    save_as_csv(df)
+    open_prompt_editor(parent, preset_id=MULTI_LABEL_PRESET_ID, execution_mode="live")