# Standard library imports
import json  # JSON file handling
from typing import List, Dict, Optional, Tuple  # Type hints for better code documentation
import time  # Time management for API rate limiting
import os  # Operating system interface

# Third-party library imports
from shapely.geometry import Polygon, Point  # Geometric operations (polygons, points)
import overpy  # Overpass API Python wrapper for OpenStreetMap queries

# Load Bari's geographical boundary coordinates from JSON file
with open("bariCoordinates.json") as f:
    COORDINATES = json.load(f)

# These coordinates define a polygon that represents the city boundaries of Bari

# Define amenities to search for in Bari
# Commented entries were tested but not found in the city

AMENITIES = {
    # Food & Drink categories
    'bar': 'Establishments selling alcoholic drinks with vibrant atmosphere',
    'pub': 'Traditional establishments with alcoholic drinks and food',
    'restaurant': 'Places serving meals with seating',
    'cafe': 'Informal places serving beverages and light meals',
    'fast_food': 'Quick service food establishments',
    'biergarten': 'Outdoor beer gardens',
    
    'ice_cream': 'Ice cream parlors',
    'food_court': 'Areas with multiple food vendors',
    
    # Entertainment & Nightlife categories
    'nightclub': 'Places for dancing and drinking at night',
    'social_club': 'Member-only social establishments',
    'casino': 'Gambling establishments',
    'cinema': 'Movie theaters',
    
    # Cultural & Community categories
    'arts_centre': 'Venues for various arts performances',
    'music_venue': 'Indoor venues for live contemporary music',
    'community_centre': 'Community centers for local events',
    # 'social_centre': 'Places for free and not-for-profit activities',
    
    'events_venue': 'Buildings specifically for organizing events',
    # 'marketplace': 'Public marketplaces for daily or weekly trading',
    
    # Religious & Other categories
    # 'place_of_worship': 'Churches, mosques, temples and other religious buildings',
    # 'monastery': 'Monasteries and religious living quarters',
    
    'internet_cafe': 'Cafes providing internet access',
}

# Initialize Overpass API client
# This object will handle all our queries to OpenStreetMap
API = overpy.Overpass()

def create_polygon(coordinates: List) -> Polygon:
    """
    Create a Polygon object from a list of coordinates.
    
    Args:
        coordinates (List): List of [longitude, latitude] pairs
    
    Returns:
        Polygon: Shapely Polygon object representing the boundary
    """
    polygon = Polygon(coordinates)
    return polygon

def get_polygon_coords_string(polygon: Polygon) -> str:
    """
    Convert polygon coordinates to Overpass API polygon format.
    
    The Overpass API requires coordinates in a specific string format:
    "lat1 lon1 lat2 lon2 lat3 lon3 ..."
    
    Args:
        polygon (Polygon): Shapely Polygon object
    
    Returns:
        str: Space-separated string of alternating latitude and longitude values
    """
    # Get polygon exterior coordinates
    coords = list(polygon.exterior.coords)
    
    # Convert to Overpass polygon format: "lat1 lon1 lat2 lon2 lat3 lon3 ..."
    coord_pairs = []
    for lon, lat in coords:
        # Note: Overpass expects latitude first, then longitude
        coord_pairs.extend([str(lat), str(lon)])
    
    # Join all coordinates into a single space-separated string
    polygon_string = " ".join(coord_pairs)
    return polygon_string

def point_in_polygon(polygon: Polygon, lat: float, lon: float) -> bool:
    """
    Check if a geographic point is within the polygon boundary.
    
    This function is used to filter results and keep only those
    that are actually within the city of Bari.
    
    Args:
        polygon (Polygon): The boundary polygon
        lat (float): Latitude of the point to check
        lon (float): Longitude of the point to check
    
    Returns:
        bool: True if point is inside polygon, False otherwise
    """
    # Create a Point object (note: Point takes lon, lat order)
    point = Point(lon, lat)
    
    # Check if the polygon contains this point
    return polygon.contains(point)

