from config import supabase from models import ( City, CityCreate, CityUpdate, District, DistrictCreate, DistrictUpdate, Street, StreetCreate, StreetUpdate, Position, PositionCreate, PositionUpdate, Group, GroupCreate, GroupUpdate, ) from typing import List, Optional import unicodedata def _next_id(table: str, id_col: str = "id") -> int: """Get max(id) + 1 from a table to avoid identity sequence conflicts.""" result = supabase.table(table).select(id_col).order(id_col, desc=True).limit(1).execute() max_id = result.data[0][id_col] if result.data else 0 return max_id + 1 class MapsService: @staticmethod def _db_pos_to_model(item: dict) -> dict: if not item: return {} return { 'id': item.get('id'), 'x': item.get('long'), 'y': item.get('lat'), 'ban': item.get('ban'), 'speed': item.get('speed'), 'park': item.get('park'), 'lane': item.get('lane'), 'tool': item.get('tool'), 'flooding': item.get('flooding'), 'streets': item.get('streets'), 'ban_turn': item.get('ban_turn'), 'ban_vehicle': item.get('ban_vehicle'), 'ban_hours': item.get('ban_hours'), 'ban_weight': item.get('ban_weight'), } @staticmethod def _flatten_positions(raw) -> list: """Streets.positions có thể là 2-D ([[id,...], ...]) hoặc 1-D cũ ([id,...]). Trả về danh sách id dương (bỏ dấu 1 chiều) đã phẳng, giữ thứ tự, không trùng.""" out = [] seen = set() for item in (raw or []): group = item if isinstance(item, list) else [item] for p in group: try: pid = abs(int(p)) except (TypeError, ValueError): continue if pid not in seen: seen.add(pid) out.append(pid) return out @staticmethod def _append_pos_id(raw, pid: int): """Thêm 1 id điểm vào Streets.positions (cấu trúc 2-D). Không thêm nếu đã tồn tại. - 2-D: append vào cạnh cuối cùng. - 1-D cũ: giữ nguyên dạng 1-D và append. - rỗng: tạo cạnh đầu tiên [[pid]].""" edges = list(raw or []) if pid in MapsService._flatten_positions(edges): return edges if edges and all(isinstance(e, list) for e in edges): edges[-1] = list(edges[-1]) + [pid] return edges if edges: # 1-D legacy return edges + [pid] return [[pid]] @staticmethod def _remove_pos_id(raw, pid: int): """Xoá id điểm khỏi Streets.positions (cả 2-D và 1-D), so khớp theo giá trị tuyệt đối. Bỏ các cạnh rỗng sau khi xoá.""" edges = list(raw or []) target = abs(int(pid)) if edges and all(isinstance(e, list) for e in edges): new = [] for e in edges: filt = [i for i in e if abs(int(i)) != target] if filt: new.append(filt) return new return [i for i in edges if abs(int(i)) != target] @staticmethod async def get_all_cities() -> List[City]: try: result = supabase.table("Cities").select("id, name").order("name").execute() return [City(**item) for item in result.data] except Exception as e: raise Exception(f"Failed to fetch cities: {str(e)}") @staticmethod async def create_city(data: CityCreate) -> City: try: next_id = _next_id("Cities") result = supabase.table("Cities").insert({"id": next_id, "name": data.name}).execute() return City(**result.data[0]) except Exception as e: raise Exception(f"Failed to create city: {str(e)}") @staticmethod async def get_city_by_id(city_id: int) -> Optional[City]: try: result = supabase.table("Cities").select("id, name").eq("id", city_id).execute() return City(**result.data[0]) if result.data else None except Exception as e: raise Exception(f"Failed to fetch city: {str(e)}") @staticmethod async def update_city(city_id: int, data: CityUpdate) -> City: try: payload = {k: v for k, v in data.model_dump().items() if v is not None} result = supabase.table("Cities").update(payload).eq("id", city_id).execute() return City(**result.data[0]) except Exception as e: raise Exception(f"Failed to update city: {str(e)}") @staticmethod async def delete_city(city_id: int) -> bool: try: supabase.table("Cities").delete().eq("id", city_id).execute() return True except Exception as e: raise Exception(f"Failed to delete city: {str(e)}") @staticmethod async def get_districts(city_id: Optional[int] = None) -> List[District]: try: query = supabase.table("Districts").select("id, name, lname, city").order("name") if city_id is not None: query = query.eq("city", city_id) result = query.execute() return [District(**item) for item in result.data] except Exception as e: raise Exception(f"Failed to fetch districts: {str(e)}") @staticmethod async def create_district(data: DistrictCreate) -> District: try: next_id = _next_id("Districts") payload = {"id": next_id, "name": data.name, "city": data.city} if data.lname: payload["lname"] = data.lname result = supabase.table("Districts").insert(payload).execute() return District(**result.data[0]) except Exception as e: raise Exception(f"Failed to create district: {str(e)}") @staticmethod async def get_district_by_id(district_id: int) -> Optional[District]: try: result = supabase.table("Districts").select("id, name, lname, city").eq("id", district_id).execute() return District(**result.data[0]) if result.data else None except Exception as e: raise Exception(f"Failed to fetch district: {str(e)}") @staticmethod async def update_district(district_id: int, data: DistrictUpdate) -> District: try: payload = {k: v for k, v in data.model_dump().items() if v is not None} result = supabase.table("Districts").update(payload).eq("id", district_id).execute() return District(**result.data[0]) except Exception as e: raise Exception(f"Failed to update district: {str(e)}") @staticmethod async def delete_district(district_id: int) -> bool: try: supabase.table("Districts").delete().eq("id", district_id).execute() return True except Exception as e: raise Exception(f"Failed to delete district: {str(e)}") @staticmethod async def get_streets(district_id: Optional[int] = None, city_id: Optional[int] = None) -> List[Street]: try: query = supabase.table("Streets").select("id, name, type, district_id, city_id").order("name") if district_id is not None: query = query.eq("district_id", district_id) elif city_id is not None: query = query.eq("city_id", city_id) result = query.execute() return [Street(**item) for item in result.data] except Exception as e: raise Exception(f"Failed to fetch streets: {str(e)}") @staticmethod async def create_street(data: StreetCreate) -> Street: try: # Normalize helper (remove diacritics, lower, collapse whitespace) import re def _normalize(s: str) -> str: if not s: return '' s = str(s).strip().lower() # NFKD to decompose accents, then remove combining marks s = unicodedata.normalize('NFKD', s) s = ''.join(ch for ch in s if not unicodedata.combining(ch)) # remove punctuation, keep letters/numbers and spaces s = re.sub(r'[^0-9a-z\s]', ' ', s) s = ' '.join(s.split()) return s def _levenshtein(a: str, b: str) -> int: # simple iterative DP if a == b: return 0 if len(a) == 0: return len(b) if len(b) == 0: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, start=1): curr = [i] for j, cb in enumerate(b, start=1): insert_cost = curr[j-1] + 1 delete_cost = prev[j] + 1 replace_cost = prev[j-1] + (0 if ca == cb else 1) curr.append(min(insert_cost, delete_cost, replace_cost)) prev = curr return prev[-1] name_norm = _normalize(data.name) # Only check duplicates WITHIN THE SAME district -> cho phép cùng tên đường ở phường khác. # Chỉ chặn khi trùng tên VÀ trùng district_id. existing_res = supabase.table("Streets").select("id, name, district_id, city_id").eq("district_id", data.district_id).order("name").execute() existing_list = existing_res.data or [] for item in existing_list: existing_name = item.get('name') or '' en_norm = _normalize(existing_name) if not en_norm: continue # exact or containment match if en_norm == name_norm or (name_norm and (name_norm in en_norm or en_norm in name_norm)): raise Exception(f"DUPLICATE_STREET:{item.get('id')}:{existing_name}") # edit distance threshold scales with length max_dist = max(2, int(max(len(en_norm), len(name_norm)) * 0.2)) dist = _levenshtein(en_norm, name_norm) if dist <= max_dist: raise Exception(f"DUPLICATE_STREET:{item.get('id')}:{existing_name}") # No similar found -> proceed to insert next_id = _next_id("Streets") payload = {"id": next_id, "name": data.name, "district_id": data.district_id, "city_id": data.city_id} if data.type: payload["type"] = data.type result = supabase.table("Streets").insert(payload).execute() return Street(**result.data[0]) except Exception as e: raise Exception(f"Failed to create street: {str(e)}") @staticmethod async def get_street_by_id(street_id: int) -> Optional[Street]: try: result = supabase.table("Streets").select("*").eq("id", street_id).execute() return Street(**result.data[0]) if result.data else None except Exception as e: raise Exception(f"Failed to fetch street: {str(e)}") @staticmethod async def update_street(street_id: int, data: StreetUpdate) -> Street: try: raw = {k: v for k, v in data.model_dump().items() if v is not None} optional = {"toll", "flooding"} core = {k: v for k, v in raw.items() if k not in optional} try: result = supabase.table("Streets").update(raw).eq("id", street_id).execute() except Exception: # toll/flooding columns may not exist yet → save core fields only result = supabase.table("Streets").update(core or raw).eq("id", street_id).execute() if not result.data: raise Exception(f"Street #{street_id} not found or update failed") return Street(**result.data[0]) except Exception as e: raise Exception(f"Failed to update street: {str(e)}") @staticmethod async def delete_street(street_id: int) -> bool: try: supabase.table("Streets").delete().eq("id", street_id).execute() return True except Exception as e: raise Exception(f"Failed to delete street: {str(e)}") @staticmethod async def search_streets_by_text(text: str, district_id: Optional[int] = None, city_id: Optional[int] = None, limit: int = 10) -> list: """Search streets by text with normalization and fuzzy matching. Returns list of dicts: {id, name, type, district_id, city_id, score} Lower score == better match. """ try: if not text or not str(text).strip(): return [] import re def _normalize(s: str) -> str: s = str(s or '').strip().lower() s = unicodedata.normalize('NFKD', s) s = ''.join(ch for ch in s if not unicodedata.combining(ch)) s = re.sub(r'[^0-9a-z\s]', ' ', s) s = ' '.join(s.split()) return s def _lev(a: str, b: str) -> int: if a == b: return 0 if len(a) == 0: return len(b) if len(b) == 0: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, start=1): curr = [i] for j, cb in enumerate(b, start=1): insert_cost = curr[j-1] + 1 delete_cost = prev[j] + 1 replace_cost = prev[j-1] + (0 if ca == cb else 1) curr.append(min(insert_cost, delete_cost, replace_cost)) prev = curr return prev[-1] q = _normalize(text) # Fetch candidate streets filtered by district or city query = supabase.table("Streets").select("id, name, type, district_id, city_id").order("name") if district_id is not None: query = query.eq("district_id", district_id) elif city_id is not None: query = query.eq("city_id", city_id) res = query.execute() candidates = res.data or [] scored = [] for item in candidates: name = item.get('name') or '' norm = _normalize(name) score = None if norm == q: score = 0.0 elif q in norm or norm in q: score = 0.1 else: dist = _lev(q, norm) max_len = max(len(q), len(norm), 1) # normalized distance score = dist / max_len scored.append({ 'id': item.get('id'), 'name': name, 'type': item.get('type'), 'district_id': item.get('district_id'), 'city_id': item.get('city_id'), 'score': round(float(score), 3) }) # sort by score asc, then by name scored.sort(key=lambda x: (x['score'], x['name'])) return scored[:limit] except Exception as e: raise Exception(f"Failed to search streets: {str(e)}") @staticmethod async def search_districts_by_text(text: str, city_id: Optional[int] = None, limit: int = 10) -> list: """Search districts by text with normalization and fuzzy matching. Matches against both `name` and `lname` (best score wins). Returns list of dicts: {id, name, lname, city, score} Lower score == better match. """ try: if not text or not str(text).strip(): return [] import re def _normalize(s: str) -> str: s = str(s or '').strip().lower() s = unicodedata.normalize('NFKD', s) s = ''.join(ch for ch in s if not unicodedata.combining(ch)) s = re.sub(r'[^0-9a-z\s]', ' ', s) s = ' '.join(s.split()) return s def _lev(a: str, b: str) -> int: if a == b: return 0 if len(a) == 0: return len(b) if len(b) == 0: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, start=1): curr = [i] for j, cb in enumerate(b, start=1): insert_cost = curr[j-1] + 1 delete_cost = prev[j] + 1 replace_cost = prev[j-1] + (0 if ca == cb else 1) curr.append(min(insert_cost, delete_cost, replace_cost)) prev = curr return prev[-1] def _score(query_norm: str, cand_norm: str) -> float: if not cand_norm: return 1.0 if cand_norm == query_norm: return 0.0 if query_norm in cand_norm or cand_norm in query_norm: return 0.1 dist = _lev(query_norm, cand_norm) max_len = max(len(query_norm), len(cand_norm), 1) return dist / max_len q = _normalize(text) # Fetch candidate districts, optionally filtered by city query = supabase.table("Districts").select("id, name, lname, city").order("name") if city_id is not None: query = query.eq("city", city_id) res = query.execute() candidates = res.data or [] scored = [] for item in candidates: name = item.get('name') or '' lname = item.get('lname') or '' score = _score(q, _normalize(name)) if lname: score = min(score, _score(q, _normalize(lname))) scored.append({ 'id': item.get('id'), 'name': name, 'lname': item.get('lname'), 'city': item.get('city'), 'score': round(float(score), 3) }) # sort by score asc, then by name scored.sort(key=lambda x: (x['score'], x['name'])) return scored[:limit] except Exception as e: raise Exception(f"Failed to search districts: {str(e)}") # ==================== POSITIONS ==================== @staticmethod async def get_positions(street_id: Optional[int] = None) -> List[Position]: try: if street_id is not None: # Filter directly on Positions.streets JSON array result = supabase.table("Positions").select("*").contains("streets", [street_id]).order("id").execute() else: result = supabase.table("Positions").select("*").order("id").execute() return [Position(**MapsService._db_pos_to_model(item)) for item in (result.data or [])] except Exception as e: raise Exception(f"Failed to fetch positions: {str(e)}") @staticmethod async def get_position_by_id(position_id: int) -> Optional[Position]: try: result = supabase.table("Positions").select("*").eq("id", position_id).execute() if result.data: return Position(**MapsService._db_pos_to_model(result.data[0])) return None except Exception as e: raise Exception(f"Failed to fetch position: {str(e)}") @staticmethod async def create_position(data: PositionCreate) -> Position: try: next_id = _next_id("Positions") # Map internal x/y to DB columns long/lat payload = {"id": next_id, "long": data.x, "lat": data.y} # include streets JSON if provided on model if getattr(data, 'street_id', None): payload['streets'] = [data.street_id] for field in ("ban", "speed", "park", "lane", "tool", "flooding", "ban_turn", "ban_vehicle", "ban_hours", "ban_weight"): val = getattr(data, field) if val is not None: payload[field] = val result = supabase.table("Positions").insert(payload).execute() if not result.data: raise Exception("Failed to insert position") pos = result.data[0] # append to Streets.positions JSON column if street_id provided if getattr(data, 'street_id', None): try: street_row = supabase.table("Streets").select("positions").eq("id", data.street_id).limit(1).execute() raw = street_row.data[0].get('positions') if street_row.data else None existing = MapsService._append_pos_id(raw, pos['id']) supabase.table("Streets").update({"positions": existing}).eq("id", data.street_id).execute() except Exception: pass return Position(**MapsService._db_pos_to_model(pos)) except Exception as e: raise Exception(f"Failed to create position: {str(e)}") @staticmethod async def update_position(position_id: int, data: PositionUpdate) -> Position: try: raw = {k: v for k, v in data.model_dump().items() if v is not None} # Map model fields -> DB columns (x->long, y->lat); 'street_id' is not a column col_map = {'x': 'long', 'y': 'lat'} payload = {col_map.get(k, k): v for k, v in raw.items() if k != 'street_id'} result = supabase.table("Positions").update(payload).eq("id", position_id).execute() return Position(**MapsService._db_pos_to_model(result.data[0])) except Exception as e: raise Exception(f"Failed to update position: {str(e)}") @staticmethod async def delete_position(position_id: int) -> bool: try: supabase.table("Positions").delete().eq("id", position_id).execute() return True except Exception as e: raise Exception(f"Failed to delete position: {str(e)}") @staticmethod async def create_positions_bulk(positions_payload: list) -> list: """positions_payload: [{lat, lon, streets: [street_id, ...], pos_id?}, ...] Each item is ONE unique position carrying its complete streets array. pos_id present → UPDATE existing Positions row and sync Streets.positions arrays. pos_id absent → INSERT new row, then add its id to all relevant Streets.positions arrays. Returns list of affected position records. """ created = [] try: for pt in positions_payload: lat = float(pt.get('lat')) lon = float(pt.get('lon')) streets = [int(s) for s in (pt.get('streets') or [])] pos_id = pt.get('pos_id') # Optional ban fields (only written when the key is present in payload) ban_fields = {k: pt[k] for k in ("ban_turn", "ban_vehicle", "ban_hours", "ban_weight") if k in pt} if pos_id: # ── UPDATE existing position ────────────────────────────────── pos_id = int(pos_id) # Fetch old streets so we can diff existing_pos = supabase.table("Positions").select("streets").eq("id", pos_id).limit(1).execute() old_streets = list((existing_pos.data[0].get('streets') or []) if existing_pos.data else []) result = supabase.table("Positions").update({"long": lon, "lat": lat, "streets": streets, **ban_fields}).eq("id", pos_id).execute() if result.data: created.append(result.data[0]) # Remove pos_id from Streets that are no longer associated for sid in old_streets: if sid not in streets: try: sr = supabase.table("Streets").select("positions").eq("id", sid).limit(1).execute() if sr.data: existing = MapsService._remove_pos_id(sr.data[0].get('positions'), pos_id) supabase.table("Streets").update({"positions": existing}).eq("id", sid).execute() except Exception: pass # Add pos_id to Streets newly associated for sid in streets: if sid not in old_streets: try: sr = supabase.table("Streets").select("positions").eq("id", sid).limit(1).execute() if sr.data: raw = sr.data[0].get('positions') if pos_id not in MapsService._flatten_positions(raw): existing = MapsService._append_pos_id(raw, pos_id) supabase.table("Streets").update({"positions": existing}).eq("id", sid).execute() except Exception: pass else: # ── INSERT new position ─────────────────────────────────────── next_id = _next_id("Positions") payload = {"id": next_id, "long": lon, "lat": lat, "streets": streets, **ban_fields} result = supabase.table("Positions").insert(payload).execute() if not result.data: raise Exception("Failed to insert position") pos = result.data[0] created.append(pos) # Append new position id to every associated Street.positions array for sid in streets: try: sr = supabase.table("Streets").select("positions").eq("id", sid).limit(1).execute() if sr.data: raw = sr.data[0].get('positions') if pos['id'] not in MapsService._flatten_positions(raw): existing = MapsService._append_pos_id(raw, pos['id']) supabase.table("Streets").update({"positions": existing}).eq("id", sid).execute() except Exception: pass return created except Exception as e: raise Exception(f"Failed to create/update positions bulk: {str(e)}") # ==================== GROUPS ==================== @staticmethod async def get_groups() -> List[Group]: try: result = supabase.table("Groups").select("*").order("id").execute() return [Group(**item) for item in (result.data or [])] except Exception as e: raise Exception(f"Failed to fetch groups: {str(e)}") @staticmethod async def get_group_by_id(group_id: int) -> Optional[Group]: try: result = supabase.table("Groups").select("*").eq("id", group_id).limit(1).execute() return Group(**result.data[0]) if result.data else None except Exception as e: raise Exception(f"Failed to fetch group: {str(e)}") @staticmethod async def create_group(data: GroupCreate) -> Group: try: next_id = _next_id("Groups") payload = { "id": next_id, "lat_center": data.lat_center, "long_center": data.long_center, "streets": [int(s) for s in (data.streets or [])], "neighbor": [int(n) for n in (data.neighbor or [])], } result = supabase.table("Groups").insert(payload).execute() if not result.data: raise Exception("Failed to insert group") return Group(**result.data[0]) except Exception as e: raise Exception(f"Failed to create group: {str(e)}") @staticmethod async def update_group(group_id: int, data: GroupUpdate) -> Group: try: payload = {} for field in ("lat_center", "long_center", "streets", "neighbor"): val = getattr(data, field) if val is None: continue payload[field] = [int(x) for x in val] if field in ("streets", "neighbor") else val if not payload: raise Exception("No valid fields to update") result = supabase.table("Groups").update(payload).eq("id", group_id).execute() if not result.data: raise Exception(f"Group #{group_id} not found or update failed") return Group(**result.data[0]) except Exception as e: raise Exception(f"Failed to update group: {str(e)}") @staticmethod async def delete_group(group_id: int) -> bool: try: supabase.table("Groups").delete().eq("id", group_id).execute() return True except Exception as e: raise Exception(f"Failed to delete group: {str(e)}") @staticmethod async def update_poi(poi_id: int, fields: dict) -> dict: """Update a single Pois row with the provided fields (lat/long/name/address/type). Returns the updated row. """ try: allowed = {"lat", "long", "name", "address", "type", "city_id", "district_id"} payload = {k: v for k, v in (fields or {}).items() if k in allowed and v is not None} if not payload: raise Exception("No valid fields to update") result = supabase.table("Pois").update(payload).eq("id", poi_id).execute() if not result.data: raise Exception(f"POI #{poi_id} not found or update failed") return result.data[0] except Exception as e: raise Exception(f"Failed to update POI: {str(e)}") @staticmethod async def update_pois_bulk(payload: list) -> list: """payload: [{poi_id, lat?, long?, name?, address?, type?}, ...] Updates each POI by id. Returns list of updated rows. """ updated = [] allowed = {"lat", "long", "name", "address", "type", "city_id", "district_id"} try: for item in payload or []: poi_id = item.get("poi_id") or item.get("id") if not poi_id: continue fields = {k: v for k, v in item.items() if k in allowed and v is not None} if not fields: continue result = supabase.table("Pois").update(fields).eq("id", int(poi_id)).execute() if result.data: updated.append(result.data[0]) return updated except Exception as e: raise Exception(f"Failed to update POIs bulk: {str(e)}") # Path where position.bin is cached on disk (backend/position.bin) _BIN_PATH: str = "" @classmethod def _bin_path(cls) -> str: if not cls._BIN_PATH: import os cls._BIN_PATH = os.path.join( os.path.dirname(os.path.abspath(__file__)), '..', 'position.bin' ) return cls._BIN_PATH @staticmethod async def export_graph_bin() -> bytes: """Serialize the position graph as a compact binary file for A* pathfinding. Format (little-endian): HEADER (16 bytes): magic b"POSBIN\x01\x00" (8B) | node_count uint32 | edge_count uint32 NODES (4B x node_count): sorted node ids as int32 EDGES (12B x edge_count): from_id int32 | to_id int32 | weight_m float32 edges are undirected and stored with from_id < to_id. Nodes contain id only — client maps back to DB for lat/lon/name. Also writes the file to disk at backend/position.bin for pathfind() to consume. """ import math import struct def haversine_m(lat1, lon1, lat2, lon2): R = 6_371_000 phi1, phi2 = math.radians(lat1), math.radians(lat2) dphi = math.radians(lat2 - lat1) dlam = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) try: result = supabase.table("Positions").select("id, lat, long, streets").execute() all_positions = result.data or [] # Index valid positions (must have lat and long) pos_by_id: dict = {} for p in all_positions: if p.get("lat") is not None and p.get("long") is not None: pos_by_id[p["id"]] = p # Build street → list of position ids street_to_pids: dict = {} for pid, pos in pos_by_id.items(): for sid in (pos.get("streets") or []): street_to_pids.setdefault(sid, []).append(pid) # Build edges: undirected, from_id < to_id, keep minimum weight per pair edge_map: dict = {} # (from_id, to_id) → weight_m for sid, pids in street_to_pids.items(): for i in range(len(pids)): for j in range(i + 1, len(pids)): a_id, b_id = pids[i], pids[j] if a_id > b_id: a_id, b_id = b_id, a_id pa = pos_by_id[a_id] pb = pos_by_id[b_id] w = haversine_m( float(pa["lat"]), float(pa["long"]), float(pb["lat"]), float(pb["long"]) ) key = (a_id, b_id) if key not in edge_map or w < edge_map[key]: edge_map[key] = w sorted_node_ids = sorted(pos_by_id.keys()) edges = [(a, b, w) for (a, b), w in edge_map.items()] node_count = len(sorted_node_ids) edge_count = len(edges) buf = bytearray() buf += struct.pack('<8sII', b"POSBIN\x01\x00", node_count, edge_count) if node_count: buf += struct.pack(f'<{node_count}i', *sorted_node_ids) for a_id, b_id, w in edges: buf += struct.pack(' dict: """Run Dijkstra on the pre-built position graph stored in position.bin. Graph is loaded from the binary file (no full DB fetch). Only queries DB for: street arrays of nodes in the found path + street details. Returns node path, ordered street ids, street details, and total distance. Requires position.bin to exist — call GET /export/position.bin first to generate it. """ import struct import heapq import os bin_path = MapsService._bin_path() if not os.path.exists(bin_path): raise Exception( "position.bin not found. Call GET /export/position.bin first to generate the graph file." ) try: with open(bin_path, 'rb') as f: data = f.read() # --- Parse header --- magic, node_count, edge_count = struct.unpack_from('<8sII', data, 0) if magic != b"POSBIN\x01\x00": raise Exception("Invalid position.bin format — unexpected magic bytes") # --- Parse node ids --- node_ids: list = list(struct.unpack_from(f'<{node_count}i', data, 16)) node_set: set = set(node_ids) if start_id not in node_set: raise Exception(f"Start position #{start_id} not found in graph") if end_id not in node_set: raise Exception(f"End position #{end_id} not found in graph") if start_id == end_id: return {"found": True, "node_path": [start_id], "street_path": [], "street_details": [], "total_m": 0} # --- Parse edges → build undirected adjacency --- adjacency: dict = {nid: {} for nid in node_ids} edge_offset = 16 + node_count * 4 for i in range(edge_count): a, b, w = struct.unpack_from(' w: adjacency[a][b] = w adjacency[b][a] = w # --- Dijkstra (no lat/lon in bin → no heuristic needed) --- g: dict = {nid: float("inf") for nid in node_ids} g[start_id] = 0.0 prev: dict = {nid: None for nid in node_ids} heap = [(0.0, start_id)] while heap: cost, u = heapq.heappop(heap) if u == end_id: break if cost > g[u]: continue for v, w in adjacency[u].items(): ng = g[u] + w if ng < g[v]: g[v] = ng prev[v] = u heapq.heappush(heap, (ng, v)) if g[end_id] == float("inf"): return {"found": False, "node_path": [], "street_path": [], "street_details": [], "total_m": 0} # --- Reconstruct node path --- path: list = [] cur = end_id while cur is not None: path.append(cur) cur = prev[cur] path.reverse() # --- Minimal DB query: only fetch streets array for nodes in path --- pos_result = supabase.table("Positions") \ .select("id, streets") \ .in_("id", path).execute() pos_streets: dict = { p["id"]: (p.get("streets") or []) for p in (pos_result.data or []) } # --- Extract ordered unique street IDs from consecutive node pairs --- street_path: list = [] seen_streets: set = set() for i in range(len(path) - 1): shared = (set(pos_streets.get(path[i], [])) & set(pos_streets.get(path[i + 1], []))) for sid in shared: if sid not in seen_streets: seen_streets.add(sid) street_path.append(sid) # --- Fetch street details for the path --- street_details: list = [] if street_path: sr = supabase.table("Streets") \ .select("id, name, type, district_id, city_id") \ .in_("id", street_path).execute() street_map = {s["id"]: s for s in (sr.data or [])} street_details = [street_map[sid] for sid in street_path if sid in street_map] return { "found": True, "node_path": path, "street_path": street_path, "street_details": street_details, "total_m": round(g[end_id]), } except Exception as e: raise Exception(f"Pathfind failed: {str(e)}") @staticmethod async def nearest_position(qlat: float, qlon: float, top_k: int = 3) -> dict: """Find the single nearest position to (qlat, qlon) with a 2-step Group strategy: 1) pick the `top_k` nearest Groups by (lat_center, long_center) 2) load ONLY those groups' positions (via Streets.positions) — not the whole DB 3) find the nearest position inside EACH group 4) return the closest among the per-group winners Falls back to a full DB scan when there are no groups / no group positions. Returns: {found, position: {id,lat,lon,streets,street_names}|None, distance_m, group_id} """ import math def haversine_m(lat1, lon1, lat2, lon2): R = 6_371_000 phi1, phi2 = math.radians(lat1), math.radians(lat2) dphi = math.radians(lat2 - lat1) dlam = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) best_pos, best_d, best_gid = None, float("inf"), None # 1) Nearest `top_k` groups by their center groups_res = supabase.table("Groups").select("id, lat_center, long_center, streets").execute() groups = [g for g in (groups_res.data or []) if g.get("lat_center") is not None and g.get("long_center") is not None] groups.sort(key=lambda g: haversine_m(qlat, qlon, float(g["lat_center"]), float(g["long_center"]))) nearest_groups = groups[:top_k] # 2) Group streets → position ids (via Streets.positions); batch-load ONLY those positions all_street_ids = sorted({int(s) for g in nearest_groups for s in (g.get("streets") or [])}) if all_street_ids: sres = supabase.table("Streets").select("id, positions").in_("id", all_street_ids).execute() street_positions = {s["id"]: MapsService._flatten_positions(s.get("positions")) for s in (sres.data or [])} group_pos_ids: dict = {} all_pos_ids: set = set() for g in nearest_groups: pids = set() for sid in (g.get("streets") or []): pids.update(street_positions.get(int(sid), [])) group_pos_ids[g["id"]] = pids all_pos_ids.update(pids) if all_pos_ids: pres = supabase.table("Positions").select("*").in_("id", list(all_pos_ids)).execute() pos_by_id = {p["id"]: p for p in (pres.data or []) if p.get("lat") is not None and p.get("long") is not None} # 3) nearest position inside each group + 4) compare the winners for g in nearest_groups: g_pos, g_d = None, float("inf") for pid in group_pos_ids.get(g["id"], ()): # chỉ xét position của group này pos = pos_by_id.get(pid) if not pos: continue d = haversine_m(qlat, qlon, float(pos["lat"]), float(pos["long"])) if d < g_d: g_d, g_pos = d, pos if g_pos is not None and g_d < best_d: best_d, best_pos, best_gid = g_d, g_pos, g["id"] # Fallback: không có group / group rỗng → quét toàn bộ DB (đúng tuyệt đối) if best_pos is None: res = supabase.table("Positions").select("id, lat, long, streets").execute() for p in (res.data or []): if p.get("lat") is None or p.get("long") is None: continue d = haversine_m(qlat, qlon, float(p["lat"]), float(p["long"])) if d < best_d: best_d, best_pos, best_gid = d, p, None if best_pos is None: return {"found": False, "position": None, "distance_m": None, "group_id": None} # Resolve street names for the chosen position street_ids = best_pos.get("streets") or [] smap = {} if street_ids: sr = supabase.table("Streets").select("id, name").in_("id", street_ids).execute() smap = {s["id"]: s["name"] for s in (sr.data or [])} position = { "id": best_pos["id"], "lat": best_pos["lat"], "lon": best_pos["long"], "streets": street_ids, "street_names": [smap.get(sid, str(sid)) for sid in street_ids], } return {"found": True, "position": position, "distance_m": round(best_d), "group_id": best_gid} @staticmethod async def search_nearest(points: list) -> dict: """ 1 query point → return nearest DB position. 2 query points → find nearest DB node to each, then run Dijkstra on the street-adjacency graph to return the shortest path array. Graph: positions are nodes; two positions are adjacent iff they share ≥1 street. Edge weight = Haversine distance (metres). Returns `tts` field: natural Vietnamese navigation text with street names, turn directions, and congestion warnings (speed.normal < 20 km/h). """ import math import heapq def haversine_m(lat1, lon1, lat2, lon2): R = 6_371_000 phi1, phi2 = math.radians(lat1), math.radians(lat2) dphi = math.radians(lat2 - lat1) dlam = math.radians(lon2 - lon1) a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlam / 2) ** 2 return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) def bearing_deg(lat1, lon1, lat2, lon2): phi1 = math.radians(lat1) phi2 = math.radians(lat2) dl = math.radians(lon2 - lon1) x = math.sin(dl) * math.cos(phi2) y = math.cos(phi1) * math.sin(phi2) - math.sin(phi1) * math.cos(phi2) * math.cos(dl) return (math.degrees(math.atan2(x, y)) + 360) % 360 def turn_vi(b_in, b_out): diff = (b_out - b_in + 360) % 360 if diff <= 30 or diff >= 330: return "đi thẳng" elif diff < 150: return "rẽ phải" elif diff < 210: return "quay đầu" else: return "rẽ trái" def is_congested(pos_row): speed = pos_row.get("speed") if isinstance(speed, dict): normal = speed.get("normal") try: return normal is not None and float(normal) < 20 except (ValueError, TypeError): pass return False def build_tts(path_pts, pos_by_id_map, s_map, total_m): if not path_pts: return "" if len(path_pts) == 1: snames = path_pts[0]["street_names"] loc = snames[0] if snames else "vị trí không xác định" return f"Vị trí gần nhất nằm trên {loc}." # Street shared between each consecutive pair of path nodes segment_streets = [] for i in range(len(path_pts) - 1): shared = set(path_pts[i]["streets"]) & set(path_pts[i + 1]["streets"]) segment_streets.append(list(shared)[0] if shared else None) instructions = [] current_sid = segment_streets[0] start_name = s_map.get(current_sid, "đường không xác định") instructions.append(f"Xuất phát từ {start_name}") for i in range(1, len(path_pts)): pt = path_pts[i] pos_row = pos_by_id_map.get(pt["id"], {}) # Congestion warning if is_congested(pos_row): loc = " - ".join(pt["street_names"]) if pt["street_names"] else "điểm này" instructions.append(f"Lưu ý kẹt xe tại {loc}") # Detect street change at this node new_sid = segment_streets[i] if i < len(segment_streets) else None if new_sid is not None and new_sid != current_sid: # Bearings: incoming (i-1 → i) and outgoing (i → i+1) p_prev = path_pts[i - 1] p_curr = pt p_next = path_pts[i + 1] # always valid: i < len(segment_streets) ensures i+1 exists b_in = bearing_deg(float(p_prev["lat"]), float(p_prev["lon"]), float(p_curr["lat"]), float(p_curr["lon"])) b_out = bearing_deg(float(p_curr["lat"]), float(p_curr["lon"]), float(p_next["lat"]), float(p_next["lon"])) turn = turn_vi(b_in, b_out) new_name = s_map.get(new_sid, "đường không xác định") inter = " - ".join(pt["street_names"]) if len(pt["street_names"]) > 1 else "" if inter: instructions.append(f"Tại ngã tư {inter}, {turn} vào {new_name}") else: instructions.append(f"{turn.capitalize()} vào {new_name}") current_sid = new_sid dist_text = ( f"khoảng {total_m} mét" if total_m < 1000 else f"khoảng {round(total_m / 1000, 1)} kilômét" ) instructions.append(f"Đã đến nơi. Tổng quãng đường {dist_text}") return ". ".join(instructions) + "." try: # ── Single-point: reuse the extracted 2-step group-based nearest search ── if len(points) < 2: res = await MapsService.nearest_position(float(points[0]["lat"]), float(points[0]["lon"])) if not res.get("found"): return {"data": [], "label": "Không có điểm nào trong CSDL", "tts": "", "warning": "", "message": ""} pt, dist_m, gid = res["position"], res["distance_m"], res["group_id"] sname = pt["street_names"][0] if pt["street_names"] else "vị trí không xác định" dl = f"~{dist_m}m" if dist_m < 1000 else f"~{round(dist_m / 1000, 1)}km" gtxt = f" | Group #{gid}" if gid is not None else "" return { "data": [pt], "label": f"Điểm gần nhất: {sname} ({dl})", "tts": f"Vị trí gần nhất nằm trên {sname}, cách bạn khoảng {dist_m} mét.", "warning": "", "message": f"Khoảng cách: ~{dist_m}m{gtxt}", } result = supabase.table("Positions").select("*").execute() all_positions = result.data or [] if not all_positions: return {"data": [], "label": "Không có điểm nào trong CSDL", "tts": "", "warning": "", "message": ""} streets_result = supabase.table("Streets").select("id, name").execute() street_map = {s["id"]: s["name"] for s in (streets_result.data or [])} pos_by_id = {} for p in all_positions: if p.get("lat") is not None and p.get("long") is not None: pos_by_id[p["id"]] = p def find_nearest_id(qlat, qlon): best_id, best_d = None, float("inf") for pid, pos in pos_by_id.items(): d = haversine_m(qlat, qlon, float(pos["lat"]), float(pos["long"])) if d < best_d: best_d, best_id = d, pid return best_id, best_d def make_result_point(pos): street_ids = pos.get("streets") or [] return { "id": pos["id"], "lat": pos["lat"], "lon": pos["long"], "streets": street_ids, "street_names": [street_map.get(sid, str(sid)) for sid in street_ids], } def dist_label(m): return f"~{m}m" if m < 1000 else f"~{round(m / 1000, 1)}km" # ── Single-point mode: return nearest ──────────────────────────── # (single-point handled earlier via MapsService.nearest_position) # ── Two-point mode: shortest path via Dijkstra ─────────────────── q1, q2 = points[0], points[1] start_id, d_start = find_nearest_id(float(q1["lat"]), float(q1["lon"])) end_id, d_end = find_nearest_id(float(q2["lat"]), float(q2["lon"])) if start_id is None or end_id is None: return {"data": [], "label": "Không tìm thấy", "tts": "", "warning": "Không có điểm nào gần query", "message": ""} if start_id == end_id: pt = make_result_point(pos_by_id[start_id]) sname = pt["street_names"][0] if pt["street_names"] else "vị trí này" return { "data": [pt], "label": f"Điểm xuất phát và đích trùng nhau tại {sname}", "tts": f"Điểm xuất phát và điểm đích đều nằm tại {sname}.", "warning": "", "message": "", } # Build adjacency: positions adjacent if they share a street street_to_pids: dict = {} for pid, pos in pos_by_id.items(): for sid in (pos.get("streets") or []): street_to_pids.setdefault(sid, set()).add(pid) adjacency: dict = {pid: {} for pid in pos_by_id} for sid, pids in street_to_pids.items(): pid_list = list(pids) for i, pid1 in enumerate(pid_list): p1 = pos_by_id[pid1] for pid2 in pid_list[i + 1:]: p2 = pos_by_id[pid2] d = haversine_m(float(p1["lat"]), float(p1["long"]), float(p2["lat"]), float(p2["long"])) if pid2 not in adjacency[pid1] or adjacency[pid1][pid2] > d: adjacency[pid1][pid2] = d adjacency[pid2][pid1] = d # Dijkstra dist = {pid: float("inf") for pid in pos_by_id} prev = {pid: None for pid in pos_by_id} dist[start_id] = 0 heap = [(0.0, start_id)] while heap: d, u = heapq.heappop(heap) if d > dist[u]: continue if u == end_id: break for v, w in adjacency[u].items(): nd = dist[u] + w if nd < dist[v]: dist[v] = nd prev[v] = u heapq.heappush(heap, (nd, v)) if dist[end_id] == float("inf"): return { "data": [], "label": "Không tìm được đường", "tts": "Không tìm được đường đi giữa hai điểm này.", "warning": "Hai điểm không có đường nối trong đồ thị", "message": "", } # Reconstruct path path_ids = [] cur = end_id while cur is not None: path_ids.append(cur) cur = prev[cur] path_ids.reverse() total_m = round(dist[end_id]) data = [make_result_point(pos_by_id[pid]) for pid in path_ids] start_name = data[0]["street_names"][0] if data[0]["street_names"] else "?" end_name = data[-1]["street_names"][0] if data[-1]["street_names"] else "?" label = f"Từ {start_name} đến {end_name}, {dist_label(total_m)}" tts = build_tts(data, pos_by_id, street_map, total_m) return { "data": data, "label": label, "tts": tts, "tts_url": "", "warning": "", "message": f"Tổng: {dist_label(total_m)} | Snap start: ~{round(d_start)}m | Snap end: ~{round(d_end)}m", } except Exception as e: raise Exception(f"Search nearest failed: {str(e)}") @staticmethod async def search_uuid(query_id: str) -> dict: """Search the utest table for the row whose id is most similar to query_id. Uses Levenshtein distance (normalised) as similarity score. Returns: {found, id, score, row} where score=0.0 is exact match. """ def _lev(a: str, b: str) -> int: if a == b: return 0 if not a: return len(b) if not b: return len(a) prev = list(range(len(b) + 1)) for i, ca in enumerate(a, start=1): curr = [i] for j, cb in enumerate(b, start=1): curr.append(min(curr[j - 1] + 1, prev[j] + 1, prev[j - 1] + (0 if ca == cb else 1))) prev = curr return prev[-1] try: result = supabase.table("utest").select("*").execute() rows = result.data or [] if not rows: return {"found": False, "id": None, "score": None, "row": None} q = str(query_id).strip() best_row = None best_score = float("inf") for row in rows: row_id = str(row.get("id") or "").strip() if not row_id: continue dist = _lev(q, row_id) max_len = max(len(q), len(row_id), 1) score = dist / max_len if score < best_score: best_score = score best_row = row if best_row is None: return {"found": False, "id": None, "score": None, "row": None} return { "found": True, "id": best_row.get("id"), "score": round(best_score, 4), "row": best_row, } except Exception as e: raise Exception(f"search_uuid failed: {str(e)}") @staticmethod async def delete_position_cascade(position_id: int) -> bool: """Delete a position and remove its id from any Streets.positions arrays listed in position.streets""" try: # fetch position to know related streets pos = None result = supabase.table("Positions").select("*").eq("id", position_id).limit(1).execute() if result.data: pos = result.data[0] # remove this position from Streets.positions JSON arrays if present if pos and 'streets' in pos and pos['streets']: for sid in pos['streets']: try: street_row = supabase.table("Streets").select("positions").eq("id", sid).limit(1).execute() if street_row.data: raw = street_row.data[0].get('positions') if position_id in MapsService._flatten_positions(raw): existing = MapsService._remove_pos_id(raw, position_id) supabase.table("Streets").update({"positions": existing}).eq("id", sid).execute() except Exception: pass # delete the position supabase.table("Positions").delete().eq("id", position_id).execute() return True except Exception as e: raise Exception(f"Failed to delete position cascade: {str(e)}")