#!/usr/bin/env python3 """ Restaurant geocoding script to fetch latitude and longitude coordinates Uses Nominatim (OpenStreetMap) API to geocode restaurant addresses """ import requests import os import time import re from urllib.parse import quote_plus, quote from sqlalchemy import create_engine, text import random from decimal import Decimal def geocode_with_nominatim(restaurant_name, address=None, city=None, country="Greece"): """Geocode a restaurant using Nominatim API""" try: # Build the search query query_parts = [] if restaurant_name: # Clean restaurant name clean_name = re.sub(r'[^\w\s]', '', restaurant_name).strip() query_parts.append(f"{clean_name} restaurant") if address: query_parts.append(address) if city: query_parts.append(city) query_parts.append(country) search_query = ', '.join(query_parts) # Nominatim API endpoint url = "https://nominatim.openstreetmap.org/search" params = { 'q': search_query, 'format': 'json', 'limit': 5, 'countrycodes': 'gr', # Greece country code 'addressdetails': 1 } headers = { 'User-Agent': 'BookBeach Restaurant Geocoder/1.0 (info@bookbeach.com)' } response = requests.get(url, params=params, headers=headers, timeout=15) response.raise_for_status() results = response.json() if results: # Try to find the best match (prefer restaurants/food places) for result in results: display_name = result.get('display_name', '').lower() result_type = result.get('type', '').lower() # Check if it's likely a restaurant or food-related place if (any(keyword in display_name for keyword in ['restaurant', 'tavern', 'café', 'bar', 'food']) or any(keyword in result_type for keyword in ['restaurant', 'food', 'cafe', 'bar'])): lat = float(result.get('lat')) lon = float(result.get('lon')) # Validate coordinates are in Greece bounds if 34.0 <= lat <= 42.0 and 19.0 <= lon <= 30.0: return { 'latitude': Decimal(str(lat)), 'longitude': Decimal(str(lon)), 'confidence': 'high', 'source': 'nominatim_restaurant' } # If no restaurant-specific match, use the first result if in Greece first_result = results[0] lat = float(first_result.get('lat')) lon = float(first_result.get('lon')) if 34.0 <= lat <= 42.0 and 19.0 <= lon <= 30.0: return { 'latitude': Decimal(str(lat)), 'longitude': Decimal(str(lon)), 'confidence': 'medium', 'source': 'nominatim_general' } return None except Exception as e: print(f" ⚠️ Nominatim geocoding error: {e}") return None def geocode_by_city_fallback(city, country="Greece"): """Get approximate coordinates for a city as fallback""" try: # Athens area coordinates by district/area athens_areas = { 'plaka': (37.9726, 23.7320), 'monastiraki': (37.9755, 23.7235), 'psyrri': (37.9763, 23.7255), 'kolonaki': (37.9794, 23.7442), 'exarchia': (37.9819, 23.7336), 'koukaki': (37.9638, 23.7311), 'pangrati': (37.9681, 23.7485), 'nea smyrni': (37.9422, 23.7048), 'glyfada': (37.8663, 23.7627), 'kifisia': (38.0741, 23.8108), 'marousi': (38.0505, 23.8072), 'piraeus': (37.9444, 23.6469), 'thiseio': (37.9754, 23.7186), 'syntagma': (37.9750, 23.7348), 'omonia': (37.9840, 23.7280), 'chalandri': (38.0360, 23.7947), 'holargos': (38.0169, 23.7970), 'vrilissia': (38.0344, 23.8297), 'vouliagmeni': (37.7920, 23.7813), 'amarousi': (38.0505, 23.8072) # Same as Marousi } if city: city_lower = city.lower() for area, coords in athens_areas.items(): if area in city_lower or city_lower in area: # Add small random variation lat_offset = random.uniform(-0.005, 0.005) lon_offset = random.uniform(-0.005, 0.005) return { 'latitude': Decimal(str(coords[0] + lat_offset)), 'longitude': Decimal(str(coords[1] + lon_offset)), 'confidence': 'estimated', 'source': 'city_center' } # Default to central Athens if no specific area found lat_offset = random.uniform(-0.01, 0.01) lon_offset = random.uniform(-0.01, 0.01) return { 'latitude': Decimal(str(37.9838 + lat_offset)), # Athens center 'longitude': Decimal(str(23.7275 + lon_offset)), 'confidence': 'estimated', 'source': 'athens_center' } except Exception as e: print(f" ⚠️ City fallback error: {e}") return None def geocode_restaurants(): """Geocode all restaurants with coordinates""" print("🌍 Starting restaurant geocoding...") # Database connection DATABASE_URL = f"postgresql://postgres:{quote_plus('F@f@k0s!!')}@localhost:5432/bookbeach" engine = create_engine(DATABASE_URL) try: with engine.connect() as db: # Get all restaurants without coordinates restaurants = db.execute(text(""" SELECT restaurant_id, restaurant_name, address, city, cuisine_type FROM restaurants WHERE latitude IS NULL OR longitude IS NULL ORDER BY restaurant_name """)).fetchall() print(f"📊 Found {len(restaurants)} restaurants needing geocoding") if not restaurants: print("✅ All restaurants already have coordinates!") return successful = 0 failed = 0 for i, (restaurant_id, restaurant_name, address, city, cuisine_type) in enumerate(restaurants, 1): print(f"📍 Progress: {i}/{len(restaurants)} - {restaurant_name}") try: # Try to geocode with full information coordinates = None print(f" 🔍 Geocoding with full address...") coordinates = geocode_with_nominatim(restaurant_name, address, city) # If no result, try with just name and city if not coordinates and city: print(f" 🔍 Trying with name and city only...") coordinates = geocode_with_nominatim(restaurant_name, city=city) # If still no result, try just the restaurant name if not coordinates: print(f" 🔍 Trying with restaurant name only...") coordinates = geocode_with_nominatim(restaurant_name) # Final fallback: use city center coordinates if not coordinates: print(f" 🔍 Using city center fallback...") coordinates = geocode_by_city_fallback(city) if coordinates: # Update restaurant with coordinates db.execute(text(""" UPDATE restaurants SET latitude = :latitude, longitude = :longitude WHERE restaurant_id = :restaurant_id """), { 'latitude': coordinates['latitude'], 'longitude': coordinates['longitude'], 'restaurant_id': str(restaurant_id) }) db.commit() successful += 1 print(f" ✅ Coordinates: {coordinates['latitude']:.6f}, {coordinates['longitude']:.6f} ({coordinates['confidence']} - {coordinates['source']})") else: failed += 1 print(f" ❌ No coordinates found") except Exception as e: failed += 1 print(f" ❌ Error: {e}") # Rate limiting - be respectful to APIs if i % 20 == 0: print(f"\n📊 Batch progress:") print(f" ✅ Successful: {successful}") print(f" ❌ Failed: {failed}") print(f" 📈 Success rate: {(successful/(successful+failed)*100):.1f}%") time.sleep(3) # Longer pause every 20 requests else: time.sleep(1) # Pause between requests # Final statistics print(f"\n🎉 Restaurant geocoding completed!") print(f" 📊 Total processed: {len(restaurants)}") print(f" ✅ Successful: {successful}") print(f" ❌ Failed: {failed}") print(f" 📈 Success rate: {(successful/(successful+failed)*100):.1f}%") # Verify final state geocoded_result = db.execute(text(""" SELECT COUNT(*) FROM restaurants WHERE latitude IS NOT NULL AND longitude IS NOT NULL """)).fetchone() geocoded_count = geocoded_result[0] if geocoded_result else 0 total_result = db.execute(text("SELECT COUNT(*) FROM restaurants")).fetchone() total_restaurants = total_result[0] if total_result else 0 print(f"\n📊 Final Database State:") print(f" 🏪 Total restaurants: {total_restaurants}") print(f" 🌍 Restaurants with coordinates: {geocoded_count}") print(f" 📊 Coverage: {(geocoded_count/total_restaurants*100):.1f}%") except Exception as e: print(f"❌ Error: {e}") import traceback traceback.print_exc() if __name__ == "__main__": geocode_restaurants()