""" Order Triage Assistant — Gradio app Paste a raw order from social/WhatsApp channels, get back a structured, catalogue-matched order ready for entry into the inventory/fulfillment system. """ import json import os from datetime import datetime import gradio as gr import pandas as pd from extraction import extract_order from matching import load_catalogue, enrich_items, DEFAULT_CONFIDENCE_THRESHOLD DEFAULT_CATALOGUE_PATH = "sample_catalogue.csv" URGENCY_BADGE = { "urgent": "🔴 URGENT", "routine": "🟢 Routine", None: "⚪ Unspecified", } def _load_default_catalogue(): try: return load_catalogue(DEFAULT_CATALOGUE_PATH) except Exception: return None def process_order(order_text, catalogue_file, confidence_threshold, api_key, state_catalogue): """Main processing function: extract -> match -> build outputs.""" if not order_text or not order_text.strip(): return ( gr.update(value="Paste an order above to get started."), None, None, None, None, state_catalogue, ) # Resolve catalogue: uploaded file takes priority, else cached state, else sample catalogue = state_catalogue if catalogue_file is not None: try: catalogue = load_catalogue(catalogue_file.name) except Exception as e: return ( gr.update(value=f"⚠️ Catalogue error: {e}"), None, None, None, None, state_catalogue, ) if catalogue is None: catalogue = _load_default_catalogue() # Resolve API key: explicit field takes priority, else env var key = api_key.strip() if api_key and api_key.strip() else None try: result = extract_order(order_text, api_key=key) except Exception as e: return ( gr.update(value=f"⚠️ Extraction failed: {e}"), None, None, None, None, catalogue, ) enriched = enrich_items(result["items"], catalogue, threshold=int(confidence_threshold)) # Build summary header sender = result.get("sender") or "—" facility = result.get("facility") or "—" req_date = result.get("requested_date") or "—" urgency = result.get("urgency") notes = result.get("notes") or "—" n_review = sum(1 for it in enriched if it["needs_review"]) summary_md = ( f"### Order Summary\n" f"| Field | Value |\n|---|---|\n" f"| Sender | {sender} |\n" f"| Facility | {facility} |\n" f"| Requested date | {req_date} |\n" f"| Urgency | {URGENCY_BADGE.get(urgency, urgency)} |\n" f"| Notes | {notes} |\n" f"| Items extracted | {len(enriched)} |\n" f"| Needing review | {'⚠️ ' + str(n_review) if n_review else '✅ 0'} |\n" ) # Build dataframe for editable table df_rows = [] for it in enriched: df_rows.append({ "Raw text": it["raw_text"], "SKU": it["sku_code"] or "", "Matched product": it["matched_product"] or "(no match)", "Category": it["category"] or "", "Quantity": it["quantity"], "Unit": it["unit"] or "", "Confidence": it["match_confidence"], "Review?": "⚠️ Yes" if it["needs_review"] else "", }) df = pd.DataFrame(df_rows) # Build API-ready JSON payload payload = { "order_meta": { "sender": result.get("sender"), "facility": result.get("facility"), "requested_date": result.get("requested_date"), "urgency": result.get("urgency"), "notes": result.get("notes"), "processed_at": datetime.utcnow().isoformat() + "Z", }, "line_items": [ { "sku_code": it["sku_code"], "product_name": it["matched_product"], "quantity": it["quantity"], "unit": it["unit"] or it["catalogue_unit"], "match_confidence": it["match_confidence"], "needs_review": it["needs_review"], "raw_text": it["raw_text"], } for it in enriched ], } json_str = json.dumps(payload, indent=2) # Save files for download out_dir = "/tmp/order_triage_outputs" os.makedirs(out_dir, exist_ok=True) json_path = os.path.join(out_dir, "order.json") csv_path = os.path.join(out_dir, "order.csv") with open(json_path, "w") as f: f.write(json_str) df.to_csv(csv_path, index=False) return summary_md, df, json_str, json_path, csv_path, catalogue def build_app(): default_catalogue = _load_default_catalogue() with gr.Blocks(title="Order Triage Assistant", theme=gr.themes.Soft()) as demo: gr.Markdown( "# 📦 Order Triage Assistant\n" "Paste a raw order as it comes in from WhatsApp / social channels. " "The assistant extracts items, matches them to your catalogue (with SKU codes), " "flags urgency, and gives you a clean table plus an API-ready JSON payload." ) catalogue_state = gr.State(default_catalogue) with gr.Row(): with gr.Column(scale=1): order_input = gr.Textbox( label="Raw order text", placeholder="Paste the order message here...", lines=12, ) with gr.Accordion("Settings", open=False): api_key_input = gr.Textbox( label="Cohere API key (optional if set as environment variable)", type="password", placeholder="co-...", ) catalogue_upload = gr.File( label="Upload your product catalogue CSV (sku_code, product_name, category, unit)", file_types=[".csv"], ) confidence_slider = gr.Slider( label="Match confidence threshold (below this = flagged for review)", minimum=0, maximum=100, step=5, value=DEFAULT_CONFIDENCE_THRESHOLD, ) process_btn = gr.Button("Process order", variant="primary") gr.Examples( examples=[ ["Pcm syrp 4\nPcm drop 3\nPcm tabs 1 pkt\nGiven set 1\nScalvein 10 mixed 23g and 21g"], ["Marmara Hc\nVaccines requesiton tomorrow 11/06/2026 Thursday please\nBCG 20 doses\nOpv40\nPenta 30\nPCV 40 doses\nRota 20\nIPV 10 doses\nMR 20 doses\nYF 10 doses\nMen'A 10 doses\nTT 20 doses\nAD Sy 100\nBCG Sy 30\nAll droppers/diluents\nPlease consider syringes more especially 0.5 and 0.05"], ], inputs=order_input, label="Example orders", ) with gr.Column(scale=2): summary_output = gr.Markdown() table_output = gr.Dataframe( label="Extracted & matched items (editable — fix anything before exporting)", interactive=True, wrap=True, ) with gr.Accordion("API-ready JSON payload", open=False): json_output = gr.Code(label="JSON", language="json") with gr.Row(): json_file_output = gr.File(label="Download JSON") csv_file_output = gr.File(label="Download CSV") process_btn.click( fn=process_order, inputs=[order_input, catalogue_upload, confidence_slider, api_key_input, catalogue_state], outputs=[summary_output, table_output, json_output, json_file_output, csv_file_output, catalogue_state], ) return demo if __name__ == "__main__": app = build_app() app.launch()