Add routecheck service and CommuteTool fast tool

routecheck/ — FastAPI service (port 8090):
  - Image captcha (PIL: arithmetic problem, noise, wave distortion)
  - POST /api/captcha/new + /api/captcha/solve → short-lived token
  - GET /api/route?from=lat,lon&to=lat,lon&token=... → Yandex Routing API
  - Internal bypass via INTERNAL_TOKEN env var (for CommuteTool)
  - HTTPS proxy forwarded to reach Yandex API from container

CommuteTool (fast_tools.py):
  - Matches commute/traffic/arrival time queries
  - Calls routecheck /api/route with ROUTECHECK_TOKEN
  - Hardcoded route: Balashikha home → Moscow center
  - Returns traffic-adjusted travel time + delay annotation

Needs: YANDEX_ROUTING_KEY + ROUTECHECK_TOKEN in .env

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Alvis
2026-03-13 07:08:48 +00:00
parent d2ca1926f8
commit 32089ed596
6 changed files with 466 additions and 2 deletions

View File

@@ -92,6 +92,68 @@ class WeatherTool(FastTool):
return "\n".join(lines) if len(lines) > 1 else ""
class CommuteTool(FastTool):
"""
Returns real-time driving time from home (Balashikha) to a destination
using Yandex traffic data via the local routecheck service.
Triggered by queries about commute time, arrival, or road traffic.
The routecheck service handles Yandex API auth and the HTTPS proxy.
"""
_PATTERN = re.compile(
r"\b(commute|arrival time|how long.{0,20}(drive|get|travel|reach)"
r"|сколько.{0,20}(ехать|добираться|минут)"
r"|пробки|traffic|road.{0,10}now|drive to (work|office|center|москва|moscow)"
r"|when (will i|do i) (arrive|get there|reach))\b",
re.IGNORECASE,
)
# Home: Balashikha. Default destination: Moscow city center.
_HOME = "55.7963,37.9382"
_DEST = "55.7558,37.6173"
def __init__(self, routecheck_url: str, internal_token: str):
self._url = routecheck_url.rstrip("/")
self._token = internal_token
@property
def name(self) -> str:
return "commute"
def matches(self, message: str) -> bool:
return bool(self._PATTERN.search(message))
async def run(self, message: str) -> str:
if not self._token:
return "[commute: ROUTECHECK_TOKEN not configured]"
try:
async with httpx.AsyncClient(timeout=15) as client:
r = await client.get(
f"{self._url}/api/route",
params={"from": self._HOME, "to": self._DEST, "token": self._token},
)
r.raise_for_status()
d = r.json()
except Exception as e:
return f"[commute error: {e}]"
traffic = d["duration_traffic_min"]
normal = d["duration_min"]
dist = d["distance_km"]
delay = traffic - normal
lines = [
f"Current drive time from Balashikha to Moscow center:",
f" With traffic: {traffic} min",
f" Without traffic: {normal} min",
f" Distance: {dist} km",
]
if delay > 5:
lines.append(f" Traffic delay: +{delay} min")
return "\n".join(lines)
class FastToolRunner:
"""
Classifier + executor for fast tools.