472 lines
16 KiB
Python
472 lines
16 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
import requests
|
|
import streamlit as st
|
|
|
|
|
|
st.set_page_config(page_title="Visione Lab", layout="wide")
|
|
|
|
|
|
def api_url(path: str) -> str:
|
|
return st.session_state.base_url.rstrip("/") + path
|
|
|
|
|
|
def headers() -> dict[str, str]:
|
|
token = st.session_state.get("token", "")
|
|
return {"Authorization": f"Bearer {token}"} if token else {}
|
|
|
|
|
|
def show_response(response: requests.Response) -> None:
|
|
st.caption(f"{response.request.method} {response.url} -> {response.status_code}")
|
|
try:
|
|
st.json(response.json())
|
|
except ValueError:
|
|
st.text(response.text)
|
|
|
|
|
|
def request(method: str, path: str, **kwargs: Any) -> requests.Response:
|
|
kwargs.setdefault("timeout", 10)
|
|
return requests.request(method, api_url(path), **kwargs)
|
|
|
|
|
|
def json_input(label: str, value: dict[str, Any], height: int = 180) -> dict[str, Any] | None:
|
|
raw = st.text_area(label, json.dumps(value, indent=2), height=height)
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError as error:
|
|
st.error(f"Invalid JSON: {error}")
|
|
return None
|
|
|
|
|
|
def pagination(prefix: str) -> dict[str, int]:
|
|
col1, col2 = st.columns(2)
|
|
with col1:
|
|
page = st.number_input(f"{prefix} page", min_value=1, value=1, step=1)
|
|
with col2:
|
|
per_page = st.number_input(
|
|
f"{prefix} per_page", min_value=1, max_value=100, value=20, step=1
|
|
)
|
|
return {"page": page, "per_page": per_page}
|
|
|
|
|
|
def response_json(response: requests.Response) -> dict[str, Any]:
|
|
try:
|
|
data = response.json()
|
|
except ValueError:
|
|
return {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
if "base_url" not in st.session_state:
|
|
st.session_state.base_url = "http://127.0.0.1:5555/api/v1"
|
|
|
|
st.title("Visione API Lab")
|
|
|
|
auth_tab, reports_tab, operator_tab, user_tab, admin_tab, public_tab, docs_tab = st.tabs(
|
|
["Auth", "Reports", "Operator", "User", "Admin", "Public", "Docs"]
|
|
)
|
|
|
|
with auth_tab:
|
|
col1, col2 = st.columns(2)
|
|
with col1:
|
|
st.subheader("Password login")
|
|
email = st.text_input("Email", "[email protected]")
|
|
password = st.text_input("Password", "ChangeMe123!", type="password")
|
|
if st.button("Login"):
|
|
response = request(
|
|
"POST",
|
|
"/auth/login/password",
|
|
json={"email": email, "password": password},
|
|
)
|
|
token = response_json(response).get("token")
|
|
if token:
|
|
st.session_state.token = token
|
|
show_response(response)
|
|
|
|
st.subheader("OTP")
|
|
otp_email = st.text_input("OTP email", "[email protected]")
|
|
if st.button("Request OTP"):
|
|
show_response(request("POST", "/auth/otp/request", json={"email": otp_email}))
|
|
otp_code = st.text_input("OTP code")
|
|
if st.button("Verify OTP"):
|
|
response = request(
|
|
"POST",
|
|
"/auth/otp/verify",
|
|
json={"email": otp_email, "code": otp_code},
|
|
)
|
|
token = response_json(response).get("token")
|
|
if token:
|
|
st.session_state.token = token
|
|
show_response(response)
|
|
|
|
with col2:
|
|
st.subheader("Register citizen")
|
|
payload = json_input(
|
|
"Register payload",
|
|
{
|
|
"first_name": "Mario",
|
|
"last_name": "Rossi",
|
|
"email": "[email protected]",
|
|
"password": "ChangeMe123!",
|
|
"fiscal_code": "RSSMRA80A01H501U",
|
|
"phone": "+391234567890",
|
|
},
|
|
)
|
|
if st.button("Register") and payload:
|
|
show_response(request("POST", "/auth/register", json=payload))
|
|
|
|
st.subheader("Me")
|
|
if st.button("GET /me"):
|
|
show_response(request("GET", "/me", headers=headers()))
|
|
|
|
st.subheader("Change password")
|
|
current_password = st.text_input(
|
|
"Current password", "ChangeMe123!", type="password"
|
|
)
|
|
new_password = st.text_input("New password", "ChangeMe123!!", type="password")
|
|
if st.button("Change password"):
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
"/auth/change-password",
|
|
json={
|
|
"current_password": current_password,
|
|
"new_password": new_password,
|
|
},
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
with reports_tab:
|
|
col1, col2 = st.columns(2)
|
|
with col1:
|
|
st.subheader("Create report")
|
|
report_payload = json_input(
|
|
"Report payload",
|
|
{
|
|
"category_id": 1,
|
|
"latitude": 41.9028,
|
|
"longitude": 12.4964,
|
|
"address": "Via Roma 1",
|
|
"description": "Lampione non funzionante",
|
|
},
|
|
)
|
|
if st.button("Create report") and report_payload:
|
|
show_response(request("POST", "/reports", json=report_payload, headers=headers()))
|
|
|
|
st.subheader("List reports")
|
|
list_col1, list_col2, list_col3 = st.columns(3)
|
|
with list_col1:
|
|
status = st.text_input("Status filter")
|
|
with list_col2:
|
|
category_id = st.text_input("Category id filter")
|
|
with list_col3:
|
|
area_id = st.text_input("Area id filter")
|
|
params = {
|
|
key: value
|
|
for key, value in {
|
|
"status": status,
|
|
"category_id": category_id,
|
|
"area_id": area_id,
|
|
}.items()
|
|
if value
|
|
}
|
|
if st.button("GET /reports"):
|
|
show_response(request("GET", "/reports", params=params, headers=headers()))
|
|
|
|
with col2:
|
|
report_id = st.number_input("Report id", min_value=1, step=1)
|
|
if st.button("Detail"):
|
|
show_response(request("GET", f"/reports/{report_id}", headers=headers()))
|
|
|
|
st.subheader("Messages")
|
|
message_payload = json_input("Message payload", {"body": "Messaggio dal lab"}, 120)
|
|
msg_col1, msg_col2 = st.columns(2)
|
|
with msg_col1:
|
|
if st.button("Add message") and message_payload:
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
f"/reports/{report_id}/messages",
|
|
json=message_payload,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
with msg_col2:
|
|
if st.button("List messages"):
|
|
show_response(
|
|
request("GET", f"/reports/{report_id}/messages", headers=headers())
|
|
)
|
|
|
|
st.subheader("Upload media")
|
|
files = st.file_uploader(
|
|
"Images or videos",
|
|
accept_multiple_files=True,
|
|
type=["jpg", "jpeg", "png", "webp", "mp4", "webm", "mov"],
|
|
)
|
|
if st.button("Upload") and files:
|
|
multipart = [
|
|
("file", (file.name, file.getvalue(), file.type or "application/octet-stream"))
|
|
for file in files
|
|
]
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
f"/reports/{report_id}/media",
|
|
headers=headers(),
|
|
files=multipart,
|
|
timeout=60,
|
|
)
|
|
)
|
|
|
|
st.subheader("Workflow")
|
|
new_status = st.selectbox(
|
|
"Status",
|
|
["UNDER_REVIEW", "ASSIGNED", "IN_PROGRESS", "RESOLVED", "REJECTED", "CLOSED"],
|
|
)
|
|
if st.button("Update status"):
|
|
show_response(
|
|
request(
|
|
"PATCH",
|
|
f"/reports/{report_id}/status",
|
|
json={"status": new_status, "note": "Updated from Visione Lab"},
|
|
headers=headers(),
|
|
)
|
|
)
|
|
if st.button("Take report"):
|
|
show_response(request("POST", f"/reports/{report_id}/take", headers=headers()))
|
|
|
|
st.subheader("Assignment and feedback")
|
|
team_id = st.number_input("Team id", min_value=1, step=1)
|
|
assign_col, feedback_col = st.columns(2)
|
|
with assign_col:
|
|
if st.button("Assign team"):
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
f"/reports/{report_id}/assignments",
|
|
json={"team_id": team_id},
|
|
headers=headers(),
|
|
)
|
|
)
|
|
with feedback_col:
|
|
feedback_payload = json_input(
|
|
"Feedback payload", {"rating": 5, "comment": "Intervento risolto"}, 120
|
|
)
|
|
if st.button("Add feedback") and feedback_payload:
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
f"/reports/{report_id}/feedback",
|
|
json=feedback_payload,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
with operator_tab:
|
|
col1, col2 = st.columns(2)
|
|
with col1:
|
|
st.subheader("Operator queues")
|
|
queue_params = pagination("Queue")
|
|
queue_endpoint = st.selectbox(
|
|
"Queue",
|
|
[
|
|
"/operator/reports/assigned",
|
|
"/operator/reports/open",
|
|
"/operator/reports/others",
|
|
],
|
|
)
|
|
if st.button("GET operator queue"):
|
|
show_response(
|
|
request("GET", queue_endpoint, params=queue_params, headers=headers())
|
|
)
|
|
|
|
st.subheader("Intervention summary")
|
|
summary_report_id = st.number_input(
|
|
"Summary report id", min_value=1, value=1, step=1
|
|
)
|
|
if st.button("GET intervention summary"):
|
|
show_response(
|
|
request(
|
|
"GET",
|
|
f"/reports/{summary_report_id}/intervention-summary",
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
with col2:
|
|
st.subheader("Flag content")
|
|
flag_report_id = st.number_input("Flag report id", min_value=1, value=1, step=1)
|
|
flag_payload = json_input(
|
|
"Flag payload",
|
|
{"reason": "abusive_content", "details": "Created from Visione Lab"},
|
|
140,
|
|
)
|
|
if st.button("Flag report") and flag_payload:
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
f"/reports/{flag_report_id}/flags",
|
|
json=flag_payload,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
message_id = st.number_input("Flag message id", min_value=1, value=1, step=1)
|
|
if st.button("Flag message") and flag_payload:
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
f"/reports/{flag_report_id}/messages/{message_id}/flags",
|
|
json=flag_payload,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
with user_tab:
|
|
col1, col2 = st.columns(2)
|
|
with col1:
|
|
st.subheader("Notifications")
|
|
notification_params = pagination("Notifications")
|
|
if st.button("List notifications"):
|
|
show_response(
|
|
request(
|
|
"GET",
|
|
"/notifications",
|
|
params=notification_params,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
notification_id = st.number_input("Notification id", min_value=1, step=1)
|
|
if st.button("Mark notification read"):
|
|
show_response(
|
|
request(
|
|
"PATCH",
|
|
f"/notifications/{notification_id}/read",
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
with col2:
|
|
st.subheader("Rewards")
|
|
rewards_params = pagination("Rewards")
|
|
if st.button("GET /me/rewards"):
|
|
show_response(
|
|
request("GET", "/me/rewards", params=rewards_params, headers=headers())
|
|
)
|
|
|
|
with admin_tab:
|
|
col1, col2 = st.columns(2)
|
|
with col1:
|
|
st.subheader("Reference data and stats")
|
|
endpoint = st.selectbox(
|
|
"Endpoint",
|
|
[
|
|
"/admin/categories",
|
|
"/admin/areas",
|
|
"/admin/teams",
|
|
"/admin/stats/overview",
|
|
"/admin/users",
|
|
],
|
|
)
|
|
if st.button("GET admin endpoint"):
|
|
show_response(request("GET", endpoint, headers=headers()))
|
|
|
|
st.subheader("Create reference data")
|
|
create_endpoint = st.selectbox(
|
|
"Create endpoint", ["/admin/categories", "/admin/areas", "/admin/teams"]
|
|
)
|
|
create_payload = json_input(
|
|
"Create payload",
|
|
{"name": "Nuova categoria", "description": "Created from Visione Lab"},
|
|
)
|
|
if st.button("POST admin endpoint") and create_payload:
|
|
show_response(
|
|
request(
|
|
"POST",
|
|
create_endpoint,
|
|
json=create_payload,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
with col2:
|
|
st.subheader("User administration")
|
|
admin_user_id = st.number_input("Admin target user id", min_value=1, step=1)
|
|
roles_payload = json_input("Roles payload", {"roles": ["operator"]}, 100)
|
|
if st.button("Update roles") and roles_payload:
|
|
show_response(
|
|
request(
|
|
"PATCH",
|
|
f"/admin/users/{admin_user_id}/roles",
|
|
json=roles_payload,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
banned = st.checkbox("Banned")
|
|
if st.button("Update ban"):
|
|
show_response(
|
|
request(
|
|
"PATCH",
|
|
f"/admin/users/{admin_user_id}/ban",
|
|
json={"banned": banned},
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
st.subheader("Moderation")
|
|
moderation_params = pagination("Moderation")
|
|
moderation_status = st.selectbox(
|
|
"Moderation status", ["", "OPEN", "RESOLVED", "DISMISSED"]
|
|
)
|
|
if moderation_status:
|
|
moderation_params["status"] = moderation_status
|
|
if st.button("List moderation flags"):
|
|
show_response(
|
|
request(
|
|
"GET",
|
|
"/admin/moderation",
|
|
params=moderation_params,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
flag_id = st.number_input("Moderation flag id", min_value=1, step=1)
|
|
moderation_payload = json_input(
|
|
"Moderation update payload",
|
|
{
|
|
"status": "RESOLVED",
|
|
"resolution_note": "Reviewed from Visione Lab",
|
|
"reject_report": False,
|
|
},
|
|
140,
|
|
)
|
|
if st.button("Update moderation flag") and moderation_payload:
|
|
show_response(
|
|
request(
|
|
"PATCH",
|
|
f"/admin/moderation/{flag_id}",
|
|
json=moderation_payload,
|
|
headers=headers(),
|
|
)
|
|
)
|
|
|
|
with public_tab:
|
|
st.subheader("Open data")
|
|
open_data_params = pagination("Open data")
|
|
if st.button("GET /open-data/reports"):
|
|
show_response(
|
|
request("GET", "/open-data/reports", params=open_data_params)
|
|
)
|
|
|
|
with docs_tab:
|
|
st.link_button("Swagger UI", api_url("/docs"))
|
|
if st.button("Fetch OpenAPI"):
|
|
show_response(request("GET", "/openapi.json"))
|
|
|
|
with st.sidebar:
|
|
st.text_input("Backend API", key="base_url")
|
|
st.text_input("Bearer token", key="token", type="password")
|
|
st.link_button("MailHog", "http://127.0.0.1:8025")
|
|
if st.button("Health check"):
|
|
show_response(request("GET", "/health", timeout=5))
|