"""在 CURSOR_CWD 启动/访问前端并截图（单页，不含多步操作）。""" from __future__ import annotations import asyncio import json import logging import re import socket import subprocess import time from dataclasses import dataclass from datetime import datetime from pathlib import Path from urllib.parse import urlparse import env_config logger = logging.getLogger(__name__) SCREENSHOT_DIR = Path(__file__).resolve().parent / ".cache" / "screenshots" @dataclass class PreviewResult: url: str screenshot_path: Path started_dev_server: bool final_url: str | None = None @dataclass class PreviewRequest: url: str | None port: int | None def _project_cwd() -> Path: raw = env_config.env("CURSOR_CWD", r"d:\LY\test\tech") or r"d:\LY\test\tech" return Path(raw).resolve() def _preview_port() -> int: raw = env_config.env("PREVIEW_PORT", "5173") or "5173" return int(raw) def _startup_timeout() -> int: raw = env_config.env("PREVIEW_STARTUP_TIMEOUT", "120") or "120" return int(raw) def _dev_command() -> str: return env_config.env("PREVIEW_DEV_COMMAND", "npm run dev") or "npm run dev" def parse_preview_command(text: str) -> tuple[str | None, int | None] | None: raw = re.sub(r"@\S+\s*", "", text).strip() if not raw: return None m = re.match( r"^(preview|截图|预览|截屏)(?:\s+(https?://\S+|/\S*))?(?:\s+(\d{2,5}))?$", raw, re.IGNORECASE, ) if not m: return None url_part = m.group(2) port_part = m.group(3) port = int(port_part) if port_part else None if url_part and url_part.startswith("/"): port = port or _preview_port() return f"http://127.0.0.1:{port}{url_part}", port return url_part, port def resolve_preview_request(text: str) -> PreviewRequest | None: explicit = parse_preview_command(text) if explicit is not None: url_override, port_override = explicit return PreviewRequest(url=url_override, port=port_override) if not is_preview_intent(text): return None url_override = extract_url_from_text(text) if not url_override: env_url = env_config.env("PREVIEW_URL") url_override = env_url.strip() if env_url else f"http://127.0.0.1:{_preview_port()}/" return PreviewRequest(url=url_override, port=None) _PREVIEW_INTENT = re.compile( r"^(preview|截图|预览|截屏)\b|" r"(页面预览|运行.*(前端|项目|页面)|" r"打开.*(前端|页面|项目)|" r"访问.*(并)?.*(截图|截屏)|" r"启动.*(前端|项目|dev|服务).*(截图|截屏)?)", re.IGNORECASE, ) def is_preview_intent(text: str) -> bool: raw = re.sub(r"@\S+\s*", "", text).strip() if parse_preview_command(text) is not None: return True return bool(_PREVIEW_INTENT.search(raw)) def extract_url_from_text(text: str) -> str | None: raw = re.sub(r"@\S+\s*", "", text) match = re.search( r"(https?://[^\s\]`\"']+|localhost:\d+[/\w\-./]*)", raw, re.IGNORECASE, ) if not match: return None url = match.group(1).rstrip(".,，。") if url.lower().startswith("localhost"): url = "http://" + url return url def _capture_screenshot_sync(url: str, output: Path) -> str: from playwright.sync_api import sync_playwright output.parent.mkdir(parents=True, exist_ok=True) with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page(viewport={"width": 1280, "height": 720}) page.goto(url, wait_until="networkidle", timeout=60_000) page.wait_for_timeout(1500) page.screenshot(path=str(output), full_page=False, type="png") final_url = page.url browser.close() return final_url def _wait_for_port(host: str, port: int, timeout: int) -> bool: deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: with socket.create_connection((host, port), timeout=2): return True except OSError: time.sleep(1) return False def _resolve_target_url(url_override: str | None, port_override: int | None) -> tuple[str, str | None]: if url_override: parsed = urlparse(url_override) if parsed.scheme and parsed.netloc: return url_override, None raise RuntimeError(f"无效 URL：{url_override}") env_url = env_config.env("PREVIEW_URL") if env_url: return env_url.strip(), None port = port_override or _preview_port() cwd = _project_cwd() dev_script = _package_dev_script(cwd) base = f"http://127.0.0.1:{port}/" return base, dev_script def _package_dev_script(cwd: Path) -> str | None: pkg = cwd / "package.json" if not pkg.exists(): return None try: data = json.loads(pkg.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None scripts = data.get("scripts") or {} for key in ("dev", "preview", "start"): if scripts.get(key): cmd = _dev_command() if key != "dev" and cmd == "npm run dev": return f"npm run {key}" return cmd return None def _capture_preview_sync(url: str, dev_command: str | None) -> PreviewResult: cwd = _project_cwd() parsed = urlparse(url) host = parsed.hostname or "127.0.0.1" port = parsed.port or (443 if parsed.scheme == "https" else 80) dev_proc: subprocess.Popen | None = None started = False if dev_command: if _wait_for_port(host, port, timeout=3): logger.info("检测到端口 %s 已监听，跳过启动 dev server", port) else: logger.info("启动 dev server: %s (cwd=%s)", dev_command, cwd) dev_proc = subprocess.Popen( dev_command, cwd=str(cwd), shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, ) started = True if not _wait_for_port(host, port, timeout=_startup_timeout()): err = "" if dev_proc.stderr: err = dev_proc.stderr.read().decode("utf-8", errors="replace")[-1000:] raise RuntimeError( f"dev server 在 {_startup_timeout()}s 内未就绪 ({url})。" f"{(' 日志: ' + err) if err else ''}" ) else: if not _wait_for_port(host, port, timeout=5): raise RuntimeError( f"无法访问 {url}。请在 CURSOR_CWD 放置前端项目，" "或先手动启动 dev server，或设置 PREVIEW_URL。" ) stamp = datetime.now().strftime("%Y%m%d-%H%M%S") output = SCREENSHOT_DIR / f"preview-{stamp}.png" try: final_url = _capture_screenshot_sync(url, output) finally: if dev_proc and dev_proc.poll() is None: dev_proc.terminate() try: dev_proc.wait(timeout=5) except subprocess.TimeoutExpired: dev_proc.kill() return PreviewResult( url=url, screenshot_path=output, started_dev_server=started, final_url=final_url, ) async def capture_preview( url_override: str | None = None, port_override: int | None = None, ) -> PreviewResult: url, dev_command = _resolve_target_url(url_override, port_override) return await asyncio.to_thread(_capture_preview_sync, url, dev_command)