Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from datetime import datetime, timedelta
3import hashlib
4import random
5import string
6from math import radians, sin, cos, sqrt, atan2
7from pydantic import BaseModel
8 
9app = FastAPI()
10 
11users = {}
12tokens = {}
13reports = {}
14report_id_counter = 1
15categories = {"pothole", "graffiti", "broken_sidewalk", "streetlight", "other"}
16 
17def generate_token():
18    return ''.join(random.choices(string.ascii_letters + string.digits, k=64))
19 
20def hash_password(password):
21    return hashlib.sha256(password.encode()).hexdigest()
22 
23def get_current_user(authorization: str = Header(...)):
24    if not authorization.startswith("Bearer "):
25        raise HTTPException(status_code=401, detail="Invalid auth header")
26    token = authorization[7:]
27    if token not in tokens:
28        raise HTTPException(status_code=401, detail="Invalid token")
29    if tokens[token]["expires"] < datetime.now():
30        del tokens[token]
31        raise HTTPException(status_code=401, detail="Token expired")
32    return tokens[token]["user_id"]
33 
34class SignupRequest(BaseModel):
35    username: str
36    password: str
37 
38class LoginRequest(BaseModel):
39    username: str
40    password: str
41 
42class ReportCreate(BaseModel):
43    lat: float
44    lon: float
45    category: str
46    description: str
47 
48class ReportUpdate(BaseModel):
49    lat: float = None
50    lon: float = None
51    category: str = None
52    description: str = None
53    status: str = None
54 
55@app.post("/signup")
56def signup(req: SignupRequest):
57    if req.username in users:
58        raise HTTPException(status_code=400, detail="Username already exists")
59    users[req.username] = {"password": hash_password(req.password)}
60    return {"message": "User created"}
61 
62@app.post("/login")
63def login(req: LoginRequest):
64    if req.username not in users or users[req.username]["password"] != hash_password(req.password):
65        raise HTTPException(status_code=401, detail="Invalid credentials")
66    token = generate_token()
67    tokens[token] = {"user_id": req.username, "expires": datetime.now() + timedelta(hours=24)}
68    return {"token": token}
69 
70@app.post("/reports")
71def create_report(report: ReportCreate, authorization: str = Header(...)):
72    user_id = get_current_user(authorization)
73    global report_id_counter
74    if report.category not in categories:
75        raise HTTPException(status_code=400, detail=f"Invalid category. Must be one of: {categories}")
76    report_id = report_id_counter
77    report_id_counter += 1
78    reports[report_id] = {
79        "id": report_id,
80        "user_id": user_id,
81        "lat": report.lat,
82        "lon": report.lon,
83        "category": report.category,
84        "description": report.description,
85        "status": "open",
86        "created_at": datetime.now().isoformat()
87    }
88    return reports[report_id]
89 
90@app.get("/reports")
91def list_reports(authorization: str = Header(...)):
92    user_id = get_current_user(authorization)
93    open_reports = [r for r in reports.values() if r["status"] == "open"]
94    # Sort by urgency: pothole > broken_sidewalk > streetlight > graffiti > other
95    urgency = {"pothole": 0, "broken_sidewalk": 1, "streetlight": 2, "graffiti": 3, "other": 4}
96    open_reports.sort(key=lambda r: urgency.get(r["category"], 5))
97    return open_reports
98 
99@app.get("/reports/{report_id}")
100def get_report(report_id: int, authorization: str = Header(...)):
101    user_id = get_current_user(authorization)
102    if report_id not in reports:
103        raise HTTPException(status_code=404, detail="Report not found")
104    return reports[report_id]
105 
106@app.patch("/reports/{report_id}")
107def update_report(report_id: int, update: ReportUpdate, authorization: str = Header(...)):
108    user_id = get_current_user(authorization)
109    if report_id not in reports:
110        raise HTTPException(status_code=404, detail="Report not found")
111    if reports[report_id]["user_id"] != user_id:
112        raise HTTPException(status_code=403, detail="Not your report")
113    report = reports[report_id]
114    if update.lat is not None:
115        report["lat"] = update.lat
116    if update.lon is not None:
117        report["lon"] = update.lon
118    if update.category is not None:
119        if update.category not in categories:
120            raise HTTPException(status_code=400, detail=f"Invalid category")
121        report["category"] = update.category
122    if update.description is not None:
123        report["description"] = update.description
124    if update.status is not None:
125        if update.status not in ("open", "closed", "in_progress"):
126            raise HTTPException(status_code=400, detail="Invalid status")
127        report["status"] = update.status
128    return report
129 
130@app.delete("/reports/{report_id}")
131def delete_report(report_id: int, authorization: str = Header(...)):
132    user_id = get_current_user(authorization)
133    if report_id not in reports:
134        raise HTTPException(status_code=404, detail="Report not found")
135    if reports[report_id]["user_id"] != user_id:
136        raise HTTPException(status_code=403, detail="Not your report")
137    del reports[report_id]
138    return {"message": "Report deleted"}
139 
140@app.get("/stats/category")
141def stats_category(authorization: str = Header(...)):
142    user_id = get_current_user(authorization)
143    counts = {}
144    for r in reports.values():
145        cat = r["category"]
146        counts[cat] = counts.get(cat, 0) + 1
147    # Ensure all categories appear
148    for c in categories:
149        if c not in counts:
150            counts[c] = 0
151    return counts

1fastapi
2uvicorn