Create providers/navitia.py

services/providers/navitia.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+import os
+import requests
+from typing import Optional
+
+# 使い方:
+# 1) https://www.navitia.io/ でAPIトークンを取得し、環境変数 NAVITIA_TOKEN に設定
+# 2) カバレッジを NAVITIA_COVERAGE（例: "jp-tokyo"）に設定
+# 取得できない場合は None を返す（フォールバックはOSRM + 概算運賃）
+
+def get_journey(from_lat: float, from_lon: float, to_lat: float, to_lon: float, when_iso: Optional[str] = None) -> Optional[dict]:
+    token = os.getenv("NAVITIA_TOKEN")
+    coverage = os.getenv("NAVITIA_COVERAGE")  # e.g., "jp-tokyo"
+    if not token or not coverage:
+        return None
+    url = f"https://api.navitia.io/v1/coverage/{coverage}/journeys"
+    params = {
+        "from": f"coord:{from_lon};{from_lat}",
+        "to": f"coord:{to_lon};{to_lat}",
+    }
+    if when_iso:
+        params["datetime"] = when_iso.replace(":", "").replace("-", "")
+    try:
+        r = requests.get(url, params=params, auth=(token, ""), timeout=25)
+        r.raise_for_status()
+        j = r.json()
+        journeys = j.get("journeys") or []
+        if not journeys:
+            return None
+        best = min(journeys, key=lambda x: x.get("duration", 1e12))
+        fare_total = None
+        if best.get("fare", {}).get("total"):
+            # navitiaの金額はcentime等の可能性があるため、そのまま返さず最小単位（value）を使う
+            total = best["fare"]["total"]
+            if isinstance(total, dict) and "value" in total:
+                fare_total = float(total["value"])
+        return {
+            "duration_s": float(best.get("duration") or 0),
+            "fare": fare_total,
+        }
+    except Exception:
+        return None