88 lines
2.6 KiB
Python
88 lines
2.6 KiB
Python
"""浏览器自动化服务：解析场景 + 启动 dev server + 执行步骤。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
from urllib.parse import urlparse
|
||
|
||
from browser_executor import run_browser_scenario_sync
|
||
from browser_models import BrowserResult, BrowserScenario
|
||
from browser_parser import parse_browser_request
|
||
from preview_service import (
|
||
_package_dev_script,
|
||
_preview_port,
|
||
_project_cwd,
|
||
_startup_timeout,
|
||
_wait_for_port,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _ensure_dev_server(base_url: str) -> bool:
|
||
parsed = urlparse(base_url)
|
||
host = parsed.hostname or "127.0.0.1"
|
||
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
||
|
||
if _wait_for_port(host, port, timeout=3):
|
||
return False
|
||
|
||
cwd = _project_cwd()
|
||
dev_command = _package_dev_script(cwd)
|
||
if not dev_command:
|
||
raise RuntimeError(
|
||
f"无法访问 {base_url}，且未找到可启动的 dev 脚本。"
|
||
"请先手动启动前端，或设置 PREVIEW_URL。"
|
||
)
|
||
|
||
import subprocess
|
||
|
||
logger.info("启动 dev server: %s (cwd=%s)", dev_command, cwd)
|
||
proc = subprocess.Popen(
|
||
dev_command,
|
||
cwd=str(cwd),
|
||
shell=True,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.PIPE,
|
||
)
|
||
if not _wait_for_port(host, port, timeout=_startup_timeout()):
|
||
err = ""
|
||
if proc.stderr:
|
||
err = proc.stderr.read().decode("utf-8", errors="replace")[-1000:]
|
||
proc.kill()
|
||
raise RuntimeError(
|
||
f"dev server 在 {_startup_timeout()}s 内未就绪 ({base_url})。"
|
||
f"{(' 日志: ' + err) if err else ''}"
|
||
)
|
||
return True
|
||
|
||
|
||
async def run_browser_automation(text: str) -> BrowserResult:
|
||
scenario = parse_browser_request(text)
|
||
if scenario is None:
|
||
raise RuntimeError("无法解析网页操作步骤")
|
||
|
||
started = await asyncio.to_thread(_ensure_dev_server, scenario.base_url)
|
||
result = await asyncio.to_thread(run_browser_scenario_sync, scenario)
|
||
result.started_dev_server = started
|
||
return result
|
||
|
||
|
||
def format_browser_caption(result: BrowserResult) -> str:
|
||
lines = [
|
||
"**网页操作完成**",
|
||
f"> 场景：`{result.scenario_name or '自定义'}`",
|
||
f"> 起始：`{result.base_url}`",
|
||
f"> 最终：`{result.final_url}`",
|
||
f"> 步骤数：{result.step_count}",
|
||
]
|
||
if result.started_dev_server:
|
||
lines.append("> dev server：已自动启动")
|
||
if result.step_log:
|
||
lines.append("")
|
||
lines.append("执行记录：")
|
||
for item in result.step_log[-8:]:
|
||
lines.append(f"- {item}")
|
||
return "\n".join(lines)
|