def is_handled_tag(tag_key: str) -> bool:
    """
    Check if a tag is already handled in the structured extraction.
    
    This function helps identify which tags are processed by our
    extraction functions and which ones are "extra" tags that should
    be stored separately.
    
    Args:
        tag_key (str): The OSM tag key to check
    
    Returns:
        bool: True if tag is already handled, False if it's an extra tag
    """
    # List of tag prefixes and exact matches that we handle
    handled_prefixes = [
        # Name variations
        'name', 'alt_name', 'official_name', 'short_name', 'loc_name', 'old_name',
        # Address information
        'addr:', 'contact:', 'phone', 'email', 'website', 'facebook', 'instagram',
        # Operating hours
        'opening_hours',
        # Service details
        'cuisine', 'diet:', 'outdoor_seating', 'wheelchair', 
        'internet_access', 'smoking', 'takeaway', 'delivery', 'capacity',
        # Business information
        'description', 'operator', 'brand', 'building', 'level', 'indoor',
        # Payment and fees
        'fee', 'charge', 'payment:', 'accepted_cards', 'cash',
        # Facilities
        'air_conditioning', 'heating', 'reservation', 'dress_code',
        'min_age', 'max_age', 'toilets', 'baby_feeding', 'changing_table',
        # Parking
        'parking', 'bicycle_parking', 'drive_through', 'drive_in',
        # Core amenity tag
        'amenity'
    ]
    
    # Check if tag matches any handled prefix or exact match
    for prefix in handled_prefixes:
        if tag_key.startswith(prefix) or tag_key == prefix.rstrip(':'):
            return True
    
    return False

def extract_address(tags: Dict) -> Optional[Dict]:
    """
    Extract comprehensive address information from OSM tags.
    
    OSM uses tags like 'addr:street', 'addr:housenumber', etc.
    This function collects all address-related tags and organizes them.
    
    Args:
        tags (Dict): Dictionary of OSM tags
    
    Returns:
        Optional[Dict]: Dictionary with cleaned address fields, or None if no address found
    """
    address = {}
    
    # Mapping of OSM address tags to clean field names
    address_keys = {
        'addr:street': 'street',
        'addr:housenumber': 'housenumber',
        'addr:housename': 'housename',
        'addr:postcode': 'postcode',
        'addr:city': 'city',
        'addr:country': 'country',
        'addr:state': 'state',
        'addr:province': 'province',
        'addr:district': 'district',
        'addr:suburb': 'suburb',
        'addr:hamlet': 'hamlet',
        'addr:place': 'place',
        'addr:unit': 'unit',
        'addr:floor': 'floor',
        'addr:door': 'door',
        'addr:full': 'full_address'
    }
    
    # Extract each address component if present
    for osm_key, clean_key in address_keys.items():
        if osm_key in tags:
            address[clean_key] = tags[osm_key]
    
    # Return None if no address information found
    return address if address else None


def extract_contact_info(tags: Dict) -> Optional[Dict]:
    """
    Extract contact information from OSM tags.
    
    Collects phone numbers, emails, websites, and social media links.
    
    Args:
        tags (Dict): Dictionary of OSM tags
    
    Returns:
        Optional[Dict]: Dictionary with contact information, or None if none found
    """
    contact = {}
    
    # List of contact-related tag keys
    contact_keys = [
        'phone', 'mobile', 'fax', 'email', 'website', 'url',
        'contact:phone', 'contact:mobile', 'contact:fax', 'contact:email', 
        'contact:website', 'contact:facebook', 'contact:instagram', 
        'contact:twitter', 'contact:youtube', 'facebook', 'instagram'
    ]
    
    # Extract each contact field if present
    for key in contact_keys:
        if key in tags:
            # Remove 'contact:' prefix for cleaner field names
            clean_key = key.replace('contact:', '')
            contact[clean_key] = tags[key]
    
    return contact if contact else None


def extract_business_info(tags: Dict) -> Optional[Dict]:
    """
    Extract business operating information from OSM tags.
    
    Includes opening hours, operator, capacity, descriptions, etc.
    
    Args:
        tags (Dict): Dictionary of OSM tags
    
    Returns:
        Optional[Dict]: Dictionary with business information, or None if none found
    """
    business = {}
    
    # List of business-related tag keys
    business_keys = [
        'opening_hours', 'opening_hours:drive_through', 'opening_hours:kitchen',
        'operator', 'owner', 'ref', 'source', 'start_date', 'end_date',
        'description', 'note', 'fixme', 'capacity', 'stars', 'rooms'
    ]
    
    # Extract each business field if present
    for key in business_keys:
        if key in tags:
            business[key] = tags[key]
    
    return business if business else None

