Spaces:
Running
Running
File size: 3,788 Bytes
00b75ab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """Place-name search: Open-Meteo geocoding first, Gemini with Google Search as a fallback."""
from __future__ import annotations
import json
import logging
import os
import re
from typing import Any
import requests
LOGGER = logging.getLogger(__name__)
OPEN_METEO_GEOCODING_URL = "https://geocoding-api.open-meteo.com/v1/search"
GEOCODING_TIMEOUT_SECONDS = 10
GEMINI_GEOCODING_TIMEOUT_MS = 20_000
MAX_RESULTS = 5
# Generous bounding box around Pakistan; anything outside is rejected.
PAKISTAN_BOUNDS = {"lat": (23.0, 37.5), "lon": (60.5, 77.5)}
def _in_pakistan(latitude: Any, longitude: Any) -> bool:
try:
lat, lon = float(latitude), float(longitude)
except (TypeError, ValueError):
return False
return (
PAKISTAN_BOUNDS["lat"][0] <= lat <= PAKISTAN_BOUNDS["lat"][1]
and PAKISTAN_BOUNDS["lon"][0] <= lon <= PAKISTAN_BOUNDS["lon"][1]
)
def search_open_meteo(query: str) -> list[dict[str, Any]]:
response = requests.get(
OPEN_METEO_GEOCODING_URL,
params={"name": query, "count": MAX_RESULTS, "language": "en", "format": "json", "countryCode": "PK"},
timeout=GEOCODING_TIMEOUT_SECONDS,
)
response.raise_for_status()
results = []
for item in response.json().get("results") or []:
if not _in_pakistan(item.get("latitude"), item.get("longitude")):
continue
parts = [item.get("name"), item.get("admin2"), item.get("admin1")]
name = ", ".join(dict.fromkeys(part for part in parts if part))
results.append({
"name": name,
"latitude": round(float(item["latitude"]), 4),
"longitude": round(float(item["longitude"]), 4),
"source": "open-meteo",
})
return results
def search_gemini(query: str) -> list[dict[str, Any]]:
api_key = os.getenv("GEMINI_API_KEY")
if not api_key:
return []
from google import genai
from google.genai import types
client = genai.Client(api_key=api_key, http_options=types.HttpOptions(timeout=GEMINI_GEOCODING_TIMEOUT_MS))
prompt = (
"Find the geographic coordinates of this place in Pakistan, using Google Search to verify: "
f"{json.dumps(query)}.\n"
'Reply with only a JSON object: {"name": "<place, district, province>", "latitude": <number>, "longitude": <number>}. '
'If you cannot identify the place confidently, reply {"name": null}.'
)
# Search grounding cannot be combined with a JSON response schema, so parse the text.
response = client.models.generate_content(
model=os.getenv("GEMINI_MODEL", "gemini-2.5-flash"),
contents=prompt,
config=types.GenerateContentConfig(
tools=[types.Tool(google_search=types.GoogleSearch())],
temperature=0,
),
)
match = re.search(r"\{.*?\}", response.text or "", re.DOTALL)
if not match:
return []
data = json.loads(match.group(0))
if not data.get("name") or not _in_pakistan(data.get("latitude"), data.get("longitude")):
return []
return [{
"name": str(data["name"])[:200],
"latitude": round(float(data["latitude"]), 4),
"longitude": round(float(data["longitude"]), 4),
"source": "gemini",
}]
def search_places(query: str) -> list[dict[str, Any]]:
"""Return matching places in Pakistan; an empty list when nothing is found."""
try:
results = search_open_meteo(query)
if results:
return results
except (requests.RequestException, ValueError, KeyError):
LOGGER.exception("Open-Meteo geocoding failed for %r", query)
try:
return search_gemini(query)
except Exception:
LOGGER.exception("Gemini geocoding failed for %r", query)
return []
|