routeopt_agent/geo_tools.py

from __future__ import annotations

import json
import math
import os
import re
import time
from pathlib import Path
from typing import Any
from urllib.parse import quote

import requests

from .models import GeoPoint, RouteMatrix


NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
OSRM_TABLE_URL = "https://router.project-osrm.org/table/v1/driving"
CACHE_DIR = Path("cache")
GEO_CACHE_PATH = CACHE_DIR / "geo_cache.json"
BUILTIN_DEMO_CITIES = ("上海", "北京")


DEMO_POINTS: dict[str, tuple[float, float, str]] = {
    "上海交通大学闵行校区": (31.0256, 121.4332, "上海交通大学闵行校区"),
    "交大闵行": (31.0256, 121.4332, "上海交通大学闵行校区"),
    "上海交大闵行校区": (31.0256, 121.4332, "上海交通大学闵行校区"),
    "外滩": (31.2400, 121.4900, "外滩"),
    "陆家嘴": (31.2381, 121.4970, "陆家嘴"),
    "人民广场": (31.2304, 121.4737, "人民广场"),
    "徐家汇": (31.1838, 121.4328, "徐家汇"),
    "豫园": (31.2272, 121.4922, "豫园"),
    "静安寺": (31.2230, 121.4454, "静安寺"),
    "南京东路": (31.2366, 121.4842, "南京东路"),
    "东方明珠": (31.2397, 121.4998, "东方明珠广播电视塔"),
    "上海虹桥站": (31.1945, 121.3188, "上海虹桥站"),
    "上海南站": (31.1546, 121.4296, "上海南站"),
    "复旦大学": (31.2989, 121.5038, "复旦大学邯郸校区"),
    "同济大学": (31.2820, 121.5062, "同济大学四平路校区"),
    "华东师范大学": (31.2282, 121.4038, "华东师范大学普陀校区"),
    "北京邮电大学": (39.9620, 116.3585, "北京邮电大学西土城路校区"),
    "北邮": (39.9620, 116.3585, "北京邮电大学西土城路校区"),
    "北京邮电大学西土城路校区": (39.9620, 116.3585, "北京邮电大学西土城路校区"),
    "国家大剧院": (39.9038, 116.3838, "国家大剧院"),
    "北京国家大剧院": (39.9038, 116.3838, "国家大剧院"),
    "鼓楼": (39.9404, 116.3972, "北京鼓楼"),
    "北京鼓楼": (39.9404, 116.3972, "北京鼓楼"),
    "鼓楼大街": (39.9470, 116.3938, "鼓楼大街"),
    "北海": (39.9255, 116.3895, "北海公园"),
    "北海公园": (39.9255, 116.3895, "北海公园"),
    "国家植物园": (39.9974, 116.2072, "国家植物园"),
    "北京国家植物园": (39.9974, 116.2072, "国家植物园"),
    "北京植物园": (39.9974, 116.2072, "国家植物园"),
}


class GeoToolError(RuntimeError):
    pass