def extract_service_features(tags: Dict) -> Optional[Dict]:
    """
    Extract service features and facilities information.
    
    Includes amenities like WiFi, outdoor seating, smoking policy,
    payment methods, etc.
    
    Args:
        tags (Dict): Dictionary of OSM tags
    
    Returns:
        Optional[Dict]: Dictionary with service features, or None if none found
    """
    services = {}
    
    # Service-related tag keys
    service_keys = [
        'internet_access', 'internet_access:fee', 'wifi',
        'outdoor_seating', 'smoking', 'takeaway', 'delivery',
        'drive_through', 'drive_in', 'air_conditioning', 'heating',
        'toilets', 'baby_feeding', 'changing_table',
        'wheelchair', 'wheelchair:description',
        'payment:cash', 'payment:cards', 'payment:bitcoin',
        'cuisine', 'diet:vegetarian', 'diet:vegan', 'diet:halal'
    ]
    
    # Extract each service feature if present
    for key in service_keys:
        if key in tags:
            services[key] = tags[key]
    
    return services if services else None


def extract_accessibility_info(tags: Dict) -> Optional[Dict]:
    """
    Extract accessibility information for people with disabilities.
    
    Args:
        tags (Dict): Dictionary of OSM tags
    
    Returns:
        Optional[Dict]: Dictionary with accessibility info, or None if none found
    """
    accessibility = {}
    
    # Accessibility-related tag keys
    access_keys = [
        'wheelchair', 'wheelchair:description',
        'blind:description', 'deaf:description',
        'tactile_paving', 'handrail', 'ramp', 'elevator',
        'disabled_parking'
    ]
    
    for key in access_keys:
        if key in tags:
            accessibility[key] = tags[key]
    
    return accessibility if accessibility else None


def extract_location_info(tags: Dict) -> Optional[Dict]:
    """
    Extract location-specific information.
    
    Includes building details, floor level, indoor/outdoor, etc.
    
    Args:
        tags (Dict): Dictionary of OSM tags
    
    Returns:
        Optional[Dict]: Dictionary with location info, or None if none found
    """
    location = {}
    
    # Location-related tag keys
    location_keys = [
        'building', 'building:levels', 'level', 'floor',
        'indoor', 'room', 'entrance'
    ]
    
    for key in location_keys:
        if key in tags:
            location[key] = tags[key]
    
    return location if location else None


def extract_extra_tags(tags: Dict) -> Optional[Dict]:
    """
    Extract tags that aren't handled by other extraction functions.
    
    This catches any additional, non-standard, or special tags
    that might be useful but don't fit into standard categories.
    
    Args:
        tags (Dict): Dictionary of OSM tags
    
    Returns:
        Optional[Dict]: Dictionary with extra tags, or None if none found
    """
    extra = {}
    
    # Collect all tags that aren't handled by standard extraction
    for key, value in tags.items():
        if not is_handled_tag(key):
            extra[key] = value
    
    return extra if extra else None

