app.py

# app.py
# Complaint Prioritization System wrapped in a Gradio app (and API)

from typing import Dict, Any
import gradio as gr

# ================================
# Configuration (default values)
# ================================
DEFAULT_PRIORITY_SCORE = {
    "water": 0.9,
    "electricity": 0.9,
    "gas": 0.6,
    "road": 0.5,
    "garbage": 0.2
}

DEPARTMENT_WEIGHT = {
    "water": 1.2,
    "electricity": 1.2,
    "gas": 1.0,
    "road": 0.9,
    "garbage": 0.8
}

# 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:
    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

# Core handler
def handle_complaint(text: str, category: str, complaints: int, upvotes: int) -> Dict[str, Any]:
    if category is None:
        category = ""
    category_key = category.strip().lower()
    default_score = DEFAULT_PRIORITY_SCORE.get(category_key, 0.3)
    weighted_score = calculate_weighted_score(upvotes=upvotes, complaints=complaints)
    department_bias = DEPARTMENT_WEIGHT.get(category_key, 1.0)

    raw_final = 0.5 * default_score + 0.5 * weighted_score
    final_score = min(raw_final * department_bias, 1.0)  # cap at 1.0

    result = {
        "text": text,
        "category": category_key if category_key else "unknown",
        "default_score": round(default_score, 2),
        "weighted_score": round(weighted_score, 2),
        "department_bias": department_bias,
        "final_score": round(final_score, 2),
        "final_label": get_priority_label(final_score)
    }
    return result

# ================================
# Gradio UI & API
# ================================
with gr.Blocks() as demo:
    gr.Markdown("# Complaint Prioritization API")
    gr.Markdown("Provide complaint details. You can also call `/api/predict` for API access.")

    with gr.Row():
        txt = gr.Textbox(lines=3, label="Complaint text", value="Pothole on main road causing damage.")
        cat = gr.Dropdown(
            choices=["water", "electricity", "gas", "road", "garbage"],
            value="road",
            label="Category"
        )

    with gr.Row():
        complaints_in = gr.Number(value=1, label="Number of distinct complaints (integer)")
        upvotes_in = gr.Number(value=0, label="Number of upvotes (integer)")

    run_btn = gr.Button("Get Priority")
    output = gr.JSON(label="Result")

    run_btn.click(fn=handle_complaint,
                  inputs=[txt, cat, complaints_in, upvotes_in],
                  outputs=output)

    # Example load
    demo.load(
        lambda: handle_complaint(
            "Water pipeline leakage near market area.",
            "water",
            15,
            8
        ),
        outputs=output
    )

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