hotfix: replace Google Maps API with local estimation in optimizer

- Disabled Google Maps `Compute Routes` and `Distance Matrix` APIs in the optimization loop to prevent billing overages.
- Swapped the underlying algorithm with a local Haversine-based distance estimator.
- This is an emergency fix to resolve the high API usage incident.

src/optimization/graph/graph_builder.py

@@ -8,7 +8,7 @@ from typing import List, Dict, Tuple, Any
 import numpy as np
 
 from src.infra.logger import get_logger
-from src.services.googlemap_api_service import GoogleMapAPIService
+from src.services.local_route_estimator import LocalRouteEstimator
 
 from ..models.internal_models import _Task, _Location, _Graph
 
@@ -25,8 +25,8 @@ class GraphBuilder:
     - 計算距離/時間矩陣
     """
 
-    def __init__(self, gmaps: GoogleMapAPIService):
-        self.gmaps = gmaps
+    def __init__(self, **kwargs):
+        self.estimator = LocalRouteEstimator()
 
     def build_graph(
             self,
@@ -59,7 +59,7 @@ class GraphBuilder:
             duration_matrix = np.zeros((1, 1), dtype=int)
             distance_matrix = np.zeros((1, 1), dtype=int)
         else:
-            duration_matrix, distance_matrix = self._calculate_matrices(locations, travel_mode=travel_mode)
+            duration_matrix, distance_matrix = self._calculate_matrices(locations)
 
         # 3. 返回圖
         return _Graph(
@@ -122,7 +122,6 @@ class GraphBuilder:
     def _calculate_matrices(
             self,
             locations: List[Dict[str, float]],
-            travel_mode="DRIVE",
     ) -> Tuple[np.ndarray, np.ndarray]:
         """
         計算距離/時間矩陣