def process_amenity(amenity_type: str, polygon: Polygon, polygon_str: str, 
                   max_retries: int = 3) -> Tuple[List[Dict], set]:
    """
    Query OpenStreetMap for a specific amenity type and process results.
    
    This is the main function that:
    1. Builds and executes Overpass API queries
    2. Extracts structured data from OSM tags
    3. Filters results to include only those within the city boundary
    4. Formats the output in a clean, consistent structure
    
    Args:
        amenity_type (str): Type of amenity to search for (e.g., 'nightclub', 'ice_cream')
        polygon (Polygon): Shapely polygon representing the city boundary
        polygon_str (str): Polygon coordinates in Overpass API format
        max_retries (int): Maximum number of retry attempts for failed queries
    
    Returns:
        Tuple[List[Dict], set]: 
            - List of processed amenity dictionaries
            - Set of extra tag keys found (for debugging/analysis)
    """
    results = []
    extra_keys = set()
    
    # Build Overpass QL query
    # This query searches for nodes and ways (areas) with the specified amenity tag
    # within the polygon boundary
    query = f"""
    [out:json];
    (
        node["amenity"="{amenity_type}"](poly:"{polygon_str}");
        way["amenity"="{amenity_type}"](poly:"{polygon_str}");
    );
    out center;
    """
    
    # Retry logic for robustness
    for attempt in range(max_retries):
        try:
            # Execute the query
            result = API.query(query)
            
            # Process nodes (point features)
            for node in result.nodes:
                # Check if the node is within our polygon boundary
                if point_in_polygon(polygon, float(node.lat), float(node.lon)):
                    # Extract all information categories
                    name = node.tags.get('name', None)
                    address = extract_address(node.tags)
                    contact = extract_contact_info(node.tags)
                    business = extract_business_info(node.tags)
                    services = extract_service_features(node.tags)
                    accessibility = extract_accessibility_info(node.tags)
                    location = extract_location_info(node.tags)
                    extra = extract_extra_tags(node.tags)
                    
                    # Track which extra tags appear (for analysis)
                    if extra:
                        extra_keys.update(extra.keys())
                    
                    # Build formatted address string if available
                    formatted_address = None
                    if address:
                        addr_parts = []
                        if 'street' in address:
                            addr_parts.append(address['street'])
                        if 'housenumber' in address:
                            addr_parts.append(address['housenumber'])
                        if 'postcode' in address:
                            addr_parts.append(address['postcode'])
                        if 'city' in address:
                            addr_parts.append(address['city'])
                        if addr_parts:
                            formatted_address = ' '.join(addr_parts)
                    
                    # Consolidate contact info into a single string if available
                    formatted_contact = None
                    if contact:
                        contact_parts = []
                        for key in ['phone', 'mobile', 'email', 'website']:
                            if key in contact:
                                contact_parts.append(contact[key])
                        if contact_parts:
                            formatted_contact = ' '.join(contact_parts)
                    
                    # Consolidate business info
                    formatted_business = None
                    if business:
                        business_parts = []
                        for key in ['opening_hours', 'operator', 'description']:
                            if key in business:
                                business_parts.append(business[key])
                        if business_parts:
                            formatted_business = ' '.join(business_parts)
                    
                    # Create amenity dictionary with all extracted data
                    amenity_dict = {
                        'Spazio': name,
                        'Categoria': amenity_type,
                        'latitudine': float(node.lat),
                        'longitudine': float(node.lon)
                    }
                    
                    # Add optional fields only if they have data
                    if formatted_address:
                        amenity_dict['address'] = formatted_address
                    if formatted_contact:
                        amenity_dict['contact'] = formatted_contact
                    if formatted_business:
                        amenity_dict['business_info'] = formatted_business
                    if services:
                        amenity_dict['services'] = 'yes'  # Simplified flag
                    if accessibility:
                        amenity_dict['accessibility'] = accessibility.get('wheelchair', 'unknown')
                    if location:
                        amenity_dict['location_info'] = location.get('level', location.get('floor', '0'))
                    
                    results.append(amenity_dict)
            
            # Process ways (area features) - similar to nodes
            for way in result.ways:
                # Ways have a center point for location
                if hasattr(way, 'center_lat') and hasattr(way, 'center_lon'):
                    center_lat = float(way.center_lat)
                    center_lon = float(way.center_lon)
                    
                    if point_in_polygon(polygon, center_lat, center_lon):
                        # Extract all information (same as nodes)
                        name = way.tags.get('name', None)
                        address = extract_address(way.tags)
                        contact = extract_contact_info(way.tags)
                        business = extract_business_info(way.tags)
                        services = extract_service_features(way.tags)
                        accessibility = extract_accessibility_info(way.tags)
                        location = extract_location_info(way.tags)
                        extra = extract_extra_tags(way.tags)
                        
                        if extra:
                            extra_keys.update(extra.keys())
                        
                        # Build formatted strings (same as nodes)
                        formatted_address = None
                        if address:
                            addr_parts = []
                            if 'street' in address:
                                addr_parts.append(address['street'])
                            if 'housenumber' in address:
                                addr_parts.append(address['housenumber'])
                            if 'postcode' in address:
                                addr_parts.append(address['postcode'])
                            if 'city' in address:
                                addr_parts.append(address['city'])
                            if addr_parts:
                                formatted_address = ' '.join(addr_parts)
                        
                        formatted_contact = None
                        if contact:
                            contact_parts = []
                            for key in ['phone', 'mobile', 'email', 'website']:
                                if key in contact:
                                    contact_parts.append(contact[key])
                            if contact_parts:
                                formatted_contact = ' '.join(contact_parts)
                        
                        formatted_business = None
                        if business:
                            business_parts = []
                            for key in ['opening_hours', 'operator', 'description']:
                                if key in business:
                                    business_parts.append(business[key])
                            if business_parts:
                                formatted_business = ' '.join(business_parts)
                        
                        amenity_dict = {
                            'Spazio': name,
                            'Categoria': amenity_type,
                            'latitudine': center_lat,
                            'longitudine': center_lon
                        }
                        
                        if formatted_address:
                            amenity_dict['address'] = formatted_address
                        if formatted_contact:
                            amenity_dict['contact'] = formatted_contact
                        if formatted_business:
                            amenity_dict['business_info'] = formatted_business
                        if services:
                            amenity_dict['services'] = 'yes'
                        if accessibility:
                            amenity_dict['accessibility'] = accessibility.get('wheelchair', 'unknown')
                        if location:
                            amenity_dict['location_info'] = location.get('level', location.get('floor', '0'))
                        
                        results.append(amenity_dict)
            
            # If we got here, query was successful - break retry loop
            break
            
        except Exception as e:
            print(f"Attempt {attempt + 1} failed for {amenity_type}: {str(e)}")
            if attempt < max_retries - 1:
                # Wait before retrying (exponential backoff)
                time.sleep(2 ** attempt)
            else:
                print(f"All attempts failed for {amenity_type}")
    
    return results, extra_keys

