# app.py """ Email Deadline Summarizer (Gradio) ---------------------------------- Drag & drop a CSV of emails, paste your OpenAI API key, and get deadline-driven summaries + next steps. Expected CSV columns (case-insensitive, best-effort mapping): - subject - received / date / datetime / timestamp - from / sender / sender_name / sender_email - body / content / text / snippet Outputs: - A table with: Subject, Received, Sender Name, Summary, Next Step, Explicit Deadline - A downloadable CSV of the results Run: pip install -r requirements.txt python app.py """ import io import os import json import time import traceback from typing import List, Dict, Any, Optional, Tuple import gradio as gr import pandas as pd # OpenAI SDK v1.x try: from openai import OpenAI except Exception: OpenAI = None # ----------------------------- # Utilities # ----------------------------- CANDIDATE_DATE_COLS = ["received", "date", "datetime", "timestamp"] CANDIDATE_FROM_COLS = ["from", "sender", "sender_name", "sender_email"] CANDIDATE_SUBJECT_COLS = ["subject", "title"] CANDIDATE_BODY_COLS = ["body", "content", "text", "snippet"] DEFAULT_MODEL = "gpt-4o-mini" # adjust as desired def _normalize_columns(df: pd.DataFrame) -> pd.DataFrame: """Map common column variations to a standard schema, if possible.""" lower_cols = {c.lower().strip(): c for c in df.columns} def pick(candidates: List[str]) -> Optional[str]: for c in candidates: if c in lower_cols: return lower_cols[c] return None col_subject = pick(CANDIDATE_SUBJECT_COLS) col_date = pick(CANDIDATE_DATE_COLS) col_from = pick(CANDIDATE_FROM_COLS) col_body = pick(CANDIDATE_BODY_COLS) # Create a new standardized dataframe with only the columns we need (if available) std = pd.DataFrame() if col_subject and col_subject in df: std["subject"] = df[col_subject].astype(str) else: std["subject"] = "" if col_date and col_date in df: std["received"] = df[col_date].astype(str) else: std["received"] = "" if col_from and col_from in df: std["sender_name"] = df[col_from].astype(str) else: std["sender_name"] = "" if col_body and col_body in df: std["body"] = df[col_body].astype(str) else: # If no body-like column found, try to assemble from other fields std["body"] = ( df.apply(lambda r: " ".join([str(x) for x in r.values if pd.notna(x)]), axis=1) if not df.empty else "" ).astype(str) return std def _build_prompt() -> str: """The instruction we send for each email.""" return ( "You are an executive assistant that triages emails for deadlines.\n" "For the given email (subject, received time, sender, and body), produce a JSON object with:\n" "- Subject: the email's subject line\n" "- Received: the time/date received (restate clearly)\n" "- Sender Name: the sender's name (or best guess from From field)\n" "- Summary: a concise 2–3 sentence summary of the content\n" "- Next Step: one concrete action item to meet the deadline\n" "- Explicit Deadline: a specific date/time. If none is stated, infer the *earliest prudent* deadline (today if urgent) and clearly label as inferred.\n\n" "Rules:\n" "1) If no action or deadline is implied, mark Explicit Deadline as 'None' and Next Step as 'Monitor only'.\n" "2) Keep JSON keys exactly as written above.\n" "3) Return ONLY valid minified JSON (no backticks, no extra text)." ) def _call_openai(client: "OpenAI", model: str, email: Dict[str, str]) -> Dict[str, Any]: """Call OpenAI to summarize a single email into deadline-driven JSON.""" system = _build_prompt() user = json.dumps({ "Subject": email.get("subject", ""), "Received": email.get("received", ""), "Sender Name": email.get("sender_name", ""), "Body": email.get("body", ""), }, ensure_ascii=False) resp = client.responses.create( model=model, input=[ {"role": "system", "content": system}, {"role": "user", "content": user}, ], temperature=0.2, ) # Extract text depending on SDK's shape; using .output_text for convenience text = getattr(resp, "output_text", None) if text is None: # Fallback: attempt to navigate the structure try: text = resp.output[0].content[0].text except Exception: text = "" # Parse JSON try: data = json.loads(text) if not isinstance(data, dict): raise ValueError("Model did not return a JSON object.") return { "Subject": data.get("Subject", email.get("subject", "")), "Received": data.get("Received", email.get("received", "")), "Sender Name": data.get("Sender Name", email.get("sender_name", "")), "Summary": data.get("Summary", ""), "Next Step": data.get("Next Step", ""), "Explicit Deadline": data.get("Explicit Deadline", ""), } except Exception: # Return a recoverable error row return { "Subject": email.get("subject", ""), "Received": email.get("received", ""), "Sender Name": email.get("sender_name", ""), "Summary": f"ERROR parsing model output. Raw: {text[:400]}", "Next Step": "—", "Explicit Deadline": "—", } def process_csv( csv_file: Optional[io.BytesIO], api_key: str, model: str, max_rows: int, assume_utc_dates: bool, ) -> Tuple[pd.DataFrame, str]: """Main pipeline: read CSV, normalize, LLM summarize, return DF + CSV bytes path.""" if OpenAI is None: raise RuntimeError("OpenAI SDK not installed. Please `pip install openai>=1.40`.") if not api_key: raise gr.Error("Please provide your OpenAI API key.") if csv_file is None: raise gr.Error("Please upload a CSV file.") # Load CSV try: df = pd.read_csv(csv_file) except Exception: # Try with ISO-8859-1 fallback csv_file.seek(0) df = pd.read_csv(csv_file, encoding="latin-1") if df.empty: raise gr.Error("The uploaded CSV appears to be empty.") # Normalize cols -> subject, received, sender_name, body std = _normalize_columns(df) # Trim to max_rows if max_rows > 0: std = std.head(max_rows) # Date normalization (optional best-effort) if assume_utc_dates and "received" in std.columns: # Just a simple pass-through; user can format later std["received"] = std["received"].astype(str) client = OpenAI(api_key=api_key) # Process rows rows = [] for _, r in std.iterrows(): email = { "subject": r.get("subject", ""), "received": r.get("received", ""), "sender_name": r.get("sender_name", ""), "body": r.get("body", ""), } try: out = _call_openai(client, model, email) except Exception as e: out = { "Subject": email["subject"], "Received": email["received"], "Sender Name": email["sender_name"], "Summary": f"ERROR calling model: {str(e)}", "Next Step": "—", "Explicit Deadline": "—", } rows.append(out) # gentle pacing to avoid rate spikes time.sleep(0.15) result_df = pd.DataFrame(rows, columns=[ "Subject", "Received", "Sender Name", "Summary", "Next Step", "Explicit Deadline" ]) # Save CSV to a temp in /mnt/data for download out_path = "/mnt/data/deadline_email_summaries.csv" result_df.to_csv(out_path, index=False) return result_df, out_path # ----------------------------- # Gradio UI # ----------------------------- with gr.Blocks(title="Email Deadline Summarizer") as demo: gr.Markdown( "# Email Deadline Summarizer\n" "Upload a CSV of emails, add your OpenAI API key, and get deadline-driven summaries + next steps.\n" "- ⚠️ Costs: Each row triggers a model call. Use the row limit to control spend.\n" "- 🔐 Your key is used only in this session." ) with gr.Row(): csv_in = gr.File(label="Drag & drop your CSV", file_types=[".csv"]) api_key_in = gr.Textbox( label="OpenAI API Key (starts with `sk-...`)", type="password", placeholder="Paste your key here" ) with gr.Row(): model_in = gr.Dropdown( label="Model", choices=[ "gpt-4o", "gpt-4o-mini", "gpt-4.1", "gpt-4.1-mini", "gpt-4o-mini-transcribe", "gpt-4o-realtime-preview" ], value=DEFAULT_MODEL ) max_rows_in = gr.Slider( label="Max rows to process (per run)", minimum=1, maximum=500, value=25, step=1 ) assume_utc_in = gr.Checkbox( label="Received timestamps are UTC strings (best-effort)", value=True ) run_btn = gr.Button("Summarize Emails") out_df = gr.Dataframe(label="Deadline-Driven Summaries", interactive=False) out_file = gr.File(label="Download results CSV") def _run(csv_file, api_key, model, max_rows, assume_utc): try: return process_csv(csv_file, api_key, model, int(max_rows), bool(assume_utc)) except Exception as e: tb = traceback.format_exc() raise gr.Error(f"{e}\n\n{tb}") run_btn.click( fn=_run, inputs=[csv_in, api_key_in, model_in, max_rows_in, assume_utc_in], outputs=[out_df, out_file] ) if __name__ == "__main__": demo.launch()