app.py

import time
import threading
from typing import Dict, Any, Optional, Tuple, List

import requests
import folium
import gradio as gr


NOMINATIM_SEARCH = "https://nominatim.openstreetmap.org/search"
OSRM_ROUTE = "https://router.project-osrm.org/route/v1/driving"

# Nominatim 公共服务要求：最多 1 req/s，且必须自定义 Referer 或 User-Agent
# 这里做一个进程级简单限速（多并发时仍建议你在生产用缓存/更严格限速）
_rate_lock = threading.Lock()
_last_nominatim_ts = 0.0


def _sleep_for_nominatim_rate_limit(min_interval_sec: float = 1.05) -> None:
    global _last_nominatim_ts
    with _rate_lock:
        now = time.time()
        wait = (_last_nominatim_ts + min_interval_sec) - now
        if wait > 0:
            time.sleep(wait)
        _last_nominatim_ts = time.time()


def geocode_nominatim(
    query: str,
    email: str,
    countrycodes: str = "cn",
    limit: int = 1,
    timeout: int = 20,
) -> Tuple[float, float, str]:
    """
    Nominatim Search API:
    - 建议大量请求时带 email 参数（文档支持 email=）
    - 必须自定义 User-Agent/Referer，且 <= 1 req/s
    """
    _sleep_for_nominatim_rate_limit()

    headers = {
        # 真实 email 更好；别用 requests 默认 UA
        "User-Agent": f"hf-gradio-osm-demo/1.0 ({email})" if email else "hf-gradio-osm-demo/1.0",
        "Accept": "application/json",
    }
    params = {
        "q": query,
        "format": "jsonv2",
        "limit": str(limit),
        "addressdetails": "1",
        "countrycodes": countrycodes,
    }
    # email 参数：对大批量/脚本化请求尤其建议填写
    if email:
        params["email"] = email

    r = requests.get(NOMINATIM_SEARCH, params=params, headers=headers, timeout=timeout)
    r.raise_for_status()
    data = r.json()
    if not data:
        raise ValueError(f"Nominatim 无结果：{query}")

    top = data[0]
    lat = float(top["lat"])
    lon = float(top["lon"])
    display_name = top.get("display_name", query)
    return lat, lon, display_name


def route_osrm(
    start_lat: float, start_lon: float,
    end_lat: float, end_lon: float,
    timeout: int = 30,
) -> Dict[str, Any]:
    """
    OSRM Route API：steps/geometries/overview 参数见官方文档
    """
    coords = f"{start_lon},{start_lat};{end_lon},{end_lat}"  # OSRM: lon,lat
    params = {"steps": "true", "geometries": "geojson", "overview": "full"}
    url = f"{OSRM_ROUTE}/{coords}"
    r = requests.get(url, params=params, timeout=timeout)
    r.raise_for_status()
    data = r.json()
    if data.get("code") != "Ok" or not data.get("routes"):
        raise ValueError(f"OSRM 路由失败：{data}")
    return data["routes"][0]


def _maneuver_cn(step: Dict[str, Any]) -> str:
    m = step.get("maneuver") or {}
    t = m.get("type", "")
    mod = m.get("modifier", "")
    name = step.get("name", "") or ""

    type_cn = {
        "depart": "出发",
        "arrive": "到达",
        "turn": "转向",
        "new name": "继续（道路改名）",
        "merge": "并入",
        "on ramp": "上匝道",
        "off ramp": "下匝道",
        "fork": "分叉",
        "roundabout": "进入环岛",
        "rotary": "进入环岛",
        "roundabout turn": "环岛转向",
        "end of road": "到道路尽头",
        "continue": "继续",
    }.get(t, t)

    mod_cn = {
        "left": "左转",
        "right": "右转",
        "straight": "直行",
        "slight left": "向左前方",
        "slight right": "向右前方",
        "sharp left": "向左急转",
        "sharp right": "向右急转",
        "uturn": "掉头",
    }.get(mod, mod)

    if t in ("depart", "arrive"):
        return f"{type_cn}（{name}）".strip()
    if mod_cn:
        return f"{type_cn}：{mod_cn} 进入 {name}".strip()
    return f"{type_cn}：进入 {name}".strip()