# Create polygon from coordinates
polygon = create_polygon(COORDINATES)
polygon_str = get_polygon_coords_string(polygon)

# Initialize result collectors
all_results = []  # Will contain all amenities from all queries
all_extra_keys = set()  # Will track all unique extra tag keys found

# Process each amenity type
for amenity_type, description in AMENITIES.items():
    print(f"Processing: {amenity_type}")
    
    # Query and process this amenity type
    amenities, extra_keys = process_amenity(amenity_type, polygon, polygon_str)
    
    # Add results to our collection
    all_results.extend(amenities)
    all_extra_keys.update(extra_keys)
    
    # Small delay between queries to be respectful to the API
    time.sleep(1)

print(f"\nQuery completate! Trovati {len(all_results)} amenities totali.")

Processing: bar
Processing: pub
Processing: restaurant
Processing: cafe
Processing: fast_food
Processing: biergarten
Processing: ice_cream
Attempt 1 failed for ice_cream: Too many requests
Processing: food_court
Processing: nightclub
Processing: social_club
Processing: casino
Processing: cinema
Processing: arts_centre
Processing: music_venue
Processing: community_centre
Processing: events_venue
Processing: internet_cafe

Query completate! Trovati 884 amenities totali.

# Display extra tags found and a sample of results
otherKeys = all_extra_keys
result = all_results

# Show what we found
otherKeys, result[:5]  # Display first 5 results as sample

({'access',
  'bar',
  'barrier',
  'branch',
  'brewery',
  'category',
  'check_date',
  'check_date:opening_hours',
  'cinema:3D',
  'cocktails',
  'community_centre',
  'community_centre:for',
  'created_by',
  'currency:XBT',
  'denomination',
  'disused:amenity',
  'dog',
  'end',
  'entrance',
  'fast_food',
  'fax',
  'fixme',
  'food',
  'guest_house',
  'highchair',
  'image',
  'kids_area',
  'landuse',
  'layer',
  'lottery',
  'microbrewery',
  'mobile',
  'note',
  'open_air',
  'outdoor',
  'pets',
  'pets_allowed',
  'ref:vatin',
  'screen',
  'self_service',
  'shop',
  'source',
  'start_date',
  'survey:date',
  'swimming_pool',
  'tickets:public_transport',
  'tobacco',
  'tourism',
  'wifi'},
 [{'Spazio': 'Città vecchia',
   'Categoria': 'bar',
   'latitudine': 41.1277817,
   'longitudine': 16.8714163,
   'services': 'yes',
   'accessibility': 'limited'},
  {'Spazio': 'Las Vegas',
   'Categoria': 'bar',
   'latitudine': 41.0873067,
   'longitudine': 17.0003784},
  {'Spazio': 'Miramare',
   'Categoria': 'bar',
   'latitudine': 41.0875655,
   'longitudine': 16.9991167,
   'services': 'yes',
   'accessibility': 'yes'},
  {'Spazio': 'Weilà Ristorante Bar Pizzeria',
   'Categoria': 'bar',
   'latitudine': 41.1133544,
   'longitudine': 16.8811288,
   'address': 'Bari',
   'services': 'yes',
   'accessibility': 'yes'},
  {'Spazio': None,
   'Categoria': 'bar',
   'latitudine': 41.0966018,
   'longitudine': 16.8587009}])