@@ -134,12 +133,9 @@ class GraphBuilder:
         """
         locations_dict = [{"lat": loc["lat"], "lng": loc["lng"]} for loc in locations]
 
-        compute_route_result = self.gmaps.compute_route_matrix(
-            locations_dict,
-            locations_dict,
-            travel_mode=travel_mode,
-            routing_preference="TRAFFIC_UNAWARE",
-        )
+        compute_route_result = self.estimator.compute_route_matrix(
+            origins=locations_dict,
+            destinations=locations_dict)
 
         duration_matrix = np.array(compute_route_result["duration_matrix"])
         distance_matrix = np.array(compute_route_result["distance_matrix"])

src/optimization/test/_solver_test.py

@@ -311,8 +311,7 @@ def main():
     from src.infra.config import get_settings
     settings = get_settings()
     # 創建求解器
-    solver = TSPTWSolver(api_key=settings.google_maps_api_key,
-                         time_limit_seconds=5, verbose=True)
+    solver = TSPTWSolver(time_limit_seconds=5, verbose=True)
 
     # 設置參數
     start_location = {"lat": 25.0400, "lng": 121.5300}

src/optimization/test/_test_convertes.py

@@ -1,255 +0,0 @@
-"""
-测试時間視窗转换和求解
-
-验证修复是否正確
-"""
-from datetime import datetime
-
-
-def test_parse_time_window():
-    """测试時間視窗解析"""
-    print("=" * 60)
-    print("测试 _parse_time_window()")
-    print("=" * 60)
-    print()
-
-    from src.optimization.models.converters import _parse_time_window
-
-    # 测试 1: Dict 格式（完整）
-    print("测试 1: Dict 格式（完整時間視窗）")
-    tw_dict = {
-        "earliest_time": "2025-11-19T16:00:00",
-        "latest_time": "2025-11-19T18:00:00"
-    }
-    result = _parse_time_window(tw_dict)
-    print(f"  input: {tw_dict}")
-    print(f"  output: {result}")
-    assert result == (datetime(2025, 11, 19, 16, 0), datetime(2025, 11, 19, 18, 0))
-    print("  ✅ passed\n")
-
-    # 测试 2: Dict 格式（部分時間視窗 - 只有结束）
-    print("测试 2: Dict 格式（只有结束时间）")
-    tw_dict = {
-        "earliest_time": None,
-        "latest_time": "2025-11-19T15:00:00"
-    }
-    result = _parse_time_window(tw_dict)
-    print(f"  input: {tw_dict}")
-    print(f"  output: {result}")
-    assert result == (None, datetime(2025, 11, 19, 15, 0))
-    print("  ✅ passed\n")
-
-    # 测试 3: Dict 格式（部分時間視窗 - 只有开始）
-    print("测试 3: Dict 格式（只有开始时间）")
-    tw_dict = {
-        "earliest_time": "2025-11-19T10:00:00",
-        "latest_time": None
-    }
-    result = _parse_time_window(tw_dict)
-    print(f"  input: {tw_dict}")
-    print(f"  output: {result}")
-    assert result == (datetime(2025, 11, 19, 10, 0), None)
-    print("  ✅ passed\n")
-
-    # 测试 4: Dict 格式（全 None）
-    print("测试 4: Dict 格式（全 None）")
-    tw_dict = {
-        "earliest_time": None,
-        "latest_time": None
-    }
-    result = _parse_time_window(tw_dict)
-    print(f"  input: {tw_dict}")
-    print(f"  output: {result}")
-    assert result is None
-    print("  ✅ passed\n")
-
-    # 测试 5: None
-    print("测试 5: None input")
-    result = _parse_time_window(None)
-    print(f"  input: None")
-    print(f"  output: {result}")
-    assert result is None
-    print("  ✅ passed\n")
-
-    # 测试 6: Tuple（兼容）
-    print("测试 6: Tuple input（兼容旧代码）")
-    tw_tuple = (datetime(2025, 11, 19, 16, 0), datetime(2025, 11, 19, 18, 0))
-    result = _parse_time_window(tw_tuple)
-    print(f"  input: {tw_tuple}")
-    print(f"  output: {result}")
-    assert result == tw_tuple
-    print("  ✅ passed\n")
-
-
-def test_convert_tasks():
-    """测试完整的任务转换"""
-    print("=" * 60)
-    print("测试 convert_tasks_to_internal()")
-    print("=" * 60)
-    print()
-
-    from converters import convert_tasks_to_internal
-
-    # 测试数据：模拟你的实际input
-    tasks = [
-        {
-            "task_id": "task_1",
-            "priority": "HIGH",
-            "service_duration_min": 60,
-            "time_window": {
-                "earliest_time": "2025-11-19T16:00:00",
-                "latest_time": "2025-11-19T18:00:00"
-            },
-            "candidates": [
-                {
-                    "poi_id": "hospital_1",
-                    "lat": 25.040978499999998,
-                    "lng": 121.51919869999999
-                }
-            ]
-        },
-        {
-            "task_id": "task_2",
-            "priority": "MEDIUM",
-            "service_duration_min": 45,
-            "time_window": None,
-            "candidates": [
-                {
-                    "poi_id": "supermarket_1",
-                    "lat": 25.0376995,
-                    "lng": 121.50622589999999
-                }
-            ]
-        },
-        {
-            "task_id": "task_3",
-            "priority": "HIGH",
-            "service_duration_min": 20,
-            "time_window": {
-                "earliest_time": None,
-                "latest_time": "2025-11-19T15:00:00"
-            },
-            "candidates": [
-                {
-                    "poi_id": "post_office_1",
-                    "lat": 25.0514598,
-                    "lng": 121.5481353
-                }
-            ]
-        }
-    ]
-
-    # 转换
-    internal_tasks = convert_tasks_to_internal(tasks)
-
-    # 验证
-    print("测试 1: task_1 (看病 16:00-18:00)")
-    print(f"  task_id: {internal_tasks[0].task_id}")
-    print(f"  priority: {internal_tasks[0].priority}")
-    print(f"  time_window: {internal_tasks[0].time_window}")
-    assert internal_tasks[0].time_window == (
-        datetime(2025, 11, 19, 16, 0),
-        datetime(2025, 11, 19, 18, 0)
-    )
-    print("  ✅ passed\n")
-
-    print("测试 2: task_2 (买菜 无限制)")
-    print(f"  task_id: {internal_tasks[1].task_id}")
-    print(f"  priority: {internal_tasks[1].priority}")
-    print(f"  time_window: {internal_tasks[1].time_window}")
-    assert internal_tasks[1].time_window is None
-    print("  ✅ passed\n")
-
-    print("测试 3: task_3 (寄包裹 None-15:00)")
-    print(f"  task_id: {internal_tasks[2].task_id}")
-    print(f"  priority: {internal_tasks[2].priority}")
-    print(f"  time_window: {internal_tasks[2].time_window}")
-    assert internal_tasks[2].time_window == (
-        None,
-        datetime(2025, 11, 19, 15, 0)
-    )
-    print("  ✅ passed\n")
-
-
-def test_time_window_seconds():
-    """测试時間視窗转换为秒"""
-    print("=" * 60)
-    print("测试時間視窗秒数计算")
-    print("=" * 60)
-    print()
-
-    from converters import _parse_time_window
-    from datetime import timedelta
-
-    # 开始时间: 08:00
-    start_time = datetime(2025, 11, 19, 8, 0)
-
-    # task_1: 16:00-18:00
-    tw1 = _parse_time_window({
-        "earliest_time": "2025-11-19T16:00:00",
-        "latest_time": "2025-11-19T18:00:00"
-    })
-
-    start_sec = int((tw1[0] - start_time).total_seconds())
-    end_sec = int((tw1[1] - start_time).total_seconds())
-
-    print("task_1 時間視窗:")
-    print(f"  原始: 16:00-18:00")
-    print(f"  相对 08:00: {start_sec}秒 - {end_sec}秒")
-    print(f"  即: {start_sec / 3600:.1f}小时 - {end_sec / 3600:.1f}小时")
-
-    assert start_sec == 8 * 3600  # 8小时 = 28800秒
-    assert end_sec == 10 * 3600  # 10小时 = 36000秒
-    print("  ✅ passed\n")
-
-    # task_3: None - 15:00
-    tw3 = _parse_time_window({
-        "earliest_time": None,
-        "latest_time": "2025-11-19T15:00:00"
-    })
-
-    print("task_3 時間視窗:")
-    print(f"  原始: None - 15:00")
-    if tw3[0] is None:
-        print(f"  start: 0秒 (无限制)")
-    if tw3[1] is not None:
-        end_sec = int((tw3[1] - start_time).total_seconds())
-        print(f"  end: {end_sec}秒 = {end_sec / 3600:.1f}小时")
-        assert end_sec == 7 * 3600  # 7小时 = 25200秒
-    print("  ✅ passed\n")
-
-
-def main():
-    """运行所有测试"""
-    print("\n")
-    print("╔" + "=" * 58 + "╗")
-    print("║" + " " * 15 + "時間視窗转换测试" + " " * 15 + "║")
-    print("╚" + "=" * 58 + "╝")
-    print("\n")
-
-    try:
-        test_parse_time_window()
-        test_convert_tasks()
-        test_time_window_seconds()
-
-        print("\n")
-        print("╔" + "=" * 58 + "╗")
-        print("║" + " " * 18 + "✅ 所有测试passed" + " " * 18 + "║")
-        print("╚" + "=" * 58 + "╝")
-        print("\n")
-
-        print("总结:")
-        print("✅ Dict 格式解析正確")
-        print("✅ 部分時間視窗支持")
-        print("✅ 完整任务转换正確")
-        print("✅ 时间秒数计算正確")
-        print("\n")
-
-    except Exception as e:
-        print(f"\n❌ 測試失敗: {e}")
-        import traceback
-        traceback.print_exc()
-
-
-if __name__ == "__main__":
-    main()

src/optimization/tsptw_solver.py

@@ -36,7 +36,6 @@ class TSPTWSolver:
 
     def __init__(
             self,
-            api_key: Optional[str] = None,
             time_limit_seconds: Optional[int] = None,
             verbose: bool = False,
     ):
@@ -54,14 +53,13 @@ class TSPTWSolver:
                 or "1"
         )
 
-        self.gmaps = GoogleMapAPIService(api_key=api_key)
         self.time_limit_seconds = (
             time_limit_seconds if time_limit_seconds is not None else int(env_limit)
         )
         self.verbose = verbose
 
         # 初始化各模塊
-        self.graph_builder = GraphBuilder(gmaps=self.gmaps)
+        self.graph_builder = GraphBuilder()
         self.ortools_solver = ORToolsSolver(
             time_limit_seconds=self.time_limit_seconds,
             verbose=verbose,

src/services/googlemap_api_service.py

@@ -174,7 +174,6 @@ class GoogleMapAPIService:
                 "places.userRatingCount,"
                 "places.businessStatus,"
                 "places.currentOpeningHours,"
-                "places.userRatingCount,"
             )
         }
 

src/tools/optimizer_toolkit.py

@@ -5,12 +5,15 @@ from datetime import datetime, timedelta
 from agno.tools import Toolkit
 from src.optimization.tsptw_solver import TSPTWSolver
 from src.infra.poi_repository import poi_repo
+from src.infra.logger import get_logger
+
+logger = get_logger(__name__)
 
 
 class OptimizationToolkit(Toolkit):
-    def __init__(self, google_maps_api_key: None):
+    def __init__(self):
         super().__init__(name="optimization_toolkit")
-        self.solver = TSPTWSolver(api_key=google_maps_api_key)
+        self.solver = TSPTWSolver()
         self.register(self.optimize_from_ref)
 
     def optimize_from_ref(self, ref_id: str) -> str:
@@ -74,7 +77,8 @@ class OptimizationToolkit(Toolkit):
                 return_to_start=False
             )
         except Exception as e:
-            return f"❌ Solver Failed: {e}"
+            logger.warning(f"Solver failed: {e}")
+            return f"❌ Solver Failed: {e}, Please check the input data."
 
         # ✅ [Critical] 將 global_info 繼承下去！
         # 如果不加這一行，Navigator 就會因為找不到 departure_time 而報錯