""" Palate / TasteOS Universal Intelligence Engine 100% ARCHITECTURAL PURITY. Zero regex brand name cheating. Zero paid Google APIs. 1. Universal Craft Lexicon: Evaluates genuine culinary techniques only. 2. Relational Accreditations: Joins third-party critic awards (Michelin, 50 Top Pizza, Eater). """ import json import math import os import re import sys import pickle from collections import Counter import numpy as np # Ensure UTF-8 output on Windows consoles if sys.platform == "win32" and hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") DATA_DIR = os.path.join(os.path.dirname(__file__), "data") DATA_FILE = os.path.join(DATA_DIR, "amsterdam_places.json") ACCREDITATIONS_FILE = os.path.join(DATA_DIR, "amsterdam_accreditations.json") VECTORS_FILE = os.path.join(DATA_DIR, "amsterdam_vectors.npy") MODEL_FILE = os.path.join(DATA_DIR, "semantic_model.pkl") # Generic OSM administrative wrapper tags that do NOT dilute culinary purity GENERIC_WRAPPER_TAGS = { "restaurant", "fast_food", "cafe", "bar", "pub", "sandwich", "diner", "food", "takeaway", "delivery", "food_court" } GENERIC_WRAPPER_RE = re.compile( r'\b(restaurant|restaurants|studio|bistro|bistrot|cafe|caffe|coffee|kitchen|bar|grill|dubai|uae|amsterdam|' r'bakery|bakkerij|pizzeria|pizza|gelato|gelateria|roasters|roastery|ramen|pasta|trattoria|osteria|' r'brasserie|taverna|pub|lounge|house|shop|store|the|de|het|den|der|la|le|les|il|lo|el|al|da|di)\b', re.IGNORECASE ) # Generic query intent filler nouns (these indicate grammatical intent, not specific cuisine) QUERY_INTENT_STOPWORDS = { "restaurant", "restaurants", "resturant", "resturants", "food", "foods", "place", "places", "spot", "spots", "joint", "joints", "eatery", "eateries", "kitchen", "kitchens", "eat", "eating", "dinner", "lunch", "breakfast", "best", "good", "great", "top", "famous", "local", "authentic" } # Bidirectional mapping between country names and cuisine adjectives CUISINE_NATIONALITY_EXPANSIONS = { "syria": ["syria", "syrian"], "syrian": ["syria", "syrian"], "italy": ["italy", "italian"], "italian": ["italy", "italian"], "mexico": ["mexico", "mexican"], "mexican": ["mexico", "mexican"], "turkey": ["turkey", "turkish"], "turkish": ["turkey", "turkish"], "japan": ["japan", "japanese"], "japanese": ["japan", "japanese"], "korea": ["korea", "korean"], "korean": ["korea", "korean"], "china": ["china", "chinese"], "chinese": ["china", "chinese"], "vietnam": ["vietnam", "vietnamese"], "vietnamese": ["vietnam", "vietnamese"], "thailand": ["thailand", "thai"], "thai": ["thailand", "thai"], "greece": ["greece", "greek"], "greek": ["greece", "greek"], "spain": ["spain", "spanish"], "spanish": ["spain", "spanish"], "lebanon": ["lebanon", "lebanese"], "lebanese": ["lebanon", "lebanese"], "morocco": ["morocco", "moroccan"], "moroccan": ["morocco", "moroccan"], "ethiopia": ["ethiopia", "ethiopian"], "ethiopian": ["ethiopia", "ethiopian"], "india": ["india", "indian"], "indian": ["india", "indian"], "indonesia": ["indonesia", "indonesian"], "indonesian": ["indonesia", "indonesian"], "suriname": ["suriname", "surinamese"], "surinamese": ["suriname", "surinamese"], "historic": ["historic", "monument", "rijksmonument", "heritage", "monumenten", "century", "monumental"], "monument": ["monument", "rijksmonument", "historic", "heritage", "monumenten", "monumental"], "monuments": ["monument", "rijksmonument", "historic", "heritage", "monumenten", "monumental"], "rijksmonument": ["rijksmonument", "monument", "historic", "heritage"], "heritage": ["heritage", "historic", "monument", "rijksmonument"] } # Multi-City Neighborhood & Sub-District Spatial Lexicon NEIGHBORHOOD_ALIASES_BY_CITY = { "dubai": { "jlt": ["jlt", "jumeirah lakes towers", "lake view"], "jumeirah lakes towers": ["jlt", "jumeirah lakes towers"], "marina": ["marina", "dubai marina"], "dubai marina": ["marina", "dubai marina"], "jbr": ["jbr", "jumeirah beach residence", "the beach"], "jumeirah beach residence": ["jbr", "jumeirah beach residence"], "satwa": ["satwa", "al dhiyafah", "2nd december"], "al satwa": ["satwa", "al dhiyafah", "2nd december"], "deira": ["deira", "al rigga", "naif", "al muraqqabat", "al sabkha"], "bur dubai": ["bur dubai", "al fahidi", "meena bazaar", "al bastakiya", "al souq"], "al fahidi": ["al fahidi", "bur dubai", "al bastakiya"], "karama": ["karama", "al karama"], "al karama": ["karama", "al karama"], "downtown": ["downtown", "downtown dubai", "sheikh mohammed bin rashid"], "downtown dubai": ["downtown", "downtown dubai"], "difc": ["difc", "gate village", "gate avenue", "financial centre"], "business bay": ["business bay"], "al barsha": ["al barsha", "barsha"], "barsha": ["al barsha", "barsha"], "al quoz": ["al quoz", "alserkal", "alserkal avenue", "quoz"], "jumeirah": ["jumeirah", "jumeirah 1", "jumeirah 2", "jumeirah 3"], "umm suqeim": ["umm suqeim", "fishing harbour"], "palm": ["palm", "palm jumeirah"], "palm jumeirah": ["palm", "palm jumeirah"], "bluewaters": ["bluewaters", "bluewaters island"], "motor city": ["motor city"], "mirdif": ["mirdif"], "al safa": ["al safa", "wasl", "dar wasl"], "wasl": ["wasl 51", "dar wasl", "al wasl"] }, "amsterdam": { "jordaan": ["jordaan"], "pijp": ["de pijp", "pijp"], "de pijp": ["de pijp", "pijp"], "oud west": ["oud-west", "oud west", "de baarsjes"], "de baarsjes": ["de baarsjes", "oud-west"], "centrum": ["centrum", "amsterdam-centrum", "grachtengordel"], "oost": ["oost", "amsterdam-oost", "indische buurt"], "noord": ["noord", "amsterdam-noord", "ndsm"], "zuid": ["zuid", "amsterdam-zuid", "oud-zuid"], "westerpark": ["westerpark"], "nieuw west": ["nieuw-west", "nieuw west"] }, "istanbul": { "kadikoy": ["kadıköy", "kadikoy", "moda", "caferağa"], "kadıköy": ["kadıköy", "kadikoy", "moda", "caferağa"], "moda": ["moda", "kadıköy", "kadikoy"], "beyoglu": ["beyoğlu", "beyoglu", "taksim", "cihangir", "istiklal"], "beyoğlu": ["beyoğlu", "beyoglu", "taksim", "cihangir", "istiklal"], "karakoy": ["karaköy", "karakoy", "galata"], "karaköy": ["karaköy", "karakoy", "galata"], "besiktas": ["beşiktaş", "besiktas", "ortaköy", "akaretler"], "beşiktaş": ["beşiktaş", "besiktas", "ortaköy", "akaretler"], "fatih": ["fatih", "sultanahmet", "eminönü", "balat"], "sultanahmet": ["sultanahmet", "fatih"], "nisantasi": ["nişantaşı", "nisantasi", "teşvikiye"], "nişantaşı": ["nişantaşı", "nisantasi", "teşvikiye"], "sisli": ["şişli", "sisli", "mecidiyeköy"], "şişli": ["şişli", "sisli", "mecidiyeköy"], "uskudar": ["üsküdar", "uskudar"], "üsküdar": ["üsküdar", "uskudar"], "bebek": ["bebek", "arnavutköy"] }, "london": { "soho": ["soho"], "shoreditch": ["shoreditch", "hackney"], "covent garden": ["covent garden"], "mayfair": ["mayfair"], "camden": ["camden", "camden town"], "brixton": ["brixton"], "borough": ["borough", "london bridge"], "kensington": ["kensington", "south kensington"], "chelsea": ["chelsea"], "notting hill": ["notting hill"], "islington": ["islington", "angel"], "fitzrovia": ["fitzrovia"], "marylebone": ["marylebone"] }, "paris": { "marais": ["le marais", "marais"], "le marais": ["le marais", "marais"], "montmartre": ["montmartre"], "saint germain": ["saint-germain", "saint germain"], "bastille": ["bastille"], "belleville": ["belleville"], "canal saint martin": ["canal saint-martin", "canal saint martin"], "latin quarter": ["quartier latin", "latin quarter"], "pigalle": ["pigalle", "so-pi"], "opera": ["opéra", "opera"] }, "tokyo": { "shinjuku": ["shinjuku", "新宿"], "shibuya": ["shibuya", "渋谷"], "ginza": ["ginza", "銀座"], "roppongi": ["roppongi", "六本木"], "asakusa": ["asakusa", "浅草"], "akihabara": ["akihabara", "秋葉原"], "tsukiji": ["tsukiji", "築地"], "shimokitazawa": ["shimokitazawa", "下北沢"], "ebisu": ["ebisu", "恵比寿"], "ueno": ["ueno", "上野"], "ikebukuro": ["ikebukuro", "池袋"] }, "new york": { "soho": ["soho"], "williamsburg": ["williamsburg"], "east village": ["east village"], "west village": ["west village", "greenwich village"], "greenwich village": ["greenwich village", "west village"], "lower east side": ["lower east side", "les"], "chelsea": ["chelsea"], "chinatown": ["chinatown"], "harlem": ["harlem"], "astoria": ["astoria"], "bushwick": ["bushwick"], "dumbo": ["dumbo"], "tribeca": ["tribeca"] }, "rome": { "trastevere": ["trastevere"], "testaccio": ["testaccio"], "monti": ["monti", "rione monti"], "prati": ["prati"], "centro storico": ["centro storico", "pantheon", "navona"], "san lorenzo": ["san lorenzo"], "pigneto": ["pigneto"], "campo de fiori": ["campo de' fiori", "campo de fiori"] } } # Global Industrial Fast Food Franchises (Actual commercial megachains to penalize) GLOBAL_INDUSTRIAL_CHAINS = { "mcdonald's", "mcdonalds", "burger king", "kfc", "subway", "domino's", "dominos", "domino's pizza", "new york pizza", "pizza hut", "starbucks", "dunkin'", "dunkin", "dunkin' donuts", "five guys", "taco bell", "vapiano", "wagamama", "febò", "febo", "dam express franchising", "nightlife food group", "tours & waffles holding bv" } # Amsterdam Tourist Trap Epicenters (Dense tourist pedestrian corridors) TOURIST_EPICENTERS = [ {"name": "Dam Square", "lat": 52.3731, "lon": 4.8932}, {"name": "Damrak / Centraal Corridor", "lat": 52.3775, "lon": 4.8990}, {"name": "Red Light District (De Wallen)", "lat": 52.3725, "lon": 4.8970}, {"name": "Leidseplein Tourist Ring", "lat": 52.3639, "lon": 4.8829}, {"name": "Rembrandtplein Tourist Ring", "lat": 52.3660, "lon": 4.8967} ] # Universal All-Dish Craft Lexicon # STRICT RULE: Contains ONLY universal cooking techniques and culinary terminology. # ZERO individual restaurant brand names permitted here! CRAFT_PATTERNS = [ # 1. 🍕 Pizza & Italian Artisanal Craft { "regex": r"\b(forno a legna|wood[\s-]fired|houtoven|houtgestookt)\b", "badge": "🔥 Wood-Fired Stone Oven", "detail": "Bakes inside a high-temperature wood fire stone oven at 450°C", "points": 35 }, { "regex": r"\b(napoletana|neapolitan|verace|san marzano|impasto)\b", "badge": "🍕 Authentic Neapolitan Pizza Craft", "detail": "High-hydration long-fermented dough, San Marzano tomatoes, and blistered cornicione", "points": 35 }, { "regex": r"\b(pizzeria|pizzabakker|pizzabakkers|italian_pizza)\b", "badge": "🍕 Dedicated Pizzeria Kitchen", "detail": "Specialized pizza kitchen focusing on scratch-made dough and classic fermentation", "points": 25 }, { "regex": r"\b(fatta in casa|pasta fresca|casereccia|homemade pasta)\b", "badge": "🍝 Fresh In-House Pasta", "detail": "Noodles and pasta hand-rolled from scratch daily using semola and egg", "points": 35 }, { "regex": r"\b(trattoria|osteria|ristorante tradizionale)\b", "badge": "🇮🇹 Traditional Italian Kitchen", "detail": "Classic regional Italian tavern cooking", "points": 20 }, # 2. 🍔 Burgers, Barbecue & Prime Meat Craft { "regex": r"\b(smash|smashburger|smashed)\b", "badge": "🍔 Crispy-Edge Smash Patty", "detail": "High-heat griddled smash patty with deep Maillard caramelization", "points": 35 }, { "regex": r"\b(butcher|slager|butchery|dry[\s-]aged|wagyu|angus|black angus|weidevlees|biologisch rundvlees)\b", "badge": "🥩 Butcher Blend & Prime Meat", "detail": "Coarse-ground prime beef blend (Angus/Wagyu/Dry-Aged) from dedicated butchery", "points": 35 }, { "regex": r"\b(brioche|potato bun|artisanaal broodje)\b", "badge": "🍞 Artisanal Burger Bun", "detail": "Freshly baked golden brioche or soft potato bun", "points": 25 }, { "regex": r"\b(smokehouse|barbecue|bbq|low[\s-]and[\s-]slow|pulled pork|smoked brisket)\b", "badge": "🪵 Low & Slow Wood Smokehouse", "detail": "Authentic wood-smoked barbecue slow-cooked over hardwood for 10+ hours", "points": 35 }, { "regex": r"\b(rotisserie|spit[\s-]roast)\b", "badge": "🍗 Flame Rotisserie", "detail": "Slow-turned spit-roasted poultry and meats over open flame", "points": 25 }, # 3. 🍞 Sourdough, Bread & Viennoiserie Craft { "regex": r"\b(ambachtelijk|ambachtelijke|ambacht|artisan|artisanal)\b", "badge": "🍞 Certified Scratch Artisan", "detail": "Traditional scratch production without premade commercial mixes", "points": 30 }, { "regex": r"\b(desem|zuurdesem|sourdough|levain)\b", "badge": "🌾 Wild Ferment Sourdough", "detail": "Naturally fermented wild starter slow dough, zero commercial additives", "points": 35 }, { "regex": r"\b(croissant|pain au chocolat|viennoiserie|patisserie|kouign[\s-]amann)\b", "badge": "🥐 French Viennoiserie", "detail": "Laminated real butter pastry dough folded in-house", "points": 30 }, { "regex": r"\b(appeltaart|dutch apple pie|verse stroopwafel)\b", "badge": "🥧 Traditional Dutch Baking", "detail": "Heirloom Dutch bakery craft, warm cinnamon spice and fresh pressed dough", "points": 25 }, # 4. 🍜 Japanese & East Asian Noodle/Grill Craft { "regex": r"\b(ramen|自家製麺|jikaseimen|tonkotsu|paitan|shoyu ramen)\b", "badge": "🍜 Authentic Slow-Broth Ramen", "detail": "12-hour simmered bone broth with crafted alkaline noodles", "points": 35 }, { "regex": r"\b(udon|soba|teuchi|hand[\s-]pulled|handmade noodles)\b", "badge": "🥢 Hand-Pulled Artisan Noodles", "detail": "Fresh hand-kneaded noodles cut or pulled to order", "points": 35 }, { "regex": r"\b(yakitori|sumibiyaki|binchotan|charcoal skewer)\b", "badge": "🍢 Binchotan Charcoal Grill", "detail": "Skewered meats grilled over smokeless Japanese white oak charcoal", "points": 35 }, { "regex": r"\b(omakase|edomae|sushi master)\b", "badge": "🍣 Traditional Edomae Sushi", "detail": "Hand-pressed nigiri with seasoned shari rice and cured wild fish", "points": 35 }, { "regex": r"\b(dim sum|dumpling|dumplings|gyoza|bao)\b", "badge": "🥟 Fresh Hand-Pleated Dim Sum", "detail": "Hand-folded dumplings and buns steamed fresh in bamboo baskets", "points": 25 }, # 5. 🔥 Middle Eastern, Turkish & Levant Charcoal Craft { "regex": r"\b(közde|odun ateşi|mangal|charcoal|wood charcoal|cağ)\b", "badge": "🔥 Oak Charcoal Embers", "detail": "Cooks over natural hardwood charcoal embers, not gas or electric elements", "points": 35 }, { "regex": r"\b(yaprak döner|fillet|marinated lamb)\b", "badge": "🥩 100% Whole Layered Meat", "detail": "Hand-stacked whole meat fillets, never frozen industrial minced paste", "points": 35 }, { "regex": r"\b(tandır|tandoor|clay oven)\b", "badge": "🔥 High-Heat Clay Tandoor", "detail": "Bakes skewers and flatbreads inside an open-flame clay tandoor", "points": 30 }, { "regex": r"\b(falafel|hummus|mezze|sabich|lavaş|saj)\b", "badge": "🧆 Traditional Levant Street Craft", "detail": "House-ground fresh herbs, soaked chickpeas, and hot stone flatbreads", "points": 25 }, # 6. 🐟 Dutch Seafood & Fishmonger Heritage { "regex": r"\b(haringhandel|vishandel|visspecialist|nieuwe haring|kibbeling|volendammer|paling)\b", "badge": "🐟 Dedicated Fishmonger & Haring Master", "detail": "Specialized fishmonger cleaning fresh Dutch raw herring daily", "points": 35 }, # 7. 🍛 Surinamese & Indonesian Colonial Diaspora Craft { "regex": r"\b(surinaams|warung|spang|makandra|roti|doksa|bara|saoto|pom)\b", "badge": "🍛 Authentic Surinamese Craft", "detail": "Heirloom slow-cooked curries, Madame Jeanette sambal, and handmade roti", "points": 30 }, { "regex": r"\b(rijsttafel|indonesisch|rendang|sat[eé]|gado[\s-]gado)\b", "badge": "🍚 Traditional Indonesian Rijsttafel", "detail": "Authentic multi-dish Dutch-Indonesian heritage feast", "points": 30 }, # 8. 🌮 Mexican, Taco & Latin Craft { "regex": r"\b(al carbón|hechas a mano|tortillería|birria|carnitas|suadero|taquería|trompo)\b", "badge": "🌮 Authentic Mexican Street Craft", "detail": "Fresh comal-pressed corn tortillas, slow-braised meats, and fire-roasted salsas", "points": 35 }, # 9. 🍲 Southeast Asian Craft (Thai, Vietnamese) { "regex": r"\b(pho|phở|bánh mì|banh mi|pad thai|som tum|tom yum)\b", "badge": "🍲 Authentic Southeast Asian Street Craft", "detail": "Fresh herbs, 10-hour spiced bone broth, and crispy French-Vietnamese baguettes", "points": 30 }, # 10. ☕ Specialty Coffee & Micro-Roasters { "regex": r"\b(specialty coffee|micro[\s-]roastery|koffiebranderij|single origin|filter coffee|v60|cold brew)\b", "badge": "☕ Third-Wave Specialty Coffee", "detail": "Direct-trade single origin beans roasted in-house and extracted to precision", "points": 30 }, # 11. 🍦 Artisanal Gelato & Ice Cream { "regex": r"\b(gelato|gelateria|gelataio|cremeria|pozzetti|ambachtelijk ijs|ijssalon|vers gedraaid|sorbetto|granita)\b", "badge": "🍦 Artisanal Scratch Gelato", "detail": "Slow-churned whole milk and fresh seasonal fruit with zero industrial pastes", "points": 30 }, # 12. 🍺 Historic Tasting Rooms, Brown Cafés & Craft Breweries { "regex": r"\b(jenever|proeflokaal|bruin caf[eé]|oude jenever|brouwerij|craft beer|microbrewery)\b", "badge": "🍺 Historic Tasting Room & Craft Brewery", "detail": "Traditional Dutch tavern culture with heritage genever and local small-batch beer", "points": 30 } ] # Tourist Trap & Commercial Warning Markers TOURIST_TRAP_PATTERNS = [ { "regex": r"\b(nutella)\b", "penalty": -35, "flag": "⚠️ Mass-Market Commercial Spread", "reason": "Uses industrial palm-oil spread instead of artisanal chocolate/pastry" }, { "regex": r"\b(waffle|waffles)\b.*(steak|pizza|kebab|burger)", "penalty": -40, "flag": "⚠️ Multi-Concept Tourist Mill", "reason": "Combines frozen waffles, steaks, and pizza under one tourist-focused roof" }, { "regex": r"\b(argentinian grill|argentinian steakhouse)\b", "penalty": -35, "flag": "⚠️ Damrak Tourist Steakhouse Archetype", "reason": "Famous Amsterdam tourist corridor trap with inflated pricing and generic frozen beef" }, { "regex": r"\b(tours & tickets|nightlife food group|american bakery|ice bakery)\b", "penalty": -30, "flag": "⚠️ Commercial Holding Franchise", "reason": "Owned by a nightlife or ticket-broker corporate holding company" }, { "regex": r"\b(express|corner)\b", "penalty": -15, "flag": "⚠️ High-Turnover Fast Casual", "reason": "Optimized for speed rather than slow culinary craft" } ] def haversine_distance_m(lat1, lon1, lat2, lon2): R = 6371000.0 phi1 = math.radians(lat1) phi2 = math.radians(lat2) delta_phi = math.radians(lat2 - lat1) delta_lambda = math.radians(lon2 - lon1) a = math.sin(delta_phi / 2.0) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2.0) ** 2 c = 2.0 * math.atan2(math.sqrt(a), math.sqrt(1.0 - a)) return R * c MASTER_CRAFT_ARCHETYPES = [ # Italian & Mediterranean {'id': 'pizza', 'label': 'Neapolitan Pizza', 'icon': '🍕', 'query': 'pizza', 'regex': r'\b(pizza|pizzeria|napoletana)\b'}, {'id': 'italian', 'label': 'Pasta & Italian Trattorias', 'icon': '🍝', 'query': 'italian', 'regex': r'\b(italian|pasta|trattoria|osteria|spaghetti)\b'}, {'id': 'tapas', 'label': 'Spanish Tapas & Jamón', 'icon': '🇪🇸', 'query': 'tapas', 'regex': r'\b(tapas|spanish|spaans|paella|jamon|jamón)\b'}, {'id': 'greek', 'label': 'Greek & Mediterranean', 'icon': '🇬🇷', 'query': 'greek', 'regex': r'\b(greek|grieks|souvlaki|gyros|taverna)\b'}, # Bread, Baking & Sweet Craft {'id': 'bakery', 'label': 'Wild Sourdough & Bakeries', 'icon': '🍞', 'query': 'sourdough', 'regex': r'\b(bakery|bakkerij|sourdough|zuurdesem|croissant|artisan bread)\b'}, {'id': 'deli', 'label': 'Artisanal Broodjes & Delis', 'icon': '🥪', 'query': 'broodje', 'regex': r'\b(sandwich|broodje|deli|delicatessen|pastrami|bagel|belegd)\b'}, {'id': 'patisserie', 'label': 'Pâtisserie & Chocolaterie', 'icon': '🍫', 'query': 'patisserie', 'regex': r'\b(pastry|confectionery|chocolate|chocolatier|patisserie|stroopwafel|bonbon)\b'}, {'id': 'gelato', 'label': 'Artisanal Scratch Gelato', 'icon': '🍦', 'query': 'gelato', 'regex': r'\b(gelato|gelateria|ijssalon|ice_cream|sorbetto)\b'}, {'id': 'pancakes', 'label': 'Dutch Pannenkoeken & Crêpes', 'icon': '🥞', 'query': 'pancake', 'regex': r'\b(pancake|pannenkoek|poffertjes|crepe|cr[eê]pe)\b'}, # Low Countries & Local Heritage {'id': 'seafood', 'label': 'Haring & North Sea Seafood', 'icon': '🐟', 'query': 'haring', 'regex': r'\b(haring|seafood|fish|vishandel|kibbeling|vis|oyster|mosselen)\b'}, {'id': 'frites', 'label': 'Artisanal Flemish Frites', 'icon': '🍟', 'query': 'friture', 'regex': r'\b(frit|friture|frites|patat|vlaams|friet)\b'}, {'id': 'tavern', 'label': 'Historic Bruin Cafés & Jenever', 'icon': '🏛️', 'query': 'jenever', 'regex': r'\b(jenever|bruin caf[eé]|brown caf[eé]|proeflokaal|historic tavern|eetcaf[eé])\b'}, {'id': 'craft_beer', 'label': 'Craft Breweries & Taprooms', 'icon': '🍺', 'query': 'craft beer', 'regex': r'\b(brewery|brouwerij|craft beer|taproom|microbrewery|draught)\b'}, # Diaspora & Heritage Kitchens {'id': 'indonesian', 'label': 'Indonesian Rijsttafel', 'icon': '🍚', 'query': 'indonesian', 'regex': r'\b(indonesisch|indonesian|rijsttafel|rendang|warung|satay|sate)\b'}, {'id': 'surinamese', 'label': 'Surinamese Roti & Warung', 'icon': '🍛', 'query': 'surinaams', 'regex': r'\b(surinaams|surinamese|roti|saoto|bara|pom|bakkeljauw)\b'}, {'id': 'moroccan', 'label': 'Moroccan Tagine & Couscous', 'icon': '🫕', 'query': 'moroccan', 'regex': r'\b(moroccan|marokkaans|tagine|tajine|couscous|pastilla)\b'}, {'id': 'ethiopian', 'label': 'Ethiopian & Eritrean Injera', 'icon': '🫓', 'query': 'ethiopian', 'regex': r'\b(ethiopian|eritrean|injera|tibs|wat|habesha)\b'}, # Middle East, Levant, Anatolia & Gulf {'id': 'esnaf', 'label': 'Historic Esnaf Lokantaları', 'icon': '🥘', 'query': 'esnaf', 'regex': r'\b(esnaf|lokanta|sulu yemek|guvec|güveç|kuru fasulye)\b'}, {'id': 'doner', 'label': 'Wood-Fired Yaprak Döner', 'icon': '🥙', 'query': 'doner', 'regex': r'\b(doner|döner|yaprak|donerci|dönerci)\b'}, {'id': 'ocakbasi', 'label': 'Ocakbaşı Charcoal Mangal', 'icon': '🔥', 'query': 'ocakbasi', 'regex': r'\b(ocakbasi|ocakbaşı|mangal|adana|urfa|cop sis|çöp şiş)\b'}, {'id': 'pide', 'label': 'Stone Oven Pide & Lahmacun', 'icon': '🥖', 'query': 'pide', 'regex': r'\b(pide|lahmacun|karadeniz|tas firin|taş fırın)\b'}, {'id': 'mandi', 'label': 'Yemeni Mandi & Khaleeji', 'icon': '🍗', 'query': 'mandi', 'regex': r'\b(mandi|kabsa|machboos|khaleeji)\b'}, {'id': 'persian', 'label': 'Persian Charcoal Grills', 'icon': '🍢', 'query': 'persian', 'regex': r'\b(persian|iranian|koobideh|sangak)\b'}, {'id': 'kofte', 'label': 'Kasap Köfte & Piyaz', 'icon': '🥩', 'query': 'kofte', 'regex': r'\b(kofte|köfte|köfteci|kofteci|piyaz)\b'}, {'id': 'baklava', 'label': 'Artisanal Antep Baklava', 'icon': '🍯', 'query': 'baklava', 'regex': r'\b(baklava|baklavacı|gulluoglu|fistikli|fıstıklı)\b'}, {'id': 'bosphorus_seafood', 'label': 'Coastal & Bosphorus Seafood', 'icon': '🐟', 'query': 'seafood', 'regex': r'\b(balik|balık|balikci|balıkçı|balik ekmek|hamour|fish fry)\b'}, {'id': 'breakfast', 'label': 'Village Breakfast & Kaymak', 'icon': '🍳', 'query': 'breakfast', 'regex': r'\b(kahvalti|kahvaltı|kaymak|menemen)\b'}, {'id': 'shawarma', 'label': 'Charcoal Kebab & Shawarma', 'icon': '🥙', 'query': 'shawarma', 'regex': r'\b(shawarma|shoarma|d[oö]ner|kebab|durum|d[uü]r[uü]m|mangal|tandur)\b'}, {'id': 'falafel', 'label': 'Levant Hummus & Falafel', 'icon': '🧆', 'query': 'falafel', 'regex': r'\b(falafel|hummus|mezze|meze|sabich|lebanese|libanees|manaeesh)\b'}, # Asian Craft {'id': 'ramen', 'label': 'Slow-Broth Ramen & Noodles', 'icon': '🍜', 'query': 'ramen', 'regex': r'\b(ramen|noodles|udon|soba)\b'}, {'id': 'sushi', 'label': 'Edomae Sushi & Izakaya', 'icon': '🍣', 'query': 'sushi', 'regex': r'\b(sushi|japanese|izakaya|yakitori|omakase)\b'}, {'id': 'chinese', 'label': 'Dim Sum & Cantonese', 'icon': '🥟', 'query': 'chinese', 'regex': r'\b(chinese|chinees|dim sum|cantonese|sichuan|dumpling)\b'}, {'id': 'thai', 'label': 'Thai Street Kitchen', 'icon': '🍲', 'query': 'thai', 'regex': r'\b(thai|pad thai|curry|tom yum)\b'}, {'id': 'vietnamese', 'label': 'Vietnamese & Pho', 'icon': '🥢', 'query': 'vietnamese', 'regex': r'\b(vietnamese|vietnamees|pho|phở|b[aá]nh m[iì])\b'}, {'id': 'korean', 'label': 'Korean & Fried Chicken', 'icon': '🍗', 'query': 'korean', 'regex': r'\b(korean|koreaans|kimchi|bibimbap|chimaek)\b'}, {'id': 'indian', 'label': 'Indian & Curry Houses', 'icon': '🥘', 'query': 'indian', 'regex': r'\b(indian|indiase|curry|tandoori|masala|biryani|dosa|thali)\b'}, # The Americas & Bistro {'id': 'burger', 'label': 'Craft & Smash Burgers', 'icon': '🍔', 'query': 'burger', 'regex': r'\b(burger|burgers|smash)\b'}, {'id': 'steak', 'label': 'Steak & Prime Grill', 'icon': '🥩', 'query': 'steak', 'regex': r'\b(steak|steakhouse|grill|ribeye|churrasco|asado)\b'}, {'id': 'bbq', 'label': 'Smokehouse BBQ & Ribs', 'icon': '🍖', 'query': 'bbq', 'regex': r'\b(barbecue|bbq|smokehouse|brisket|pulled pork|spareribs)\b'}, {'id': 'mexican', 'label': 'Mexican & Taquerías', 'icon': '🌮', 'query': 'mexican', 'regex': r'\b(mexican|mexicaans|taco|taqueria|birria|burrito)\b'}, {'id': 'french', 'label': 'French Bistro & Brasserie', 'icon': '🍷', 'query': 'french', 'regex': r'\b(french|frans|bistro|brasserie|charcuterie)\b'}, # Roast & Sips {'id': 'coffee', 'label': 'Specialty Coffee Micro-Roasters', 'icon': '☕', 'query': 'coffee', 'regex': r'\b(coffee|koffie|espresso|roastery|filter coffee|pourover)\b'}, {'id': 'bubble_tea', 'label': 'Taiwanese Bubble Tea & Boba', 'icon': '🧋', 'query': 'bubble tea', 'regex': r'\b(bubble tea|boba|tapioca)\b'}, # Broad Regional, Heritage & Evening Dining (Collectively Exhaustive MECE Buckets) {'id': 'dutch', 'label': 'Dutch Heritage & Eetcafés', 'icon': '🇳🇱', 'query': 'dutch', 'regex': r'\b(dutch|nederlands|stamppot|snert|bitterballen|poffertjes|hollands|eetcaf[eé])\b'}, {'id': 'bistro', 'label': 'Modern European & Chef Bistros', 'icon': '🍽️', 'query': 'bistro', 'regex': r'\b(bistro|european|contemporary|gastronomy|fine dining|seasonal|farm[\s-]to[\s-]table|international)\b'}, {'id': 'gastropub', 'label': 'British & Irish Gastropubs', 'icon': '🍺', 'query': 'pub', 'regex': r'\b(british|irish|cask_ale|sunday_roast|gastropub|pub)\b'}, {'id': 'wine_bar', 'label': 'Natural Wine Bars & Cellars', 'icon': '🍷', 'query': 'wine bar', 'regex': r'\b(wine_bar|wine bar|vinotheque|natural wine|wijnbar|cocktails|drinks)\b'}, {'id': 'latin', 'label': 'Latin American & Caribbean', 'icon': '🌮', 'query': 'latin', 'regex': r'\b(argentinian|caribbean|peruvian|colombian|empanada|arepa|ceviche|latin)\b'} ] ARCHETYPE_BY_ID = {a['id']: a for a in MASTER_CRAFT_ARCHETYPES} for _a in MASTER_CRAFT_ARCHETYPES: _a['_re'] = re.compile(_a['regex'], re.IGNORECASE) # MECE Mutually Exclusive Priority Order (Specific Craft Specialist > Broad Regional Dining) SPECIFICITY_PRIORITY = [ # Level 1: Dedicated sweet & beverage specialists 'gelato', 'pancakes', 'patisserie', 'bubble_tea', 'coffee', # Level 2: Dedicated craft mono-specialists 'pizza', 'ramen', 'sushi', 'doner', 'ocakbasi', 'pide', 'kofte', 'baklava', 'shawarma', 'falafel', 'mandi', 'persian', 'esnaf', 'frites', 'bbq', 'burger', 'steak', # Level 3: Distinct national heritage kitchens 'surinamese', 'indonesian', 'ethiopian', 'moroccan', 'thai', 'vietnamese', 'korean', 'indian', 'chinese', 'mexican', 'tapas', 'greek', # Level 4: Core traditional culinary categories 'seafood', 'deli', 'bakery', 'italian', 'french', 'craft_beer', 'tavern', # Level 5: Broad European & Regional heritage 'latin', 'gastropub', 'wine_bar', 'bistro', 'dutch' ] class PalateEngine: def __init__(self, city_id="amsterdam", data_path=None, accreditations_path=None, vectors_path=None, model_path=None): self.city_id = (city_id or "amsterdam").lower().strip() self.city_config = self.load_city_config() city_dir = os.path.join(DATA_DIR, self.city_id) self.data_path = data_path or (os.path.join(city_dir, "places.json") if os.path.exists(os.path.join(city_dir, "places.json")) else DATA_FILE) self.accreditations_path = accreditations_path or (os.path.join(city_dir, "accreditations.json") if os.path.exists(os.path.join(city_dir, "accreditations.json")) else ACCREDITATIONS_FILE) self.vectors_path = vectors_path or (os.path.join(city_dir, "vectors.npy") if os.path.exists(os.path.join(city_dir, "vectors.npy")) else VECTORS_FILE) self.model_path = model_path or (os.path.join(city_dir, "semantic_model.pkl") if os.path.exists(os.path.join(city_dir, "semantic_model.pkl")) else MODEL_FILE) self.epicenters = self.city_config.get("epicenters", TOURIST_EPICENTERS) self.places = [] self.accreditations = [] self.vectors = None self.vectorizer = None self.svd = None self.load_accreditations() self.load_data() self.load_semantic_model() self.neighborhood_map = NEIGHBORHOOD_ALIASES_BY_CITY.get(self.city_id, {}) def load_city_config(self): cities_file = os.path.join(DATA_DIR, "cities.json") if os.path.exists(cities_file): try: with open(cities_file, "r", encoding="utf-8") as f: cities = json.load(f) for c in cities: if c.get("id") == self.city_id: return c except Exception as e: print(f"Error loading cities.json: {e}") return { "id": self.city_id, "name": self.city_id.capitalize(), "epicenters": TOURIST_EPICENTERS, "waterfront_label": "⛵ Canal Terrace", "heritage_badge_label": "🏛️ Historic Monuments" } def load_semantic_model(self): """Load the precomputed 128-dim dense semantic vector matrix and SVD pipeline.""" if os.path.exists(self.vectors_path) and os.path.exists(self.model_path): try: self.vectors = np.load(self.vectors_path) with open(self.model_path, "rb") as f: model = pickle.load(f) self.vectorizer = model.get("vectorizer") self.svd = model.get("svd") if self.vectorizer is not None and self.svd is not None and len(self.places) > len(self.vectors): extra_places = self.places[len(self.vectors):] extra_texts = [f"{p.get('name', '')} {p.get('primary_category', '')} {p.get('cuisine', '')} {' '.join(p.get('craft_badges', []))}" for p in extra_places] extra_tfidf = self.vectorizer.transform(extra_texts) extra_dense = self.svd.transform(extra_tfidf) norms = np.linalg.norm(extra_dense, axis=1, keepdims=True) norms[norms == 0] = 1.0 extra_dense = extra_dense / norms self.vectors = np.vstack([self.vectors, extra_dense]) print(f"Loaded {self.vectors.shape[0]} dense semantic vectors (128-dim) for TasteOS semantic search.") except Exception as e: print(f"Warning: Failed to load semantic model: {e}") self.vectors = None self.vectorizer = None self.svd = None def load_accreditations(self): """Load the structured Guide of Guides awards table (place_accreditations).""" if os.path.exists(self.accreditations_path): with open(self.accreditations_path, "r", encoding="utf-8") as f: raw_accs = json.load(f) self.accreditations = [] for acc in raw_accs: item = dict(acc) item["_m_name"] = acc.get("match_name", "").lower().strip() item["_m_norm"] = self._normalize_name(item["_m_name"]) m_core = GENERIC_WRAPPER_RE.sub(' ', item["_m_norm"]) item["_m_core"] = re.sub(r'\s+', ' ', m_core).strip() item["_re_norm"] = re.compile(r'\b' + re.escape(item["_m_norm"]) + r'\b') if len(item["_m_norm"]) >= 6 else None item["_re_core"] = re.compile(r'\b' + re.escape(item["_m_core"]) + r'\b') if len(item["_m_core"]) >= 6 else None item["_ex_patterns"] = [] if acc.get("exclude_match"): raw_ex = acc["exclude_match"] patterns = raw_ex if isinstance(raw_ex, list) else [p.strip() for p in str(raw_ex).split("|") if p.strip()] for p in patterns: item["_ex_patterns"].append((p.lower(), self._normalize_name(p))) self.accreditations.append(item) print(f"Loaded {len(self.accreditations)} official third-party accreditations.") def _normalize_name(self, text): s = text.lower().strip() replacements = { 'ö': 'o', 'ç': 'c', 'ş': 's', 'ı': 'i', 'ğ': 'g', 'ü': 'u', 'é': 'e', 'è': 'e', 'ê': 'e', 'ë': 'e', 'à': 'a', 'á': 'a', 'â': 'a', 'ä': 'a', 'î': 'i', 'ï': 'i', 'ô': 'o', 'ó': 'o', 'û': 'u', 'ú': 'u', 'ñ': 'n', 'ō': 'o', 'ū': 'u', 'ī': 'i', 'ā': 'a', 'ē': 'e' } for k, v in replacements.items(): s = s.replace(k, v) s = s.replace('\ufffd', 'e') # Strip apostrophes and punctuation for fuzzy token joining s = re.sub(r"['’`\.\-]", "", s) s = re.sub(r"\s+", " ", s) return s.strip() def match_all_accreditations(self, place_name): """Find ALL matching third-party accreditations for a place (e.g. Michelin + Lekker 500).""" p_name = place_name.lower().strip() p_norm = self._normalize_name(place_name) matches = [] source_priority = { "michelin": 10, "michelin_bib": 9, "worlds_50_best": 10, "50_top_pizza": 9, "avpn_napoli": 9, "gault_millau": 9, "vedat_milor": 9, "incili_gastronomi": 8, "oad": 8, "la_liste": 8, "grandes_tables": 8, "relais_chateaux": 8, "50_best_discovery": 9, "gouden_pollepel": 8, "entree_awards": 7, "lekker_500": 8, "european_coffee_trip": 8, "world_barista_championship": 8, "specialty_coffee_association": 8, "fact_awards": 7, "timeout_awards": 7, "whats_on": 7, "eater_38": 7, "worlds_101_best_steak": 9, "burgerdudes_worlds_best_burgers": 8, "gelato_festival_world_masters": 8, "tokyo_ramen_of_the_year": 9, "coupe_du_monde_boulangerie": 9, "texas_monthly_bbq": 9, "concurso_paella_valenciana": 8, "tabelog": 9, "gambero_rosso": 9, "slow_food": 8, "guia_repsol": 8, "le_fooding": 8, "good_food_guide": 8, "james_beard": 9, "het_parool": 6, "amsterdam_heritage": 5 } p_core = GENERIC_WRAPPER_RE.sub(' ', p_norm) p_core = re.sub(r'\s+', ' ', p_core).strip() for acc in self.accreditations: if acc.get("_ex_patterns"): if any(ex_norm in p_norm or ex_raw in p_name for ex_raw, ex_norm in acc["_ex_patterns"]): continue m_raw = acc["_m_name"] m_norm = acc["_m_norm"] # 1. Exact match on raw name or normalized form if m_raw == p_name or m_norm == p_norm: matches.append(acc) continue # 2. Distinctive core equality (e.g. "bocca" == "bocca", "massimo" == "massimo") m_core = acc["_m_core"] if len(m_core) >= 4 and len(p_core) >= 4 and m_core == p_core: matches.append(acc) # 3. Whole-phrase word-bounded match (e.g. "\bfort negen\b" in "fort negen bakery") elif acc["_re_norm"] and acc["_re_norm"].search(p_norm): matches.append(acc) # 4. Multi-word or long distinctive core inclusion (>= 6 chars) elif acc["_re_core"] and len(p_core) >= 6 and acc["_re_core"].search(p_core): matches.append(acc) # Sort so highest prestige comes first matches.sort(key=lambda x: source_priority.get(x.get("source", ""), 1), reverse=True) return [{k: v for k, v in x.items() if not k.startswith("_")} for x in matches] def match_accreditation(self, place_name): """Simulate SQL JOIN place_accreditations ON place_id / name.""" matches = self.match_all_accreditations(place_name) return matches[0] if matches else None def get_place_archetypes(self, p): badges = " ".join(p.get("craft_badges") or []).lower() name = (p.get("name") or "").lower() cat = (p.get("primary_category") or "").lower() cuisine = (p.get("cuisine") or "").lower() must_try = (p.get("must_try") or "").lower() # Generic OSM tag sanitize: cafes, bars, and bistros frequently have "coffee;tea;pastry;espresso" sanitized_cuisine = cuisine if cat in ("cafe", "bar", "pub", "restaurant") or "coffee" in cuisine: sanitized_cuisine = re.sub(r'\bpastry\b', '', sanitized_cuisine) text = f"{name} {cat} {sanitized_cuisine} {badges} {must_try}".lower() matched_candidates = set() for a in MASTER_CRAFT_ARCHETYPES: aid = a["id"] if aid == "patisserie": if re.search(r'\b(p[aâ]tisserie|chocolatier|chocolade|chocolate|bonbon|stroopwafel|banketbakkerij|confectionery|macaron|viennoiserie|eclair|éclair|croissant)\b', f"{name} {badges} {must_try}"): matched_candidates.add("patisserie") elif cat in ["bakery", "pastry", "chocolatier", "confectionery"]: if re.search(r'\b(pastry|cake|taart|baking|sweet|viennoiserie|patisserie|chocolate|bonbon|stroopwafel|croissant)\b', f"{name} {cuisine} {must_try} {badges}") or cat in ["pastry", "chocolatier", "confectionery"]: matched_candidates.add("patisserie") elif re.search(r'\b(p[aâ]tisserie|chocolatier|bonbon|stroopwafel)\b', sanitized_cuisine): matched_candidates.add("patisserie") elif aid == "pizza": if re.search(r'\b(pizza|pizzeria|napoletana)\b', f"{name} {cat} {cuisine}"): matched_candidates.add("pizza") elif aid == "dutch": if self.city_id == "amsterdam" and a["_re"].search(text): matched_candidates.add("dutch") else: if a["_re"].search(text): matched_candidates.add(aid) # MECE Selection: Mutually exclusive priority assignment (most specific craft wins) for prio_id in SPECIFICITY_PRIORITY: if prio_id in matched_candidates: return [prio_id] # MECE Fallback: 100% Collective Exhaustiveness (every authentic venue has a home) if cat in ("cafe", "coffee_shop"): return ["coffee"] if cat in ("bakery", "pastry"): return ["bakery"] if cat == "pub" or "pub" in cuisine or "cask_ale" in cuisine: return ["gastropub"] if cat == "bar" or "wine" in cuisine or "cocktails" in cuisine: return ["wine_bar"] if cat == "pizza": return ["pizza"] if "argentinian" in cuisine or "latin" in cuisine: return ["latin"] if self.city_id == "amsterdam": return ["dutch"] return ["bistro"] def load_data(self): if not os.path.exists(self.data_path): print(f"Data file not found: {self.data_path}") return with open(self.data_path, "r", encoding="utf-8") as f: raw_places = json.load(f) # Deduplication safeguard: eliminate duplicate venue records with same normalized (name, address) seen_keys = set() deduped = [] for p in raw_places: addr = (p.get("address") or "").lower().strip() name = (p.get("name") or "").lower().strip() key = (name, addr) if addr and addr not in ["amsterdam, netherlands", "dubai, united arab emirates", "istanbul, turkey"] and key in seen_keys: continue if addr and addr not in ["amsterdam, netherlands", "dubai, united arab emirates", "istanbul, turkey"]: seen_keys.add(key) deduped.append(p) self.places = [self.evaluate_place(p) for p in deduped] for p in self.places: mece_clusters = self.get_place_archetypes(p) p["_cluster_ids"] = mece_clusters primary_id = mece_clusters[0] if mece_clusters else "bistro" p["primary_cluster_id"] = primary_id arch_meta = ARCHETYPE_BY_ID.get(primary_id, {}) p["primary_cluster_label"] = arch_meta.get("label", primary_id.capitalize()) p["primary_cluster_icon"] = arch_meta.get("icon", "✨") print(f"PalateEngine [{self.city_id}] initialized with {len(self.places)} analyzed venues (100% MECE architecture).") def evaluate_place(self, p): name = p.get("name", "") cats = p.get("alternate_categories", []) combined_text = f"{name} {p.get('primary_category', '')} {' '.join(cats)} {p.get('cuisine', '')} {p.get('must_try', '')}".lower() # 1. Smarter Category Purity # Filters out generic administrative tags that do NOT dilute culinary craft specialized_alts = [ c for c in cats if c.lower() not in GENERIC_WRAPPER_TAGS and c.lower() != p.get("primary_category", "").lower() ] total_relevant_categories = 1 + len(specialized_alts) category_purity = 1.0 / total_relevant_categories purity_score = round(category_purity * 25.0, 1) # 2. Universal Craft Lexicon Matching craft_score = 0 craft_badges = [] craft_details = [] for rule in CRAFT_PATTERNS: if re.search(rule["regex"], combined_text, re.IGNORECASE): craft_badges.append(rule["badge"]) craft_details.append(rule["detail"]) craft_score += rule["points"] craft_score = min(craft_score, 35) # 3. Relational Critic Accreditations (The 'Guide of Guides' Join) accreditations = self.match_all_accreditations(name) primary_accreditation = accreditations[0] if accreditations else p.get("accreditation") accreditation = primary_accreditation critic_score = 0 if accreditations: # 35 base for primary guide + 5 consensus bonus for each additional authority (max 45) critic_score = min(45, 35 + (len(accreditations) - 1) * 5) for acc in reversed(accreditations): tier_name = acc.get("award_tier") or acc.get("award") or "Culinary Distinction" award_badge = tier_name if any(tier_name.startswith(sym) for sym in ["🏅", "⭐", "🍴", "🏆", "🟢", "🍕", "☕", "🇳🇱", "🎖️", "💎"]) else f"🏅 {tier_name}" if award_badge not in craft_badges: craft_badges.insert(0, award_badge) if len(accreditations) >= 2: consensus_badge = f"🏆 Critical Consensus ({len(accreditations)} Independent Guides)" if consensus_badge not in craft_badges: craft_badges.insert(0, consensus_badge) elif p.get("accreditation"): critic_score = 35 tier_name = p["accreditation"].get("award_tier") or p["accreditation"].get("award") or "Culinary Distinction" award_badge = tier_name if any(tier_name.startswith(sym) for sym in ["🏅", "⭐", "🍴", "🏆", "🟢", "🍕", "☕", "🇳🇱", "🎖️", "💎"]) else f"🏅 {tier_name}" if award_badge not in craft_badges: craft_badges.insert(0, award_badge) # 4. Tourist Trap Negative Patterns trap_flags = [] trap_penalty = 0 for pat in TOURIST_TRAP_PATTERNS: if re.search(pat["regex"], combined_text, re.IGNORECASE): trap_flags.append(pat["flag"]) trap_penalty += pat["penalty"] # 5. Tourist Zone Spatial Penalty lat = p.get("latitude") lon = p.get("longitude") min_epicenter_dist = 999999.0 nearest_epicenter = "" if lat and lon: for epic in self.epicenters: d = haversine_distance_m(lat, lon, epic["lat"], epic["lon"]) if d < min_epicenter_dist: min_epicenter_dist = d nearest_epicenter = epic["name"] spatial_score = 0 is_tourist_trap = False if min_epicenter_dist < 85: if craft_score < 20 and not accreditation: spatial_score = -30 trap_flags.append(f"⚠️ Epicenter Alert: Located within {int(min_epicenter_dist)}m of {nearest_epicenter}") is_tourist_trap = True elif min_epicenter_dist < 180: if craft_score < 15 and not accreditation: spatial_score = -15 elif min_epicenter_dist > 350: spatial_score = 10 # 6. Independence vs. Global Industrial Chains ownership = p.get("ownership", {}) brand = p.get("brand") brand_clean = (brand or "").strip().lower() is_industrial_chain = brand_clean in GLOBAL_INDUSTRIAL_CHAINS or (not ownership.get("is_independent", True) if ownership else False) if ownership.get("type") == "independent": independence_score = 25 elif ownership.get("type") == "mini_group": independence_score = 15 elif is_industrial_chain: independence_score = 0 chain_label = ownership.get("badge") or f"Owned by '{brand}'" trap_flags.append(f"🏢 Multi-Unit Chain: {chain_label}") else: independence_score = 15 # 7. Lindy Longevity Score start_year = p.get("start_year") longevity_score = 0 longevity_badge = None if start_year: age = 2026 - start_year if age >= 100: longevity_score = 30 longevity_badge = f"🏛️ Century-Old Institution ({age} yrs, est. {start_year})" elif age >= 25: longevity_score = 20 longevity_badge = f"⏳ Lindy Survivor ({age} yrs, est. {start_year})" elif age >= 10: longevity_score = 10 longevity_badge = f"Established Local Favorite (est. {start_year})" # 8. Operational Rhythm & In-House Production Signal rhythm_score = 0 schedule_rhythm = p.get("schedule_rhythm", "standard") if schedule_rhythm == "artisan_rhythm": rhythm_score = 10 elif schedule_rhythm == "tourist_continuous": rhythm_score = -15 trap_flags.append("⚠️ Factory Schedule: Continuous 7-day all-hours commercial mass production") amenities = p.get("amenities", {}) in_house_score = 0 if amenities.get("microbrewery"): in_house_score += 15 craft_badges.append("🍺 In-House Craft Microbrewery") if amenities.get("in_house_roastery"): in_house_score += 15 craft_badges.append("☕ In-House Specialty Coffee Roastery") # 10. Classic Cash-Only Lindy Marker cash_score = 5 if p.get("cash_only") else 0 if p.get("cash_only") and "💶 Old-School Cash Only" not in craft_badges: craft_badges.append("💶 Old-School Cash Only") # 11. Architectural Monument & Heritage Signal (Gemeente Amsterdam / BAG) heritage = p.get("heritage", {}) heritage_score = 0 if heritage.get("is_monument"): if heritage.get("badge") and heritage["badge"] not in craft_badges: craft_badges.append(heritage["badge"]) if heritage.get("authenticity_boost"): # Meaningful quality boost for authentic operators in protected monuments heritage_score = 15 if heritage.get("status") == "Rijksmonument" else 10 # Dietary Badges dietary = p.get("dietary", {}) dietary_badges = [] if dietary.get("halal"): dietary_badges.append("🥩 Halal Friendly") if dietary.get("vegan"): dietary_badges.append("🌿 100% Vegan Friendly") elif dietary.get("vegetarian"): dietary_badges.append("🌱 Vegetarian Friendly") if dietary.get("gluten_free"): dietary_badges.append("🌾 Gluten-Free Options") if dietary.get("kosher"): dietary_badges.append("✡️ Kosher") # Amenity Badges amenity_badges = [] if amenities.get("canal_terrace"): waterfront_badge = "⛵ Canal Terrace" if self.city_id == "amsterdam" else ("🌊 Bosphorus Terrace" if self.city_id == "istanbul" else "🌊 Waterfront Terrace") amenity_badges.append(waterfront_badge) elif amenities.get("outdoor_seating"): amenity_badges.append("☀️ Outdoor Seating") if amenities.get("wheelchair_accessible"): amenity_badges.append("♿ Wheelchair Accessible") # Celiac & Medical Kitchen Safety celiac_safety = p.get("celiac_safety", {}) celiac_score = 0 if celiac_safety.get("is_safe"): celiac_score = 15 if celiac_safety.get("certified_by") == "NCV Horeca Alliantie" else 10 if celiac_safety.get("badge") and celiac_safety["badge"] not in dietary_badges: dietary_badges.insert(0, celiac_safety["badge"]) # Scratch Kitchen Craft scratch = p.get("scratch_kitchen", {}) scratch_score = 0 if scratch.get("is_scratch"): scratch_score = 10 if scratch.get("badge") and scratch["badge"] not in craft_badges: craft_badges.append(scratch["badge"]) # Honest Hospitality / Tap Water hospitality = p.get("hospitality", {}) hospitality_score = 0 if hospitality.get("free_tap_water"): hospitality_score = 5 if hospitality.get("badge") and hospitality["badge"] not in amenity_badges: amenity_badges.append(hospitality["badge"]) # 12. Urban Food Habitat Matrix (Local Resident Enclave vs Tourist Corridor) habitat = p.get("habitat", {}) habitat_score = 0 resident_density = habitat.get("resident_food_density", 0) tourist_kiosks = habitat.get("tourist_kiosk_density", 0) if resident_density >= 2: habitat_score = 10 if "🏘️ Resident Food Habitat" not in craft_badges: craft_badges.append("🏘️ Resident Food Habitat") elif tourist_kiosks >= 3 and not accreditation: spatial_score -= 15 trap_flags.append("⚠️ Tourist Funnel Corridor: Surrounded by souvenir & exchange kiosks") # 13. Multilingual Over-Translation Distortion Index translations_count = p.get("translations_count", len(p.get("name_translations", []))) if translations_count >= 5 and not accreditation and not p.get("is_luxury_hotel"): trap_penalty -= 15 trap_flags.append("🚩 Multilingual Tourist Lure: 5+ languages printed to capture tourist foot-traffic") # 14. Dedicated Experiential Dining vs Ghost / Virtual Dark Kitchen is_ghost = p.get("is_ghost_kitchen") or ( amenities.get("indoor_seating") is False and (amenities.get("takeaway_only") or amenities.get("delivery_only")) ) if is_ghost: p["is_ghost_kitchen"] = True trap_flags.append("📦 Dark / Ghost Kitchen: Virtual delivery brand without experiential dining room") # 15. Direct Digital Independence (Custom Domain vs Aggregator) website = p.get("website", "") digital_score = 0 if website: domain = website.lower() if not any(agg in domain for agg in ["deliveroo", "talabat", "ubereats", "yemeksepeti", "thuisbezorgd", "linktr.ee", "facebook.com", "instagram.com"]): digital_score = 5 # 16. Government Food Safety & Clean Kitchen Inspection food_safety = p.get("food_safety") or {} fs_grade = food_safety.get("inspection_grade", "") food_safety_score = 0 if fs_grade in ["Grade A", "5/5", "Très satisfaisant", "Elite Smiley"]: food_safety_score = 15 fs_badge = food_safety.get("badge") or "🏆 Grade A Clean Kitchen" if fs_badge not in amenity_badges: amenity_badges.append(fs_badge) elif fs_grade in ["Grade B", "4/5", "Satisfaisant"]: food_safety_score = 5 fs_badge = food_safety.get("badge") or "🥈 Grade B Clean Kitchen" if fs_badge not in amenity_badges: amenity_badges.append(fs_badge) elif fs_grade in ["Grade D", "Grade E", "0/5", "1/5", "Violation"]: trap_penalty -= 50 trap_flags.append(f"⚠️ Municipal Hygiene Warning: Critical sanitation failure ({fs_grade})") # Confidence Factor conf = p.get("confidence", 0.85) conf_multiplier = min(1.0, 0.85 + (conf * 0.15)) # Final Composite Quality Score positive_components = ( purity_score + craft_score + independence_score + longevity_score + critic_score + rhythm_score + in_house_score + cash_score + heritage_score + celiac_score + scratch_score + hospitality_score + habitat_score + digital_score + food_safety_score ) adjusted_positive = positive_components * conf_multiplier raw_score = adjusted_positive + spatial_score + trap_penalty final_score = max(5, min(99, int(raw_score))) if trap_penalty <= -30 or (min_epicenter_dist < 85 and craft_score < 15 and not accreditation) or is_industrial_chain: is_tourist_trap = True if final_score >= 80: tier_label = "🏆 Artisanal Culinary Master" elif final_score >= 60: tier_label = "⭐ Authentic Local Favorite" elif final_score >= 40: tier_label = "☕ Decent Neighborhood Spot" else: tier_label = "☕ Neighborhood Spot" # 17. Rule-Based Culinary Price Tiering (Option B) # Tier 4: Fine Dining / Michelin Stars / World's 50 Best / Luxury Omakase # Tier 3: Upscale Dining / Bib Gourmand / Premium Steakhouses & Bistros # Tier 2: Casual Sit-down / Neighborhood Favorites / Ramen / Trattorias / Burgers # Tier 1: Street Food / Budget / Döner / Bakeries / Falafel / Fast Casual award_sources = [a.get("source", "").lower() for a in accreditations] award_tiers = [a.get("award_tier", "").lower() for a in accreditations] primary_cat = p.get("primary_category", "").lower() amenity_type = p.get("amenity", "").lower() # Explicit budget / street food signals has_budget_or_street_signal = ( any(k in t for t in award_tiers for k in ["budget", "street food", "fish fry", "cheap", "shack", "karahi", "kebab master"]) or primary_cat in ["haringhandel", "doner", "shawarma", "falafel", "bakery", "tea", "ice_cream", "gelato", "fast_food", "soup", "deli", "sandwich", "mandi"] or amenity_type in ["fast_food"] or p.get("is_ghost_kitchen") ) is_tier_4 = ( not has_budget_or_street_signal and ( any("star" in t and "green star" not in t.lower() for t in award_tiers) or any(s in ["worlds_50_best", "oad", "grandes_tables"] for s in award_sources if not any("discovery" in t for t in award_tiers)) or primary_cat in ["omakase", "fine_dining"] or any("michelin 3 star" in b.lower() or "michelin 2 star" in b.lower() or "michelin 1 star" in b.lower() for b in craft_badges) ) ) is_tier_3 = ( not has_budget_or_street_signal and not is_tier_4 and ( any("bib" in t for t in award_tiers) or any(s in ["gault_millau", "50_top_pizza", "lekker_500", "vedat_milor"] for s in award_sources) or primary_cat in ["steak", "french", "contemporary", "bistro", "seafood", "bosphorus_seafood", "spanish", "tapas"] or (primary_cat == "italian" and "pizza" not in p.get("cuisine", "")) ) ) is_tier_1 = ( has_budget_or_street_signal or any("budget" in t or "street food" in t for t in award_tiers) ) if is_tier_4: price_tier = 4 elif is_tier_3: price_tier = 3 elif is_tier_1: price_tier = 1 else: price_tier = 2 currency_sym = self.city_config.get("currency_symbol", "€") tier_names = { 1: "Budget / Street Food", 2: "Casual Dining", 3: "Upscale Dining", 4: "Fine Dining / Tasting" } price_label = currency_sym * price_tier if currency_sym != "AED" else ("AED" if price_tier == 1 else f"AED {price_tier}") return { **p, "quality_score": final_score, "category_purity": round(category_purity, 2), "purity_score": purity_score, "craft_badges": craft_badges, "craft_details": craft_details, "dietary_badges": dietary_badges, "amenity_badges": amenity_badges, "trap_flags": trap_flags, "is_tourist_trap": is_tourist_trap, "longevity_badge": longevity_badge, "is_chain": is_industrial_chain, "tier_label": tier_label, "accreditation": accreditation, "all_accreditations": accreditations, "confidence": conf, "dist_to_tourist_hub_m": int(min_epicenter_dist) if min_epicenter_dist < 999999 else None, "nearest_tourist_hub": nearest_epicenter, "food_safety": food_safety, "price_tier": price_tier, "price_label": price_label, "price_tier_name": tier_names[price_tier] } def parse_geo_intent(self, query): """Extract neighborhood entity and decouple location terms from food query.""" q_clean = (query or "").strip() q_lower = q_clean.lower() if not q_lower: return query, None, [] sorted_keys = sorted(self.neighborhood_map.keys(), key=lambda k: len(k), reverse=True) for key in sorted_keys: aliases = self.neighborhood_map[key] for alias in aliases: # 1. Prepositional phrases: e.g. "in marina", "near jlt", "around satwa", "bij de pijp" prep_pattern = rf'\b(in|at|near|around|by|bij)\s+{re.escape(alias)}\b' prep_match = re.search(prep_pattern, q_lower) if prep_match: food_q = re.sub(prep_pattern, '', q_clean, flags=re.IGNORECASE).strip() food_q = re.sub(r'\s+', ' ', food_q) return food_q, key, aliases # 2. Standalone neighborhood token: e.g. "shawarma satwa" or "satwa" token_pattern = rf'\b{re.escape(alias)}\b' if re.search(token_pattern, q_lower): food_q = re.sub(token_pattern, '', q_clean, flags=re.IGNORECASE).strip() food_q = re.sub(r'\b(in|at|near|around|by|bij)\b', '', food_q, flags=re.IGNORECASE).strip() food_q = re.sub(r'\s+', ' ', food_q) return food_q, key, aliases return query, None, [] def search( self, query="", user_lat=52.3680, user_lon=4.8900, offset=0, max_results=40, cuisine=None, diet=None, outdoor=False, canal=False, wheelchair=False, accredited_only=False, monuments_only=False, independent_only=False, celiac_safe_only=False, scratch_only=False, tap_water_only=False, clean_only=False, late_night_only=False, roastery_only=False, historic_only=False, price_tiers=None, min_score=0, sort_by="quality", has_user_coords=False, return_total=False, bbox=None, max_dist_m=None ): # Geographic Entity Decoupling food_query, geo_target, geo_aliases = self.parse_geo_intent(query) effective_query = food_query if geo_target else query q_clean = effective_query.strip() q_lower = q_clean.lower() results = [] arch = ARCHETYPE_BY_ID.get(cuisine) if cuisine else None arch_regex = re.compile(arch["regex"], re.IGNORECASE) if arch else None # Vector semantic inference sims = None if q_lower and self.vectorizer is not None and self.svd is not None and self.vectors is not None: try: q_tfidf = self.vectorizer.transform([q_clean]) q_dense = self.svd.transform(q_tfidf) q_norm = np.linalg.norm(q_dense) if q_norm > 0: q_dense = q_dense / q_norm sims = np.dot(self.vectors, q_dense.T).flatten() except Exception as e: print(f"Semantic inference error: {e}") sims = None tokens = q_lower.split() if q_lower else [] intent_tokens = [t for t in tokens if t not in QUERY_INTENT_STOPWORDS] if not intent_tokens: intent_tokens = tokens # fallback if query was purely generic (e.g., "restaurants") # Expand nationality / cuisine stems (e.g., syria <-> syrian, italy <-> italian) expanded_intent = set(intent_tokens) for t in intent_tokens: if t in CUISINE_NATIONALITY_EXPANSIONS: for exp in CUISINE_NATIONALITY_EXPANSIONS[t]: expanded_intent.add(exp) for i, p in enumerate(self.places): # Exclude low-quality tourist traps completely if p.get("is_tourist_trap"): continue # Exclude delivery-only dark / ghost kitchens from experiential dining discovery if p.get("is_ghost_kitchen"): continue # Viewport Bounding Box Filter (min_lat, min_lon, max_lat, max_lon) if bbox and len(bbox) == 4: min_lat, min_lon, max_lat, max_lon = bbox plat = p.get("latitude") plon = p.get("longitude") if plat is None or plon is None or not (min_lat <= plat <= max_lat and min_lon <= plon <= max_lon): continue # Calculate Distance dist_m = 0 if p.get("latitude") and p.get("longitude"): dist_m = int(haversine_distance_m(user_lat, user_lon, p["latitude"], p["longitude"])) # Walking / Transit Radius Filter if max_dist_m is not None and dist_m > max_dist_m: continue # Geographic Neighborhood Intent Filtering venue_in_geo = False if geo_target and geo_aliases: venue_geo_text = f"{p.get('neighborhood') or ''} {p.get('address') or ''} {p.get('name') or ''}".lower() venue_in_geo = any(a in venue_geo_text for a in geo_aliases) # If query was solely a neighborhood (e.g. 'satwa', 'jordaan') without a food term, exclude other districts if not q_lower and not venue_in_geo: continue # Dedicated craft archetype / cuisine cluster filter if arch: if arch["id"] not in p.get("_cluster_ids", []): continue elif arch_regex: badges = " ".join(p.get("craft_badges") or []) text = f"{p['name']} {p.get('primary_category', '')} {p.get('cuisine', '')} {badges}".lower() if not arch_regex.search(text): continue # 1. Dietary filter if diet: diet_key = diet.lower().strip() if diet_key in ("celiac", "celiac_safe"): if not p.get("celiac_safety", {}).get("is_safe"): continue elif diet_key in ("gluten_free", "glutenfree", "gf"): if not (p.get("dietary", {}).get("gluten_free") or p.get("celiac_safety", {}).get("is_safe")): continue elif not p.get("dietary", {}).get(diet_key): continue # 2. Amenity, Heritage & Transparency filters if outdoor and not p.get("amenities", {}).get("outdoor_seating"): continue if canal and not p.get("amenities", {}).get("canal_terrace"): continue if wheelchair and not p.get("amenities", {}).get("wheelchair_accessible"): continue if accredited_only and not p.get("accreditation"): continue if monuments_only and not p.get("heritage", {}).get("is_monument"): continue if independent_only and not p.get("ownership", {}).get("is_independent"): continue if celiac_safe_only and not p.get("celiac_safety", {}).get("is_safe"): continue if scratch_only and not p.get("scratch_kitchen", {}).get("is_scratch"): continue if tap_water_only and not p.get("hospitality", {}).get("free_tap_water"): continue if clean_only: fs = p.get("food_safety") or {} if fs.get("inspection_grade") not in ["Grade A", "5/5", "Très satisfaisant", "Elite Smiley"]: continue if late_night_only and not p.get("amenities", {}).get("late_night"): continue if roastery_only and not p.get("amenities", {}).get("in_house_roastery"): continue if historic_only and not ((p.get("heritage") or {}).get("is_historic") or (p.get("start_year") and p["start_year"] <= 2010)): continue if price_tiers: tiers_set = {price_tiers} if isinstance(price_tiers, int) else set(price_tiers) if p.get("price_tier") not in tiers_set: continue if min_score and p["quality_score"] < min_score: continue sim = float(sims[i]) if (sims is not None and i < len(sims)) else 0.0 exact_bonus = 0.0 intent_matched = False if not q_lower: match_weight = 1.0 else: p_name = p["name"].lower() p_norm = self._normalize_name(p["name"]) q_norm = self._normalize_name(q_lower) p_cat = p.get("primary_category", "").lower() p_cuisine = (p.get("cuisine") or "").lower() p_alts = [c.lower() for c in p.get("alternate_categories", [])] p_badges = [b.lower() for b in p.get("craft_badges", [])] # 1. Exact entity matches on place name (with unicode diacritic tolerance) if q_lower == p_name or q_norm == p_norm: exact_bonus += 50.0 intent_matched = True elif q_lower in p_name or q_norm in p_norm: exact_bonus += 35.0 intent_matched = True elif any(t in p_name or self._normalize_name(t) in p_norm for t in intent_tokens if len(t) > 2): exact_bonus += 20.0 intent_matched = True # 2. Specific Category / Cuisine / Craft match (NEVER match generic stops like 'restaurant') for t in expanded_intent: if t == p_cat or (len(t) >= 4 and t in p_cat and p_cat not in QUERY_INTENT_STOPWORDS): exact_bonus += 25.0 intent_matched = True if t in p_cuisine: exact_bonus += 20.0 intent_matched = True if any(t in c for c in p_alts if c not in QUERY_INTENT_STOPWORDS): exact_bonus += 15.0 intent_matched = True if any(t in b for b in p_badges): exact_bonus += 15.0 intent_matched = True dossier = p.get("dossier") if dossier: sig_lower = dossier.get("signature_dishes", "").lower() if len(t) >= 4 and t in sig_lower: exact_bonus += 25.0 intent_matched = True # 2b. Monument / Heritage queries boost authentic monuments if any(h in expanded_intent for h in ["historic", "monument", "rijksmonument", "heritage", "century", "oldest", "brown"]): if p.get("heritage", {}).get("is_monument") and p.get("heritage", {}).get("authenticity_boost"): exact_bonus += 35.0 intent_matched = True # 2c. Geographic Neighborhood Intent Match / Mismatch Penalty if geo_target: if venue_in_geo: exact_bonus += 250.0 intent_matched = True else: exact_bonus -= 100.0 # 3. If user specified a clear intent (e.g. "syrian") and venue does NOT match it: # require strong semantic similarity (>= 0.65) to even be considered. has_clear_intent = len(intent_tokens) > 0 and len(intent_tokens) < len(tokens) if has_clear_intent and not intent_matched and sim < 0.65: continue # Standard semantic cutoff if not intent_matched and sim < 0.20: continue dist_penalty = min(15, (dist_m / 450.0)) if not q_lower: if sort_by == "distance": effective_rank = 10000 - dist_m + (p["quality_score"] * 0.5) elif sort_by == "endorsed": effective_rank = (p.get("endorsement_count", 0) * 15.0) + p["quality_score"] elif sort_by == "heritage": is_monument = 1 if p.get("heritage", {}).get("is_monument") else 0 dist_bias = dist_penalty if has_user_coords else 0 effective_rank = (is_monument * 500.0) + (p["quality_score"] * 0.85) - dist_bias else: # quality (default) # When sorting by Culinary Quality across a city, quality score is king. # Endorsement count is a subtle tie-breaker; distance is only a gentle tie-breaker if user gave real GPS. tie_breaker = (p.get("endorsement_count", 0) * 0.05) if has_user_coords: tie_breaker -= min(1.0, dist_m / 10000.0) effective_rank = (p["quality_score"] * 10.0) + tie_breaker else: # Active Search Query: Hybrid Vector Cosine Similarity (75x) + Palate Quality (0.40x) + Intent Exact Bonuses if sort_by == "distance": effective_rank = 10000 - dist_m + (p["quality_score"] * 0.4) + (sim * 50.0) + exact_bonus elif sort_by == "endorsed": effective_rank = (p.get("endorsement_count", 0) * 15.0) + (p["quality_score"] * 0.7) + (sim * 50.0) + exact_bonus elif sort_by == "heritage": is_monument = 1 if p.get("heritage", {}).get("is_monument") else 0 dist_bias = dist_penalty if has_user_coords else 0 effective_rank = (is_monument * 500.0) + (sim * 75.0) + (p["quality_score"] * 0.40) + exact_bonus - dist_bias else: # quality (default) dist_bias = min(2.0, dist_penalty) if has_user_coords else 0 effective_rank = (sim * 75.0) + (p["quality_score"] * 0.40) + exact_bonus - dist_bias results.append({ **p, "distance_m": dist_m, "similarity_score": round(sim, 3) if q_lower else None, "effective_rank": round(effective_rank, 1), "matched_neighborhood": geo_target if venue_in_geo else None }) results.sort(key=lambda x: x["effective_rank"], reverse=True) total_count = len(results) paginated = results[offset:offset + max_results] if return_total: return paginated, total_count return paginated def get_facets(self, authentic_only=False): """Return aggregations for UI filter buttons and system stats.""" pool = [p for p in self.places if not p.get('is_tourist_trap') and not p.get('is_ghost_kitchen')] if authentic_only else self.places total = len(pool) return { "total_eateries": total, "halal_count": sum(1 for p in pool if p.get("dietary", {}).get("halal")), "vegan_count": sum(1 for p in pool if p.get("dietary", {}).get("vegan")), "vegetarian_count": sum(1 for p in pool if p.get("dietary", {}).get("vegetarian")), "gluten_free_count": sum(1 for p in pool if p.get("dietary", {}).get("gluten_free")), "canal_terrace_count": sum(1 for p in pool if p.get("amenities", {}).get("canal_terrace")), "outdoor_count": sum(1 for p in pool if p.get("amenities", {}).get("outdoor_seating")), "wheelchair_count": sum(1 for p in pool if p.get("amenities", {}).get("wheelchair_accessible")), "accredited_count": sum(1 for p in pool if p.get("accreditation")), "craft_artisans_detected": sum(1 for p in pool if p.get("craft_badges")), "tourist_traps_flagged": sum(1 for p in self.places if p.get("is_tourist_trap")), "mono_specialists": sum(1 for p in pool if p.get("category_purity", 0) == 1.0), "high_confidence_count": sum(1 for p in pool if p.get("confidence", 0) >= 0.85), "monuments_count": sum(1 for p in pool if p.get("heritage", {}).get("is_monument")), "rijksmonumenten_count": sum(1 for p in pool if p.get("heritage", {}).get("status") == "Rijksmonument"), "independent_count": sum(1 for p in pool if p.get("ownership", {}).get("is_independent")), "corporate_chain_count": sum(1 for p in pool if not p.get("ownership", {}).get("is_independent")), "celiac_safe_count": sum(1 for p in pool if p.get("celiac_safety", {}).get("is_safe")), "scratch_count": sum(1 for p in pool if p.get("scratch_kitchen", {}).get("is_scratch")), "tap_water_count": sum(1 for p in pool if p.get("hospitality", {}).get("free_tap_water")), "clean_kitchen_count": sum(1 for p in pool if (p.get("food_safety") or {}).get("inspection_grade") in ["Grade A", "5/5", "Très satisfaisant", "Elite Smiley"]), "late_night_count": sum(1 for p in pool if p.get("amenities", {}).get("late_night")), "roastery_count": sum(1 for p in pool if p.get("amenities", {}).get("in_house_roastery")), "historic_count": sum(1 for p in pool if (p.get("heritage") or {}).get("is_historic") or (p.get("start_year") and p["start_year"] <= 2010)), "price_1_count": sum(1 for p in pool if p.get("price_tier") == 1), "price_2_count": sum(1 for p in pool if p.get("price_tier") == 2), "price_3_count": sum(1 for p in pool if p.get("price_tier") == 3), "price_4_count": sum(1 for p in pool if p.get("price_tier") == 4) } def place_matches_filters(self, p, filters, exclude_keys=None): if not filters: return True if exclude_keys is None: exclude_keys = set() # 0. Exclude low-quality tourist traps and ghost kitchens if p.get("is_tourist_trap") or p.get("is_ghost_kitchen"): return False # 1. Price tier if "price" not in exclude_keys and filters.get("price") is not None: pt = filters["price"] tiers = {pt} if isinstance(pt, int) else set(pt) if p.get("price_tier") not in tiers: return False # 2. Cuisine cluster if "cuisine" not in exclude_keys and filters.get("cuisine"): target_c = filters["cuisine"] if target_c not in p.get("_cluster_ids", []): return False # 3. Diet if "diet" not in exclude_keys and filters.get("diet"): d = str(filters["diet"]).lower().strip() if d in ("celiac", "celiac_safe"): if not p.get("celiac_safety", {}).get("is_safe"): return False elif d in ("gluten_free", "glutenfree", "gf"): if not (p.get("dietary", {}).get("gluten_free") or p.get("celiac_safety", {}).get("is_safe")): return False elif not p.get("dietary", {}).get(d): return False # 4. Amenities / Flags amenities_map = { "outdoor": ("amenities", "outdoor_seating"), "canal": ("amenities", "canal_terrace"), "wheelchair": ("amenities", "wheelchair_accessible"), "late_night": ("amenities", "late_night"), "roastery": ("amenities", "in_house_roastery"), "scratch": ("scratch_kitchen", "is_scratch"), "water": ("hospitality", "free_tap_water"), "celiac": ("celiac_safety", "is_safe"), "monument": ("heritage", "is_monument"), "independent": ("ownership", "is_independent"), } for key, (sec, prop) in amenities_map.items(): if key not in exclude_keys and filters.get(key): if not p.get(sec, {}).get(prop): return False if "accredited" not in exclude_keys and filters.get("accredited"): if not p.get("accreditation"): return False if "clean" not in exclude_keys and filters.get("clean"): fs = p.get("food_safety") or {} if fs.get("inspection_grade") not in ["Grade A", "5/5", "Très satisfaisant", "Elite Smiley"]: return False if "historic" not in exclude_keys and filters.get("historic"): if not ((p.get("heritage") or {}).get("is_historic") or (p.get("start_year") and p["start_year"] <= 2010)): return False # 5. Spatial bounding box if "bbox" not in exclude_keys and filters.get("bbox"): bbox = filters["bbox"] if len(bbox) == 4: min_lat, min_lon, max_lat, max_lon = bbox plat, plon = p.get("latitude"), p.get("longitude") if plat is None or plon is None or not (min_lat <= plat <= max_lat and min_lon <= plon <= max_lon): return False # 6. Max distance if "max_dist" not in exclude_keys and filters.get("max_dist") is not None and filters.get("user_lat") and filters.get("user_lon"): plat, plon = p.get("latitude"), p.get("longitude") if plat and plon: dist = haversine_distance_m(filters["user_lat"], filters["user_lon"], plat, plon) if dist > filters["max_dist"]: return False return True def get_city_taxonomy(self, city_name=None, active_filters=None): """ Dynamic City Taxonomy Engine with Cascading Facets: Computes top craft clusters, price tiers, and environmental facets for any city based on ground-truth data. When active_filters is provided, dynamically calculates cross-facet and disjunctive counts. """ if not city_name: city_name = self.city_config.get("name", self.city_id.capitalize()) authentic_places = [p for p in self.places if not p.get('is_tourist_trap') and not p.get('is_ghost_kitchen')] has_active_filters = False if active_filters: for k, v in active_filters.items(): if v: has_active_filters = True break if not has_active_filters: base_pool = authentic_places cuisine_pool = authentic_places price_pool = authentic_places else: base_pool = [p for p in self.places if self.place_matches_filters(p, active_filters)] cuisine_pool = [p for p in self.places if self.place_matches_filters(p, active_filters, exclude_keys={"cuisine"})] price_pool = [p for p in self.places if self.place_matches_filters(p, active_filters, exclude_keys={"price"})] currency_sym = self.city_config.get("currency_symbol", "€") price_labels = { 1: currency_sym if currency_sym != "AED" else "AED", 2: currency_sym * 2 if currency_sym != "AED" else "AED 2", 3: currency_sym * 3 if currency_sym != "AED" else "AED 3", 4: currency_sym * 4 if currency_sym != "AED" else "AED 4" } price_tiers_data = [ { "tier": 1, "symbol": price_labels[1], "label": "Budget / Street Food", "count": sum(1 for p in price_pool if p.get("price_tier") == 1) }, { "tier": 2, "symbol": price_labels[2], "label": "Casual Dining", "count": sum(1 for p in price_pool if p.get("price_tier") == 2) }, { "tier": 3, "symbol": price_labels[3], "label": "Upscale Dining", "count": sum(1 for p in price_pool if p.get("price_tier") == 3) }, { "tier": 4, "symbol": price_labels[4], "label": "Fine Dining / Stars", "count": sum(1 for p in price_pool if p.get("price_tier") == 4) } ] # Calculate cluster counts from cuisine_pool (Disjunctive faceting) cluster_counts = Counter() for p in cuisine_pool: for cid in p.get("_cluster_ids", []): cluster_counts[cid] += 1 all_clusters = [] for arch in MASTER_CRAFT_ARCHETYPES: aid = arch["id"] cnt = cluster_counts[aid] label = arch["label"] if aid == "seafood": if self.city_id == "dubai": label = "Coastal & Fresh Seafood" elif self.city_id == "istanbul": label = "Bosphorus & Coastal Seafood" elif aid == "deli" and self.city_id != "amsterdam": label = "Artisanal Delis & Sandwiches" # Check if this cluster exists in this city at all city_has_cluster = any(aid in p.get("_cluster_ids", []) for p in authentic_places) if city_has_cluster: all_clusters.append({ "id": aid, "label": label, "icon": arch["icon"], "query": arch["query"], "count": cnt, "avg_score": 85.0 }) # Rank clusters by count descending all_clusters.sort(key=lambda x: x["count"], reverse=True) row1_clusters = all_clusters[:10] waterfront_label = self.city_config.get("waterfront_label") or ("Canal Terrace" if self.city_id == "amsterdam" else "Waterfront Terrace") heritage_label = self.city_config.get("heritage_badge_label") or "Historic Monuments" amenity_counts = { "independent": sum(1 for p in base_pool if p.get("ownership", {}).get("is_independent")), "clean": sum(1 for p in base_pool if (p.get("food_safety") or {}).get("inspection_grade") in ["Grade A", "5/5", "Très satisfaisant", "Elite Smiley"]), "accredited": sum(1 for p in base_pool if p.get("accreditation")), "late_night": sum(1 for p in base_pool if p.get("amenities", {}).get("late_night")), "roastery": sum(1 for p in base_pool if p.get("amenities", {}).get("in_house_roastery")), "historic": sum(1 for p in base_pool if ((p.get("heritage") or {}).get("is_historic") or (p.get("start_year") and p["start_year"] <= 2010))), "wheelchair": sum(1 for p in base_pool if p.get("amenities", {}).get("wheelchair_accessible")), "canal": sum(1 for p in base_pool if p.get("amenities", {}).get("canal_terrace")), "outdoor": sum(1 for p in base_pool if p.get("amenities", {}).get("outdoor_seating")), "scratch": sum(1 for p in base_pool if p.get("scratch_kitchen", {}).get("is_scratch")), "halal": sum(1 for p in base_pool if p.get("dietary", {}).get("halal")), "vegan": sum(1 for p in base_pool if p.get("dietary", {}).get("vegan")), "vegetarian": sum(1 for p in base_pool if p.get("dietary", {}).get("vegetarian")), "celiac": sum(1 for p in base_pool if p.get("celiac_safety", {}).get("is_safe")), "water": sum(1 for p in base_pool if p.get("hospitality", {}).get("free_tap_water")), "monument": sum(1 for p in base_pool if p.get("heritage", {}).get("is_monument")) } # Baseline counts over all authentic places in city to know what options exist in the city baseline_facets = self.get_facets(authentic_only=True) city_supported_keys = { "independent": baseline_facets["independent_count"] > 0, "clean": baseline_facets.get("clean_kitchen_count", 0) > 0, "accredited": baseline_facets["accredited_count"] > 0, "late_night": baseline_facets.get("late_night_count", 0) > 0, "roastery": baseline_facets.get("roastery_count", 0) > 0, "historic": baseline_facets.get("historic_count", 0) > 0, "wheelchair": baseline_facets.get("wheelchair_count", 0) > 0, "canal": baseline_facets["canal_terrace_count"] > 0, "outdoor": baseline_facets["outdoor_count"] > 0, "scratch": baseline_facets["scratch_count"] > 0, "halal": baseline_facets["halal_count"] > 0, "vegan": baseline_facets["vegan_count"] > 0, "vegetarian": baseline_facets["vegetarian_count"] > 0, "celiac": baseline_facets["celiac_safe_count"] > 0, "water": baseline_facets["tap_water_count"] > 0, "monument": baseline_facets.get("monuments_count", 0) > 0 } row2_facets = [ {"id": "independent", "type": "amenity", "label": "100% Independent", "icon": "👨‍🍳", "count": amenity_counts["independent"]}, {"id": "clean", "type": "amenity", "label": "Grade A Clean Kitchen" if self.city_id == "dubai" else "Gov Hygiene Certified", "icon": "🏆", "count": amenity_counts["clean"]}, {"id": "accredited", "type": "amenity", "label": "Awarded by Guides", "icon": "🏅", "count": amenity_counts["accredited"]}, {"id": "late_night", "type": "amenity", "label": "Late-Night Dining (Post-Midnight)", "icon": "🌙", "count": amenity_counts["late_night"]}, {"id": "roastery", "type": "amenity", "label": "On-Site Micro-Roaster", "icon": "☕", "count": amenity_counts["roastery"]}, {"id": "historic", "type": "amenity", "label": "Historic Icons (Est. Pre-2010)", "icon": "⏳", "count": amenity_counts["historic"]}, {"id": "wheelchair", "type": "amenity", "label": "Wheelchair Accessible", "icon": "♿", "count": amenity_counts["wheelchair"]}, {"id": "canal", "type": "amenity", "label": waterfront_label, "icon": "⛵" if "Canal" in waterfront_label else "🌊", "count": amenity_counts["canal"]}, {"id": "outdoor", "type": "amenity", "label": "Outdoor Seating", "icon": "☀️", "count": amenity_counts["outdoor"]}, {"id": "scratch", "type": "amenity", "label": "Scratch Kitchen", "icon": "🌾", "count": amenity_counts["scratch"]}, {"id": "halal", "type": "diet", "label": "Halal Verified", "icon": "🥩", "count": amenity_counts["halal"]}, {"id": "vegan", "type": "diet", "label": "100% Vegan", "icon": "🌿", "count": amenity_counts["vegan"]}, {"id": "vegetarian", "type": "diet", "label": "Vegetarian", "icon": "🌱", "count": amenity_counts["vegetarian"]}, {"id": "celiac", "type": "amenity", "label": "Celiac Safe Kitchen", "icon": "🩺", "count": amenity_counts["celiac"]}, {"id": "water", "type": "amenity", "label": "Free Tap Water", "icon": "💧", "count": amenity_counts["water"]}, {"id": "monument", "type": "amenity", "label": heritage_label, "icon": "🏛️", "count": amenity_counts["monument"]} ] active_facets = [f for f in row2_facets if city_supported_keys.get(f["id"], False)] return { "city": city_name, "city_id": self.city_id, "currency_symbol": currency_sym, "price_tiers": price_tiers_data, "total_authentic": len(base_pool), "row1_clusters": row1_clusters, "all_clusters": all_clusters, "row2_facets": active_facets, "dining_guide": self.city_config.get("dining_guide", {}), "is_dynamic": has_active_filters } def compare_head_to_head(self, id1, id2): p1 = next((p for p in self.places if p["id"] == id1), None) p2 = next((p for p in self.places if p["id"] == id2), None) if not p1 or not p2: return None winner = p1 if p1["quality_score"] >= p2["quality_score"] else p2 loser = p2 if winner == p1 else p1 dist_between_m = int(haversine_distance_m( p1["latitude"], p1["longitude"], p2["latitude"], p2["longitude"] )) advantages = [] if winner.get("ownership", {}).get("is_independent") and not loser.get("ownership", {}).get("is_independent"): advantages.append(f"Ownership Advantage: 100% independent artisan shop vs multi-unit commercial chain ({loser.get('ownership', {}).get('badge', 'Chain')}).") if winner.get("accreditation") and not loser.get("accreditation"): tier = winner['accreditation'].get('award_tier') or winner['accreditation'].get('award') advantages.append(f"Holds official culinary accreditation ({tier}).") if winner.get("celiac_safety", {}).get("is_safe") and not loser.get("celiac_safety", {}).get("is_safe"): advantages.append(f"Medical Safety Advantage: {winner['celiac_safety'].get('badge')} with verified cross-contamination protocols.") if winner.get("scratch_kitchen", {}).get("is_scratch") and not loser.get("scratch_kitchen", {}).get("is_scratch"): advantages.append("Culinary Craft Advantage: Verified scratch production (scratch dough, in-house butchery, fresh prep).") if winner.get("heritage", {}).get("is_monument") and not loser.get("heritage", {}).get("is_monument"): badge = winner['heritage'].get('badge') or 'Historic Monument' advantages.append(f"Architectural Heritage Advantage: Housed in authentic protected monument ({badge}) vs modern commercial space.") if winner.get("hospitality", {}).get("free_tap_water") and not loser.get("hospitality", {}).get("free_tap_water"): advantages.append("Honest Hospitality: Pledged complimentary fresh tap water carafes.") if winner["craft_badges"] and not loser["craft_badges"]: advantages.append(f"Uses verified craft techniques ({', '.join(winner['craft_badges'][:2])}) whereas {loser['name']} has none.") if winner.get("amenities", {}).get("canal_terrace") and not loser.get("amenities", {}).get("canal_terrace"): advantages.append("Offers scenic outdoor canal terrace seating along historic waterways.") if winner.get("schedule_rhythm") == "artisan_rhythm" and loser.get("schedule_rhythm") == "tourist_continuous": advantages.append("Follows traditional artisan kitchen rhythms with fresh prep and rest days instead of continuous factory heating.") if winner["category_purity"] > loser["category_purity"]: advantages.append(f"100% focused on its core specialty ({winner['primary_category']}) instead of a diluted multi-menu.") if not winner["is_chain"] and loser["is_chain"]: advantages.append(f"Independent local artisan shop vs. a multi-unit corporate franchise ({loser['brand']}).") if winner.get("start_year") and (not loser.get("start_year") or winner["start_year"] < loser["start_year"]): advantages.append(f"Proven Lindy longevity: established in {winner['start_year']}.") if loser.get("dist_to_tourist_hub_m") and loser["dist_to_tourist_hub_m"] < 100: advantages.append(f"{winner['name']} avoids high-density commercial corridor congestion.") if not advantages: advantages.append(f"{winner['name']} ranks higher on authentic neighborhood presence and category focus.") return { "place_a": p1, "place_b": p2, "winner_id": winner["id"], "winner_name": winner["name"], "loser_name": loser["name"], "distance_between_m": dist_between_m, "advantages": advantages, "recommendation": f"Walk {dist_between_m}m to {winner['name']} for authentic quality." if dist_between_m > 0 else f"Choose {winner['name']}." } class MultiCityEngineManager: def __init__(self): self.engines = {} self.cities = self.load_cities() self._cached_city_list = None def load_cities(self): cities_file = os.path.join(DATA_DIR, "cities.json") if os.path.exists(cities_file): try: with open(cities_file, "r", encoding="utf-8") as f: return json.load(f) except Exception as e: print(f"Error loading cities.json: {e}") return [{"id": "amsterdam", "name": "Amsterdam", "country": "Netherlands", "flag": "🇳🇱"}] def get_engine(self, city_id="amsterdam"): cid = (city_id or "amsterdam").lower().strip() valid_ids = [c["id"] for c in self.cities] if cid not in valid_ids: cid = "amsterdam" if cid not in self.engines: print(f"Initializing PalateEngine for city: {cid}...") self.engines[cid] = PalateEngine(city_id=cid) return self.engines[cid] def list_cities(self): result = [] for c in self.cities: cid = c["id"] eng = self.engines.get(cid) total = len(eng.places) if eng else c.get("total_eateries", 3500) authentic = sum(1 for p in eng.places if not p.get("is_tourist_trap")) if eng else total craft = sum(1 for p in eng.places if p.get("craft_badges")) if eng else int(total * 0.4) independent = sum(1 for p in eng.places if p.get("ownership", {}).get("is_independent")) if eng else int(total * 0.7) result.append({ "id": cid, "name": c["name"], "country": c["country"], "country_code": c.get("country_code", ""), "flag": c.get("flag", "🌐"), "center": c["center"], "default_zoom": c.get("default_zoom", 13), "tagline": c.get("tagline", ""), "currency": c.get("currency", "EUR"), "currency_symbol": c.get("currency_symbol", "€"), "total_eateries": total, "authentic_eateries": authentic, "craft_artisans_detected": craft, "independent_count": independent, "dining_guide": c.get("dining_guide", {}) }) return result if __name__ == "__main__": engine = PalateEngine() print("\n--- TEST: 'burger' TOP RESULTS ---") for r in engine.search("burger", max_results=5): print(f"[{r['quality_score']} pts] {r['name']} ({r['neighborhood']})") print(f" Badges: {r['craft_badges']}") print("\n--- TEST: 'pizza' TOP RESULTS ---") for r in engine.search("pizza", max_results=6): print(f"[{r['quality_score']} pts] {r['name']} ({r['neighborhood']})") print(f" Badges: {r['craft_badges']}")