import math from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple from uuid import uuid4 from fastapi import APIRouter, Body, Header, HTTPException import requests from app.core.config import ( SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, SUPABASE_TIMEOUT_SECONDS, PARKING_HOURLY_RATE_EGP, PARKING_BILLING_MODE, PARKING_DAILY_RATE_EGP, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP, PARKING_LOCATION_HOURLY_RATES, PARKING_LOCATION_CAPACITIES, PARKING_LOCATION_PRICING, GARAGE_TOTAL_CAPACITY, APP_SERVICE_FEE_EGP, supabase_configured, is_staff_role, can_view_global_records, LOGGER, ) from app.security.auth import require_authenticated_user, require_developer_user, _supabase_headers, _normalize_spaces from app.services.supabase_client import supabase_get_profile_by_user_id from app.api.helpers import _normalize_parking_location, _normalize_location_key_for_config, _split_arabic_plate, _split_english_plate, _compose_full_name TZ_UTC = timezone.utc router = APIRouter(tags=["parking"]) def _parse_iso_datetime(value: Optional[Any]) -> Optional[datetime]: if value is None: return None raw = str(value).strip() if not raw: return None try: parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=TZ_UTC) return parsed.astimezone(TZ_UTC) def _resolve_location_pricing_policy(parking_location: Optional[Any]) -> Dict[str, Any]: candidates: List[str] = [] if parking_location is not None: raw = str(parking_location).strip() if raw: for part in raw.split(","): location_key = _normalize_location_key_for_config(part) if location_key: candidates.append(location_key) for location_key in candidates: location_policy = PARKING_LOCATION_PRICING.get(location_key) if isinstance(location_policy, dict): return { "pricing_location": location_key, "billing_mode": location_policy.get("billing_mode", "hourly"), "base_rate_egp": max(0.0, float(location_policy.get("base_rate_egp", 0.0))), "extra_after_first_hour_egp": max(0.0, float(location_policy.get("extra_after_first_hour_egp", 0.0))), "rate_source": "location_pricing", "location_capacity": max(0, int(location_policy.get("capacity", GARAGE_TOTAL_CAPACITY))), } hourly_rate_override = PARKING_LOCATION_HOURLY_RATES.get(location_key) if hourly_rate_override is not None: cap = PARKING_LOCATION_CAPACITIES.get(location_key, GARAGE_TOTAL_CAPACITY) return { "pricing_location": location_key, "billing_mode": "hourly", "base_rate_egp": max(0.0, float(hourly_rate_override)), "extra_after_first_hour_egp": max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP), "rate_source": "location_hourly_legacy", "location_capacity": max(0, int(cap)), } fallback_location = candidates[0] if candidates else None default_base = PARKING_HOURLY_RATE_EGP if PARKING_BILLING_MODE == "hourly" else PARKING_DAILY_RATE_EGP return { "pricing_location": fallback_location, "billing_mode": PARKING_BILLING_MODE, "base_rate_egp": max(0.0, default_base), "extra_after_first_hour_egp": max(0.0, PARKING_EXTRA_AFTER_FIRST_HOUR_EGP), "rate_source": "default", "location_capacity": GARAGE_TOTAL_CAPACITY, } def _compute_parking_fee_cents(*, duration_seconds: float, billing_mode: str, base_rate_egp: float, extra_after_first_hour_egp: float) -> Tuple[int, int, int]: if duration_seconds <= 0: return 0, 0, 0 base_rate_cents = max(0, int(round(max(0.0, base_rate_egp) * 100.0))) extra_rate_cents = max(0, int(round(max(0.0, extra_after_first_hour_egp) * 100.0))) base_window = 86400 if billing_mode == "daily" else 3600 extra_hours = 0 if duration_seconds > base_window and extra_rate_cents > 0: extra_hours = int(math.ceil((duration_seconds - base_window) / 3600.0)) fee = base_rate_cents + (extra_hours * extra_rate_cents) return max(0, fee), extra_hours, base_window // 3600 def compute_session_time_and_fee(session_row: Optional[Dict[str, Any]], now_utc: Optional[datetime] = None, parking_location_override: Optional[str] = None) -> Dict[str, Any]: reference_now = now_utc or datetime.now(TZ_UTC) location_for_pricing = parking_location_override if isinstance(session_row, dict): location_for_pricing = session_row.get("parking_location") or parking_location_override pricing_policy = _resolve_location_pricing_policy(location_for_pricing) billing_mode = str(pricing_policy.get("billing_mode") or "hourly") base_rate_egp = max(0.0, float(pricing_policy.get("base_rate_egp") or 0.0)) extra_after_first_hour_egp = max(0.0, float(pricing_policy.get("extra_after_first_hour_egp") or 0.0)) hourly_rate_egp = base_rate_egp if billing_mode == "hourly" else round(base_rate_egp / 24.0, 2) app_fee_egp = max(0.0, APP_SERVICE_FEE_EGP) app_fee_cents = int(round(app_fee_egp * 100.0)) result = { "pricing_location": pricing_policy.get("pricing_location"), "rate_source": pricing_policy.get("rate_source"), "billing_mode": billing_mode, "base_rate_egp": round(base_rate_egp, 2), "extra_after_first_hour_egp": round(extra_after_first_hour_egp, 2), "location_capacity": max(0, int(pricing_policy.get("location_capacity") or 0)), "base_window_hours": 24 if billing_mode == "daily" else 1, "hourly_rate_egp": round(hourly_rate_egp, 2), "app_fee_egp": round(app_fee_egp, 2), "app_fee_cents": app_fee_cents, "duration_minutes": 0.0, "extra_hours_billed": 0, "parking_fee_egp": 0.0, "parking_fee_cents": 0, "total_fee_egp": 0.0, "total_fee_cents": 0, "billing_reference_time": reference_now.isoformat(), } if not isinstance(session_row, dict): return result check_in_dt = _parse_iso_datetime(session_row.get("check_in_at")) if check_in_dt is None: return result check_out_dt = _parse_iso_datetime(session_row.get("check_out_at")) end_dt = check_out_dt or reference_now duration_seconds = max(0.0, (end_dt - check_in_dt).total_seconds()) duration_minutes = round(duration_seconds / 60.0, 2) fee_cents, extra_hours, base_window_hours = _compute_parking_fee_cents( duration_seconds=duration_seconds, billing_mode=billing_mode, base_rate_egp=base_rate_egp, extra_after_first_hour_egp=extra_after_first_hour_egp, ) total_fee_cents = fee_cents + app_fee_cents result["duration_minutes"] = duration_minutes result["extra_hours_billed"] = extra_hours result["base_window_hours"] = base_window_hours result["parking_fee_cents"] = fee_cents result["parking_fee_egp"] = round(fee_cents / 100.0, 2) result["total_fee_cents"] = total_fee_cents result["total_fee_egp"] = round(total_fee_cents / 100.0, 2) return result 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: return 0, {} rows = response.json() if not isinstance(rows, list): return 0, {} by_loc: Dict[str, int] = {} total = 0 for row in rows: if not isinstance(row, dict): continue total += 1 loc = _normalize_location_key_for_config(str(row.get("parking_location") or "")) if loc: by_loc[loc] = by_loc.get(loc, 0) + 1 return total, by_loc def _build_parking_locations_snapshot() -> Dict[str, Any]: total_inside, inside_by_location = _fetch_active_inside_counts_by_location() 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) cap = max(0, int(policy.get("location_capacity") or 0)) inside = inside_by_location.get(location_key, 0) left = max(0, cap - inside) 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, "total_capacity": cap, "left": left, "display": f"{inside}/{cap}"}, }) unknown_inside = max(0, total_inside - sum(inside_by_location.values())) garage_left = max(0, GARAGE_TOTAL_CAPACITY - total_inside) return { "garage": {"inside": total_inside, "total_capacity": GARAGE_TOTAL_CAPACITY, "left": garage_left, "display": f"{total_inside}/{GARAGE_TOTAL_CAPACITY}", "unknown_location_inside": unknown_inside}, "locations": locations, } @router.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() requester_role = (request_user.get("role") or "").strip().lower() snapshot = _build_parking_locations_snapshot() return { "status": "ok", "requester": {"id": requester_id, "role": requester_role or "user", "is_staff": is_staff_role(requester_role)}, "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"], } @router.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.") snapshot = _build_parking_locations_snapshot() return { "status": "ok", "garage": snapshot["garage"], "locations": [{"parking_location": row.get("parking_location"), "occupancy": row.get("occupancy")} for row in snapshot["locations"]], } @router.get("/parking/history") def parking_history(limit: int = 100, for_user_id: Optional[str] = None, 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 = (request_user.get("role") or "").strip().lower() requester_has_global = 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 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: 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", "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="Failed to fetch parking history.") sessions = sessions_response.json() if isinstance(sessions_response.json(), list) else [] history = [{"session": s, "pricing": compute_session_time_and_fee(s)} for s in sessions if isinstance(s, dict)] return {"status": "ok", "history": history[:limit]} def _fetch_vehicle_ids_for_owner(owner_id: str) -> List[str]: if not owner_id or not supabase_configured(): return [] response = requests.get( f"{SUPABASE_URL}/rest/v1/vehicles", params={"select": "id", "owner_id": f"eq.{owner_id}"}, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY), timeout=SUPABASE_TIMEOUT_SECONDS, ) if response.status_code != 200: return [] rows = response.json() if not isinstance(rows, list): return [] return [str(r["id"]) for r in rows if isinstance(r, dict) and r.get("id")]