File size: 3,461 Bytes
bddaee4
 
f400493
 
 
bddaee4
 
f400493
bddaee4
6424ce3
bddaee4
 
6424ce3
 
 
 
 
 
 
 
 
bddaee4
 
 
 
 
6424ce3
 
 
bddaee4
f400493
6424ce3
 
 
f400493
 
 
 
6424ce3
 
f400493
 
bddaee4
 
 
 
 
f400493
bddaee4
 
 
 
f400493
bddaee4
 
f400493
 
bddaee4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
f400493
bddaee4
 
f400493
bddaee4
f400493
bddaee4
 
 
 
 
 
f400493
bddaee4
 
f400493
 
 
 
 
bddaee4
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
# 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()