"""
web_app.py — Complaint Auto-Routing System · Gradio Web Interface
────────────────────────────────────────────────────────────────
Run:
python app/web_app.py
# → opens at http://localhost:7860
Install Gradio:
pip install gradio
No external API keys required.
"""
import os
import sys
import glob
import json
import textwrap
# ─── Proactive Windows FFmpeg PATH Discovery ────────────────────
if sys.platform == "win32":
# Standard winget installation directory
winget_packages = os.path.expandvars(r"%LOCALAPPDATA%\Microsoft\WinGet\Packages")
if os.path.exists(winget_packages):
# Find any Gyan.FFmpeg bin folder
ffmpeg_bins = glob.glob(os.path.join(winget_packages, "Gyan.FFmpeg*", "**", "bin"), recursive=True)
if ffmpeg_bins:
ffmpeg_path = ffmpeg_bins[0]
if ffmpeg_path not in os.environ["PATH"]:
os.environ["PATH"] = ffmpeg_path + os.pathsep + os.environ["PATH"]
print(f"[Startup] Automatically resolved system FFmpeg path at: {ffmpeg_path}")
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
from inference.engine import ComplaintRoutingEngine, SAVE_DIR
# ─── Automatic Offline Model Training & Setup ───────────────────
def ensure_models_trained():
required_files = [
"embedding_engine.pkl",
"officer_classifier.pkl",
"priority_classifier.pkl",
"eta_regressor.pkl",
"label_encoders.pkl",
"vector_store.pkl"
]
all_exist = all(os.path.exists(os.path.join(SAVE_DIR, f)) for f in required_files)
if not all_exist:
print("[Startup] Missing trained models. Initiating automatic data generation and training...")
# 1. Generate data if missing
data_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "synthetic_complaints.csv")
if not os.path.exists(data_path):
print("[Startup] Generating synthetic complaints dataset...")
from data.generate_data import generate_complaints
df = generate_complaints(n_per_officer=100)
os.makedirs(os.path.dirname(data_path), exist_ok=True)
df.to_csv(data_path, index=False)
print(f"[Startup] Generated {len(df)} complaints.")
# 2. Train models
print("[Startup] Training models offline...")
from models.train import main as train_main
train_main()
print("[Startup] Model training complete.")
ensure_models_trained()
engine = ComplaintRoutingEngine().load(SAVE_DIR)
# Priority colours (HTML)
PRIORITY_BADGE = {
"High": 'High',
"Medium": 'Med',
"Low": 'Low',
}
DEPT_ICON = {
"Infrastructure & Roads": "🛣️",
"Water & Sanitation": "💧",
"Electricity & Utilities": "⚡",
"Public Safety & Security": "🛡️",
"Health & Environment": "🌿",
"Land & Property": "🏠",
"Transport & Traffic": "🚌",
"Administrative Services": "📋",
}
def build_output_html(result: dict) -> str:
o = result["officer"]
p = result["priority"]
eta = result["eta_days"]
sim = result.get("similar_complaints", [])
icon = DEPT_ICON.get(o["department"], "🏛️")
badge = PRIORITY_BADGE.get(p["level"], p["level"])
# ── Similar complaints table
sim_rows = ""
for s in sim:
snip = textwrap.shorten(s["text_snippet"].replace("…", ""), width=80)
sb = PRIORITY_BADGE.get(s["priority"], s["priority"])
sim_rows += f"""
| {s['complaint_id']} |
{snip} |
{sb} |
{s['eta_days']}d |
{s['similarity_score']:.3f} |
"""
# Check for audio/video transcription text
transcription_section = ""
if result.get("source_text"):
transcription_section = f"""
Transcribed Text
"{result['source_text']}"
"""
html = f"""
{transcription_section}
Assigned Officer
{o['name']}
{o['department']}
{o['id']} · {o['confidence']}% conf.
Priority
{badge}
{p['confidence']}% confidence
Est. Resolution
{eta}
day(s)
Similar Past Complaints (Top {len(sim)})
| ID |
Snippet |
Priority |
ETA |
Score |
{sim_rows}
Please enter complaint text.
", ""
result = engine.predict(text.strip(), top_k_similar=int(top_k))
html = build_output_html(result)
raw = json.dumps(result, indent=2, ensure_ascii=False)
return html, raw
def route_audio_complaint(audio_file, top_k: int) -> tuple:
if audio_file is None:
return "Please upload an audio file.
", ""
try:
result = engine.process(audio_path=audio_file, top_k=int(top_k))
except ImportError as e:
return f"Dependency Error: {e}
", ""
except Exception as e:
err_msg = str(e)
if "ffmpeg" in err_msg.lower() or "winerror 2" in err_msg.lower():
return (
""
"System Configuration Error: FFmpeg not detected!
"
"Whisper requires FFmpeg to process and decode audio uploads.
"
"To fix this on Windows:
"
"1. Open PowerShell as Administrator and run: winget install Gyan.FFmpeg
"
"2. Crucial: Close and restart your IDE (VS Code), terminal, or command prompt so Windows reloads the new PATH system variable.
"
"3. Restart the web app and try again."
"
",
f"Error details: {err_msg}"
)
return f"Error processing audio: {err_msg}
", f"Error details: {err_msg}"
html = build_output_html(result)
raw = json.dumps(result, indent=2, ensure_ascii=False)
return html, raw
def route_video_complaint(video_file, top_k: int) -> tuple:
if video_file is None:
return "Please upload a video file.
", ""
try:
result = engine.process(video_path=video_file, top_k=int(top_k))
except ImportError as e:
return f"Dependency Error: {e}
", ""
except Exception as e:
err_msg = str(e)
if "ffmpeg" in err_msg.lower() or "winerror 2" in err_msg.lower():
return (
""
"System Configuration Error: FFmpeg not detected!
"
"Whisper requires FFmpeg to extract audio from video uploads.
"
"To fix this on Windows:
"
"1. Open PowerShell as Administrator and run: winget install Gyan.FFmpeg
"
"2. Crucial: Close and restart your IDE (VS Code), terminal, or command prompt so Windows reloads the new PATH system variable.
"
"3. Restart the web app and try again."
"
",
f"Error details: {err_msg}"
)
return f"Error processing video: {err_msg}
", f"Error details: {err_msg}"
html = build_output_html(result)
raw = json.dumps(result, indent=2, ensure_ascii=False)
return html, raw
def create_app():
try:
import gradio as gr
except ImportError:
raise ImportError("pip install gradio # then retry")
EXAMPLE_COMPLAINTS = [
["There is a massive pothole on Brigade Road near the hospital causing accidents. URGENT! People are in immediate danger.", 5],
["My ration card application has been pending for 45 days. Not urgent, but the matter needs attention when convenient.", 5],
["Sewage water overflowing near Central Park. This is a serious problem affecting daily life. Requesting action at the earliest.", 5],
["This is a very serious problem. A live electric wire has fallen near the school. People are in immediate danger. Urgent action needed!", 5],
["Bus route 42 from East Colony has been suspended for 10 days without notice. Multiple families are affected.", 5],
["Illegal construction is happening on government land near the Railway Station. Not urgent but needs attention.", 5],
]
with gr.Blocks(
title="Complaint Auto-Routing System",
theme=gr.themes.Soft(),
css="""
* { font-family: 'Inter', 'Segoe UI', system-ui, sans-serif !important; }
.gradio-container { max-width: 900px !important; margin: 0 auto; }
footer { display: none; }
""",
) as demo:
gr.Markdown("""
# Complaint Auto-Routing System
AI/ML system that automatically routes complaints to the right officer, predicts priority and resolution time, and retrieves similar past complaints — **fully offline, no external APIs**.
""")
with gr.Tabs():
# ── Text Tab
with gr.Tab("Text Complaint"):
with gr.Row():
text_input = gr.Textbox(
label="Complaint Text",
placeholder="Describe your complaint here in English…",
lines=4,
)
top_k_text = gr.Slider(1, 10, value=5, step=1, label="Similar complaints")
text_btn = gr.Button("Route Complaint", variant="primary")
text_output = gr.HTML(label="Routing Result")
with gr.Accordion("Show raw JSON", open=False):
text_json = gr.Code(label="JSON", language="json")
gr.Markdown("
💡 **Tip:** Click on any of the examples below to instantly test the routing system!")
gr.Examples(
examples=EXAMPLE_COMPLAINTS,
inputs=[text_input, top_k_text],
)
text_btn.click(
fn=route_text_complaint,
inputs=[text_input, top_k_text],
outputs=[text_output, text_json],
)
# ── Audio Tab
with gr.Tab("Audio Complaint"):
gr.Markdown("""
Upload an audio recording of the complaint in English.
**Requires:** `pip install openai-whisper` (local model, English)
""")
audio_input = gr.Audio(type="filepath", label="Audio File (.wav, .mp3, .m4a)")
top_k_audio = gr.Slider(1, 10, value=5, step=1, label="Similar complaints")
audio_btn = gr.Button("Transcribe & Route", variant="primary")
audio_out = gr.HTML(label="Result")
audio_json = gr.Code(label="JSON", language="json")
gr.Markdown("
💡 **Tip:** Click the sample audio file below to test the transcription and routing without needing your own file!")
gr.Examples(
examples=[["data/sample_audio.mp3", 5]],
inputs=[audio_input, top_k_audio],
)
audio_btn.click(
fn=route_audio_complaint,
inputs=[audio_input, top_k_audio],
outputs=[audio_out, audio_json],
)
# ── Video Tab
with gr.Tab("Video Complaint"):
gr.Markdown("""
Upload a video of the complainant speaking.
**Requires:** `pip install openai-whisper` + `ffmpeg` installed system-wide.
""")
video_input = gr.Video(label="Video File (.mp4, .mkv, .avi)")
top_k_video = gr.Slider(1, 10, value=5, step=1, label="Similar complaints")
video_btn = gr.Button("Extract Audio & Route", variant="primary")
video_out = gr.HTML(label="Result")
video_json = gr.Code(label="JSON", language="json")
video_btn.click(
fn=route_video_complaint,
inputs=[video_input, top_k_video],
outputs=[video_out, video_json],
)
# ── About Tab
with gr.Tab("About"):
gr.Markdown("""
## Architecture
| Component | Model | Notes |
|-----------|-------|-------|
| **Embeddings** | TF-IDF + SVD (256-dim) | Offline baseline; swap for `all-MiniLM-L6-v2` (English SentenceTransformer) |
| **Officer Routing** | SVM (RBF kernel) | 8-class, probability calibrated |
| **Priority** | Random Forest | High / Medium / Low |
| **ETA Prediction** | Gradient Boosting Regressor | MAE ≈ 5–8 days |
| **Similarity Search** | Cosine over NumPy matrix | FAISS drop-in available |
| **Audio/Video** | Whisper (local) | English, fully offline |
## Officers
| ID | Name | Department |
|----|------|-----------|
| OFF001 | Rahul Sharma | Infrastructure & Roads |
| OFF002 | Priya Mehta | Water & Sanitation |
| OFF003 | Amit Verma | Electricity & Utilities |
| OFF004 | Sunita Patel | Public Safety & Security |
| OFF005 | Vijay Kumar | Health & Environment |
| OFF006 | Anjali Singh | Land & Property |
| OFF007 | Ravi Nair | Transport & Traffic |
| OFF008 | Meena Reddy | Administrative Services |
## No External APIs
All inference happens locally. Models are trained from scratch on synthetic data.
For production, replace synthetic data with real complaint records.
""")
return demo
if __name__ == "__main__":
app = create_app()
app.launch(server_name="0.0.0.0", server_port=7860, share=False)