# Translation dictionary from English amenity types to Italian
traduzioni = {
    'arts centre': 'Centro culturale',
    'bar': 'Bar',
    'cafe': 'Caffè',
    'casino': 'Casinò',
    'cinema': 'Cinema',
    'community centre': 'Centro sociale',
    'events venue': 'Sede eventi',
    'fast food': 'Fast food',
    'food court': 'Area ristorazione',
    'ice cream': 'Gelateria',
    'internet cafe': 'Internet café',
    'marketplace': 'Mercato',
    'monastery': 'Monastero',
    'nightclub': 'Discoteca',
    'place of worship': 'Luogo di culto',
    'pub': 'Pub',
    'restaurant': 'Ristorante',
    'music venue': 'Locale musicale',
    'social centre': 'Centro sociale',
}

# Apply translations to all results
for x in result:
    # Use translated name if available, otherwise keep original
    x["Categoria"] = traduzioni.get(x["Categoria"], x["Categoria"])

[
  {
    "Spazio": "Nome del locale",
    "Categoria": "Gelateria",
    "latitudine": 41.1234,
    "longitudine": 16.8765,
    "address": "Via Roma 123 70122 Bari",
    "contact": "+39 080 1234567 info@example.com",
    "business_info": "Mo-Su 10:00-22:00"
  },
  ...
]

# Export all amenities to JSON file
with open("Amenities.json", "w", encoding='utf-8') as f:
    json.dump(result, f, indent=4, ensure_ascii=False)

# Success message
print(f"✅ File 'Amenities.json' creato con successo!")
print(f"📊 Totale amenities esportate: {len(result)}")
print(f"🗂️ Categorie presenti: {len(set(x['Categoria'] for x in result))}")

✅ File 'Amenities.json' creato con successo!
📊 Totale amenities esportate: 884
🗂️ Categorie presenti: 14

🗺️ Query OpenStreetMap: Raccolta Dati Amenities Bari¶

Obiettivo del Notebook¶

Cos'è OpenStreetMap?¶

Cosa fa questo notebook?¶

Tecnologie utilizzate:¶

File necessari:¶

File generati:¶

1. Importazione Librerie¶

2. Caricamento Coordinate Geografiche¶

Perché usiamo coordinate?¶

3. Definizione Amenities da Cercare¶

Cosa sono le "Amenities"?¶

Note sul dizionario:¶

4. Inizializzazione API Overpass¶

Cos'è l'API Overpass?¶

5. Funzioni Helper per Geometria¶

Perché sono importanti?¶

6. Funzione per Identificare Tag Gestiti¶

Cosa sono i "Tag" in OpenStreetMap?¶

Perché questa funzione?¶

7. Funzioni di Estrazione Dati Strutturati¶

Perché organizzare i dati?¶

Categorie di estrazione:¶

8. Funzioni di Estrazione Avanzata¶

9. Funzione Principale di Elaborazione¶

Processo passo-passo:¶

Gestione errori:¶

10. Esecuzione Query per Tutte le Amenities¶

Processo:¶

Nota:¶

11. Analisi Risultati¶

12. Traduzione Categorie in Italiano¶

Dizionario di traduzione:¶

13. Esportazione Risultati Finali¶

Struttura del file:¶

Parametri di esportazione:¶

🎉 Processo Completato!¶