import os from dotenv import load_dotenv BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) load_dotenv(os.path.join(BASE_DIR, ".env")) GOOGLE_API_KEY = os.getenv("GOOGLE_MAPS_API_KEY") def get_gmaps(): """Lazy load Google Maps client.""" if not GOOGLE_API_KEY: return None try: import googlemaps return googlemaps.Client(key=GOOGLE_API_KEY) except Exception: return None # Keep all existing TWIN_CITY_MAP, FERRY_CITIES, HILL_CITIES etc. TWIN_CITY_MAP = { "hubli": ["hubli", "hubballi", "dharwad", "hubli-dharwad"], "dharwad": ["hubli", "hubballi", "dharwad", "hubli-dharwad"], } FERRY_CITIES = ["mumbai", "kochi", "guwahati", "port blair", "panaji", "kavaratti"] HILL_CITIES = ["gangtok", "shillong", "shimla", "darjeeling", "mussoorie", "nainital", "manali", "ooty"] WATER_KEYWORDS = { "ferry": ["jetty", "ghat", "wharf", "pier", "dock", "ferry", "harbour", "harbor", "port", "gateway", "island", "creek"], "shikara": ["lake", "dal", "nagin", "ghat", "houseboat", "boulevard", "water", "floating", "char chinar"], } def get_location_info(place: str, city: str) -> dict: gmaps = get_gmaps() if not gmaps: return {"city": city.lower(), "district": city.lower(), "state": "unknown", "sublocality": "", "lat": 0, "lng": 0} try: results = gmaps.geocode(f"{place}, {city}, India") if not results: results = gmaps.geocode(f"{place}, India") if not results: return {"city": city.lower(), "district": city.lower(), "state": "unknown", "sublocality": "", "lat": 0, "lng": 0} components = results[0]["address_components"] location = results[0]["geometry"]["location"] data = {"lat": location["lat"], "lng": location["lng"], "city": "", "district": "", "state": "", "sublocality": "", "locality": ""} for comp in components: types = comp["types"] if "locality" in types: data["city"] = comp["long_name"].lower() elif "administrative_area_level_2" in types: data["district"] = comp["long_name"].lower() elif "administrative_area_level_1" in types: data["state"] = comp["long_name"].lower() elif "sublocality" in types or "sublocality_level_1" in types: data["sublocality"] = comp["long_name"].lower() return data except Exception as e: print(f"⚠️ Geocoding failed: {e}") return {"city": city.lower(), "district": city.lower(), "state": "unknown", "sublocality": "", "lat": 0, "lng": 0} def is_within_city(place: str, expected_city: str) -> bool: gmaps = get_gmaps() if not gmaps: return True loc = get_location_info(place, expected_city) expected = expected_city.lower() return ( expected in loc.get("city", "") or expected in loc.get("district", "") or loc.get("city", "").replace(" ", "") in expected.replace(" ", "") or loc.get("district", "").replace(" ", "") in expected.replace(" ", "") ) def _is_in_twin_city(loc_data: dict, valid_cities: list) -> bool: city_str = (loc_data.get("city", "") + " " + loc_data.get("district", "") + " " + loc_data.get("sublocality", "")).lower() return any(vc in city_str for vc in valid_cities) def validate_corridor(transport_type: str, source: str, destination: str, city: str) -> dict: t = transport_type.lower() city_lower = city.lower() if t in ["bus", "metro", "train"]: return {"valid": True, "reason": None} if t == "chigari": valid_twin = TWIN_CITY_MAP.get(city_lower) if not valid_twin: return {"valid": False, "reason": f"Chigari only operates in Hubli-Dharwad, not {city}."} src_loc = get_location_info(source, city) dest_loc = get_location_info(destination, city) src_valid = _is_in_twin_city(src_loc, valid_twin) dest_valid = _is_in_twin_city(dest_loc, valid_twin) if src_valid and dest_valid: return {"valid": True, "reason": None} elif not src_valid: return {"valid": False, "reason": f"'{source}' is outside Hubli-Dharwad corridor."} else: return {"valid": False, "reason": f"'{destination}' is outside Hubli-Dharwad corridor."} elif t == "tram": if city_lower != "kolkata": return {"valid": False, "reason": f"Tram only in Kolkata, not {city}."} return {"valid": True, "reason": None} elif t == "ferry": if city_lower not in FERRY_CITIES: return {"valid": False, "reason": f"Ferry not available in {city}."} if city_lower == "kavaratti": return {"valid": True, "reason": None} water_kw = WATER_KEYWORDS["ferry"] near_water = any(kw in source.lower() or kw in destination.lower() for kw in water_kw) if near_water: return {"valid": True, "reason": None} return {"valid": True, "reason": None} elif t == "toy_train": if city_lower != "shimla": return {"valid": False, "reason": f"Toy Train only on Kalka-Shimla route, not {city}."} return {"valid": True, "reason": None} elif t == "shikara": if city_lower != "srinagar": return {"valid": False, "reason": f"Shikara only in Srinagar, not {city}."} water_kw = WATER_KEYWORDS["shikara"] near_water = any(kw in source.lower() or kw in destination.lower() for kw in water_kw) if near_water: return {"valid": True, "reason": None} return {"valid": True, "reason": None} elif t == "shared_cab": if city_lower not in HILL_CITIES: return {"valid": False, "reason": f"Shared cab not standard in {city}."} return {"valid": True, "reason": None} return {"valid": True, "reason": None}