Spaces:
Running on Zero
Running on Zero
File size: 9,212 Bytes
34a66f3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 | import json
import logging
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional, Any, Tuple
from uuid import uuid4
import requests
from fastapi import HTTPException
from app.core.config import (
SUPABASE_URL, SUPABASE_ANON_KEY, SUPABASE_SERVICE_ROLE_KEY,
SUPABASE_TIMEOUT_SECONDS, SUPABASE_RAW_BUCKET, SUPABASE_PROCESSED_BUCKET,
supabase_configured, PROFILE_SELECT_FIELDS, VEHICLE_SELECT_FIELDS,
CORE_TABLES, STAFF_ROLES, PHONE_PATTERN, USERNAME_PATTERN, NATIONAL_ID_PATTERN,
INTERNAL_LOGIN_EMAIL_SUFFIX, LOGGER,
)
from app.security.auth import (
_supabase_headers, _extract_response_error_message, _normalize_spaces,
is_valid_redirect_url, MOBILE_REDIRECT_SCHEME_PATTERN,
)
TZ_UTC = timezone.utc
def supabase_get_profile_by_user_id(user_id: Optional[str]) -> Optional[Dict[str, Any]]:
if not user_id or not supabase_configured():
return None
response = requests.get(
f"{SUPABASE_URL}/rest/v1/profiles",
params={"select": PROFILE_SELECT_FIELDS, "id": f"eq.{user_id}", "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:
return None
rows = response.json()
if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
return None
return rows[0]
def supabase_get_profile_by_field(field: str, value: str, *, case_insensitive: bool = False) -> Optional[Dict[str, Any]]:
if not supabase_configured() or not value:
return None
operator = "ilike" if case_insensitive else "eq"
response = requests.get(
f"{SUPABASE_URL}/rest/v1/profiles",
params={"select": PROFILE_SELECT_FIELDS, field: f"{operator}.{value}", "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:
return None
rows = response.json()
if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
return None
return rows[0]
def supabase_find_profile_by_identifier(identifier: Optional[str]) -> Optional[Dict[str, Any]]:
if not supabase_configured():
return None
normalized = _normalize_spaces(str(identifier or ""))
if not normalized:
return None
lookup_chain: List[Tuple[str, str, bool]] = []
if "@" in normalized:
lookup_chain.append(("email", normalized.lower(), True))
lookup_chain.extend([
("username", normalized, False),
("phone_number", normalized, False),
("national_id", normalized, False),
])
if "@" not in normalized:
lookup_chain.append(("email", normalized.lower(), True))
seen = set()
for field_name, field_value, ci in lookup_chain:
key = (field_name, field_value, ci)
if key in seen:
continue
seen.add(key)
profile = supabase_get_profile_by_field(field_name, field_value, case_insensitive=ci)
if isinstance(profile, dict):
return profile
return None
def supabase_password_login(*, password: str, email: Optional[str] = None, phone_number: Optional[str] = None) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
if not supabase_configured():
return None, "Supabase is not configured."
login_payload: Dict[str, str] = {"password": password}
if email:
login_payload["email"] = email.strip().lower()
elif phone_number:
login_payload["phone"] = phone_number.strip()
else:
return None, "No login identifier was provided."
try:
response = requests.post(
f"{SUPABASE_URL}/auth/v1/token?grant_type=password",
json=login_payload,
headers={"apikey": SUPABASE_ANON_KEY, "Content-Type": "application/json"},
timeout=SUPABASE_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
return None, str(exc)
if response.status_code == 200:
payload = response.json()
if isinstance(payload, dict):
return payload, None
return None, "Unexpected response payload."
error_msg = _extract_response_error_message(response) or f"Login failed with status {response.status_code}."
return None, error_msg
def supabase_list_buckets() -> List[str]:
if not supabase_configured():
return []
response = requests.get(
f"{SUPABASE_URL}/storage/v1/bucket",
headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code != 200:
raise RuntimeError(f"Bucket list failed ({response.status_code})")
rows = response.json()
if not isinstance(rows, list):
return []
return [str(row.get("id")) for row in rows if isinstance(row, dict) and row.get("id")]
def supabase_table_exists(table_name: str) -> Tuple[bool, Optional[str]]:
if not supabase_configured():
return False, "Supabase not configured"
try:
response = requests.get(
f"{SUPABASE_URL}/rest/v1/{table_name}",
params={"select": "*", "limit": "1"},
headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
except requests.RequestException as exc:
return False, str(exc)
if response.status_code == 200:
return True, None
return False, f"HTTP {response.status_code}: {response.text[:220]}"
def upload_to_supabase_storage(bucket: str, object_path: str, content: bytes, content_type: str = "image/jpeg") -> None:
if not supabase_configured():
raise RuntimeError("Supabase is not configured.")
endpoint = f"{SUPABASE_URL}/storage/v1/object/{bucket}/{object_path}"
headers = _supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type=content_type)
headers["x-upsert"] = "true"
response = requests.post(endpoint, data=content, headers=headers, timeout=SUPABASE_TIMEOUT_SECONDS)
if response.status_code not in {200, 201}:
raise RuntimeError(f"Storage upload failed ({response.status_code}): {response.text[:400]}")
def delete_from_supabase_storage(bucket: str, object_path: str) -> None:
if not supabase_configured() or not bucket or not object_path:
return
endpoint = f"{SUPABASE_URL}/storage/v1/object/{bucket}/{object_path}"
response = requests.delete(endpoint, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY), timeout=SUPABASE_TIMEOUT_SECONDS)
if response.status_code not in {200, 204, 404}:
raise RuntimeError(f"Storage delete failed ({response.status_code}): {response.text[:400]}")
def ensure_profile_row_exists(auth_user: Dict[str, Any]) -> None:
if not supabase_configured() or not isinstance(auth_user, dict):
return
user_id = str(auth_user.get("id") or "").strip()
if not user_id:
return
if supabase_get_profile_by_user_id(user_id):
return
email = _normalize_spaces(str(auth_user.get("email") or "")).lower()
payload: Dict[str, Any] = {"id": user_id}
if email:
payload["email"] = email
try:
response = requests.post(
f"{SUPABASE_URL}/rest/v1/profiles",
json=payload,
headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json", prefer="resolution=merge-duplicates,return=minimal"),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
except requests.RequestException:
pass
def upsert_profile(*, user_id: str, payload: Dict[str, Any]) -> Optional[Dict[str, Any]]:
if not user_id:
return None
allowed_keys = {"id", "first_name", "last_name", "full_name", "username", "phone_number", "email", "national_id", "role", "staff_id"}
upsert_payload: Dict[str, Any] = {"id": user_id}
for key in allowed_keys:
if key == "id":
continue
value = payload.get(key)
if value is None:
continue
upsert_payload[key] = _normalize_spaces(str(value)) if isinstance(value, str) else value
if len(upsert_payload) == 1:
return supabase_get_profile_by_user_id(user_id)
response = requests.post(
f"{SUPABASE_URL}/rest/v1/profiles",
json=upsert_payload,
headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json", prefer="resolution=merge-duplicates,return=representation"),
timeout=SUPABASE_TIMEOUT_SECONDS,
)
if response.status_code not in {200, 201}:
raise HTTPException(status_code=502, detail=f"Failed to upsert profile ({response.status_code})")
rows = response.json() if response.text else []
if isinstance(rows, list) and rows and isinstance(rows[0], dict):
return rows[0]
return supabase_get_profile_by_user_id(user_id)
|