app.py

# app.py
"""
Complaint Prioritization API (no category input).
Inputs: text, complaints, upvotes
Outputs: score + label
"""

from typing import Dict, Any
import gradio as gr

# Helper to convert 0–1 score into labels
def get_priority_label(score: float) -> str:
    if score >= 0.75:
        return "Critical"
    elif score >= 0.55:
        return "High"
    elif score >= 0.35:
        return "Medium"
    else:
        return "Low"

# Calculate weighted score (0–1)
def calculate_weighted_score(upvotes: int, complaints: int, alpha: float = 0.6, beta: float = 0.4) -> float:
    # Normalize to 0–1 scale (cap at 1)
    upvote_score = min(upvotes / 50.0, 1.0)       # assume 50 upvotes = max
    complaint_score = min(complaints / 20.0, 1.0) # assume 20 complaints = max
    weighted = (alpha * complaint_score) + (beta * upvote_score)
    return weighted

# Handle a single complaint
def handle_complaint(text: str, complaints: int, upvotes: int) -> Dict[str, Any]:
    weighted_score = calculate_weighted_score(upvotes, complaints)
    return {
        "text": text,
        "complaints": complaints,
        "upvotes": upvotes,
        "final_score": round(weighted_score, 2),
        "final_label": get_priority_label(weighted_score)
    }

# API entrypoint for single complaint (for UI)
def predict_single(text: str, complaints: int, upvotes: int):
    try:
        complaints = int(complaints)
        upvotes = int(upvotes)
    except Exception:
        return {"error": "complaints and upvotes must be integers"}
    return handle_complaint(text, complaints, upvotes)

# API entrypoint for batch complaints (JSON text input)
def predict_batch(json_string: str):
    """
    Accepts a JSON array string (list of dicts with keys: text, complaints, upvotes)
    Example:
    [
      {"text":"Pothole on main road","complaints":2,"upvotes":5},
      {"text":"Water leakage","complaints":15,"upvotes":8}
    ]
    """
    import json
    try:
        items = json.loads(json_string)
        if not isinstance(items, list):
            return {"error": "Expected a JSON array/list of complaints"}
    except Exception as e:
        return {"error": f"Invalid JSON: {str(e)}"}

    results = []
    for it in items:
        t = it.get("text", "")
        complaints = int(it.get("complaints", 0))
        upvotes = int(it.get("upvotes", 0))
        results.append(handle_complaint(t, complaints, upvotes))
    return {"results": results}

# Small UI using Gradio
with gr.Blocks() as demo:
    gr.Markdown("## Complaint Prioritization API (No Category Input)")
    with gr.Tab("Single complaint"):
        txt = gr.Textbox(label="Complaint text", value="Huge pothole on main road, damaging cars daily.")
        comp = gr.Number(label="Number of complaints", value=2, precision=0)
        upv = gr.Number(label="Number of upvotes", value=5, precision=0)
        out = gr.JSON(label="Result")
        btn = gr.Button("Predict")
        btn.click(fn=predict_single, inputs=[txt, comp, upv], outputs=out)

    with gr.Tab("Batch (JSON array)"):
        batch_in = gr.Textbox(
            label="JSON array of complaints", 
            lines=10, 
            value='[{"text":"Leak near market","complaints":15,"upvotes":8}]'
        )
        batch_out = gr.JSON(label="Batch results")
        batch_btn = gr.Button("Predict batch")
        batch_btn.click(fn=predict_batch, inputs=batch_in, outputs=batch_out)

if __name__ == "__main__":
    demo.launch()