crowdroute / app /services /corridor_validator.py
UAVDETECTION's picture
Fix lazy loading to prevent startup hang
6455d34
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}