def geocode_places(places: list[str], city_hint: str = "", timeout: int = 12) -> list[GeoPoint]:
    cache = load_json_cache(GEO_CACHE_PATH)
    results: list[GeoPoint] = []
    last_live_query_at = 0.0

    for place in places:
        place = place.strip()
        if not place:
            raise GeoToolError("地点列表中存在空项。请删除空行，确保起点和每个目的地都有具体名称。")

        manual = parse_manual_point(place)
        if manual:
            point = manual
            results.append(point)
            continue

        demo = lookup_demo_point(place)
        if demo:
            lat, lon, display_name = demo
            point = GeoPoint(place, place, lat, lon, display_name, "built-in-demo")
            cache_key = normalize_key(place, city_hint)
            cache[cache_key] = point.__dict__
            results.append(point)
            continue

        cache_key = normalize_key(place, city_hint)
        if cache_key in cache:
            cached = cache[cache_key]
            results.append(GeoPoint(**cached))
            continue

        query = f"{place}, {city_hint}".strip(" ,")
        sleep_for_nominatim(last_live_query_at)
        last_live_query_at = time.time()
        try:
            point = geocode_with_nominatim(place, query, timeout)
        except GeoToolError:
            raise
        except requests.exceptions.Timeout as exc:
            raise GeoToolError(build_geocode_failure_message(place, query, "在线地理编码服务响应超时")) from exc
        except requests.exceptions.ConnectionError as exc:
            raise GeoToolError(build_geocode_failure_message(place, query, "无法连接在线地理编码服务")) from exc
        except requests.exceptions.HTTPError as exc:
            raise GeoToolError(build_geocode_failure_message(place, query, f"在线地理编码 HTTP 错误：{exc.response.status_code if exc.response else 'unknown'}")) from exc
        except requests.exceptions.RequestException as exc:
            raise GeoToolError(build_geocode_failure_message(place, query, f"在线地理编码请求失败：{exc}")) from exc
        cache[cache_key] = point.__dict__
        results.append(point)

    save_json_cache(GEO_CACHE_PATH, cache)
    return results


def build_route_matrix(points: list[GeoPoint], timeout: int = 15) -> RouteMatrix:
    if len(points) < 2:
        raise GeoToolError("至少需要起点和 1 个目的地。")

    coords = ";".join(f"{point.lon:.6f},{point.lat:.6f}" for point in points)
    url = f"{OSRM_TABLE_URL}/{coords}"
    params = {"annotations": "duration,distance"}

    try:
        response = requests.get(url, params=params, timeout=timeout)
        response.raise_for_status()
        payload = response.json()
        durations = payload.get("durations")
        distances = payload.get("distances")
        if not matrix_is_complete(durations) or not matrix_is_complete(distances):
            raise GeoToolError("OSRM 返回的矩阵不完整。")
        return RouteMatrix(
            points=points,
            durations=durations,
            distances=distances,
            source="OSRM public demo server",
        )
    except Exception:
        durations, distances = approximate_city_matrix(points)
        return RouteMatrix(
            points=points,
            durations=durations,
            distances=distances,
            source="haversine fallback, speed=35km/h, road factor=1.30",
        )


def geocode_with_nominatim(place: str, query: str, timeout: int) -> GeoPoint:
    headers = {
        "User-Agent": "routeopt-agent-homework/1.0 (public classroom demo)",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.6",
    }
    params = {
        "q": query,
        "format": "jsonv2",
        "limit": 1,
        "addressdetails": 1,
    }
    response = requests.get(NOMINATIM_URL, params=params, headers=headers, timeout=timeout)
    response.raise_for_status()
    items = response.json()
    if not items:
        raise GeoToolError(build_geocode_failure_message(place, query, "在线地理编码没有找到匹配地点"))
    item = items[0]
    return GeoPoint(
        name=place,
        query=query,
        lat=float(item["lat"]),
        lon=float(item["lon"]),
        display_name=item.get("display_name", query),
        source="OpenStreetMap Nominatim",
    )


def lookup_demo_point(place: str) -> tuple[float, float, str] | None:
    normalized = place.strip().lower()
    for key, value in DEMO_POINTS.items():
        key_norm = key.lower()
        if normalized == key_norm or normalized in key_norm or key_norm in normalized:
            return value
    return None


def parse_manual_point(place: str) -> GeoPoint | None:
    patterns = [
        r"^(?P<name>.+?)[@|]\s*(?P<lat>-?\d+(?:\.\d+)?)\s*[,，]\s*(?P<lon>-?\d+(?:\.\d+)?)$",
        r"^(?P<name>.+?)[（(]\s*(?P<lat>-?\d+(?:\.\d+)?)\s*[,，]\s*(?P<lon>-?\d+(?:\.\d+)?)\s*[)）]$",
    ]
    for pattern in patterns:
        match = re.match(pattern, place.strip())
        if not match:
            continue
        name = match.group("name").strip()
        lat = float(match.group("lat"))
        lon = float(match.group("lon"))
        if not name:
            raise GeoToolError("手动坐标格式缺少地点名称。正确格式示例：未知景点@39.90,116.40")
        if not (-90 <= lat <= 90 and -180 <= lon <= 180):
            raise GeoToolError(f"手动坐标超出范围：{place}。纬度应在 -90 到 90，经度应在 -180 到 180。")
        return GeoPoint(
            name=name,
            query=place,
            lat=lat,
            lon=lon,
            display_name=f"{name}（手动坐标）",
            source="manual-coordinate",
        )
    return None