def build_map_html(
    start: Tuple[float, float, str],
    end: Tuple[float, float, str],
    route: Dict[str, Any],
) -> str:
    s_lat, s_lon, s_name = start
    e_lat, e_lon, e_name = end

    geom = route.get("geometry")  # GeoJSON LineString with [lon, lat]
    coords = geom.get("coordinates", []) if isinstance(geom, dict) else []
    latlngs = [(lat, lon) for lon, lat in coords]  # folium expects (lat, lon)

    # 初始中心用起终点中点
    m = folium.Map(location=[(s_lat + e_lat) / 2, (s_lon + e_lon) / 2], zoom_start=10, control_scale=True)

    folium.Marker([s_lat, s_lon], popup=f"起点：{s_name}", tooltip="起点").add_to(m)
    folium.Marker([e_lat, e_lon], popup=f"终点：{e_name}", tooltip="终点").add_to(m)

    if latlngs:
        folium.PolyLine(latlngs, weight=6, opacity=0.85).add_to(m)
        min_lat = min(lat for lat, _ in latlngs)
        min_lon = min(lon for _, lon in latlngs)
        max_lat = max(lat for lat, _ in latlngs)
        max_lon = max(lon for _, lon in latlngs)
        m.fit_bounds([[min_lat, min_lon], [max_lat, max_lon]])

    # folium 自带 OSM attribution
    # _repr_html_ returns an iframe embed that works better inside Gradio.
    return m._repr_html_()


def plan(start_query: str, end_query: str, email: str) -> Tuple[str, str, str]:
    try:
        if not email or "@" not in email:
            return ("", "❌ 请填写一个可联系的 email（会用于 Nominatim 的 `email=` 参数与 User-Agent）。", "")

        start = geocode_nominatim(start_query, email=email)
        end = geocode_nominatim(end_query, email=email)

        route = route_osrm(start[0], start[1], end[0], end[1])

        dist_km = route["distance"] / 1000.0
        dur_min = route["duration"] / 60.0

        summary_md = (
            f"**起点：** {start[2]}\n\n"
            f"**终点：** {end[2]}\n\n"
            f"**总里程：** {dist_km:.2f} km　　**预计用时：** {dur_min:.1f} 分钟\n\n"
            f"> 注：Nominatim 公共服务要求 **≤ 1 req/s** 且 **自定义 User-Agent/Referer**。"
        )

        # Steps
        steps_lines: List[str] = []
        step_idx = 0
        for leg in route.get("legs", []):
            for step in leg.get("steps", []):
                step_idx += 1
                d = step.get("distance", 0.0)
                steps_lines.append(f"{step_idx}. {_maneuver_cn(step)}（约 {d:.0f} m）")
                if step_idx >= 60:
                    break
            if step_idx >= 60:
                break
        steps_md = "\n".join(steps_lines) if steps_lines else "_（OSRM 未返回 steps）_"

        map_html = build_map_html(start, end, route)
        return map_html, summary_md, steps_md

    except requests.HTTPError as e:
        return ("", f"❌ HTTP 错误：{e}", "")
    except Exception as e:
        return ("", f"❌ 失败：{e}", "")


with gr.Blocks(title="OSM 路线 Demo（Nominatim + OSRM + Folium）") as demo:
    gr.Markdown(
        "## OSM 路线 Demo（Nominatim + OSRM）\n"
        "- 地理编码：Nominatim Search API（会用你填写的 email 作为 `email=` 参数并拼进 User-Agent）  \n"
        "- 路线：OSRM Route API（`steps=true&geometries=geojson&overview=full`）  \n"
        "> 仅用于调试/演示：公共服务无 SLA，别高频。"
    )

    with gr.Row():
        start_in = gr.Textbox(label="起点", value="北京 大恒科技大厦")
        end_in = gr.Textbox(label="终点", value="北京大兴国际机场")
    email_in = gr.Textbox(label="Email（建议填真实可联系邮箱）", placeholder="your_name@example.com")

    btn = gr.Button("规划并绘制")

    with gr.Row():
        map_out = gr.HTML(label="地图（路线）", sanitize=False)
    with gr.Row():
        summary_out = gr.Markdown(label="摘要")
    with gr.Row():
        steps_out = gr.Markdown(label="导航步骤（前 60 条）")

    btn.click(plan, inputs=[start_in, end_in, email_in], outputs=[map_out, summary_out, steps_out])

if __name__ == "__main__":
    demo.launch()