Open the app to reset password
Preparing your secure reset link...
If the app does not open automatically, tap \"Open Ain El Aql App\".
"""
return HTMLResponse(content=html)
@app.post("/auth/forgot-password")
def auth_forgot_password(
request: Request,
payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
identifier = _normalize_spaces(
str(
payload.get("identifier")
or payload.get("email")
or payload.get("phone")
or payload.get("phone_number")
or payload.get("username")
or payload.get("nid")
or payload.get("national_id")
or ""
)
)
if not identifier:
raise HTTPException(status_code=400, detail="identifier is required.")
profile = _supabase_find_profile_by_identifier(identifier)
resolved_email = _normalize_spaces(str((profile or {}).get("email") or "")).lower()
resolved_phone = _normalize_spaces(str((profile or {}).get("phone_number") or ""))
if not resolved_email and "@" in identifier:
resolved_email = _normalize_spaces(identifier).lower()
has_real_email = bool(resolved_email and "@" in resolved_email and not _is_internal_login_email(resolved_email))
# Do not leak whether account exists.
if not has_real_email and not resolved_phone:
return {
"status": "ok",
"message": "If the account exists, reset instructions were sent.",
}
if has_real_email:
request_payload: Dict[str, Any] = {"email": resolved_email}
payload_redirect_to = _normalize_spaces(str(payload.get("redirect_to") or ""))
redirect_to = payload_redirect_to or _resolve_default_password_reset_redirect_url(request)
request_host = _normalize_spaces(urlparse(_resolve_public_base_url(request)).hostname or "").lower()
allowed_hosts = [request_host] if request_host else None
if not _is_valid_redirect_url(redirect_to, allowed_http_hosts=allowed_hosts):
raise HTTPException(status_code=400, detail="redirect_to must be an allow-listed URL or deep-link.")
request_payload["redirect_to"] = redirect_to
try:
response = requests.post(
f"{SUPABASE_URL}/auth/v1/recover",
json=request_payload,
headers={
"apikey": SUPABASE_ANON_KEY,
"Content-Type": "application/json",
},
timeout=SUPABASE_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
raise HTTPException(status_code=503, detail=f"Supabase forgot-password failed: {exc}") from exc
if response.status_code not in {200, 201, 204}:
error_data = response.json() if response.text else {}
error_msg = ""
if isinstance(error_data, dict):
error_msg = str(error_data.get("message") or error_data.get("error_description") or "").strip()
if not error_msg:
error_msg = response.text[:220]
LOGGER.warning("Forgot-password provider response %s: %s", response.status_code, error_msg)
else:
try:
_request_supabase_otp(
channel="phone",
identifier=resolved_phone,
create_user=False,
redirect_to=None,
purpose="password_reset",
)
except HTTPException as exc:
LOGGER.warning("Forgot-password phone OTP response %s: %s", exc.status_code, exc.detail)
return {
"status": "ok",
"message": "If the account exists, reset instructions were sent.",
}
@app.post("/auth/reset-password/resolve-token")
def auth_reset_password_resolve_token(
payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
direct_access_token = _normalize_spaces(str(payload.get("access_token") or payload.get("token") or ""))
if direct_access_token:
return {
"status": "ok",
"session": {
"access_token": direct_access_token,
"refresh_token": _normalize_spaces(str(payload.get("refresh_token") or "")),
"expires_in": payload.get("expires_in"),
},
}
verify_type = _normalize_spaces(str(payload.get("type") or "recovery")).lower() or "recovery"
resolved_email = _normalize_spaces(str(payload.get("email") or "")).lower()
token_candidates: List[str] = []
token_hash = _normalize_spaces(str(payload.get("token_hash") or payload.get("tokenHash") or ""))
if token_hash:
token_candidates.append(token_hash)
code = _normalize_spaces(str(payload.get("code") or ""))
if code and code not in token_candidates:
token_candidates.append(code)
if not token_candidates:
raise HTTPException(status_code=400, detail="token_hash or code is required when access_token is missing.")
last_error = ""
for candidate in token_candidates:
verify_payload: Dict[str, Any] = {
"type": verify_type,
"token_hash": candidate,
}
if resolved_email:
verify_payload["email"] = resolved_email
try:
response = requests.post(
f"{SUPABASE_URL}/auth/v1/verify",
json=verify_payload,
headers={
"apikey": SUPABASE_ANON_KEY,
"Content-Type": "application/json",
},
timeout=SUPABASE_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
raise HTTPException(status_code=503, detail=f"Supabase verify failed: {exc}") from exc
if response.status_code == 200:
try:
data = response.json() if response.text else {}
except ValueError:
data = {}
if not isinstance(data, dict):
last_error = "Supabase verify returned an unexpected payload."
continue
session_payload = _extract_supabase_session_tokens(data)
if session_payload.get("access_token"):
return {
"status": "ok",
"session": session_payload,
}
last_error = "Supabase verify did not include an access token."
continue
try:
error_data = response.json() if response.text else {}
except ValueError:
error_data = {}
error_msg = ""
if isinstance(error_data, dict):
error_msg = str(error_data.get("message") or error_data.get("error_description") or "").strip()
if not error_msg:
error_msg = response.text[:220]
last_error = error_msg or f"Supabase verify failed with status {response.status_code}."
raise HTTPException(status_code=401, detail=f"Could not resolve reset token: {last_error or 'Invalid or expired link.'}")
@app.post("/auth/reset-password")
def auth_reset_password(
payload: Dict[str, Any] = Body(...),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
access_token = _normalize_spaces(str(payload.get("access_token") or payload.get("token") or ""))
new_password = str(payload.get("new_password") or payload.get("password") or "").strip()
if not access_token:
raise HTTPException(status_code=400, detail="access_token is required.")
if len(new_password) < 6:
raise HTTPException(status_code=400, detail="new_password must be at least 6 characters.")
try:
response = requests.put(
f"{SUPABASE_URL}/auth/v1/user",
json={"password": new_password},
headers={
"apikey": SUPABASE_ANON_KEY,
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
},
timeout=SUPABASE_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
raise HTTPException(status_code=503, detail=f"Supabase reset-password failed: {exc}") from exc
if response.status_code != 200:
error_data = response.json() if response.text else {}
error_msg = ""
if isinstance(error_data, dict):
error_msg = str(error_data.get("message") or error_data.get("error_description") or "").strip()
if not error_msg:
error_msg = response.text[:220]
raise HTTPException(status_code=response.status_code, detail=f"Reset-password failed: {error_msg}")
return {
"status": "ok",
"message": "Password reset successful.",
}
@app.post("/auth/otp/request")
def auth_otp_request(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
purpose = _normalize_spaces(str(payload.get("purpose") or "register")).lower() or "register"
raw_channel = payload.get("channel") or payload.get("otp_channel")
if raw_channel is None:
has_email = bool(_normalize_spaces(str(payload.get("email") or "")))
has_phone = bool(_normalize_spaces(str(payload.get("phone") or payload.get("phone_number") or "")))
raw_channel = "email" if has_email else ("phone" if has_phone else "email")
channel = _normalize_otp_channel(str(raw_channel))
request_user: Optional[Dict[str, Any]] = None
requester_id: Optional[str] = None
profile: Optional[Dict[str, Any]] = None
if purpose in {"vehicle_verify", "profile_update"}:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required for this OTP purpose.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
profile = _supabase_get_profile_by_user_id(requester_id)
identifier = _resolve_otp_identifier(channel=channel, payload=payload, profile=profile)
raw_create_user = payload.get("create_user")
if isinstance(raw_create_user, bool):
create_user = raw_create_user
elif raw_create_user is None:
create_user = purpose in {"register", "signup"}
else:
create_user = str(raw_create_user).strip().lower() in {"1", "true", "yes", "on"}
# Registration OTP must allow creating missing auth users, even if client sends create_user=false.
if purpose in {"register", "signup"}:
create_user = True
raw_redirect_to = _normalize_spaces(str(payload.get("redirect_to") or ""))
redirect_to: Optional[str] = raw_redirect_to or None
if channel == "email" and not redirect_to:
default_redirect = _normalize_spaces(PASSWORD_RESET_REDIRECT_URL)
if default_redirect:
redirect_to = default_redirect
_request_supabase_otp(
channel=channel,
identifier=identifier,
create_user=create_user,
redirect_to=redirect_to,
purpose=purpose,
)
return {
"status": "ok",
"purpose": purpose,
"channel": channel,
"identifier": identifier,
"message": "OTP was sent successfully.",
}
@app.post("/auth/otp/verify")
def auth_otp_verify(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
purpose = _normalize_spaces(str(payload.get("purpose") or "register")).lower() or "register"
channel = _normalize_otp_channel(str(payload.get("channel") or payload.get("otp_channel") or "email"))
request_user: Optional[Dict[str, Any]] = None
token = _extract_bearer_token(authorization)
if token:
barrier_user = _verify_barrier_token(token)
if barrier_user is not None:
request_user = barrier_user
else:
request_user = _verify_supabase_user_token(token)
if purpose in {"vehicle_verify", "profile_update"} and request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required for this OTP purpose.")
requester_profile = None
if request_user is not None:
requester_id = str(request_user.get("id") or "").strip()
if requester_id:
requester_profile = _supabase_get_profile_by_user_id(requester_id)
identifier = _resolve_otp_identifier(
channel=channel,
payload=payload,
profile=requester_profile,
)
verify_payload = _verify_supabase_otp(
channel=channel,
identifier=identifier,
token=str(payload.get("otp_token") or payload.get("token") or ""),
verify_type=str(payload.get("type") or payload.get("verify_type") or "").strip() or None,
)
user_payload = verify_payload.get("user") if isinstance(verify_payload.get("user"), dict) else {}
return {
"status": "ok",
"purpose": purpose,
"channel": channel,
"user": {
"id": user_payload.get("id"),
"email": user_payload.get("email"),
"phone": user_payload.get("phone"),
},
"session": {
"access_token": verify_payload.get("access_token"),
"refresh_token": verify_payload.get("refresh_token"),
"expires_in": verify_payload.get("expires_in"),
"token_type": verify_payload.get("token_type"),
},
"message": "OTP verified successfully.",
}
@app.put("/profile/update")
def profile_update(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
current_profile = _supabase_get_profile_by_user_id(requester_id) or {}
current_first_name, current_last_name, _ = _resolve_profile_name_fields(
current_profile if isinstance(current_profile, dict) else None,
fallback_full_name=(current_profile or {}).get("full_name") if isinstance(current_profile, dict) else None,
)
name_fields_provided = any(
key in payload
for key in ["first_name", "firstName", "last_name", "lastName", "full_name", "name"]
)
requested_first_name, requested_last_name, _ = _resolve_profile_name_fields(payload)
email = _normalize_spaces(str(payload.get("email") or "")).lower()
phone_number = _normalize_spaces(str(payload.get("phone") or payload.get("phone_number") or ""))
profile_patch: Dict[str, Any] = {}
if name_fields_provided:
next_first_name = requested_first_name or current_first_name
next_last_name = requested_last_name or current_last_name
next_full_name = _compose_full_name(next_first_name, next_last_name)
if not next_full_name:
raise HTTPException(status_code=400, detail="Name update requires at least first_name or full_name.")
profile_patch["first_name"] = next_first_name
profile_patch["last_name"] = next_last_name
profile_patch["full_name"] = next_full_name
if email:
if "@" not in email:
raise HTTPException(status_code=400, detail="email format is invalid.")
other_with_email = _supabase_get_profile_by_field("email", email, case_insensitive=True)
if isinstance(other_with_email, dict) and str(other_with_email.get("id") or "") != requester_id:
raise HTTPException(status_code=409, detail="email is already in use.")
profile_patch["email"] = email
if phone_number:
if not PHONE_PATTERN.fullmatch(phone_number):
raise HTTPException(status_code=400, detail="phone must be 7-15 digits and may start with +.")
other_with_phone = _supabase_get_profile_by_field("phone_number", phone_number)
if isinstance(other_with_phone, dict) and str(other_with_phone.get("id") or "") != requester_id:
raise HTTPException(status_code=409, detail="phone is already in use.")
profile_patch["phone_number"] = phone_number
if not profile_patch:
raise HTTPException(status_code=400, detail="No profile fields were provided.")
response = requests.patch(
f"{SUPABASE_URL}/rest/v1/profiles",
params={
"id": f"eq.{requester_id}",
"select": PROFILE_SELECT_FIELDS,
},
json=profile_patch,
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
content_type="application/json",
prefer="return=representation",
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code not in {200, 204}:
raise HTTPException(status_code=400, detail=f"Failed to update profile: {_missing_schema_hint(response.text)}")
updated_rows = response.json() if response.text else []
updated_profile = (
updated_rows[0]
if isinstance(updated_rows, list) and updated_rows and isinstance(updated_rows[0], dict)
else (_supabase_get_profile_by_user_id(requester_id) or current_profile)
)
auth_patch: Dict[str, Any] = {}
user_metadata_patch: Dict[str, Any] = {}
if "email" in profile_patch:
auth_patch["email"] = profile_patch["email"]
if "first_name" in profile_patch:
user_metadata_patch["first_name"] = profile_patch["first_name"]
if "last_name" in profile_patch:
user_metadata_patch["last_name"] = profile_patch["last_name"]
if "full_name" in profile_patch:
user_metadata_patch["full_name"] = profile_patch["full_name"]
if "phone_number" in profile_patch:
user_metadata_patch["phone_number"] = profile_patch["phone_number"]
user_metadata_patch["phone"] = profile_patch["phone_number"]
if user_metadata_patch:
auth_patch["user_metadata"] = user_metadata_patch
if auth_patch:
try:
auth_response = requests.put(
f"{SUPABASE_URL}/auth/v1/admin/users/{requester_id}",
json=auth_patch,
headers={
"apikey": SUPABASE_SERVICE_ROLE_KEY,
"Authorization": f"Bearer {SUPABASE_SERVICE_ROLE_KEY}",
"Content-Type": "application/json",
},
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if auth_response.status_code not in {200, 201}:
LOGGER.warning("Auth metadata update failed for %s: %s", requester_id, auth_response.text[:220])
except requests.RequestException as exc:
LOGGER.warning("Auth metadata update request failed for %s: %s", requester_id, exc)
return {
"status": "ok",
"profile": updated_profile,
"message": "Profile updated successfully.",
}
@app.get("/users")
def get_all_users(
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_role = _resolve_requester_role(request_user)
if not _is_staff_role(requester_role):
raise HTTPException(status_code=403, detail="Admin/Security role required to view users.")
try:
response = requests.get(
f"{SUPABASE_URL}/rest/v1/profiles",
params={"select": PROFILE_SELECT_FIELDS},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
raise HTTPException(status_code=502, detail="Failed to fetch users from database.")
users = response.json()
if isinstance(users, dict) and "error" in users:
users = []
return {"status": "ok", "users": users}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/vehicles/add")
def vehicles_add(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
vehicle_payload = _extract_vehicle_payload(payload, require_plate=True)
created_vehicle = _create_vehicle_for_owner(owner_id=requester_id, vehicle_payload=vehicle_payload)
if not isinstance(created_vehicle, dict):
raise HTTPException(status_code=502, detail="Vehicle was not created.")
return {
"status": "ok",
"vehicle": created_vehicle,
}
@app.get("/vehicles/my")
def vehicles_my(
include_inactive: bool = False,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
rows = _list_vehicles_for_owner(requester_id)
if not include_inactive:
rows = [row for row in rows if bool(row.get("is_active", True))]
return {
"status": "ok",
"count": len(rows),
"vehicles": rows,
}
@app.put("/vehicles/{vehicle_id}")
def vehicles_update(
vehicle_id: str,
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
update_payload: Dict[str, Any] = {}
try:
update_payload.update(_extract_vehicle_payload(payload, require_plate=False))
except HTTPException as exc:
if str(exc.detail) != "No vehicle fields were provided.":
raise
if "is_active" in payload:
raw_is_active = payload.get("is_active")
if isinstance(raw_is_active, bool):
update_payload["is_active"] = raw_is_active
elif raw_is_active is not None:
update_payload["is_active"] = str(raw_is_active).strip().lower() in {"1", "true", "yes", "on"}
if not update_payload:
raise HTTPException(status_code=400, detail="No updatable vehicle fields were provided.")
updated_vehicle = _update_vehicle_for_owner(requester_id, vehicle_id, update_payload)
if not isinstance(updated_vehicle, dict):
raise HTTPException(status_code=404, detail="Vehicle was not found.")
return {
"status": "ok",
"vehicle": updated_vehicle,
}
@app.delete("/vehicles/{vehicle_id}")
def vehicles_delete(
vehicle_id: str,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
vehicle_record = _fetch_vehicle_for_owner(requester_id, vehicle_id)
if not isinstance(vehicle_record, dict):
raise HTTPException(status_code=404, detail="Vehicle was not found.")
deleted = _delete_vehicle_for_owner(requester_id, vehicle_id)
if not deleted:
raise HTTPException(status_code=404, detail="Vehicle was not found.")
return {
"status": "ok",
"deleted_vehicle_id": vehicle_id,
}
@app.post("/vehicles/verify")
def vehicles_verify(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
profile = _supabase_get_profile_by_user_id(requester_id)
if not isinstance(profile, dict):
raise HTTPException(status_code=404, detail="Profile was not found.")
vehicle_id = _normalize_spaces(str(payload.get("vehicle_id") or ""))
if vehicle_id:
vehicle = _fetch_vehicle_for_owner(requester_id, vehicle_id)
else:
vehicle_plate_payload = _extract_vehicle_payload(payload, require_plate=True)
vehicle = _fetch_vehicle_by_plate_for_owner(
requester_id,
vehicle_plate_payload["plate_letters_ar"],
vehicle_plate_payload["plate_numbers_ar"],
)
if not isinstance(vehicle, dict):
raise HTTPException(status_code=404, detail="Vehicle was not found.")
channel = _normalize_otp_channel(str(payload.get("channel") or payload.get("otp_channel") or "email"))
identifier = _resolve_otp_identifier(channel=channel, payload=payload, profile=profile)
_verify_supabase_otp(
channel=channel,
identifier=identifier,
token=str(payload.get("otp_token") or payload.get("token") or ""),
verify_type=str(payload.get("type") or payload.get("verify_type") or "").strip() or None,
)
resolved_vehicle_id = str(vehicle.get("id") or "").strip()
updated_vehicle = _update_vehicle_for_owner(
requester_id,
resolved_vehicle_id,
{
"is_verified": True,
"verified_at": datetime.now(timezone.utc).isoformat(),
"verified_by": requester_id,
},
)
return {
"status": "ok",
"channel": channel,
"vehicle": updated_vehicle or vehicle,
"message": "Vehicle ownership verified.",
}
@app.post("/notifications/fcm-token")
def notifications_register_fcm_token(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if not _supabase_configured():
raise HTTPException(status_code=503, detail="Supabase is not configured.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
fcm_token = _normalize_spaces(str(payload.get("fcm_token") or payload.get("token") or ""))
if not fcm_token:
raise HTTPException(status_code=400, detail="fcm_token is required.")
platform = _normalize_spaces(str(payload.get("platform") or "")) or None
device_id = _normalize_spaces(str(payload.get("device_id") or "")) or None
row = _upsert_notification_device_token(
user_id=requester_id,
fcm_token=fcm_token,
platform=platform,
device_id=device_id,
)
return {
"status": "ok",
"device": row,
"message": "FCM token registered.",
}
@app.get("/notifications/my")
def notifications_my(
limit: int = 50,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
resolved_limit = max(1, min(200, int(limit)))
rows = _list_notification_events_for_user(requester_id, resolved_limit)
return {
"status": "ok",
"count": len(rows),
"notifications": rows,
}
@app.get("/health")
def health() -> Dict[str, Any]:
local_models_required = _should_load_local_models()
local_models_loaded = False
local_model_error: Optional[str] = None
if local_models_required:
try:
_ensure_models_loaded()
local_models_loaded = True
except Exception as exc: # pragma: no cover
local_model_error = str(exc)
status_value = "ok" if (not local_models_required or local_models_loaded) else "degraded"
model_inference_payload: Dict[str, Any] = {
"provider": MODEL_INFERENCE_PROVIDER,
"local_models_required": local_models_required,
"local_models_loaded": local_models_loaded,
"local_endpoint_enabled": MODEL_SERVICE_ENABLE_LOCAL_ENDPOINT,
"remote_service_configured": _model_service_configured(),
"remote_infer_url": _resolve_model_service_infer_url(),
"header_name": MODEL_SERVICE_HEADER_NAME,
}
if local_model_error:
model_inference_payload["local_model_error"] = local_model_error
return {
"status": status_value,
"models_dir": str(MODELS_DIR),
"ocr_model_file": MODEL_PATHS["ocr"].name,
"model_inference": model_inference_payload,
"supabase_configured": _supabase_configured(),
"supabase_auth_enforced": SUPABASE_ENFORCE_AUTH,
"pricing": {
"default_billing_mode": PARKING_BILLING_MODE,
"default_hourly_rate_egp": round(max(0.0, PARKING_HOURLY_RATE_EGP), 2),
"default_daily_rate_egp": round(max(0.0, PARKING_DAILY_RATE_EGP), 2),
"default_extra_after_first_hour_egp": round(max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP), 2),
"location_hourly_rates": PARKING_LOCATION_HOURLY_RATES,
"location_pricing": PARKING_LOCATION_PRICING,
"location_capacities": PARKING_LOCATION_CAPACITIES,
"garage_total_capacity": GARAGE_TOTAL_CAPACITY,
"app_service_fee_egp": round(max(0.0, APP_SERVICE_FEE_EGP), 2),
},
}
@app.get("/supabase/health")
def supabase_health() -> Dict[str, Any]:
if not _supabase_configured():
return {
"status": "disabled",
"reason": "Set SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_SERVICE_ROLE_KEY",
}
try:
response = requests.get(
f"{SUPABASE_URL}/auth/v1/settings",
headers={"apikey": SUPABASE_ANON_KEY},
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
return {
"status": "error",
"http_status": response.status_code,
"detail": response.text[:400],
}
existing_buckets = set(_supabase_list_buckets())
required_buckets = {SUPABASE_RAW_BUCKET, SUPABASE_PROCESSED_BUCKET}
missing_buckets = sorted(required_buckets - existing_buckets)
table_checks: Dict[str, Dict[str, Any]] = {}
all_tables_ready = True
for table_name in sorted(CORE_TABLES):
exists, detail = _supabase_table_exists(table_name)
table_checks[table_name] = {
"exists": exists,
"detail": detail,
}
if not exists:
all_tables_ready = False
optional_checks: Dict[str, Dict[str, Any]] = {}
for table_name in ["payment_transactions", "notification_device_tokens", "notification_events", "app_links"]:
exists, detail = _supabase_table_exists(table_name)
optional_checks[table_name] = {
"exists": exists,
"detail": detail,
}
ready_for_persistence = all_tables_ready and not missing_buckets
except requests.RequestException as exc:
return {
"status": "error",
"detail": str(exc),
}
except Exception as exc:
return {
"status": "error",
"detail": str(exc),
}
return {
"status": "ok" if ready_for_persistence else "partial",
"url": SUPABASE_URL,
"raw_bucket": SUPABASE_RAW_BUCKET,
"processed_bucket": SUPABASE_PROCESSED_BUCKET,
"auth_enforced": SUPABASE_ENFORCE_AUTH,
"readiness": {
"ready_for_persistence": ready_for_persistence,
"missing_buckets": missing_buckets,
"tables": table_checks,
"optional_tables": optional_checks,
},
}
@app.get("/models")
def models() -> Dict[str, Any]:
local_models_required = _should_load_local_models()
if local_models_required:
_ensure_models_loaded()
return {
"inference_provider": MODEL_INFERENCE_PROVIDER,
"local_models_required": local_models_required,
"local_endpoint_enabled": MODEL_SERVICE_ENABLE_LOCAL_ENDPOINT,
"remote_service_configured": _model_service_configured(),
"remote_infer_url": _resolve_model_service_infer_url(),
"plate_detector": str(MODEL_PATHS["plate"]) if local_models_required else None,
"ocr_model": str(MODEL_PATHS["ocr"]) if local_models_required else None,
}
@app.get("/developer/models")
def developer_models(
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user, requester_role = _require_developer_user(authorization)
payload = models()
payload["request_user"] = {
"id": request_user.get("id"),
"email": request_user.get("email"),
"role": requester_role,
}
return payload
@app.post("/model/infer")
async def model_infer(
image: UploadFile = File(...),
model_service_secret: Optional[str] = Header(default=None, alias=MODEL_SERVICE_HEADER_NAME),
) -> Dict[str, Any]:
# Always use remote HuggingFace Space for inference
if image.content_type and not image.content_type.startswith("image/") and image.content_type != "application/octet-stream":
raise HTTPException(status_code=400, detail="Only image uploads are supported.")
image_bytes = await image.read()
if not image_bytes:
raise HTTPException(status_code=400, detail="Uploaded file is empty.")
response = _run_pipeline_remote(
image_bytes=image_bytes,
filename=image.filename,
content_type=image.content_type,
)
response["filename"] = image.filename
response["inference_provider"] = "remote"
return response
def _resolve_history_scope(requester_id: str, requester_has_global_scope: bool, for_user_id: Optional[str]) -> Optional[str]:
target_user_id = (for_user_id or "").strip() or None
if target_user_id and not requester_has_global_scope and target_user_id != requester_id:
raise HTTPException(status_code=403, detail="You can only view your own records.")
if not requester_has_global_scope:
target_user_id = requester_id
return target_user_id
def _fetch_active_inside_counts_by_location() -> Tuple[int, Dict[str, int]]:
if not _supabase_configured():
return 0, {}
response = requests.get(
f"{SUPABASE_URL}/rest/v1/parking_sessions",
params={
"select": "parking_location",
"check_out_at": "is.null",
"limit": "10000",
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
LOGGER.warning("Failed to fetch inside occupancy counts: %s", response.text[:280])
return 0, {}
rows = response.json()
if not isinstance(rows, list):
return 0, {}
inside_by_location: Dict[str, int] = {}
total_inside = 0
for row in rows:
if not isinstance(row, dict):
continue
total_inside += 1
raw_location = str(row.get("parking_location") or "").strip()
if not raw_location:
continue
location_key = _normalize_location_key_for_config(raw_location.split(",", 1)[0])
if not location_key:
continue
inside_by_location[location_key] = inside_by_location.get(location_key, 0) + 1
return total_inside, inside_by_location
def _build_parking_locations_snapshot() -> Dict[str, Any]:
total_inside_from_sessions, inside_by_location_from_sessions = _fetch_active_inside_counts_by_location()
total_inside_inferred, inside_by_location_inferred = _count_inferred_inside_by_location()
total_inside = total_inside_from_sessions + total_inside_inferred
inside_by_location: Dict[str, int] = dict(inside_by_location_from_sessions)
for location_key, inferred_count in inside_by_location_inferred.items():
inside_by_location[location_key] = inside_by_location.get(location_key, 0) + inferred_count
location_keys = set(PARKING_LOCATION_PRICING.keys())
location_keys.update(PARKING_LOCATION_HOURLY_RATES.keys())
location_keys.update(PARKING_LOCATION_CAPACITIES.keys())
location_keys.update(inside_by_location.keys())
if not location_keys:
location_keys.add("DEFAULT")
locations: List[Dict[str, Any]] = []
for location_key in sorted(location_keys):
policy = _resolve_location_pricing_policy(location_key)
location_capacity = max(0, int(policy.get("location_capacity") or 0))
inside_count = inside_by_location.get(location_key, 0)
left_count = max(0, location_capacity - inside_count)
base_window_hours = 24 if str(policy.get("billing_mode") or "hourly") == "daily" else 1
locations.append(
{
"parking_location": location_key,
"pricing": {
"billing_mode": str(policy.get("billing_mode") or "hourly"),
"base_rate_egp": round(max(0.0, float(policy.get("base_rate_egp") or 0.0)), 2),
"extra_after_first_hour_egp": round(max(0.0, float(policy.get("extra_after_first_hour_egp") or 0.0)), 2),
"app_service_fee_egp": round(max(0.0, APP_SERVICE_FEE_EGP), 2),
"base_window_hours": base_window_hours,
"rate_source": str(policy.get("rate_source") or "default"),
},
"occupancy": {
"inside": inside_count,
"inside_from_registered_sessions": inside_by_location_from_sessions.get(location_key, 0),
"inside_from_inferred_unmatched": inside_by_location_inferred.get(location_key, 0),
"total_capacity": location_capacity,
"left": left_count,
"display": f"{inside_count}/{location_capacity}",
"capacity_source": str(policy.get("location_capacity_source") or "default"),
},
}
)
unknown_location_inside = max(0, total_inside - sum(inside_by_location.values()))
garage_left = max(0, GARAGE_TOTAL_CAPACITY - total_inside)
return {
"garage": {
"inside": total_inside,
"inside_from_registered_sessions": total_inside_from_sessions,
"inside_from_inferred_unmatched": total_inside_inferred,
"total_capacity": GARAGE_TOTAL_CAPACITY,
"left": garage_left,
"display": f"{total_inside}/{GARAGE_TOTAL_CAPACITY}",
"unknown_location_inside": unknown_location_inside,
},
"locations": locations,
}
@app.get("/parking/locations")
def parking_locations(
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
snapshot = _build_parking_locations_snapshot()
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"pricing_defaults": {
"billing_mode": PARKING_BILLING_MODE,
"hourly_rate_egp": round(max(0.0, PARKING_HOURLY_RATE_EGP), 2),
"daily_rate_egp": round(max(0.0, PARKING_DAILY_RATE_EGP), 2),
"extra_after_first_hour_egp": round(max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP), 2),
"app_service_fee_egp": round(max(0.0, APP_SERVICE_FEE_EGP), 2),
},
"garage": snapshot["garage"],
"locations": snapshot["locations"],
}
@app.get("/parking/occupancy")
def parking_occupancy(
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
snapshot = _build_parking_locations_snapshot()
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"garage": snapshot["garage"],
"locations": [
{
"parking_location": row.get("parking_location"),
"occupancy": row.get("occupancy"),
}
for row in snapshot["locations"]
],
}
def _fetch_vehicle_map_by_ids(vehicle_ids: List[str]) -> Dict[str, Dict[str, Any]]:
if not vehicle_ids or not _supabase_configured():
return {}
response = requests.get(
f"{SUPABASE_URL}/rest/v1/vehicles",
params={
"select": "id,owner_id,plate_letters_ar,plate_numbers_ar,plate_letters_en,plate_numbers_en,car_name,car_model,car_color,is_active,is_verified,verified_at,verified_by,created_at,updated_at",
"id": f"in.({','.join(sorted(set(vehicle_ids)))})",
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
LOGGER.warning("Failed to fetch vehicle map: %s", response.text[:280])
return {}
rows = response.json()
if not isinstance(rows, list):
return {}
result: Dict[str, Dict[str, Any]] = {}
for row in rows:
if isinstance(row, dict) and row.get("id"):
result[str(row["id"])] = row
return result
def _fetch_session_map_by_ids(session_ids: List[str]) -> Dict[str, Dict[str, Any]]:
if not session_ids or not _supabase_configured():
return {}
response = requests.get(
f"{SUPABASE_URL}/rest/v1/parking_sessions",
params={
"select": "id,vehicle_id,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
"id": f"in.({','.join(sorted(set(session_ids)))})",
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
LOGGER.warning("Failed to fetch session map: %s", response.text[:280])
return {}
rows = response.json()
if not isinstance(rows, list):
return {}
result: Dict[str, Dict[str, Any]] = {}
for row in rows:
if isinstance(row, dict) and row.get("id"):
result[str(row["id"])] = row
return result
def _fetch_latest_event_by_session_ids(session_ids: List[str]) -> Dict[str, Dict[str, Any]]:
if not session_ids or not _supabase_configured():
return {}
response = requests.get(
f"{SUPABASE_URL}/rest/v1/car_events",
params={
"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
"session_id": f"in.({','.join(sorted(set(session_ids)))})",
"order": "captured_at.desc",
"limit": str(max(len(session_ids) * 6, 200)),
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
LOGGER.warning("Failed to fetch latest events by session ids: %s", response.text[:280])
return {}
rows = response.json()
if not isinstance(rows, list):
return {}
result: Dict[str, Dict[str, Any]] = {}
for row in rows:
if not isinstance(row, dict):
continue
session_id = str(row.get("session_id") or "").strip()
if not session_id or session_id in result:
continue
result[session_id] = row
return result
def _event_plate_value(value: Any) -> Optional[str]:
raw = _normalize_spaces(str(value or ""))
if not raw or raw.upper() == "N/A":
return None
return raw
def _build_unmatched_plate_key(event_row: Dict[str, Any]) -> Tuple[Optional[str], Optional[str], Optional[str]]:
plate_arabic = _event_plate_value(event_row.get("ocr_arabic"))
plate_english = _event_plate_value(event_row.get("ocr_english"))
if plate_arabic:
return f"AR::{plate_arabic}", plate_arabic, plate_english
if plate_english:
plate_english_upper = plate_english.upper()
return f"EN::{plate_english_upper}", None, plate_english_upper
return None, None, None
def _resolve_event_timestamp(event_row: Dict[str, Any]) -> Optional[datetime]:
for field_name in ["captured_at", "created_at"]:
parsed = _parse_iso_datetime(event_row.get(field_name))
if parsed is not None:
return parsed
return None
def _resolve_event_location_key(event_row: Dict[str, Any]) -> Optional[str]:
raw_location = _normalize_spaces(str(event_row.get("parking_location") or ""))
if not raw_location:
event_metadata = event_row.get("event_metadata") if isinstance(event_row.get("event_metadata"), dict) else {}
raw_location = _normalize_spaces(str(event_metadata.get("parking_location") or ""))
if not raw_location:
return None
return _normalize_location_key_for_config(raw_location.split(",", 1)[0])
def _fetch_unmatched_car_events(limit: int = 4000) -> List[Dict[str, Any]]:
if not _supabase_configured():
return []
response = requests.get(
f"{SUPABASE_URL}/rest/v1/car_events",
params={
"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,event_metadata,captured_at,created_at",
"session_id": "is.null",
"order": "captured_at.desc",
"limit": str(max(limit, 1)),
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
LOGGER.warning("Failed to fetch unmatched car events: %s", response.text[:280])
return []
rows = response.json()
if not isinstance(rows, list):
return []
return [row for row in rows if isinstance(row, dict)]
def _build_inferred_unmatched_session_rows(limit: int = 300, *, inside_only: bool) -> List[Dict[str, Any]]:
events = _fetch_unmatched_car_events(limit=max(limit * 8, 400))
if not events:
return []
epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
sorted_events = sorted(events, key=lambda row: _resolve_event_timestamp(row) or epoch)
state_by_plate: Dict[str, Dict[str, Any]] = {}
for event in sorted_events:
event_type = str(event.get("event_type") or "").strip().lower()
if event_type not in SUPPORTED_EVENT_TYPES:
continue
event_timestamp = _resolve_event_timestamp(event)
if event_timestamp is None:
continue
plate_key, plate_arabic, plate_english = _build_unmatched_plate_key(event)
if not plate_key:
continue
state = state_by_plate.setdefault(
plate_key,
{
"plate_key": plate_key,
"plate_arabic": plate_arabic,
"plate_english": plate_english,
"inside": False,
"has_transition": False,
"check_in_at": None,
"check_out_at": None,
"parking_location": None,
"latest_event": None,
"latest_event_at": None,
"event_count": 0,
},
)
if plate_arabic and not state.get("plate_arabic"):
state["plate_arabic"] = plate_arabic
if plate_english and not state.get("plate_english"):
state["plate_english"] = plate_english
state["event_count"] = int(state.get("event_count") or 0) + 1
state["latest_event"] = event
state["latest_event_at"] = event_timestamp
resolved_location = _resolve_event_location_key(event)
if resolved_location:
state["parking_location"] = resolved_location
if event_type == "entry":
state["inside"] = True
state["has_transition"] = True
state["check_in_at"] = event_timestamp
state["check_out_at"] = None
if resolved_location:
state["parking_location"] = resolved_location
elif event_type == "exit":
state["inside"] = False
state["has_transition"] = True
state["check_out_at"] = event_timestamp
if state.get("check_in_at") is None:
state["check_in_at"] = event_timestamp
inferred_rows: List[Dict[str, Any]] = []
for state in state_by_plate.values():
if not bool(state.get("has_transition")):
continue
is_inside = bool(state.get("inside"))
if inside_only and not is_inside:
continue
latest_event = state.get("latest_event") if isinstance(state.get("latest_event"), dict) else None
latest_event_at = state.get("latest_event_at") if isinstance(state.get("latest_event_at"), datetime) else None
if latest_event is None or latest_event_at is None:
continue
check_in_at = state.get("check_in_at") if isinstance(state.get("check_in_at"), datetime) else latest_event_at
check_out_at = state.get("check_out_at") if isinstance(state.get("check_out_at"), datetime) else None
if not is_inside and check_out_at is None:
check_out_at = latest_event_at
parking_location = state.get("parking_location")
if not parking_location:
parking_location = _resolve_event_location_key(latest_event)
plate_key = str(state.get("plate_key") or "")
session_id = f"inferred:{hashlib.sha1(plate_key.encode('utf-8')).hexdigest()[:20]}"
session_row: Dict[str, Any] = {
"id": session_id,
"vehicle_id": None,
"check_in_at": check_in_at.isoformat(),
"payment_confirmed_at": None,
"check_out_at": check_out_at.isoformat() if check_out_at else None,
"left_within_5_minutes": False,
"status": "inferred_inside_unmatched" if is_inside else "inferred_exited_unmatched",
"parking_location": parking_location,
"created_by": None,
"notes": "Inferred from unmatched OCR events.",
"created_at": check_in_at.isoformat(),
"updated_at": latest_event_at.isoformat(),
"inferred": True,
}
pricing = _compute_session_time_and_fee(session_row, parking_location_override=parking_location)
inferred_rows.append(
{
"session": session_row,
"vehicle": None,
"latest_event": latest_event,
"events": [latest_event],
"plate": {
"key": plate_key,
"arabic": state.get("plate_arabic"),
"english": state.get("plate_english"),
},
"event_count": int(state.get("event_count") or 0),
"inferred_unmatched": True,
"payment_status": _derive_payment_status_from_session(session_row),
"left_within_5_minutes": False,
"pricing": pricing,
"gate_decision_if_exit_attempted": _build_gate_decision(
event_type="exit",
session_row=session_row,
plate_detected=True,
),
}
)
inferred_rows.sort(
key=lambda row: _resolve_event_timestamp(row.get("latest_event") if isinstance(row.get("latest_event"), dict) else {}) or epoch,
reverse=True,
)
return inferred_rows[:limit]
def _count_inferred_inside_by_location() -> Tuple[int, Dict[str, int]]:
inferred_inside_rows = _build_inferred_unmatched_session_rows(limit=1200, inside_only=True)
inside_by_location: Dict[str, int] = {}
total_inside = 0
for row in inferred_inside_rows:
session_row = row.get("session") if isinstance(row.get("session"), dict) else {}
total_inside += 1
location_key = _normalize_location_key_for_config(str(session_row.get("parking_location") or ""))
if not location_key:
continue
inside_by_location[location_key] = inside_by_location.get(location_key, 0) + 1
return total_inside, inside_by_location
def _list_event_feed(
*,
limit: int,
target_user_id: Optional[str],
event_type: str,
within_5_only: bool,
) -> List[Dict[str, Any]]:
fetch_limit = max(limit * 4, 300) if within_5_only else limit
event_params: Dict[str, str] = {
"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
"event_type": f"eq.{event_type}",
"order": "captured_at.desc",
"limit": str(fetch_limit),
}
if target_user_id:
vehicle_ids = _fetch_vehicle_ids_for_owner(target_user_id)
if not vehicle_ids:
return []
event_params["vehicle_id"] = f"in.({','.join(vehicle_ids)})"
response = requests.get(
f"{SUPABASE_URL}/rest/v1/car_events",
params=event_params,
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
raise HTTPException(status_code=502, detail=f"Failed to load event feed: {response.text[:320]}")
events_rows = response.json()
events = events_rows if isinstance(events_rows, list) else []
session_ids: List[str] = []
vehicle_ids: List[str] = []
for event in events:
if not isinstance(event, dict):
continue
sid = str(event.get("session_id") or "").strip()
vid = str(event.get("vehicle_id") or "").strip()
if sid:
session_ids.append(sid)
if vid:
vehicle_ids.append(vid)
session_map = _fetch_session_map_by_ids(session_ids)
vehicle_map = _fetch_vehicle_map_by_ids(vehicle_ids)
filtered_events: List[Dict[str, Any]] = []
for event in events:
if not isinstance(event, dict):
continue
event_metadata = event.get("event_metadata") if isinstance(event.get("event_metadata"), dict) else {}
session_id = str(event.get("session_id") or "").strip()
session_row = session_map.get(session_id)
left_within_5 = bool(
(session_row or {}).get("left_within_5_minutes")
or event_metadata.get("left_within_5_minutes")
)
if within_5_only and not left_within_5:
continue
filtered_events.append(event)
if len(filtered_events) >= limit:
break
items: List[Dict[str, Any]] = []
for event in filtered_events:
session_id = str(event.get("session_id") or "").strip()
vehicle_id = str(event.get("vehicle_id") or "").strip()
session_row = session_map.get(session_id)
event_metadata = event.get("event_metadata") if isinstance(event.get("event_metadata"), dict) else {}
payment_status = event_metadata.get("payment_status") or _derive_payment_status_from_session(session_row)
pricing = (
event_metadata.get("pricing")
if isinstance(event_metadata.get("pricing"), dict)
else _compute_session_time_and_fee(
session_row,
parking_location_override=(
str(event.get("parking_location") or "").strip()
or str(event_metadata.get("parking_location") or "").strip()
or None
),
)
)
plate_detected = str(event.get("ocr_arabic") or "").strip().upper() not in {"", "N/A"} or str(event.get("ocr_english") or "").strip().upper() not in {"", "N/A"}
gate_decision = event_metadata.get("gate_decision") if isinstance(event_metadata.get("gate_decision"), dict) else _build_gate_decision(
event_type=event_type,
session_row=session_row,
plate_detected=plate_detected,
)
items.append(
{
"event": event,
"vehicle": vehicle_map.get(vehicle_id),
"session": session_row,
"payment_status": payment_status,
"left_within_5_minutes": bool(
(session_row or {}).get("left_within_5_minutes")
or event_metadata.get("left_within_5_minutes")
),
"pricing": pricing,
"gate_decision": gate_decision,
}
)
return items
@app.get("/events/entered-cars")
def entered_cars_feed(
limit: int = 100,
for_user_id: Optional[str] = None,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if limit < 1 or limit > 500:
raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
requester_has_global_scope = _can_view_global_records(requester_role)
target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)
items = _list_event_feed(
limit=limit,
target_user_id=target_user_id,
event_type="entry",
within_5_only=False,
)
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"scope": {
"for_user_id": target_user_id,
"limit": limit,
"event_type": "entry",
},
"items": items,
}
@app.get("/events/leaving-cars-within-5-minutes")
def leaving_within_5_minutes_feed(
limit: int = 100,
for_user_id: Optional[str] = None,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if limit < 1 or limit > 500:
raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
requester_has_global_scope = _can_view_global_records(requester_role)
target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)
items = _list_event_feed(
limit=limit,
target_user_id=target_user_id,
event_type="exit",
within_5_only=True,
)
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"scope": {
"for_user_id": target_user_id,
"limit": limit,
"event_type": "exit",
"within_5_minutes_after_payment": True,
},
"items": items,
}
@app.get("/events/new-cars")
def new_cars_feed(
limit: int = 100,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if limit < 1 or limit > 500:
raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
if not _is_staff_role(requester_role):
raise HTTPException(status_code=403, detail="Only admin/security can view new-car alerts.")
fetch_limit = max(limit * 8, 400)
response = requests.get(
f"{SUPABASE_URL}/rest/v1/car_events",
params={
"select": "id,session_id,vehicle_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
"order": "captured_at.desc",
"limit": str(fetch_limit),
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
raise HTTPException(status_code=502, detail=f"Failed to fetch new-car alerts: {response.text[:320]}")
events_rows = response.json()
events = events_rows if isinstance(events_rows, list) else []
session_ids: List[str] = []
vehicle_ids: List[str] = []
for event in events:
if not isinstance(event, dict):
continue
sid = str(event.get("session_id") or "").strip()
vid = str(event.get("vehicle_id") or "").strip()
if sid:
session_ids.append(sid)
if vid:
vehicle_ids.append(vid)
session_map = _fetch_session_map_by_ids(session_ids)
vehicle_map = _fetch_vehicle_map_by_ids(vehicle_ids)
items: List[Dict[str, Any]] = []
for event in events:
if not isinstance(event, dict):
continue
event_metadata = event.get("event_metadata") if isinstance(event.get("event_metadata"), dict) else {}
new_car_alert = event_metadata.get("new_car_alert") if isinstance(event_metadata.get("new_car_alert"), dict) else {}
if new_car_alert.get("is_new_car") is not True:
app_registration = event_metadata.get("app_registration") if isinstance(event_metadata.get("app_registration"), dict) else {}
if app_registration.get("is_registered") is False:
new_car_alert = {
"is_new_car": True,
"notify_roles": ["admin", "security"],
"action": "vendor_outreach_recommended",
"reason": "car_not_registered_by_any_user",
"message": "New car detected. Admin/security should contact a vendor to call the driver and introduce the app.",
}
else:
continue
session_id = str(event.get("session_id") or "").strip()
vehicle_id = str(event.get("vehicle_id") or "").strip()
items.append(
{
"event": event,
"vehicle": vehicle_map.get(vehicle_id),
"session": session_map.get(session_id),
"new_car_alert": new_car_alert,
}
)
if len(items) >= limit:
break
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": True,
},
"scope": {
"limit": limit,
"event_type": "new_car_alert",
"notify_roles": ["admin", "security"],
"action": "vendor_outreach_recommended",
},
"items": items,
}
@app.get("/events/left-cars")
def left_cars_feed(
limit: int = 100,
for_user_id: Optional[str] = None,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if limit < 1 or limit > 500:
raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
requester_has_global_scope = _can_view_global_records(requester_role)
target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)
items = _list_event_feed(
limit=limit,
target_user_id=target_user_id,
event_type="exit",
within_5_only=False,
)
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"scope": {
"for_user_id": target_user_id,
"limit": limit,
"event_type": "exit",
"within_5_minutes_after_payment": False,
},
"items": items,
}
@app.get("/events/inside-cars")
def inside_cars_feed(
limit: int = 100,
for_user_id: Optional[str] = None,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if limit < 1 or limit > 500:
raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
requester_has_global_scope = _can_view_global_records(requester_role)
target_user_id = _resolve_history_scope(requester_id, requester_has_global_scope, for_user_id)
session_params: Dict[str, str] = {
"select": "id,vehicle_id,plate_arabic,plate_english,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
"check_out_at": "is.null",
"status": "not.in.(exited,left_without_payment)",
"order": "check_in_at.desc",
"limit": str(limit),
}
if target_user_id:
vehicle_ids = _fetch_vehicle_ids_for_owner(target_user_id)
if not vehicle_ids:
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"scope": {
"for_user_id": target_user_id,
"limit": limit,
"event_type": "inside",
},
"items": [],
}
session_params["vehicle_id"] = f"in.({','.join(vehicle_ids)})"
sessions_response = requests.get(
f"{SUPABASE_URL}/rest/v1/parking_sessions",
params=session_params,
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if sessions_response.status_code != 200:
raise HTTPException(status_code=502, detail=f"Failed to fetch inside sessions: {sessions_response.text[:320]}")
sessions_rows = sessions_response.json()
sessions = sessions_rows if isinstance(sessions_rows, list) else []
vehicle_ids_in_sessions = [
str(row.get("vehicle_id"))
for row in sessions
if isinstance(row, dict) and row.get("vehicle_id")
]
vehicle_map = _fetch_vehicle_map_by_ids(vehicle_ids_in_sessions)
session_ids = [
str(row.get("id"))
for row in sessions
if isinstance(row, dict) and row.get("id")
]
latest_event_by_session = _fetch_latest_event_by_session_ids(session_ids)
items: List[Dict[str, Any]] = []
for session in sessions:
if not isinstance(session, dict):
continue
session_id = str(session.get("id") or "").strip()
vehicle_id = str(session.get("vehicle_id") or "").strip()
# Self-healing: Hide and close old ghost sessions created before the fix
v_id = vehicle_id.upper()
plate_ar = str(session.get("plate_arabic") or "").strip().upper()
plate_en = str(session.get("plate_english") or "").strip().upper()
if v_id in {"", "NONE", "NULL"} and plate_ar in {"", "N/A", "NONE", "NULL", "-"} and plate_en in {"", "N/A", "NONE", "NULL", "-"}:
LOGGER.warning("Self-healing: Closing invalid ghost session %s", session_id)
_update_parking_session(session_id, {
"status": "exited",
"check_out_at": datetime.now(timezone.utc).isoformat(),
"notes": "Closed invalid ghost session automatically."
})
continue
latest_event = latest_event_by_session.get(session_id)
# Pricing and Status calculation
pricing = _compute_session_time_and_fee(session)
payment_status = _derive_payment_status_from_session(session)
plate_detected_in_latest_event = bool(
isinstance(latest_event, dict)
and (
str(latest_event.get("ocr_arabic") or "").strip().upper() not in {"", "N/A"}
or str(latest_event.get("ocr_english") or "").strip().upper() not in {"", "N/A"}
)
)
items.append(
{
"session": session,
"vehicle": vehicle_map.get(vehicle_id),
"latest_event": latest_event,
"payment_status": payment_status,
"pricing": pricing,
"gate_decision_if_exit_attempted": _build_gate_decision(
event_type="exit",
session_row=session,
plate_detected=True,
),
"plate_detected_in_latest_event": plate_detected_in_latest_event,
"inferred_unmatched": False,
}
)
if target_user_id is None:
real_session_plates = set()
for item in items:
s = item.get("session") or {}
plate_ar = _normalize_plate_text(s.get("plate_arabic"))
plate_en = _normalize_plate_text(s.get("plate_english"))
if plate_ar: real_session_plates.add(plate_ar)
if plate_en: real_session_plates.add(plate_en)
inferred_items = _build_inferred_unmatched_session_rows(limit=max(limit * 3, 300), inside_only=True)
for inferred in inferred_items:
inf_plate = inferred.get("plate")
inf_plate_norm = _normalize_plate_text(inf_plate)
if inf_plate_norm in real_session_plates:
continue
items.append(
{
"session": inferred.get("session"),
"vehicle": None,
"latest_event": inferred.get("latest_event"),
"payment_status": inferred.get("payment_status"),
"pricing": inferred.get("pricing"),
"gate_decision_if_exit_attempted": inferred.get("gate_decision_if_exit_attempted"),
"plate_detected_in_latest_event": True,
"inferred_unmatched": True,
"plate": inf_plate,
"event_count": inferred.get("event_count"),
}
)
epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
items.sort(
key=lambda row: (
_parse_iso_datetime(((row.get("session") if isinstance(row.get("session"), dict) else {}).get("check_in_at")))
or _resolve_event_timestamp((row.get("latest_event") if isinstance(row.get("latest_event"), dict) else {}))
or epoch
),
reverse=True,
)
items = items[:limit]
parking_snapshot = _build_parking_locations_snapshot()
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"scope": {
"for_user_id": target_user_id,
"limit": limit,
"event_type": "inside",
},
"garage_occupancy": parking_snapshot["garage"],
"location_pricing": parking_snapshot["locations"],
"items": items,
}
@app.get("/parking/history")
def parking_history(
limit: int = 100,
for_user_id: Optional[str] = None,
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if limit < 1 or limit > 500:
raise HTTPException(status_code=400, detail="limit must be between 1 and 500.")
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
requester_has_global_scope = _can_view_global_records(requester_role)
target_user_id = (for_user_id or "").strip() or None
if target_user_id and not requester_has_global_scope and target_user_id != requester_id:
raise HTTPException(status_code=403, detail="You can only view your own parking history.")
if not requester_has_global_scope:
target_user_id = requester_id
session_params: Dict[str, str] = {
"select": "id,vehicle_id,plate_arabic,plate_english,check_in_at,payment_confirmed_at,check_out_at,left_within_5_minutes,status,parking_location,created_by,notes,created_at,updated_at",
"order": "check_in_at.desc",
"limit": str(limit),
}
if target_user_id:
vehicle_ids = _fetch_vehicle_ids_for_owner(target_user_id)
if not vehicle_ids:
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"scope": {
"for_user_id": target_user_id,
"limit": limit,
},
"history": [],
}
session_params["vehicle_id"] = f"in.({','.join(vehicle_ids)})"
sessions_response = requests.get(
f"{SUPABASE_URL}/rest/v1/parking_sessions",
params=session_params,
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if sessions_response.status_code != 200:
raise HTTPException(
status_code=502,
detail=f"Failed to fetch parking history: {sessions_response.text[:320]}",
)
sessions_rows = sessions_response.json()
sessions = sessions_rows if isinstance(sessions_rows, list) else []
vehicle_ids_in_history: List[str] = []
for row in sessions:
if isinstance(row, dict) and row.get("vehicle_id"):
vehicle_ids_in_history.append(str(row["vehicle_id"]))
vehicle_map: Dict[str, Dict[str, Any]] = {}
if vehicle_ids_in_history:
vehicle_response = requests.get(
f"{SUPABASE_URL}/rest/v1/vehicles",
params={
"select": "id,owner_id,plate_letters_ar,plate_numbers_ar,plate_letters_en,plate_numbers_en,car_model,car_color,is_active",
"id": f"in.({','.join(sorted(set(vehicle_ids_in_history)))})",
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if vehicle_response.status_code == 200:
vehicle_rows = vehicle_response.json()
if isinstance(vehicle_rows, list):
for vehicle in vehicle_rows:
if isinstance(vehicle, dict) and vehicle.get("id"):
vehicle_map[str(vehicle["id"])] = vehicle
session_ids: List[str] = []
for row in sessions:
if isinstance(row, dict) and row.get("id"):
session_ids.append(str(row["id"]))
events_by_session: Dict[str, List[Dict[str, Any]]] = {}
if session_ids:
events_response = requests.get(
f"{SUPABASE_URL}/rest/v1/car_events",
params={
"select": "id,session_id,event_type,ocr_arabic,ocr_english,ocr_confidence,parking_location,raw_image_path,user_split_image_path,admin_annotated_image_path,event_metadata,captured_at,created_at",
"session_id": f"in.({','.join(sorted(set(session_ids)))})",
"order": "captured_at.desc",
"limit": str(max(limit * 6, 200)),
},
headers=_supabase_headers(
api_key=SUPABASE_SERVICE_ROLE_KEY,
bearer=SUPABASE_SERVICE_ROLE_KEY,
),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if events_response.status_code == 200:
events_rows = events_response.json()
if isinstance(events_rows, list):
for event in events_rows:
if not isinstance(event, dict):
continue
sid = str(event.get("session_id") or "").strip()
if not sid:
continue
events_by_session.setdefault(sid, []).append(event)
history_items: List[Dict[str, Any]] = []
for session in sessions:
if not isinstance(session, dict):
continue
session_id = str(session.get("id") or "").strip()
vehicle_id = str(session.get("vehicle_id") or "").strip()
pricing = _compute_session_time_and_fee(session)
payment_status = _derive_payment_status_from_session(session)
history_items.append(
{
"session": session,
"payment_status": payment_status,
"left_within_5_minutes": bool(session.get("left_within_5_minutes")),
"pricing": pricing,
"gate_decision_if_exit_attempted": _build_gate_decision(
event_type="exit",
session_row=session,
plate_detected=True,
),
"vehicle": vehicle_map.get(vehicle_id),
"events": events_by_session.get(session_id, []),
"inferred_unmatched": False,
}
)
if target_user_id is None:
inferred_history = _build_inferred_unmatched_session_rows(limit=max(limit * 3, 300), inside_only=False)
for inferred in inferred_history:
history_items.append(
{
"session": inferred.get("session"),
"payment_status": inferred.get("payment_status"),
"left_within_5_minutes": False,
"pricing": inferred.get("pricing"),
"gate_decision_if_exit_attempted": inferred.get("gate_decision_if_exit_attempted"),
"vehicle": None,
"events": inferred.get("events", []),
"inferred_unmatched": True,
"plate": inferred.get("plate"),
"event_count": inferred.get("event_count"),
}
)
epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
history_items.sort(
key=lambda row: (
_parse_iso_datetime(((row.get("session") if isinstance(row.get("session"), dict) else {}).get("check_in_at")))
or _resolve_event_timestamp(((row.get("events") or [None])[0] if isinstance((row.get("events") or [None])[0], dict) else {}))
or epoch
),
reverse=True,
)
history_items = history_items[:limit]
parking_snapshot = _build_parking_locations_snapshot()
return {
"status": "ok",
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
"scope": {
"for_user_id": target_user_id,
"limit": limit,
},
"garage_occupancy": parking_snapshot["garage"],
"location_pricing": parking_snapshot["locations"],
"history": history_items,
}
@app.post("/payments/manual/cash-confirm")
def manual_cash_confirm_payment(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
if not _is_staff_role(requester_role):
raise HTTPException(status_code=403, detail="Only admin/security can confirm manual cash payments.")
session_id = str(payload.get("session_id") or "").strip()
if not session_id:
raise HTTPException(status_code=400, detail="session_id is required.")
parking_location = _normalize_parking_location(payload.get("parking_location"))
# === GHOST FIX: Handle Inferred Sessions (Unregistered Cars) ===
if session_id.startswith("inferred:"):
LOGGER.info("Handling inferred session checkout for: %s", session_id)
# Extract plate info from inferred session (if possible) or use provided plate
# For simplicity, we search for the latest unmatched event for this 'inferred' car
# and create a real session for it.
plate_ar = str(payload.get("plate_arabic") or payload.get("plate") or "").strip()
if not plate_ar:
# Fallback: try to find it in the inside feed cache or inferred logic
# But usually the dashboard will send the plate now.
raise HTTPException(status_code=400, detail="plate is required for inferred session checkout.")
# 1. Create a REAL session
new_session = _insert_parking_session({
"plate_arabic": plate_ar,
"status": "inside",
"check_in_at": datetime.now(timezone.utc).isoformat(),
"parking_location": parking_location or "OPERA",
"notes": f"Inferred promote from {session_id}"
})
if not new_session:
raise HTTPException(status_code=500, detail="Failed to promote inferred session to real session.")
real_sid = str(new_session.get("id"))
LOGGER.info("Promoted inferred session to real ID: %s", real_sid)
# 2. Link any unmatched events for this plate to the new session
_link_unmatched_events_to_session(plate_ar, real_sid)
# 3. Use the REAL session ID for the rest of the flow
session_id = real_sid
session_row = new_session
else:
session_row = _fetch_parking_session_by_id(session_id)
if not isinstance(session_row, dict):
raise HTTPException(status_code=404, detail="parking session was not found.")
if parking_location:
_set_parking_session_location(session_id, parking_location)
session_row = _fetch_parking_session_by_id(session_id) or session_row
pricing = _compute_session_time_and_fee(session_row)
amount_raw = payload.get("amount_cents")
if amount_raw is None:
amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)
else:
try:
amount_cents = int(amount_raw)
except Exception:
raise HTTPException(status_code=400, detail="amount_cents must be a positive integer.")
if amount_cents <= 0:
raise HTTPException(status_code=400, detail="amount_cents must be greater than zero.")
currency = str(payload.get("currency") or PAYMOB_CURRENCY).strip().upper() or PAYMOB_CURRENCY
manual_reference = str(payload.get("manual_reference") or f"cash-{uuid4().hex[:18]}").strip()
manual_reference = manual_reference or f"cash-{uuid4().hex[:18]}"
already_paid = _parse_iso_datetime(session_row.get("payment_confirmed_at")) is not None
force_exit = bool(payload.get("force_exit", False))
if not already_paid:
_mark_parking_session_paid(session_id, payment_reference=manual_reference)
if force_exit:
now_str = datetime.now(timezone.utc).isoformat()
_update_parking_session(session_id, {
"status": "exited",
"check_out_at": now_str,
"left_within_5_minutes": True,
"notes": f"Staff cash-confirm checkout ({requester_id})"
})
plate_label = _resolve_plate_label_for_session(session_row)
if not plate_label:
plate_label = _normalize_spaces(str(payload.get("plate_arabic") or payload.get("plate") or ""))
if plate_label:
_notification_title = "✅ Cash Payment Received"
_notification_body = f"تم استلام الدفع كاش للعربية {plate_label} — تم فتح البوابة"
_notify_security_staff(
title=_notification_title,
body=_notification_body,
data={
"session_id": session_id,
"plate": plate_label,
"payment_method": "cash",
},
)
# Force checkout for manual cash confirmations (Guard Hub workflow)
_update_parking_session(session_id, {
"check_out_at": datetime.now(timezone.utc).isoformat(),
"status": "exited",
"left_within_5_minutes": True
})
payment_transaction_id: Optional[str] = None
if not already_paid:
payment_transaction_id = _insert_payment_transaction_record(
created_by=requester_id,
session_id=session_id,
amount_cents=amount_cents,
currency=currency,
merchant_order_id=f"cash-{session_id[:8]}-{uuid4().hex[:12]}",
paymob_order_id=None,
parking_location=parking_location,
request_payload={
"session_id": session_id,
"amount_cents": amount_cents,
"currency": currency,
"manual_reference": manual_reference,
"parking_location": parking_location,
},
response_payload={
"manual_reference": manual_reference,
"confirmed_by": requester_id,
"confirmed_by_role": requester_role,
},
provider="cash",
status="paid",
payment_reference=manual_reference,
)
updated_session = _fetch_parking_session_by_id(session_id) or session_row
updated_pricing = _compute_session_time_and_fee(updated_session)
return {
"status": "ok",
"provider": "cash",
"already_paid": already_paid,
"session_id": session_id,
"payment_transaction_id": payment_transaction_id,
"amount_cents": amount_cents,
"currency": currency,
"payment_reference": manual_reference,
"pricing": updated_pricing,
"payment_status": _derive_payment_status_from_session(updated_session),
"gate_decision": _build_gate_decision(
event_type="exit",
session_row=updated_session,
plate_detected=True,
),
}
@app.post("/payments/demo/confirm")
def demo_confirm_payment(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
session_id = str(payload.get("session_id") or "").strip()
if not session_id:
raise HTTPException(status_code=400, detail="session_id is required.")
session_row = _fetch_parking_session_by_id(session_id)
if not isinstance(session_row, dict):
raise HTTPException(status_code=404, detail="parking session was not found.")
owner_id = _get_parking_session_owner_id(session_id)
if owner_id and not requester_is_staff and owner_id != requester_id:
raise HTTPException(status_code=403, detail="You can only confirm payments for your own sessions.")
already_paid = _parse_iso_datetime(session_row.get("payment_confirmed_at")) is not None
if not already_paid:
_mark_parking_session_paid(session_id)
pricing = _compute_session_time_and_fee(session_row)
amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)
_insert_payment_transaction_record(
created_by=requester_id,
session_id=session_id,
amount_cents=amount_cents,
currency=PAYMOB_CURRENCY,
merchant_order_id=f"demo-{session_id[:8]}-{uuid4().hex[:10]}",
paymob_order_id=None,
parking_location=_normalize_parking_location(session_row.get("parking_location")),
request_payload={
"session_id": session_id,
"provider": "demo_online",
},
response_payload={
"status": "paid",
},
provider="demo_online",
status="paid",
)
if owner_id:
_insert_notification_event(
user_id=owner_id,
event_type="payment_confirmed",
title="Payment Confirmed :white_check_mark:",
body="تم تأكيد دفعك — عندك 5 دقايق للخروج",
data={
"session_id": session_id,
"payment_method": "demo_online",
},
)
updated_session = _fetch_parking_session_by_id(session_id) or session_row
return {
"session_id": session_id,
"payment_status": _derive_payment_status_from_session(updated_session),
"gate_decision": _build_gate_decision(
event_type="exit",
session_row=updated_session,
plate_detected=True,
),
"left_within_5_minutes": bool(updated_session.get("left_within_5_minutes")),
}
@app.post("/payments/cash/request")
def request_cash_payment(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
session_id = str(payload.get("session_id") or "").strip()
if not session_id:
raise HTTPException(status_code=400, detail="session_id is required.")
session_row = _fetch_parking_session_by_id(session_id)
if not isinstance(session_row, dict):
raise HTTPException(status_code=404, detail="parking session was not found.")
owner_id = _get_parking_session_owner_id(session_id)
if owner_id and not requester_is_staff and owner_id != requester_id:
raise HTTPException(status_code=403, detail="You can only request cash for your own sessions.")
profile = _supabase_get_profile_by_user_id(requester_id) or {}
requester_name = _normalize_spaces(
str(profile.get("full_name") or profile.get("username") or profile.get("first_name") or "")
)
if not requester_name:
requester_name = "اليوزر"
plate_label = _resolve_plate_label_for_session(session_row)
if not plate_label:
plate_label = _normalize_spaces(str(payload.get("plate_arabic") or payload.get("plate") or ""))
if not plate_label:
plate_label = "غير معروف"
_notification_title = ":dollar: Cash Payment Needed"
_notification_body = f"اليوزر {requester_name} عايز يدفع كاش — {plate_label}"
_notify_security_staff(
title=_notification_title,
body=_notification_body,
data={
"session_id": session_id,
"plate": plate_label,
"requester_id": requester_id,
"payment_method": "cash",
},
)
old_notes = session_row.get("notes") or ""
new_notes = old_notes
if "intent_cash" not in old_notes:
new_notes = f"{old_notes} | intent_cash".strip(" |")
update_payload = {"notes": new_notes}
if session_row.get("status") == "paid":
update_payload["status"] = "entered"
update_payload["payment_confirmed_at"] = None
update_payload["notes"] = f"{new_notes} | switched_to_cash".strip(" |")
_update_parking_session(session_id, update_payload)
return {
"status": "notified",
"message": "تم إبلاغ الأمن",
}
@app.post("/gate/decision")
def gate_decision(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
raise HTTPException(status_code=401, detail="Authenticated user id is missing.")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
requester_has_global_scope = _can_view_global_records(requester_role)
session_id = str(payload.get("session_id") or "").strip()
if not session_id:
raise HTTPException(status_code=400, detail="session_id is required.")
plate_detected = bool(payload.get("plate_detected", True))
session_row = _fetch_parking_session_by_id(session_id)
if not isinstance(session_row, dict):
raise HTTPException(status_code=404, detail="parking session was not found.")
owner_id = _get_parking_session_owner_id(session_id)
if not requester_has_global_scope and owner_id and requester_id != owner_id:
raise HTTPException(status_code=403, detail="You can only access your own session gate decision.")
decision = _build_gate_decision(
event_type="exit",
session_row=session_row,
plate_detected=plate_detected,
)
return {
"status": "ok",
"session_id": session_id,
"payment_status": _derive_payment_status_from_session(session_row),
"pricing": _compute_session_time_and_fee(session_row),
"gate_decision": decision,
"requester": {
"id": requester_id,
"role": requester_role or "user",
"is_staff": requester_is_staff,
},
}
@app.post("/payments/paymob/create")
def create_paymob_payment(
payload: Dict[str, Any] = Body(...),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
if not _paymob_configured():
return _error_response(
503,
"Payment gateway not configured",
"PAYMOB_NOT_CONFIGURED",
)
try:
request_user = _require_authenticated_user(authorization)
except HTTPException as exc:
return _error_response(exc.status_code, str(exc.detail), "AUTH_REQUIRED")
if request_user is None:
return _error_response(401, "Authentication is required.", "AUTH_REQUIRED")
requester_id = str(request_user.get("id") or "").strip()
if not requester_id:
return _error_response(401, "Authenticated user id is missing.", "AUTH_REQUIRED")
requester_role = _resolve_requester_role(request_user)
requester_is_staff = _is_staff_role(requester_role)
session_id = str(payload.get("session_id") or "").strip()
if not session_id:
return _error_response(400, "session_id is required.", "SESSION_ID_REQUIRED")
session_row = _fetch_parking_session_by_id(session_id)
if not isinstance(session_row, dict):
return _error_response(404, "parking session was not found.", "SESSION_NOT_FOUND")
parking_location = _normalize_parking_location(payload.get("parking_location"))
pricing = _compute_session_time_and_fee(session_row, parking_location_override=parking_location)
amount_raw = payload.get("amount_cents")
if amount_raw is None:
amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)
else:
try:
amount_cents = int(amount_raw)
except Exception:
return _error_response(400, "amount_cents must be a positive integer.", "INVALID_AMOUNT")
if amount_cents <= 0:
amount_cents = max(int(pricing.get("total_fee_cents") or 0), 1)
currency = str(payload.get("currency") or PAYMOB_CURRENCY).strip().upper() or PAYMOB_CURRENCY
owner_id = _get_parking_session_owner_id(session_id)
if not owner_id:
return _error_response(404, "parking session was not found.", "SESSION_NOT_FOUND")
if not requester_is_staff and requester_id != owner_id:
return _error_response(
403,
"You can only create payments for your own sessions.",
"FORBIDDEN_SESSION",
)
if _parse_iso_datetime(session_row.get("payment_confirmed_at")) is not None:
return _error_response(409, "Session is already paid.", "ALREADY_PAID")
try:
if parking_location:
_set_parking_session_location(session_id, parking_location)
merchant_order_id = f"ain-{session_id[:8]}-{uuid4().hex[:12]}"
checkout = _paymob_create_checkout(
amount_cents=amount_cents,
currency=currency,
merchant_order_id=merchant_order_id,
request_user=request_user,
)
payment_id = _insert_payment_transaction_record(
created_by=requester_id,
session_id=session_id,
amount_cents=amount_cents,
currency=currency,
merchant_order_id=merchant_order_id,
paymob_order_id=checkout.get("paymob_order_id"),
parking_location=parking_location,
request_payload={
"session_id": session_id,
"amount_cents": amount_cents,
"currency": currency,
"merchant_order_id": merchant_order_id,
"parking_location": parking_location,
},
response_payload={
"paymob_order_id": checkout.get("paymob_order_id"),
"iframe_url": checkout.get("iframe_url"),
},
)
return {
"status": "ok",
"provider": "paymob",
"payment_transaction_id": payment_id,
"merchant_order_id": merchant_order_id,
"paymob_order_id": checkout.get("paymob_order_id"),
"session_id": session_id,
"amount_cents": amount_cents,
"currency": currency,
"pricing": pricing,
"parking_location": parking_location,
"payment_url": checkout.get("iframe_url"),
}
except HTTPException as exc:
return _error_response(exc.status_code, str(exc.detail), "PAYMOB_CREATE_FAILED")
except Exception as exc:
return _error_response(502, f"Failed to create Paymob payment: {exc}", "PAYMOB_CREATE_FAILED")
@app.post("/payments/paymob/webhook")
def paymob_webhook(payload: Dict[str, Any] = Body(...)) -> Dict[str, Any]:
obj = payload.get("obj") if isinstance(payload, dict) else None
if not isinstance(obj, dict):
obj = payload if isinstance(payload, dict) else {}
order_obj = obj.get("order") if isinstance(obj.get("order"), dict) else {}
merchant_order_id = str(order_obj.get("merchant_order_id") or payload.get("merchant_order_id") or "").strip()
if not merchant_order_id:
raise HTTPException(status_code=400, detail="merchant_order_id was not found in webhook payload.")
paymob_order_id = str(order_obj.get("id") or obj.get("order_id") or payload.get("order_id") or "").strip() or None
payment_reference = str(obj.get("id") or payload.get("id") or "").strip() or None
is_pending = bool(obj.get("pending"))
is_success = bool(obj.get("success")) and not is_pending
status = "pending"
if is_success:
status = "paid"
elif not is_pending:
status = "failed"
try:
_update_payment_transaction_by_merchant_order_id(
merchant_order_id=merchant_order_id,
status=status,
payment_reference=payment_reference,
paymob_order_id=paymob_order_id,
response_payload=payload if isinstance(payload, dict) else {},
)
session_id = _get_payment_session_id_by_merchant_order_id(merchant_order_id)
if is_success and session_id:
_mark_parking_session_paid(session_id, payment_reference=payment_reference)
return {
"status": "ok",
"payment_status": status,
"merchant_order_id": merchant_order_id,
"paymob_order_id": paymob_order_id,
"session_id": session_id,
}
except HTTPException:
raise
except Exception as exc:
raise HTTPException(status_code=502, detail=f"Failed to process Paymob webhook: {exc}") from exc
@app.post("/admin/maintenance/weekly-refresh")
def admin_weekly_refresh(
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
if request_user is None:
raise HTTPException(status_code=401, detail="Authentication is required.")
requester_id = str(request_user.get("id") or "").strip()
requester_role = _resolve_requester_role(request_user)
if not _is_staff_role(requester_role):
raise HTTPException(status_code=403, detail="Only admin/security can run weekly refresh.")
try:
refresh_result = _run_weekly_refresh_non_core()
return {
"status": "ok",
"requested_by": requester_id,
"requested_by_role": requester_role,
"core_tables_protected": sorted(CORE_TABLES),
"refresh": refresh_result,
}
except Exception as exc:
raise HTTPException(status_code=502, detail=f"Weekly refresh failed: {exc}") from exc
@app.post("/predict")
async def predict(
image: UploadFile = File(...),
event_type: Optional[str] = Form(default="entry"),
parking_location: Optional[str] = Form(default=None),
camera_source: Optional[str] = Form(default=None),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user = _require_authenticated_user(authorization)
normalized_parking_location = _normalize_parking_location(parking_location)
normalized_camera_source = _normalize_camera_source(camera_source)
normalized_event_type = _normalize_event_type(event_type)
if image.content_type and not image.content_type.startswith("image/") and image.content_type != "application/octet-stream":
raise HTTPException(status_code=400, detail="Only image uploads are supported.")
image_bytes = await image.read()
if not image_bytes:
raise HTTPException(status_code=400, detail="Uploaded file is empty.")
# Always use remote HuggingFace Space for inference
response = _run_pipeline_remote(
image_bytes=image_bytes,
filename=image.filename,
content_type=image.content_type,
)
if not isinstance(response, dict):
raise HTTPException(status_code=502, detail="Inference pipeline returned an invalid payload.")
response["inference_provider"] = "remote"
response["filename"] = image.filename
response["event_type"] = normalized_event_type
response["parking_location"] = normalized_parking_location
response["camera_source"] = normalized_camera_source
plate_payload = _build_plate_payload(response.get("plate_info", {}))
response["plate_payload"] = plate_payload
response["plate"] = plate_payload.get("arabic_text") or "N/A"
response["plate_text_ar"] = plate_payload.get("arabic_text") or "N/A"
response["plate_text_en"] = plate_payload.get("english_text") or "N/A"
# Flat plate fields for Flutter
_ar_parts = plate_payload.get("arabic") if isinstance(plate_payload.get("arabic"), dict) else {}
_en_parts = plate_payload.get("english") if isinstance(plate_payload.get("english"), dict) else {}
response["plate_letters_ar"] = _ar_parts.get("letters") or ""
response["plate_numbers_ar"] = _ar_parts.get("numbers") or ""
response["plate_letters_en"] = _en_parts.get("letters") or ""
response["plate_numbers_en"] = _en_parts.get("numbers") or ""
original_b64 = base64.b64encode(image_bytes).decode("utf-8")
response["processed_image_base64"] = str(
(response.get("user_page", {}) or {}).get("split_image_base64") or original_b64
)
response["annotated_image_base64"] = str(
(response.get("admin_page", {}) or {}).get("annotated_image_base64") or original_b64
)
if request_user:
response["request_user"] = {
"id": request_user.get("id"),
"email": request_user.get("email"),
}
response["supabase"] = _persist_prediction_to_supabase(
original_image_bytes=image_bytes,
response_payload=response,
request_user=request_user,
parking_location=normalized_parking_location,
event_type=normalized_event_type,
camera_source=normalized_camera_source,
)
supabase_payload = response.get("supabase") if isinstance(response.get("supabase"), dict) else {}
response["app_registration"] = supabase_payload.get(
"app_registration",
{
"is_registered": False,
"vehicle_id": None,
"owner_id": None,
},
)
response["new_car_alert"] = supabase_payload.get(
"new_car_alert",
{
"is_new_car": None,
"notify_roles": ["admin", "security"],
"action": "lookup_unavailable",
"reason": "supabase_unavailable",
"message": "Could not verify registration status.",
"vehicle_id": None,
"owner_id": None,
"plate_arabic": None,
"plate_english": None,
},
)
response["payment_status"] = supabase_payload.get("payment_status", "unknown")
response["left_within_5_minutes"] = bool(supabase_payload.get("left_within_5_minutes", False))
response["pricing"] = supabase_payload.get(
"pricing",
_compute_session_time_and_fee(None, parking_location_override=normalized_parking_location),
)
response["gate_decision"] = supabase_payload.get(
"gate_decision",
_build_gate_decision(
event_type=normalized_event_type,
session_row=None,
plate_detected=_is_plate_detected(response.get("plate_info", {})),
),
)
response["session_id"] = supabase_payload.get("session_id")
response["event_id"] = supabase_payload.get("event_id")
# === DEBUG LOGS (remove after bugs confirmed fixed) ===
LOGGER.info("=== PREDICT START ===")
LOGGER.info("event_type: %s", normalized_event_type)
LOGGER.info("plate_ar: %s", response.get("plate_text_ar"))
LOGGER.info("plate_en: %s", response.get("plate_text_en"))
LOGGER.info("vehicle_found: %s", response.get("app_registration", {}).get("is_registered"))
LOGGER.info("vehicle_id: %s", response.get("app_registration", {}).get("vehicle_id"))
LOGGER.info("session_id: %s", response.get("session_id"))
LOGGER.info("is_registered: %s", response.get("app_registration", {}).get("is_registered"))
LOGGER.info("=== PREDICT END ===")
return response
@app.post("/developer/predict")
async def developer_predict(
image: UploadFile = File(...),
event_type: Optional[str] = Form(default="entry"),
parking_location: Optional[str] = Form(default=None),
camera_source: Optional[str] = Form(default=None),
authorization: Optional[str] = Header(default=None),
) -> Dict[str, Any]:
request_user, requester_role = _require_developer_user(authorization)
response = await predict(
image=image,
event_type=event_type,
parking_location=parking_location,
camera_source=camera_source,
authorization=authorization,
)
response["developer_test"] = {
"enabled": True,
"request_user": {
"id": request_user.get("id"),
"email": request_user.get("email"),
"role": requester_role,
},
}
return response
# ─── ESP32 Gate Command Queue ────────────────────────────────────
# In-memory command queue for ESP32 gate controllers.
# POST /gate/trigger → queues an "open" command (called by security dashboard)
# GET /gate/poll → ESP32 polls this; returns & clears the pending command
_gate_command_lock = Lock()
_gate_pending_commands: List[Dict[str, Any]] = []
@app.post("/gate/trigger")
def gate_trigger(
payload: Dict[str, Any] = Body(default={}),
) -> Dict[str, Any]:
"""Queue a gate-open command for ESP32 devices to pick up."""
location = str(payload.get("parking_location") or "OPERA").strip().upper()
duration_sec = int(payload.get("duration_seconds") or 10)
plate = str(payload.get("plate") or "").strip()
cmd = {
"action": "open",
"parking_location": location,
"duration_seconds": duration_sec,
"plate": plate,
"queued_at": datetime.now(timezone.utc).isoformat(),
}
with _gate_command_lock:
_gate_pending_commands.append(cmd)
# Keep only last 20 commands to prevent memory leak
if len(_gate_pending_commands) > 20:
_gate_pending_commands.pop(0)
LOGGER.info("Gate trigger queued: %s", cmd)
return {"status": "queued", "command": cmd}
@app.get("/gate/poll")
def gate_poll(
location: str = "OPERA",
) -> Dict[str, Any]:
"""ESP32 polls this endpoint. Returns the oldest pending command and removes it."""
loc = location.strip().upper()
with _gate_command_lock:
for i, cmd in enumerate(_gate_pending_commands):
if cmd.get("parking_location") == loc:
_gate_pending_commands.pop(i)
return {"status": "command", "command": cmd}
return {"status": "idle"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"API:app",
host=os.getenv("API_HOST", "0.0.0.0"),
port=int(os.getenv("PORT", os.getenv("API_PORT", "8000"))),
reload=False,
)