110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
"""消息路由：skills 快查 / 网页操作 / 截图预览 / Cursor 通用任务。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
|
||
import env_config
|
||
from browser_parser import is_browser_intent, parse_browser_request
|
||
from browser_service import format_browser_caption, run_browser_automation
|
||
from cursor_runner import run_cursor_task, strip_mention
|
||
from image_extract import find_image_paths, strip_fake_image_markdown
|
||
from preview_service import capture_preview, is_preview_intent, resolve_preview_request
|
||
from skills_service import handle_command, parse_command
|
||
from bot_types import RouteResult
|
||
|
||
|
||
def routing_mode() -> str:
|
||
return (env_config.env("ROUTING_MODE", "hybrid") or "hybrid").lower()
|
||
|
||
|
||
def _normalize(text: str) -> str:
|
||
return re.sub(r"@\S+\s*", "", text).strip().lower()
|
||
|
||
|
||
def is_skills_fast_command(text: str) -> bool:
|
||
raw = _normalize(text)
|
||
if not raw:
|
||
return True
|
||
if raw in {"help", "帮助", "?", "h"}:
|
||
return True
|
||
|
||
cmd = parse_command(text)
|
||
if cmd.kind in {"help", "list", "detail"}:
|
||
return True
|
||
if cmd.kind == "search" and re.match(r"^(search|搜索|find|查)\s+", raw):
|
||
return True
|
||
return False
|
||
|
||
|
||
async def _run_browser(text: str, on_progress=None) -> RouteResult:
|
||
if parse_browser_request(text) is None:
|
||
raise RuntimeError("无法解析网页操作步骤")
|
||
|
||
if on_progress:
|
||
await on_progress("正在按步骤执行网页操作…")
|
||
|
||
result = await run_browser_automation(text)
|
||
return RouteResult(
|
||
source="browser",
|
||
text=format_browser_caption(result),
|
||
image_path=str(result.screenshot_path),
|
||
)
|
||
|
||
|
||
async def _run_preview(text: str, on_progress=None) -> RouteResult:
|
||
preview_req = resolve_preview_request(text)
|
||
if preview_req is None:
|
||
raise RuntimeError("无法解析截图请求")
|
||
|
||
if on_progress:
|
||
await on_progress(f"正在访问并截图：{preview_req.url or '默认地址'}…")
|
||
|
||
result = await capture_preview(preview_req.url, preview_req.port)
|
||
caption = (
|
||
f"**页面预览**\n"
|
||
f"> URL：`{result.final_url or result.url}`\n"
|
||
f"> 项目：`{env_config.env('CURSOR_CWD', '')}`\n"
|
||
f"> dev server：{'已自动启动' if result.started_dev_server else '使用已有服务'}"
|
||
)
|
||
return RouteResult(
|
||
source="preview",
|
||
text=caption,
|
||
image_path=str(result.screenshot_path),
|
||
)
|
||
|
||
|
||
async def route_message(text: str, on_progress=None) -> RouteResult:
|
||
task = strip_mention(text)
|
||
if not task:
|
||
return RouteResult("skills", handle_command("help"))
|
||
|
||
if is_browser_intent(text):
|
||
return await _run_browser(text, on_progress=on_progress)
|
||
|
||
if resolve_preview_request(text) is not None:
|
||
return await _run_preview(text, on_progress=on_progress)
|
||
|
||
mode = routing_mode()
|
||
if mode == "skills":
|
||
return RouteResult("skills", handle_command(text))
|
||
|
||
if mode == "cursor" or not is_skills_fast_command(text):
|
||
reply = await run_cursor_task(task, on_progress=on_progress)
|
||
reply = strip_fake_image_markdown(reply)
|
||
|
||
image_path: str | None = None
|
||
paths = find_image_paths(reply)
|
||
if paths:
|
||
image_path = str(paths[0])
|
||
elif is_preview_intent(text) or is_browser_intent(text):
|
||
if on_progress:
|
||
await on_progress("未找到截图文件，改用 Playwright 自动执行…")
|
||
if is_browser_intent(text):
|
||
return await _run_browser(text, on_progress=on_progress)
|
||
return await _run_preview(text, on_progress=on_progress)
|
||
|
||
return RouteResult("cursor", reply, image_path=image_path)
|
||
|
||
return RouteResult("skills", handle_command(text))
|