Spaces:
Running on Zero
Running on Zero
| import hashlib | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, List, Optional, Tuple | |
| from fastapi import APIRouter, 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, SUPPORTED_EVENT_TYPES, | |
| supabase_configured, is_staff_role, can_view_global_records, LOGGER, | |
| ) | |
| from app.security.auth import require_authenticated_user, _supabase_headers, _normalize_spaces | |
| from app.api.helpers import _normalize_location_key_for_config | |
| from app.api.parking import compute_session_time_and_fee, _parse_iso_datetime, _fetch_vehicle_ids_for_owner, _build_parking_locations_snapshot | |
| from app.api.predict import _normalize_plate_text | |
| TZ_UTC = timezone.utc | |
| router = APIRouter(tags=["events"]) | |
| 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_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 _derive_payment_status_from_session(session_row: Optional[Dict[str, Any]]) -> str: | |
| if not isinstance(session_row, dict): | |
| return "no_active_session" | |
| payment_confirmed_at = session_row.get("payment_confirmed_at") | |
| check_out_at = session_row.get("check_out_at") | |
| left_within_5 = bool(session_row.get("left_within_5_minutes")) | |
| status_value = str(session_row.get("status") or "").strip().lower() | |
| if payment_confirmed_at is None and check_out_at is None: | |
| return "unpaid" | |
| if payment_confirmed_at is not None and check_out_at is None: | |
| if status_value == "overstayed": | |
| return "paid_grace_period_expired" | |
| return "paid_waiting_exit" | |
| if payment_confirmed_at is None and check_out_at is not None: | |
| return "left_without_payment" | |
| if left_within_5: | |
| return "left_within_5_minutes_after_payment" | |
| return "left_after_5_minutes_after_payment" | |
| def _build_gate_decision(*, event_type: str, session_row: Optional[Dict[str, Any]], plate_detected: bool) -> Dict[str, Any]: | |
| if event_type != "exit": | |
| return {"allowed": True, "action": "open_gate", "reason": "entry_access_granted"} | |
| if not plate_detected: | |
| return {"allowed": False, "action": "manual_open_only", "reason": "plate_not_detected"} | |
| payment_status = _derive_payment_status_from_session(session_row) | |
| if payment_status in {"paid_waiting_exit", "left_within_5_minutes_after_payment"}: | |
| return {"allowed": True, "action": "open_gate", "reason": "paid_within_grace_period", "payment_status": payment_status} | |
| if payment_status == "paid_grace_period_expired": | |
| return {"allowed": False, "action": "deny_and_alert_security", "reason": "grace_period_expired", "payment_status": payment_status} | |
| if payment_status == "no_active_session": | |
| return {"allowed": False, "action": "manual_review_required", "reason": "no_active_session", "payment_status": payment_status} | |
| return {"allowed": False, "action": "deny_gate", "reason": "payment_not_confirmed", "payment_status": payment_status} | |
| 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_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 _update_parking_session(session_id: str, patch_payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: | |
| if not supabase_configured() or not session_id: | |
| return None | |
| response = requests.patch( | |
| f"{SUPABASE_URL}/rest/v1/parking_sessions", | |
| params={"id": f"eq.{session_id}"}, | |
| json=patch_payload, | |
| 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 RuntimeError(f"Failed to update parking session ({response.status_code}): {response.text[:300]}") | |
| if response.status_code == 204: | |
| return None | |
| rows = response.json() | |
| if isinstance(rows, list) and rows and isinstance(rows[0], dict): | |
| return rows[0] | |
| return None | |
| 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=TZ_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 | |
| 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 = (request_user.get("role") or "").strip().lower() | |
| 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, | |
| } | |
| 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 = (request_user.get("role") or "").strip().lower() | |
| 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, | |
| } | |
| 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 = (request_user.get("role") or "").strip().lower() | |
| 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, | |
| } | |
| 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 = (request_user.get("role") or "").strip().lower() | |
| 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, | |
| } | |
| 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 = (request_user.get("role") or "").strip().lower() | |
| 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() | |
| 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(TZ_UTC).isoformat(), "notes": "Closed invalid ghost session automatically."}) | |
| continue | |
| latest_event = latest_event_by_session.get(session_id) | |
| 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_norm = _normalize_plate_text(s.get("plate_arabic")) | |
| plate_en_norm = _normalize_plate_text(s.get("plate_english")) | |
| if plate_ar_norm: | |
| real_session_plates.add(plate_ar_norm) | |
| if plate_en_norm: | |
| real_session_plates.add(plate_en_norm) | |
| 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=TZ_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, | |
| } | |