""" Normalize raw scraped restaurant data → cuisine_database.json schema. Merges into existing cuisine_database.json (append, no overwrite). Deduplicates by name + coordinates. """ import json import logging import re from typing import Optional try: from rapidfuzz import fuzz HAS_FUZZ = True except ImportError: HAS_FUZZ = False from scripts.scrapers.config import ( CUISINE_TYPE_MAP, DIETARY_KEYWORD_MAP, PRICE_TIER_MAP, PATHS, SCRAPER_SETTINGS ) logger = logging.getLogger(__name__) VALID_CUISINE_TYPES = { "local_specialty", "street_food", "seafood", "fine_dining", "cafe", "vegetarian", "bbq_grill", "hotpot", "international", } VALID_MEAL_TYPES = {"breakfast", "lunch", "dinner", "brunch", "snack"} MEAL_TYPE_HINTS = { "cafe": ["breakfast", "brunch"], "street_food": ["breakfast", "lunch", "snack"], "fine_dining": ["dinner"], "hotpot": ["dinner"], "bbq_grill": ["lunch", "dinner"], "seafood": ["lunch", "dinner"], "vegetarian": ["lunch", "dinner"], "international":["lunch", "dinner"], "local_specialty": ["lunch", "dinner"], } def map_cuisine_type(raw_cuisines: list) -> list: """Map raw cuisine strings to internal cuisine type list.""" result = set() for c in raw_cuisines: if not c: continue c_lower = c.lower().strip().replace("-", "_").replace(" ", "_") if c_lower in VALID_CUISINE_TYPES: result.add(c_lower) continue for keyword, mapped in CUISINE_TYPE_MAP.items(): if keyword in c_lower: result.update(mapped) break else: result.add("local_specialty") return list(result) or ["local_specialty"] def detect_dietary_tags(cuisines: list, name: str, description: str = "") -> list: """Detect dietary tags from cuisine types, name, and description.""" text = " ".join([name, description] + cuisines).lower() tags = [] for tag, keywords in DIETARY_KEYWORD_MAP.items(): if any(kw in text for kw in keywords): tags.append(tag) return tags def infer_meal_type(cuisine_types: list) -> list: """Infer meal types from cuisine types.""" meal_types = set() for ct in cuisine_types: hints = MEAL_TYPE_HINTS.get(ct, ["lunch", "dinner"]) meal_types.update(hints) return list(meal_types) or ["lunch", "dinner"] def convert_price_to_vnd(price_level: Optional[int]) -> list: """Convert universal price level (1-4) to VND range. price_level is a 1-4 scale (as used by Google Maps / TripAdvisor): 1 = budget (~$2-8 USD equivalent) 2 = mid-range (~$8-25) 3 = upscale (~$25-60) 4 = fine dining (~$60-200) Always outputs VND (Vietnamese Dong) as the primary currency. Conversion: USD × 25,000 VND/USD (standard reference rate). """ # Reference prices in USD per person, then converted to VND usd_ranges = { 1: (2, 8), # budget → 50,000 – 200,000 VND 2: (8, 25), # mid-range → 200,000 – 625,000 VND 3: (25, 60), # upscale → 625,000 – 1,500,000 VND 4: (60, 200), # fine dining → 1,500,000 – 5,000,000 VND } low_usd, high_usd = usd_ranges.get(price_level or 2, (5, 20)) vnd_per_usd = 25_000 return [int(low_usd * vnd_per_usd), int(high_usd * vnd_per_usd)] def map_price_tier(price_level: Optional[int]) -> str: return PRICE_TIER_MAP.get(price_level or 2, "mid_range") def normalize_restaurant(raw: dict) -> Optional[dict]: """ Convert a raw scraped restaurant dict to cuisine_database.json schema. Returns None if the entry lacks minimum required fields. """ name = (raw.get("name") or "").strip() if not name or len(name) < 2: return None lat = raw.get("lat") lon = raw.get("lon") cuisine_raw = raw.get("cuisine") or [] if isinstance(cuisine_raw, str): cuisine_raw = [c.strip() for c in cuisine_raw.split(";") if c.strip()] cuisine_types = map_cuisine_type(cuisine_raw) dietary_tags = detect_dietary_tags( cuisine_raw, name, raw.get("description", "") ) price_level = raw.get("price_level") rating = raw.get("rating") if rating is not None: try: rating = float(rating) # Normalize: Google/TripAdvisor use 1-5, Booking uses 1-10 if rating > 5: rating = round(rating / 2, 1) rating = round(min(max(rating, 1.0), 5.0), 1) except (TypeError, ValueError): rating = 4.0 else: rating = 4.0 opening_hours = (raw.get("opening_hours") or "").strip() if not opening_hours: opening_hours = "08:00-22:00" return { "name": name, "name_en": (raw.get("name_en") or name).strip(), "destination_id": raw.get("destination_id", ""), "coordinates": { "lat": float(lat) if lat is not None else None, "lon": float(lon) if lon is not None else None, }, "cuisine_type": cuisine_types, "price_range_vnd": convert_price_to_vnd(price_level), "price_tier": map_price_tier(price_level), "dietary_tags": dietary_tags, "specialties": [], "rating": rating, "opening_hours": opening_hours, "meal_type": infer_meal_type(cuisine_types), "description_vi": "", "description_en": (raw.get("description") or "")[:300].strip(), "_source": raw.get("source", "unknown"), } def _haversine_m(lat1, lon1, lat2, lon2) -> float: """Distance in metres between two lat/lon points.""" import math 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 2 * R * math.asin(math.sqrt(a)) class _DeduplicatorIndex: """ O(1) average-case deduplication using: 1. Exact name hash set (instant reject of exact duplicates) 2. Spatial grid (1° × 1° cells) to limit fuzzy candidates to nearby restaurants only """ GRID_SIZE = 1.0 # degrees (~100 km) def __init__(self, existing: list): self.exact_names: set = set() # grid_cell → list of (name_lower, lat, lon) self.grid: dict = {} for ex in existing: name = ex.get("name", "").lower().strip() self.exact_names.add(name) lat = ex.get("coordinates", {}).get("lat") lon = ex.get("coordinates", {}).get("lon") if lat is not None and lon is not None: cell = (int(lat / self.GRID_SIZE), int(lon / self.GRID_SIZE)) self.grid.setdefault(cell, []).append((name, lat, lon)) def _candidates(self, lat, lon) -> list: """Return names of restaurants in the same and adjacent grid cells.""" cx = int(lat / self.GRID_SIZE) cy = int(lon / self.GRID_SIZE) result = [] for dx in (-1, 0, 1): for dy in (-1, 0, 1): result.extend(self.grid.get((cx + dx, cy + dy), [])) return result def is_duplicate(self, new: dict) -> bool: threshold_m = SCRAPER_SETTINGS["dedup_distance_meters"] name_threshold = SCRAPER_SETTINGS["dedup_name_threshold"] new_name = new["name"].lower().strip() new_lat = new["coordinates"].get("lat") new_lon = new["coordinates"].get("lon") # Fast path: exact name match if new_name in self.exact_names: return True # Spatial fuzzy check only among nearby candidates if new_lat is not None and new_lon is not None: for ex_name, ex_lat, ex_lon in self._candidates(new_lat, new_lon): if HAS_FUZZ: score = fuzz.ratio(new_name, ex_name) else: score = 100 if new_name == ex_name else 0 if score >= name_threshold: dist = _haversine_m(new_lat, new_lon, ex_lat, ex_lon) if dist < threshold_m: return True return False def add(self, normalized: dict): """Register a newly-added restaurant into the index.""" name = normalized["name"].lower().strip() self.exact_names.add(name) lat = normalized["coordinates"].get("lat") lon = normalized["coordinates"].get("lon") if lat is not None and lon is not None: cell = (int(lat / self.GRID_SIZE), int(lon / self.GRID_SIZE)) self.grid.setdefault(cell, []).append((name, lat, lon)) def merge_into_cuisine_db(new_restaurants: list) -> dict: """ Load existing cuisine_database.json, append new restaurants (deduped), save and return stats. Uses spatial grid index for O(n) deduplication instead of O(n²). """ cuisine_db_path = PATHS["cuisine_db"] with open(cuisine_db_path, "r", encoding="utf-8") as f: db = json.load(f) existing = db.get("restaurants", []) index = _DeduplicatorIndex(existing) added = 0 skipped_invalid = 0 skipped_dup = 0 for raw in new_restaurants: normalized = normalize_restaurant(raw) if normalized is None: skipped_invalid += 1 continue if index.is_duplicate(normalized): skipped_dup += 1 continue existing.append(normalized) index.add(normalized) added += 1 db["restaurants"] = existing with open(cuisine_db_path, "w", encoding="utf-8") as f: json.dump(db, f, ensure_ascii=False, indent=2) stats = { "total_in_db": len(existing), "added": added, "skipped_invalid": skipped_invalid, "skipped_duplicate": skipped_dup, } logger.info(f"Restaurant merge: +{added} added, {skipped_dup} dupes, {skipped_invalid} invalid → {len(existing)} total") return stats