def build_geocode_failure_message(place: str, query: str, reason: str) -> str:
    return (
        f"地点解析失败：`{place}`。\n\n"
        f"失败原因：{reason}。\n\n"
        f"本次查询语句：`{query}`。\n\n"
        f"当前内置演示坐标覆盖城市：{builtin_city_summary()}。如果你输入的是其他城市或较冷门地点，"
        "系统会尝试在线地理编码；当在线服务超时、限流或找不到地点时，就需要你把地名写得更具体，"
        "例如加上城市、区县、校区/景区全称。\n\n"
        "也可以直接使用手动坐标格式绕过在线地理编码：`地点名@纬度,经度`，例如 `某景点@39.9042,116.4074`。"
    )


def builtin_city_summary() -> str:
    return "、".join(BUILTIN_DEMO_CITIES)


def builtin_place_examples(limit: int = 12) -> str:
    names = list(DEMO_POINTS.keys())[:limit]
    return "、".join(names)


def approximate_city_matrix(points: list[GeoPoint]) -> tuple[list[list[float]], list[list[float]]]:
    distances: list[list[float]] = []
    durations: list[list[float]] = []
    road_factor = 1.30
    speed_mps = 35_000 / 3600
    for origin in points:
        distance_row: list[float] = []
        duration_row: list[float] = []
        for dest in points:
            if origin is dest:
                distance = 0.0
            else:
                distance = haversine_meters(origin.lat, origin.lon, dest.lat, dest.lon) * road_factor
            distance_row.append(distance)
            duration_row.append(distance / speed_mps if speed_mps else 0.0)
        distances.append(distance_row)
        durations.append(duration_row)
    return durations, distances


def haversine_meters(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    radius = 6_371_000
    phi1 = math.radians(lat1)
    phi2 = math.radians(lat2)
    d_phi = math.radians(lat2 - lat1)
    d_lambda = math.radians(lon2 - lon1)
    a = (
        math.sin(d_phi / 2) ** 2
        + math.cos(phi1) * math.cos(phi2) * math.sin(d_lambda / 2) ** 2
    )
    return 2 * radius * math.atan2(math.sqrt(a), math.sqrt(1 - a))


def matrix_is_complete(matrix: Any) -> bool:
    if not isinstance(matrix, list) or not matrix:
        return False
    size = len(matrix)
    return all(isinstance(row, list) and len(row) == size and all(value is not None for value in row) for row in matrix)


def format_osm_link(point: GeoPoint) -> str:
    return f"https://www.openstreetmap.org/?mlat={point.lat:.6f}&mlon={point.lon:.6f}#map=14/{point.lat:.6f}/{point.lon:.6f}"


def make_osrm_route_link(points: list[GeoPoint], route_indices: list[int]) -> str:
    coords = ";".join(f"{points[idx].lon:.6f},{points[idx].lat:.6f}" for idx in route_indices)
    return f"https://router.project-osrm.org/route/v1/driving/{quote(coords, safe=';,')}"


def sleep_for_nominatim(last_live_query_at: float) -> None:
    elapsed = time.time() - last_live_query_at
    if last_live_query_at and elapsed < 1.05:
        time.sleep(1.05 - elapsed)


def load_json_cache(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}


def save_json_cache(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def normalize_key(place: str, city_hint: str) -> str:
    return f"{place.strip().lower()}|{city_hint.strip().lower()}"