Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from datetime import datetime, timedelta
3import secrets
4import time
5 
6app = FastAPI()
7 
8users = {}
9tickets = {}
10ticket_id_counter = 1
11tokens = {}
12 
13SIMULATED_AGENTS = ["Alice", "Bob", "Charlie", "Diana"]
14 
15def get_current_user(authorization: str = Header(None)):
16    if not authorization:
17        raise HTTPException(status_code=401, detail="Missing auth header")
18    token = authorization.replace("Bearer ", "")
19    user = tokens.get(token)
20    if not user:
21        raise HTTPException(status_code=401, detail="Invalid token")
22    return user
23 
24@app.post("/signup")
25def signup(username: str, password: str):
26    if username in users:
27        raise HTTPException(status_code=400, detail="User exists")
28    users[username] = {"password": password, "id": len(users) + 1}
29    return {"message": "User created", "user_id": users[username]["id"]}
30 
31@app.post("/login")
32def login(username: str, password: str):
33    user = users.get(username)
34    if not user or user["password"] != password:
35        raise HTTPException(status_code=401, detail="Invalid credentials")
36    token = secrets.token_hex(16)
37    tokens[token] = username
38    return {"token": token}
39 
40@app.get("/tickets")
41def get_tickets(authorization: str = Header(None)):
42    get_current_user(authorization)
43    now = time.time()
44    open_tickets = []
45    for tid, t in tickets.items():
46        if t["status"] == "open":
47            last_update_delta = now - t["last_updated"]
48            hours = int(last_update_delta // 3600)
49            minutes = int((last_update_delta % 3600) // 60)
50            if hours > 0:
51                time_since = f"{hours}h {minutes}m ago"
52            else:
53                time_since = f"{minutes}m ago"
54            open_tickets.append({
55                "id": tid,
56                "subject": t["subject"],
57                "priority": t["priority"],
58                "assigned_agent": t["assigned_agent"],
59                "time_since_last_update": time_since
60            })
61    return {"tickets": open_tickets}
62 
63@app.post("/escalate")
64def escalate_ticket(ticket_id: int, authorization: str = Header(None)):
65    get_current_user(authorization)
66    ticket = tickets.get(ticket_id)
67    if not ticket:
68        raise HTTPException(status_code=404, detail="Ticket not found")
69    if ticket["priority"] != "high":
70        raise HTTPException(status_code=400, detail="Only high priority tickets can be escalated")
71    ticket["priority"] = "critical"
72    ticket["last_updated"] = time.time()
73    return {"message": f"Ticket {ticket_id} escalated to critical", "ticket": ticket}
74 
75@app.get("/tickets/{ticket_id}")
76def get_ticket(ticket_id: int, authorization: str = Header(None)):
77    get_current_user(authorization)
78    ticket = tickets.get(ticket_id)
79    if not ticket:
80        raise HTTPException(status_code=404, detail="Ticket not found")
81    return ticket
82 
83@app.post("/tickets")
84def create_ticket(subject: str, priority: str = "medium", authorization: str = Header(None)):
85    get_current_user(authorization)
86    global ticket_id_counter
87    if priority not in ["low", "medium", "high"]:
88        priority = "medium"
89    import random
90    agent = random.choice(SIMULATED_AGENTS)
91    ticket = {
92        "id": ticket_id_counter,
93        "subject": subject,
94        "priority": priority,
95        "assigned_agent": agent,
96        "status": "open",
97        "created_at": time.time(),
98        "last_updated": time.time()
99    }
100    tickets[ticket_id_counter] = ticket
101    ticket_id_counter += 1
102    return ticket

1fastapi
2uvicorn