chore: 移除独立企微 bot/浏览器桥接系统

两套系统并存，bot/（browser_service、preview_service、wecom_media、
router、xiaobao 场景等）与早报生成是独立的一条线，不再维护，整体删除。
同时将 __pycache__/*.pyc/.cursor 加入 .gitignore。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.gitignore

@@ -2,7 +2,7 @@
 .env.local
 logs/
 .cache/
-bot/.env
-bot/.venv/
-bot/.cache/
 output
+__pycache__/
+*.pyc
+.cursor/

bot/.env.example

@@ -1,29 +0,0 @@
-# 企业微信智能机器人（API 模式 · 长连接）
-# 管理后台 → 安全与管理 → 管理工具 → 智能机器人 → 创建 → API 模式 → 使用长连接
-WECOM_BOT_ID=your-bot-id
-WECOM_BOT_SECRET=your-bot-secret
-
-# Cursor SDK（@ 机器人后的通用任务由 Cursor 执行）
-CURSOR_API_KEY=cursor_...
-CURSOR_CWD=d:\LY\test\tech
-CURSOR_MODEL=composer-2.5
-CURSOR_TIMEOUT=600
-
-# 前端截图预览（基于 CURSOR_CWD）
-PREVIEW_PORT=5173
-PREVIEW_URL=http://127.0.0.1:5173/
-# PREVIEW_DEV_COMMAND=npm run dev
-# PREVIEW_STARTUP_TIMEOUT=120
-
-# 登录后截图（账号密码只放 .env，切勿发到企微群）
-# PREVIEW_LOGIN_USER=your_account_or_phone
-# PREVIEW_LOGIN_PASSWORD=your_password
-# PREVIEW_AFTER_LOGIN_URL=/app/dashboard
-# PREVIEW_AUTO_LOGIN=true
-
-# 网页操作场景目录（可选，默认 bot/scenarios 与 CURSOR_CWD/.browser-scenarios）
-# BROWSER_SCENARIOS_DIR=d:\path\to\scenarios
-# BROWSER_DEFAULT_SCENARIO=xiaobao-agent-manage
-
-# hybrid=快查走本地 / 其余走 Cursor | cursor=全部 Cursor | skills=仅本地
-ROUTING_MODE=hybrid

bot/.gitignore

@@ -1,4 +0,0 @@
-.cache/
-.env
-.venv/
-.cache/screenshots/

bot/bot_types.py

@@ -1,12 +0,0 @@
-"""Bot 内部数据结构。"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-
-
-@dataclass
-class RouteResult:
-    source: str
-    text: str
-    image_path: str | None = None

bot/bridge_manager.py

@@ -1,156 +0,0 @@
-"""Windows 兼容的 Cursor SDK bridge 管理。"""
-
-from __future__ import annotations
-
-import codecs
-import json
-import logging
-import os
-import subprocess
-import threading
-import time
-from pathlib import Path
-from typing import Any, Mapping
-
-import env_config
-
-logger = logging.getLogger(__name__)
-
-READY_LINE_PREFIX = "cursor-sdk-bridge ready "
-_bridge_lock = threading.Lock()
-_bridge_process: subprocess.Popen[bytes] | None = None
-
-
-def _cursor_cwd() -> str:
-    return env_config.env("CURSOR_CWD", r"d:\LY\test\tech") or r"d:\LY\test\tech"
-
-
-def _parse_discovery_line(line: str) -> Mapping[str, Any] | None:
-    if not line.startswith(READY_LINE_PREFIX):
-        return None
-    payload = line[len(READY_LINE_PREFIX) :].strip()
-    loaded = json.loads(payload)
-    if not isinstance(loaded, dict):
-        raise RuntimeError("Bridge discovery payload must be an object")
-    return loaded
-
-
-def _read_discovery_polling(process: subprocess.Popen[bytes], timeout: float = 60) -> Mapping[str, Any]:
-    """不用 selectors，避免 Windows 上 WinError 10038。"""
-    if process.stderr is None:
-        raise RuntimeError("Bridge stderr unavailable")
-
-    fd = process.stderr.fileno()
-    was_blocking = os.get_blocking(fd)
-    os.set_blocking(fd, False)
-    decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
-    pending = ""
-    stderr_lines: list[str] = []
-    deadline = time.monotonic() + timeout
-
-    try:
-        while time.monotonic() < deadline:
-            try:
-                chunk = os.read(fd, 8192)
-            except BlockingIOError:
-                chunk = b""
-
-            if chunk:
-                pending += decoder.decode(chunk)
-                while "\n" in pending:
-                    line, pending = pending.split("\n", 1)
-                    stderr_lines.append(line)
-                    discovery = _parse_discovery_line(line)
-                    if discovery is not None:
-                        return discovery
-            else:
-                code = process.poll()
-                if code is not None:
-                    pending += decoder.decode(b"", final=True)
-                    if pending.strip():
-                        stderr_lines.append(pending.strip())
-                    joined = "\n".join(stderr_lines)[-2000:]
-                    raise RuntimeError(
-                        f"Bridge 启动失败 exit={code}: {joined or '无 stderr 输出'}"
-                    )
-                time.sleep(0.05)
-    finally:
-        os.set_blocking(fd, was_blocking)
-
-    raise RuntimeError("等待 Cursor bridge 就绪超时")
-
-
-def _auth_token_from_discovery(discovery: Mapping[str, Any]) -> str:
-    token = str(discovery.get("authToken") or "").strip()
-    if token:
-        return token
-    token_file = discovery.get("authTokenFile")
-    if token_file:
-        return Path(str(token_file)).read_text(encoding="utf-8").strip()
-    raise RuntimeError("Bridge discovery 缺少 auth token")
-
-
-def warm_cursor_bridge(force: bool = False) -> None:
-    """启动 cursor-sdk-bridge 并写入 CURSOR_SDK_BRIDGE_* 环境变量。"""
-    global _bridge_process
-
-    with _bridge_lock:
-        if (
-            not force
-            and _bridge_process is not None
-            and _bridge_process.poll() is None
-            and os.environ.get("CURSOR_SDK_BRIDGE_URL")
-            and os.environ.get("CURSOR_SDK_BRIDGE_TOKEN")
-        ):
-            return
-
-        if _bridge_process is not None and _bridge_process.poll() is None:
-            _bridge_process.terminate()
-            try:
-                _bridge_process.wait(timeout=5)
-            except subprocess.TimeoutExpired:
-                _bridge_process.kill()
-
-        from cursor_sdk._vendor import resolve_bridge_path
-
-        cwd = _cursor_cwd()
-        argv = [resolve_bridge_path(), "--workspace", cwd]
-        logger.info("启动 Cursor bridge workspace=%s", cwd)
-
-        process = subprocess.Popen(
-            argv,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.PIPE,
-        )
-        try:
-            discovery = _read_discovery_polling(process)
-        except Exception:
-            process.kill()
-            process.wait(timeout=5)
-            raise
-
-        url = str(discovery.get("url") or "").strip()
-        if not url:
-            host = str(discovery.get("host") or "127.0.0.1")
-            port = discovery.get("port")
-            url = f"http://{host}:{port}"
-
-        token = _auth_token_from_discovery(discovery)
-        os.environ["CURSOR_SDK_BRIDGE_URL"] = url
-        os.environ["CURSOR_SDK_BRIDGE_TOKEN"] = token
-        _bridge_process = process
-        logger.info("Cursor bridge 就绪: %s", url)
-
-
-def shutdown_cursor_bridge() -> None:
-    global _bridge_process
-    with _bridge_lock:
-        if _bridge_process is None:
-            return
-        if _bridge_process.poll() is None:
-            _bridge_process.terminate()
-            try:
-                _bridge_process.wait(timeout=5)
-            except subprocess.TimeoutExpired:
-                _bridge_process.kill()
-        _bridge_process = None

bot/browser_env.py

@@ -1,49 +0,0 @@
-"""浏览器场景变量替换与 base URL 解析。"""
-
-from __future__ import annotations
-
-import re
-from urllib.parse import urlparse
-
-import env_config
-
-_VAR_PATTERN = re.compile(r"\{\{([A-Z0-9_]+)\}\}")
-
-
-def interpolate(value: str) -> str:
-    def repl(match: re.Match[str]) -> str:
-        key = match.group(1)
-        resolved = env_config.env(key)
-        if resolved is None:
-            raise RuntimeError(f"场景变量未配置：{key}")
-        return resolved
-
-    return _VAR_PATTERN.sub(repl, value)
-
-
-def default_base_url() -> str:
-    explicit = (env_config.env("PREVIEW_BASE_URL") or "").strip()
-    if explicit:
-        return interpolate(explicit.rstrip("/"))
-
-    preview = (env_config.env("PREVIEW_URL") or "").strip()
-    if preview:
-        parsed = urlparse(preview)
-        scheme = parsed.scheme or "http"
-        host = parsed.hostname or "127.0.0.1"
-        port = parsed.port
-        if port and port not in (80, 443):
-            return f"{scheme}://{host}:{port}"
-        return f"{scheme}://{host}"
-
-    port = env_config.env("PREVIEW_PORT", "5173") or "5173"
-    return f"http://127.0.0.1:{port}"
-
-
-def resolve_url(base_url: str, target: str) -> str:
-    target = interpolate(target.strip())
-    if target.startswith("http://") or target.startswith("https://"):
-        return target
-    if not target.startswith("/"):
-        target = "/" + target
-    return base_url.rstrip("/") + target

bot/browser_executor.py

@@ -1,269 +0,0 @@
-"""通用 Playwright 步骤执行器（不写死业务页面）。"""
-
-from __future__ import annotations
-
-import logging
-import re
-import time
-from datetime import datetime
-from pathlib import Path
-from typing import Any
-
-from browser_env import interpolate, resolve_url
-from browser_models import BrowserResult, BrowserScenario
-
-logger = logging.getLogger(__name__)
-
-SCREENSHOT_DIR = Path(__file__).resolve().parent / ".cache" / "screenshots"
-
-FIELD_HINTS: dict[str, list[str]] = {
-    "账号": [
-        "#login-username",
-        "input#login-username",
-        "input[autocomplete='username']",
-        "username",
-        "account",
-        "phone",
-        "账号",
-        "手机号",
-        "企业账号",
-    ],
-    "密码": [
-        "#login-password input",
-        "#login-password",
-        "input#login-password",
-        "input[type='password']",
-        "password",
-        "密码",
-    ],
-    "用户名": ["#login-username", "input#login-username", "username", "account", "账号"],
-}
-
-def _step_label(step: dict[str, Any], index: int) -> str:
-    action = step.get("action", "?")
-    target = step.get("target") or step.get("field") or step.get("url") or ""
-    return f"{index + 1}. {action} {target}".strip()
-
-
-def _resolve_fill_locator(page, field: str, step: dict[str, Any]):
-    if step.get("selector"):
-        return page.locator(interpolate(str(step["selector"])))
-
-    field_key = interpolate(str(field))
-    if step.get("label"):
-        return page.get_by_label(interpolate(str(step["label"])), exact=False)
-    if step.get("placeholder"):
-        return page.get_by_placeholder(interpolate(str(step["placeholder"])), exact=False)
-
-    hints = FIELD_HINTS.get(field_key, [field_key])
-    for hint in hints:
-        if hint.startswith("#") or hint.startswith(".") or hint.startswith("["):
-            locator = page.locator(hint)
-            if locator.count() > 0:
-                return locator.first
-        for getter in (
-            lambda h=hint: page.get_by_label(h, exact=False),
-            lambda h=hint: page.get_by_placeholder(h, exact=False),
-        ):
-            locator = getter()
-            if locator.count() > 0:
-                return locator.first
-
-    return page.locator("input, textarea").filter(has_text=field_key).first
-
-
-def _fill_field(page, field: str, step: dict[str, Any]) -> None:
-    value = interpolate(str(step.get("value", "")))
-    locator = _resolve_fill_locator(page, field, step)
-    locator.click(timeout=10_000)
-    locator.fill("", timeout=5_000)
-    locator.fill(value, timeout=10_000)
-
-
-def _page_error_text(page) -> str | None:
-    for selector in (
-        ".ant-message-error",
-        ".ant-form-item-explain-error",
-        ".ant-alert-error",
-    ):
-        try:
-            locator = page.locator(selector).first
-            if locator.is_visible(timeout=300):
-                text = locator.inner_text(timeout=1_000).strip()
-                if text:
-                    return text
-        except Exception:
-            continue
-    return None
-
-
-def _pathname_matches(pattern: str, pathname: str) -> bool:
-    pattern = pattern.strip()
-    if pattern in {"**/app/**", "**/app/*", "/app/**"}:
-        return pathname.startswith("/app")
-    if pattern.endswith("/**"):
-        prefix = pattern[:-3].rstrip("/")
-        if prefix.startswith("**/"):
-            prefix = prefix[3:]
-        if not prefix.startswith("/"):
-            prefix = "/" + prefix
-        return pathname.startswith(prefix)
-    if "**" in pattern or "*" in pattern:
-        regex = "^" + re.escape(pattern).replace(r"\*\*", ".*").replace(r"\*", "[^/]*") + "$"
-        return re.search(regex, pathname) is not None
-    return pathname == pattern or pathname.startswith(pattern)
-
-
-def _wait_for_url_pattern(page, pattern: str, timeout: int = 60_000) -> None:
-    """SPA 路由用 pathname 轮询；glob 模式不依赖 navigation 事件。"""
-    deadline = time.monotonic() + timeout / 1000
-    last_error: str | None = None
-
-    while time.monotonic() < deadline:
-        pathname = page.evaluate("() => window.location.pathname")
-        if _pathname_matches(pattern, pathname):
-            try:
-                page.wait_for_load_state("networkidle", timeout=8_000)
-            except Exception:
-                page.wait_for_timeout(800)
-            return
-
-        err = _page_error_text(page)
-        if err and err != last_error:
-            last_error = err
-            logger.warning("页面提示：%s", err)
-            if "/login" in pathname:
-                raise RuntimeError(f"登录失败：{err}")
-
-        page.wait_for_timeout(400)
-
-    err = _page_error_text(page)
-    hint_parts = [f"当前 URL：`{page.url}`"]
-    if err:
-        hint_parts.append(f"页面错误：{err}")
-    elif last_error:
-        hint_parts.append(f"页面错误：{last_error}")
-    hint_parts.append("请确认 PREVIEW_LOGIN_USER/PASSWORD 正确，且登录 API（内网网关）可达。")
-    raise RuntimeError(f"等待 URL 匹配 `{pattern}` 超时（{timeout}ms）。{' '.join(hint_parts)}")
-
-
-def _click_target(page, target: str) -> None:
-    target = interpolate(target.strip())
-    if target.lower() in {"登录", "login"}:
-        for selector in ("button.login-submit", "button[type='submit']"):
-            locator = page.locator(selector)
-            if locator.count() > 0:
-                locator.first.click(timeout=10_000)
-                return
-
-    candidates = [
-        page.get_by_role("menuitem", name=target, exact=True),
-        page.get_by_role("button", name=target, exact=True),
-        page.get_by_role("link", name=target, exact=True),
-        page.get_by_text(target, exact=True),
-        page.get_by_text(target, exact=False),
-    ]
-    for locator in candidates:
-        if locator.count() > 0:
-            locator.first.click(timeout=10_000)
-            return
-    raise RuntimeError(f"未找到可点击元素：{target}")
-
-
-def _execute_step(page, base_url: str, step: dict[str, Any]) -> None:
-    action = str(step.get("action", "")).lower()
-    if action == "goto":
-        target = step.get("target") or step.get("url") or "/"
-        url = resolve_url(base_url, str(target))
-        page.goto(url, wait_until="networkidle", timeout=60_000)
-        return
-
-    if action == "fill":
-        field = str(step.get("field") or step.get("target") or "账号")
-        _fill_field(page, field, step)
-        return
-
-    if action == "click":
-        target = step.get("target") or step.get("text")
-        if not target:
-            raise RuntimeError("click 步骤缺少 target")
-        _click_target(page, str(target))
-        page.wait_for_timeout(800)
-        return
-
-    if action == "wait":
-        timeout = int(step.get("timeout") or 60_000)
-        if step.get("url"):
-            _wait_for_url_pattern(page, str(step["url"]), timeout=timeout)
-            return
-        if step.get("selector"):
-            page.locator(interpolate(str(step["selector"]))).wait_for(timeout=30_000)
-            return
-        if step.get("text"):
-            page.get_by_text(interpolate(str(step["text"])), exact=False).wait_for(timeout=30_000)
-            return
-        ms = int(step.get("ms") or 1500)
-        page.wait_for_timeout(ms)
-        return
-
-    if action == "press":
-        key = str(step.get("key") or step.get("target") or "Enter")
-        page.keyboard.press(key)
-        return
-
-    if action == "screenshot":
-        return
-
-    raise RuntimeError(f"未知步骤 action={action}")
-
-
-def run_browser_scenario_sync(scenario: BrowserScenario) -> BrowserResult:
-    from playwright.sync_api import sync_playwright
-
-    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
-    slug = (scenario.name or "browser").replace(" ", "-")
-    output = SCREENSHOT_DIR / f"{slug}-{stamp}.png"
-    output.parent.mkdir(parents=True, exist_ok=True)
-
-    step_log: list[str] = []
-    final_url = scenario.base_url
-
-    with sync_playwright() as playwright:
-        browser = playwright.chromium.launch(headless=True)
-        page = browser.new_page(viewport={"width": 1280, "height": 720})
-
-        steps = list(scenario.steps)
-        if steps and steps[-1].get("action") != "screenshot" and not any(
-            s.get("action") == "screenshot" for s in steps
-        ):
-            steps.append({"action": "screenshot"})
-
-        for index, step in enumerate(steps):
-            label = _step_label(step, index)
-            logger.info("执行步骤 %s", label)
-            action = str(step.get("action", "")).lower()
-            if action == "screenshot":
-                page.wait_for_timeout(int(step.get("ms") or 1500))
-                page.screenshot(path=str(output), full_page=False, type="png")
-                final_url = page.url
-                step_log.append(label + " ✓")
-                continue
-            try:
-                _execute_step(page, scenario.base_url, step)
-                final_url = page.url
-                step_log.append(label + " ✓")
-            except Exception as exc:
-                err = _page_error_text(page)
-                detail = f"（{err}）" if err else ""
-                raise RuntimeError(f"步骤失败：{label} @ {page.url}{detail}") from exc
-
-        browser.close()
-
-    return BrowserResult(
-        scenario_name=scenario.name,
-        base_url=scenario.base_url,
-        final_url=final_url,
-        screenshot_path=output,
-        step_count=len(steps),
-        step_log=step_log,
-    )

bot/browser_models.py

@@ -1,26 +0,0 @@
-"""浏览器自动化步骤模型。"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from pathlib import Path
-from typing import Any
-
-
-@dataclass
-class BrowserScenario:
-    name: str | None
-    base_url: str
-    steps: list[dict[str, Any]]
-    source: str = "natural"
-
-
-@dataclass
-class BrowserResult:
-    scenario_name: str | None
-    base_url: str
-    final_url: str
-    screenshot_path: Path
-    step_count: int
-    started_dev_server: bool = False
-    step_log: list[str] = field(default_factory=list)

bot/browser_parser.py

@@ -1,307 +0,0 @@
-"""解析自然语言 / YAML / 场景名 → 浏览器步骤。"""
-
-from __future__ import annotations
-
-import re
-from pathlib import Path
-from typing import Any
-
-import yaml
-
-import env_config
-from browser_env import default_base_url, interpolate
-from browser_models import BrowserScenario
-
-SCENARIO_DIRS = [
-    Path(__file__).resolve().parent / "scenarios",
-    Path(__file__).resolve().parent.parent / "scenarios",
-]
-
-
-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 _strip_mention(text: str) -> str:
-    return re.sub(r"@\S+\s*", "", text).strip()
-
-
-def _scenario_search_dirs() -> list[Path]:
-    dirs = list(SCENARIO_DIRS)
-    dirs.append(_project_cwd() / ".browser-scenarios")
-    custom = (env_config.env("BROWSER_SCENARIOS_DIR") or "").strip()
-    if custom:
-        dirs.append(Path(custom).resolve())
-    return dirs
-
-
-def is_browser_intent(text: str) -> bool:
-    raw = _strip_mention(text)
-    if not raw:
-        return False
-    if re.match(r"^(browser|网页|网页操作|操作)\b", raw, re.IGNORECASE):
-        return True
-    if re.search(r"```(?:yaml|yml)", raw, re.IGNORECASE):
-        return True
-    if re.search(r"(?m)^browser\s*:", raw, re.IGNORECASE):
-        return True
-
-    if re.match(r"^(preview|截图|预览|截屏)\s", raw, re.IGNORECASE):
-        if not re.search(r"[，,。；;]|然后|输入|点击|填写|访问|打开|登录", raw):
-            return False
-
-    if len(_split_segments(text)) >= 2:
-        return True
-
-    verbs = 0
-    for pattern in (r"访问", r"打开", r"输入", r"填写", r"点击", r"点选", r"选择", r"登录"):
-        if re.search(pattern, raw):
-            verbs += 1
-    return verbs >= 2
-
-
-def _load_yaml_scenario(path: Path) -> BrowserScenario:
-    data = yaml.safe_load(path.read_text(encoding="utf-8"))
-    if not isinstance(data, dict):
-        raise RuntimeError(f"场景文件格式错误：{path}")
-    base_url = interpolate(str(data.get("base_url") or default_base_url()))
-    steps = data.get("steps")
-    if not isinstance(steps, list) or not steps:
-        raise RuntimeError(f"场景缺少 steps：{path}")
-    return BrowserScenario(
-        name=data.get("name") or path.stem,
-        base_url=base_url,
-        steps=_normalize_steps(steps),
-        source=f"file:{path.name}",
-    )
-
-
-def _find_scenario_file(name: str) -> Path | None:
-    slug = name.strip().replace(" ", "-")
-    for directory in _scenario_search_dirs():
-        for candidate in (directory / f"{slug}.yaml", directory / f"{slug}.yml"):
-            if candidate.exists():
-                return candidate
-    return None
-
-
-def _normalize_steps(raw_steps: list[Any]) -> list[dict[str, Any]]:
-    normalized: list[dict[str, Any]] = []
-    for item in raw_steps:
-        if isinstance(item, str):
-            normalized.append({"action": item})
-            continue
-        if not isinstance(item, dict) or not item:
-            raise RuntimeError(f"无效步骤：{item!r}")
-        if "action" in item:
-            normalized.append(dict(item))
-            continue
-        if len(item) == 1:
-            action, payload = next(iter(item.items()))
-            step = {"action": action}
-            if payload is not None:
-                if isinstance(payload, dict):
-                    step.update(payload)
-                elif action == "wait" and isinstance(payload, int):
-                    step["ms"] = payload
-                elif action == "wait" and isinstance(payload, str) and payload.isdigit():
-                    step["ms"] = int(payload)
-                else:
-                    step["target"] = payload
-            normalized.append(step)
-            continue
-        raise RuntimeError(f"无效步骤：{item!r}")
-    return normalized
-
-
-def _parse_inline_dsl(text: str) -> BrowserScenario | None:
-    raw = _strip_mention(text)
-    match = re.search(r"(?ms)^browser\s*:\s*\n(.+)$", raw, re.IGNORECASE)
-    if not match:
-        return None
-
-    steps: list[dict[str, Any]] = []
-    for line in match.group(1).splitlines():
-        line = line.strip()
-        if not line or line.startswith("#"):
-            continue
-        line = re.sub(r"^[-*]\s*", "", line)
-        if not line:
-            continue
-        steps.append(_parse_dsl_line(line))
-
-    if not steps:
-        return None
-    return BrowserScenario(
-        name="inline",
-        base_url=default_base_url(),
-        steps=steps,
-        source="inline-dsl",
-    )
-
-
-def _parse_dsl_line(line: str) -> dict[str, Any]:
-    parts = line.split(None, 2)
-    action = parts[0].lower()
-    if action == "goto":
-        return {"action": "goto", "target": parts[1] if len(parts) > 1 else "/"}
-    if action == "click":
-        return {"action": "click", "target": " ".join(parts[1:])}
-    if action == "fill":
-        if len(parts) < 3:
-            raise RuntimeError(f"fill 语法：fill 字段 值（{line}）")
-        return {"action": "fill", "field": parts[1], "value": parts[2]}
-    if action == "wait":
-        payload = parts[1] if len(parts) > 1 else "1500"
-        if payload.isdigit():
-            return {"action": "wait", "ms": int(payload)}
-        return {"action": "wait", "url": payload}
-    if action in {"screenshot", "shot"}:
-        return {"action": "screenshot"}
-    raise RuntimeError(f"未知 DSL 步骤：{line}")
-
-
-def _split_segments(text: str) -> list[str]:
-    raw = _strip_mention(text)
-    raw = re.sub(r"^(browser|网页|网页操作|操作)\s*[:：]?\s*", "", raw, flags=re.IGNORECASE)
-    raw = re.sub(r"然后截图|再截图|最后截图", "截图", raw)
-    chunks = re.split(r"[，,。；;]\s*|\s+然后\s+|\s+接着\s+|\s+并\s*", raw)
-    expanded: list[str] = []
-    for chunk in chunks:
-        chunk = chunk.strip()
-        if not chunk:
-            continue
-        subchunks = re.split(r"\s+然后\s+", chunk)
-        if "后" in chunk and len(subchunks) == 1:
-            subchunks = re.split(r"(?<=[登录页表单])后(?=[进入打开等待点击访问])", chunk)
-        for part in subchunks:
-            part = part.strip()
-            if part:
-                expanded.append(part)
-    return expanded
-
-
-def _parse_segment(segment: str) -> list[dict[str, Any]]:
-    seg = segment.strip()
-    if not seg or seg.lower() in {"browser", "网页操作"}:
-        return []
-
-    if re.fullmatch(r"截图|截屏", seg, re.IGNORECASE):
-        return [{"action": "screenshot"}]
-
-    match = re.search(r"访问登录页|打开登录页|进入登录页|运行登录页", seg, re.IGNORECASE)
-    if match:
-        return [{"action": "goto", "target": "/login"}]
-
-    match = re.search(r"输入账号密码|填写账号密码|输入账号和密码", seg, re.IGNORECASE)
-    if match:
-        return [
-            {"action": "fill", "field": "账号", "value": "{{PREVIEW_LOGIN_USER}}"},
-            {"action": "fill", "field": "密码", "value": "{{PREVIEW_LOGIN_PASSWORD}}"},
-        ]
-
-    match = re.search(r"输入账号|填写账号|输入用户名|填写用户名", seg, re.IGNORECASE)
-    if match:
-        return [{"action": "fill", "field": "账号", "value": "{{PREVIEW_LOGIN_USER}}"}]
-
-    match = re.search(r"输入密码|填写密码", seg, re.IGNORECASE)
-    if match:
-        return [{"action": "fill", "field": "密码", "value": "{{PREVIEW_LOGIN_PASSWORD}}"}]
-
-    match = re.search(r"进入主页|进入首页|打开主页|打开首页|等待主页", seg, re.IGNORECASE)
-    if match:
-        return [{"action": "wait", "url": "**/app/**"}]
-
-    match = re.search(r"等待\s*(\d+)\s*秒", seg, re.IGNORECASE)
-    if match:
-        return [{"action": "wait", "ms": int(match.group(1)) * 1000}]
-
-    match = re.search(
-        r"(?:点击|点选|选择)\s*(.+?)(?:菜单|按钮|链接)?$",
-        seg,
-        re.IGNORECASE,
-    )
-    if match:
-        target = match.group(1).strip()
-        target = re.sub(r"(然后|再|并)?\s*(截图|截屏).*$", "", target, flags=re.IGNORECASE).strip()
-        target = re.sub(r"(然后|再|之后)$", "", target).strip()
-        target = re.sub(r"(菜单|按钮|链接)$", "", target).strip()
-        if target:
-            return [{"action": "click", "target": target}]
-
-    match = re.search(
-        r"(?:访问|打开|进入)\s*(https?://\S+|/\S+|登录页|主页|首页)",
-        seg,
-        re.IGNORECASE,
-    )
-    if match:
-        target = match.group(1)
-        mapping = {"登录页": "/login", "主页": "/app/dashboard", "首页": "/app/dashboard"}
-        return [{"action": "goto", "target": mapping.get(target, target)}]
-
-    return []
-
-
-def parse_natural_language(text: str) -> BrowserScenario | None:
-    segments = _split_segments(text)
-    steps: list[dict[str, Any]] = []
-    for segment in segments:
-        steps.extend(_parse_segment(segment))
-
-    if not steps:
-        return None
-    if not any(step.get("action") == "screenshot" for step in steps):
-        if re.search(r"截图|截屏", text, re.IGNORECASE):
-            steps.append({"action": "screenshot"})
-    if not steps:
-        return None
-
-    return BrowserScenario(
-        name="natural",
-        base_url=default_base_url(),
-        steps=steps,
-        source="natural-language",
-    )
-
-
-def parse_browser_request(text: str) -> BrowserScenario | None:
-    if not is_browser_intent(text):
-        return None
-
-    raw = _strip_mention(text)
-    yaml_block = re.search(r"```(?:yaml|yml)\s*\n(.+?)```", raw, re.IGNORECASE | re.DOTALL)
-    if yaml_block:
-        data = yaml.safe_load(yaml_block.group(1))
-        if isinstance(data, dict):
-            base_url = interpolate(str(data.get("base_url") or default_base_url()))
-            steps = data.get("steps") or []
-            return BrowserScenario(
-                name=data.get("name") or "yaml-inline",
-                base_url=base_url,
-                steps=_normalize_steps(steps),
-                source="yaml-inline",
-            )
-
-    inline = _parse_inline_dsl(text)
-    if inline:
-        return inline
-
-    match = re.match(r"^(browser|网页|网页操作|操作)\s+([\w\-./]+)\s*$", raw, re.IGNORECASE)
-    if match:
-        path = _find_scenario_file(match.group(2))
-        if not path:
-            raise RuntimeError(f"未找到场景文件：{match.group(2)}.yaml")
-        return _load_yaml_scenario(path)
-
-    scenario = parse_natural_language(text)
-    if scenario:
-        return scenario
-
-    default_name = (env_config.env("BROWSER_DEFAULT_SCENARIO") or "").strip()
-    if default_name:
-        path = _find_scenario_file(default_name)
-        if path:
-            return _load_yaml_scenario(path)
-
-    return None

bot/browser_service.py

@@ -1,87 +0,0 @@
-"""浏览器自动化服务：解析场景 + 启动 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)

bot/cursor_runner.py

@@ -1,106 +0,0 @@
-"""通过 Cursor SDK 执行用户任务。"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-import re
-from typing import Awaitable, Callable
-
-import env_config
-from bridge_manager import warm_cursor_bridge
-
-logger = logging.getLogger(__name__)
-
-_cursor_lock = asyncio.Lock()
-
-WECHAT_SYSTEM_PREFIX = """你是企业微信群里的 Skills 助手，正在回复群成员的消息。
-
-要求：
-- 用简洁的中文回答（除非用户用其他语言提问）
-- 使用企业微信支持的 Markdown 子集（加粗、链接、列表；避免复杂表格）
-- 直接给出结论，不要冗长铺垫
-- 若任务涉及 skills.sh，可说明安装命令 `npx skills add owner/repo/skill-name`
-- **不要**在回复里写 `[图片]` 占位符；企微无法通过 Markdown 显示图片
-- 若用户要页面截图，请明确告知其发送：`截图` 或 `preview`（由 bot 自动发图）
-
-用户任务：
-"""
-
-
-def _cursor_settings() -> dict[str, str | int]:
-    timeout_raw = env_config.env("CURSOR_TIMEOUT", "600") or "600"
-    return {
-        "api_key": env_config.env("CURSOR_API_KEY"),
-        "cwd": env_config.env("CURSOR_CWD", r"d:\LY\test\tech") or r"d:\LY\test\tech",
-        "model": env_config.env("CURSOR_MODEL", "composer-2.5") or "composer-2.5",
-        "timeout": int(timeout_raw),
-    }
-
-
-def strip_mention(text: str) -> str:
-    return re.sub(r"@\S+\s*", "", text).strip()
-
-
-def _build_prompt(task: str) -> str:
-    return WECHAT_SYSTEM_PREFIX + task.strip()
-
-
-def execute_cursor_task_sync(task: str) -> str:
-    from cursor_sdk import Agent, AgentOptions, CursorAgentError, LocalAgentOptions
-
-    settings = _cursor_settings()
-    api_key = settings["api_key"]
-    if not api_key:
-        raise RuntimeError(
-            "未配置 CURSOR_API_KEY。请在 bot/.env 中设置，"
-            "密钥见 https://cursor.com/dashboard/integrations"
-        )
-
-    warm_cursor_bridge()
-
-    cwd = str(settings["cwd"])
-    prompt = _build_prompt(task)
-    logger.info("Cursor 执行任务 cwd=%s model=%s", cwd, settings["model"])
-
-    try:
-        result = Agent.prompt(
-            prompt,
-            AgentOptions(
-                api_key=api_key,
-                model=settings["model"],
-                local=LocalAgentOptions(cwd=cwd),
-            ),
-        )
-    except CursorAgentError as exc:
-        raise RuntimeError(
-            f"Cursor 启动失败：{exc.message}"
-            + ("（可重试）" if exc.is_retryable else "")
-        ) from exc
-
-    if result.status == "error":
-        detail = result.result or "运行失败，无详细错误"
-        raise RuntimeError(f"Cursor 执行失败：{detail}")
-
-    text = (result.result or "").strip()
-    if not text:
-        return "Cursor 已完成任务，但没有返回文本内容。"
-    return text
-
-
-async def run_cursor_task(
-    task: str,
-    on_progress: Callable[[str], Awaitable[None]] | None = None,
-) -> str:
-    timeout = int(_cursor_settings()["timeout"])
-    if on_progress:
-        await on_progress("Cursor 正在执行任务，请稍候…")
-
-    async with _cursor_lock:
-        try:
-            return await asyncio.wait_for(
-                asyncio.to_thread(execute_cursor_task_sync, task),
-                timeout=timeout,
-            )
-        except asyncio.TimeoutError as exc:
-            raise RuntimeError(f"Cursor 执行超时（>{timeout}s）") from exc

bot/env_config.py

@@ -1,16 +0,0 @@
-"""加载 bot/.env，供各模块在 import 时统一读取环境变量。"""
-
-from __future__ import annotations
-
-import os
-from pathlib import Path
-
-from dotenv import load_dotenv
-
-_BOT_DIR = Path(__file__).resolve().parent
-load_dotenv(_BOT_DIR / ".env")
-load_dotenv(_BOT_DIR / ".env.local", override=True)
-
-
-def env(key: str, default: str | None = None) -> str | None:
-    return os.getenv(key, default)

bot/image_extract.py

@@ -1,58 +0,0 @@
-"""从文本/Cursor 回复中解析本地截图路径。"""
-
-from __future__ import annotations
-
-import re
-from pathlib import Path
-
-import env_config
-
-IMAGE_SUFFIXES = (".png", ".jpg", ".jpeg", ".webp")
-
-
-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 _resolve_candidate(raw: str, cwd: Path) -> Path | None:
-    cleaned = raw.strip().strip("`\"'[]()")
-    if not cleaned or cleaned.startswith("http"):
-        return None
-    path = Path(cleaned)
-    if not path.is_absolute():
-        path = cwd / path
-    try:
-        resolved = path.resolve()
-    except OSError:
-        return None
-    if resolved.is_file() and resolved.suffix.lower() in IMAGE_SUFFIXES:
-        return resolved
-    return None
-
-
-def find_image_paths(text: str) -> list[Path]:
-    cwd = _project_cwd()
-    seen: set[Path] = set()
-    found: list[Path] = []
-
-    patterns = [
-        r"(?:保存(?:至|到)|saved\s+to|screenshot\s*[:：])\s*([^\s\n\]]+\.(?:png|jpe?g|webp))",
-        r"([A-Za-z]:\\[^\s\n\]]+\.(?:png|jpe?g|webp))",
-        r"([^\s\n\]]+\.(?:png|jpe?g|webp))",
-    ]
-
-    for pattern in patterns:
-        for match in re.finditer(pattern, text, re.IGNORECASE):
-            path = _resolve_candidate(match.group(1), cwd)
-            if path and path not in seen:
-                seen.add(path)
-                found.append(path)
-
-    return found
-
-
-def strip_fake_image_markdown(text: str) -> str:
-    text = re.sub(r"^\s*\[图片\]\s*$", "", text, flags=re.MULTILINE)
-    text = re.sub(r"\n{3,}", "\n\n", text)
-    return text.strip()

bot/main.py

@@ -1,156 +0,0 @@
-"""企业微信智能机器人 · Skills 助手（skills 快查 + 截图预览 + Cursor 执行任务）。"""
-
-from __future__ import annotations
-
-import logging
-import sys
-
-import env_config
-from bridge_manager import shutdown_cursor_bridge, warm_cursor_bridge
-from router import route_message, routing_mode
-from skills_service import handle_command, warm_feed_cache
-from wecom_media import reply_image, upload_image
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(message)s",
-)
-logger = logging.getLogger("skills-bot")
-
-BOT_ID = env_config.env("WECOM_BOT_ID") or env_config.env("WECHAT_BOT_ID")
-BOT_SECRET = env_config.env("WECOM_BOT_SECRET") or env_config.env("WECHAT_BOT_SECRET")
-
-
-def _require_credentials() -> None:
-    if not BOT_ID or not BOT_SECRET:
-        print(
-            "请设置环境变量 WECOM_BOT_ID 和 WECOM_BOT_SECRET\n"
-            "（企业微信 → 智能机器人 → API 模式 → 长连接）",
-            file=sys.stderr,
-        )
-        sys.exit(1)
-
-
-def create_client():
-    from aibot import WSClient, WSClientOptions, generate_req_id
-
-    ws_client = WSClient(
-        WSClientOptions(
-            bot_id=BOT_ID,
-            secret=BOT_SECRET,
-        )
-    )
-
-    @ws_client.on("authenticated")
-    def on_authenticated():
-        logger.info("企业微信长连接认证成功，路由模式=%s", routing_mode())
-        cursor_key = env_config.env("CURSOR_API_KEY")
-        if cursor_key:
-            logger.info("CURSOR_API_KEY 已加载（%s…）", cursor_key[:8])
-            try:
-                warm_cursor_bridge()
-                logger.info("Cursor bridge 预启动完成")
-            except Exception as exc:
-                logger.warning("Cursor bridge 预启动失败（Cursor 任务时会重试）: %s", exc)
-        else:
-            logger.warning("CURSOR_API_KEY 未配置，Cursor 任务将失败")
-        try:
-            warm_feed_cache()
-            logger.info("skills 数据预加载完成")
-        except Exception as exc:
-            logger.warning("skills 数据预加载失败: %s", exc)
-
-    @ws_client.on("event.enter_chat")
-    async def on_enter_chat(frame):
-        help_text = handle_command("help")
-        extra = (
-            "\n\n---\n"
-            "**单页截图**：`preview` / `截图` / `预览 [路径或URL]`\n"
-            "**网页操作**：自然语言多步操作，或 `browser 场景名`\n"
-            "例：`访问登录页，输入账号密码，点击登录，点击智能体管理，截图`\n"
-            "场景文件：`bot/scenarios/*.yaml`（可用 `browser xiaobao-agent-manage`）"
-        )
-        await ws_client.reply_welcome(
-            frame,
-            {
-                "msgtype": "markdown",
-                "markdown": {"content": help_text + extra},
-            },
-        )
-
-    @ws_client.on("message.text")
-    async def on_text(frame):
-        body = frame.get("body", {})
-        content = body.get("text", {}).get("content", "")
-        logger.info("收到消息: %s", content)
-
-        stream_id = generate_req_id("stream")
-        last_progress = ""
-
-        async def on_progress(message: str) -> None:
-            nonlocal last_progress
-            if message != last_progress:
-                last_progress = message
-                await ws_client.reply_stream(frame, stream_id, message, False)
-
-        await ws_client.reply_stream(frame, stream_id, "收到，正在处理…", False)
-
-        try:
-            result = await route_message(content, on_progress=on_progress)
-            reply = result.text
-            logger.info(
-                "回复来源: %s, 文本长度=%d, 图片=%s",
-                result.source,
-                len(reply),
-                result.image_path or "-",
-            )
-        except Exception as exc:
-            logger.exception("处理失败")
-            reply = f"处理失败：{exc}"
-            result = None
-
-        if len(reply) > 3800:
-            reply = reply[:3800] + "\n\n> …内容已截断"
-
-        await ws_client.reply_stream(frame, stream_id, reply, True)
-
-        if result and result.image_path:
-            try:
-                media_id = await upload_image(ws_client, result.image_path)
-                await reply_image(ws_client, frame, media_id)
-                logger.info("图片已发送到企微: %s", result.image_path)
-            except Exception as exc:
-                logger.exception("发送图片失败")
-                await ws_client.reply(
-                    frame,
-                    {
-                        "msgtype": "markdown",
-                        "markdown": {
-                            "content": f"截图文件：`{result.image_path}`\n发图失败：{exc}\n\n请确认 bot 已重启，或发送 `截图` 重试。",
-                        },
-                    },
-                )
-
-    @ws_client.on("error")
-    def on_error(error):
-        logger.error("连接错误: %s", error)
-
-    @ws_client.on("disconnected")
-    def on_disconnected(reason):
-        logger.warning("连接断开: %s", reason)
-
-    return ws_client
-
-
-def main() -> None:
-    import atexit
-
-    atexit.register(shutdown_cursor_bridge)
-    _require_credentials()
-    client = create_client()
-    logger.info("启动 Skills 助手，Bot ID=%s…", BOT_ID[:8] if BOT_ID else "?")
-    client.run()
-
-
-if __name__ == "__main__":
-    main()

bot/preview_service.py

@@ -1,255 +0,0 @@
-"""在 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)

bot/requirements.txt

@@ -1,7 +0,0 @@
-wecom-aibot-python-sdk>=1.0.2
-python-dotenv>=1.0.0
-httpx>=0.27.0
-certifi>=2024.0.0
-cursor-sdk>=0.1.0
-playwright>=1.49.0
-PyYAML>=6.0.0

bot/router.py

@@ -1,109 +0,0 @@
-"""消息路由：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))

bot/scenarios/xiaobao-agent-manage.yaml

@@ -1,17 +0,0 @@
-name: xiaobao-agent-manage
-description: 登录后打开智能体管理并截图
-steps:
-  - goto: /login
-  - fill:
-      field: 账号
-      value: "{{PREVIEW_LOGIN_USER}}"
-  - fill:
-      field: 密码
-      value: "{{PREVIEW_LOGIN_PASSWORD}}"
-  - click: 登录
-  - wait:
-      url: "**/app/**"
-      timeout: 60000
-  - click: 智能体管理
-  - wait: 1500
-  - screenshot

bot/skills_service.py

@@ -1,317 +0,0 @@
-"""skills.sh 数据查询与命令解析。"""
-
-from __future__ import annotations
-
-import json
-import logging
-import re
-import time
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any, Literal
-
-import certifi
-import httpx
-
-logger = logging.getLogger(__name__)
-
-FEED_URLS = [
-    # jsDelivr 在国内通常比 raw.githubusercontent.com 更稳定
-    "https://cdn.jsdelivr.net/gh/NeverSight/skills.sh_feed@main/data/feed.json",
-    "https://raw.githubusercontent.com/NeverSight/skills.sh_feed/main/data/feed.json",
-]
-CACHE_TTL_SECONDS = 600
-CACHE_DIR = Path(__file__).resolve().parent / ".cache"
-CACHE_FILE = CACHE_DIR / "feed.json"
-
-_cache: dict[str, Any] = {"data": None, "fetched_at": 0.0}
-
-Board = Literal["trending", "hot", "all"]
-
-
-@dataclass
-class Command:
-    kind: Literal["help", "list", "search", "detail"]
-    board: Board = "trending"
-    limit: int = 10
-    query: str = ""
-
-
-def _fetch_json(url: str) -> dict[str, Any]:
-    headers = {
-        "User-Agent": "skills-hot-bot/1.0",
-        "Accept": "application/json",
-    }
-    with httpx.Client(
-        timeout=httpx.Timeout(20.0, connect=10.0),
-        verify=certifi.where(),
-        follow_redirects=True,
-    ) as client:
-        resp = client.get(url, headers=headers)
-        resp.raise_for_status()
-        return resp.json()
-
-
-def _load_disk_cache() -> dict[str, Any] | None:
-    if not CACHE_FILE.exists():
-        return None
-    try:
-        return json.loads(CACHE_FILE.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError) as exc:
-        logger.warning("读取本地缓存失败: %s", exc)
-        return None
-
-
-def _save_disk_cache(data: dict[str, Any]) -> None:
-    CACHE_DIR.mkdir(parents=True, exist_ok=True)
-    CACHE_FILE.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
-
-
-def load_feed(force: bool = False) -> dict[str, Any]:
-    now = time.time()
-    if not force and _cache["data"] and now - _cache["fetched_at"] < CACHE_TTL_SECONDS:
-        return _cache["data"]
-
-    errors: list[str] = []
-    for url in FEED_URLS:
-        for attempt in range(3):
-            try:
-                data = _fetch_json(url)
-                _cache["data"] = data
-                _cache["fetched_at"] = now
-                _save_disk_cache(data)
-                logger.info("skills 数据已更新: %s", url)
-                return data
-            except Exception as exc:
-                msg = f"{url} (#{attempt + 1}): {exc}"
-                errors.append(msg)
-                logger.debug("拉取失败 %s", msg)
-                time.sleep(0.5 * (attempt + 1))
-
-    stale = _load_disk_cache()
-    if stale:
-        logger.warning("网络不可用，回退到本地缓存")
-        _cache["data"] = stale
-        _cache["fetched_at"] = now
-        return stale
-
-    raise RuntimeError(f"无法获取 skills 数据。最近错误: {errors[-1]}")
-
-
-def warm_feed_cache() -> None:
-    """启动时预加载，避免首条消息才触发网络请求。"""
-    load_feed(force=True)
-
-
-def _normalize_text(text: str) -> str:
-    text = re.sub(r"@\S+\s*", "", text)
-    return text.strip().lower()
-
-
-def _parse_limit(raw: str | None, default: int = 10) -> int:
-    if not raw:
-        return default
-    try:
-        n = int(raw)
-    except ValueError:
-        return default
-    return max(1, min(n, 30))
-
-
-def _match_list(raw: str, board: Board, aliases: str) -> Command | None:
-    m = re.match(rf"^({aliases})(?:\s+top)?\s*(\d+)?$", raw)
-    if m:
-        return Command(kind="list", board=board, limit=_parse_limit(m.group(2)))
-    m = re.match(rf"^(查|查询)\s+({aliases})(?:\s+top)?\s*(\d+)?$", raw)
-    if m:
-        return Command(kind="list", board=board, limit=_parse_limit(m.group(3)))
-    return None
-
-
-def parse_command(text: str) -> Command:
-    raw = _normalize_text(text)
-    if not raw or raw in {"help", "帮助", "?", "h"}:
-        return Command(kind="help")
-
-    for board, aliases in (
-        ("trending", "trending|趋势|top"),
-        ("hot", "hot|实时|热门"),
-        ("all", "all|总榜|alltime|all-time"),
-    ):
-        cmd = _match_list(raw, board, aliases)
-        if cmd:
-            return cmd
-
-    m = re.match(r"^(search|搜索|find|查)\s+(.+)$", raw)
-    if m:
-        return Command(kind="search", query=m.group(2).strip(), limit=5)
-
-    m = re.match(r"^(detail|详情|skill|info)\s+(.+)$", raw)
-    if m:
-        return Command(kind="detail", query=m.group(2).strip())
-
-    if raw.startswith("trending") or raw.startswith("趋势"):
-        parts = raw.split(maxsplit=1)
-        return Command(kind="list", board="trending", limit=_parse_limit(parts[1] if len(parts) > 1 else None))
-
-    return Command(kind="search", query=raw, limit=5)
-
-
-def _format_installs(n: int | float) -> str:
-    if n >= 1_000_000:
-        return f"{n / 1_000_000:.1f}M"
-    if n >= 1_000:
-        return f"{n / 1_000:.1f}K"
-    return str(int(n))
-
-
-def _board_items(feed: dict[str, Any], board: Board) -> list[dict[str, Any]]:
-    key = {"trending": "topTrending", "hot": "topHot", "all": "topAllTime"}[board]
-    return feed.get(key, [])
-
-
-def _board_title(board: Board) -> str:
-    return {
-        "trending": "Trending（近期增长）",
-        "hot": "Hot（实时热度）",
-        "all": "All Time（总安装榜）",
-    }[board]
-
-
-def format_list(board: Board, limit: int) -> str:
-    feed = load_feed()
-    items = _board_items(feed, board)[:limit]
-    updated = feed.get("updatedAt", "未知")[:10]
-
-    lines = [
-        f"**skills.sh {_board_title(board)} Top {limit}**",
-        f"> 数据更新：{updated}",
-        "",
-    ]
-
-    for i, item in enumerate(items, 1):
-        title = item.get("title", "?")
-        source = item.get("source", "?")
-        installs = _format_installs(item.get("installs", 0))
-        desc = item.get("description", "")
-        if len(desc) > 80:
-            desc = desc[:77] + "..."
-        link = item.get("link", "")
-        lines.append(f"{i}. **{title}** · {installs}")
-        lines.append(f"   `{source}`")
-        if desc:
-            lines.append(f"   {desc}")
-        if link:
-            lines.append(f"   [查看]({link})")
-        lines.append("")
-
-    return "\n".join(lines).strip()
-
-
-def format_search(query: str, limit: int) -> str:
-    feed = load_feed()
-    q = query.lower()
-    seen: set[str] = set()
-    matches: list[dict[str, Any]] = []
-
-    for board in ("topTrending", "topHot", "topAllTime"):
-        for item in feed.get(board, []):
-            item_id = item.get("id") or item.get("title", "")
-            if item_id in seen:
-                continue
-            haystack = " ".join(
-                [
-                    item.get("title", ""),
-                    item.get("source", ""),
-                    item.get("description", ""),
-                ]
-            ).lower()
-            if q in haystack:
-                seen.add(item_id)
-                matches.append(item)
-            if len(matches) >= limit:
-                break
-        if len(matches) >= limit:
-            break
-
-    if not matches:
-        return f"未找到与 **{query}** 相关的 skill。\n\n试试：`trending 10` / `hot 10` / `搜索 react`"
-
-    lines = [f"**搜索「{query}」** 共 {len(matches)} 条", ""]
-    for i, item in enumerate(matches, 1):
-        title = item.get("title", "?")
-        source = item.get("source", "?")
-        installs = _format_installs(item.get("installs", 0))
-        link = item.get("link", "")
-        lines.append(f"{i}. **{title}** · {installs} · `{source}`")
-        if link:
-            lines.append(f"   [查看]({link})")
-    return "\n".join(lines)
-
-
-def format_detail(name: str) -> str:
-    feed = load_feed()
-    q = name.lower().strip()
-    best: dict[str, Any] | None = None
-
-    for board in ("topTrending", "topHot", "topAllTime"):
-        for item in feed.get(board, []):
-            title = (item.get("title") or "").lower()
-            item_id = (item.get("id") or "").lower()
-            if title == q or q in title or q in item_id:
-                if best is None or item.get("installs", 0) > best.get("installs", 0):
-                    best = item
-
-    if not best:
-        return f"未找到 skill：**{name}**\n\n试试：`搜索 {name}`"
-
-    desc = best.get("description", "无描述")
-    return "\n".join(
-        [
-            f"**{best.get('title', '?')}**",
-            f"`{best.get('source', '?')}`",
-            f"安装量：**{_format_installs(best.get('installs', 0))}**",
-            "",
-            desc,
-            "",
-            f"[skills.sh 详情]({best.get('link', 'https://skills.sh')})",
-            "",
-            f"安装：`npx skills add {best.get('source', '')}/{best.get('title', '')}`",
-        ]
-    )
-
-
-def format_help() -> str:
-    return "\n".join(
-        [
-            "**Skills 助手 · 命令帮助**",
-            "",
-            "`trending 10` / `趋势 10` — 近期增长榜",
-            "`hot 10` / `实时 10` — 实时热度榜",
-            "`all 10` / `总榜 10` — 历史总安装榜",
-            "`搜索 react` / `search tdd` — 关键词搜索",
-            "`详情 find-skills` — 查看单个 skill",
-            "`preview` / `截图` / `预览` — 单页截图",
-            "`browser 场景名` — 执行 YAML 场景（见 bot/scenarios/）",
-            "自然语言 — 如：访问登录页，输入账号密码，点击登录，点击智能体管理，截图",
-            "`preview /about 5173` — 指定路径和端口",
-            "",
-            "示例：",
-            "• trending top10",
-            "• 查 grill",
-            "• 详情 remotion-render",
-        ]
-    )
-
-
-def handle_command(text: str) -> str:
-    cmd = parse_command(text)
-    if cmd.kind == "help":
-        return format_help()
-    if cmd.kind == "list":
-        return format_list(cmd.board, cmd.limit)
-    if cmd.kind == "search":
-        return format_search(cmd.query, cmd.limit)
-    if cmd.kind == "detail":
-        return format_detail(cmd.query)
-    return format_help()

bot/wecom_media.py

@@ -1,96 +0,0 @@
-"""企业微信 API 模式：上传图片并回复。"""
-
-from __future__ import annotations
-
-import base64
-import hashlib
-import logging
-from pathlib import Path
-from typing import Any
-
-from aibot import generate_req_id
-
-logger = logging.getLogger(__name__)
-
-CHUNK_SIZE = 512 * 1024
-MAX_IMAGE_BYTES = 9 * 1024 * 1024
-
-
-def _ensure_image_size(path: Path) -> bytes:
-    data = path.read_bytes()
-    if len(data) > MAX_IMAGE_BYTES:
-        raise RuntimeError(
-            f"截图过大（{len(data) // 1024}KB），请缩小页面或使用 viewport 截图（上限 9MB）"
-        )
-    return data
-
-
-def _response_body(frame: dict[str, Any]) -> dict[str, Any]:
-    if frame.get("errcode", 0) != 0:
-        raise RuntimeError(
-            f"企微接口错误 errcode={frame.get('errcode')} errmsg={frame.get('errmsg')}"
-        )
-    body = frame.get("body")
-    return body if isinstance(body, dict) else {}
-
-
-async def upload_image(ws_client: Any, image_path: str | Path) -> str:
-    path = Path(image_path)
-    if not path.exists():
-        raise RuntimeError(f"截图不存在: {path}")
-
-    data = _ensure_image_size(path)
-    md5 = hashlib.md5(data).hexdigest()
-    chunks = [data[i : i + CHUNK_SIZE] for i in range(0, len(data), CHUNK_SIZE)]
-    total_chunks = len(chunks)
-
-    manager = ws_client._ws_manager
-
-    init_frame = await manager.send_reply(
-        generate_req_id("upload_init"),
-        {
-            "type": "image",
-            "filename": path.name,
-            "total_size": len(data),
-            "total_chunks": total_chunks,
-            "md5": md5,
-        },
-        "aibot_upload_media_init",
-    )
-    upload_id = _response_body(init_frame).get("upload_id")
-    if not upload_id:
-        raise RuntimeError("上传初始化失败：未返回 upload_id")
-
-    for index, chunk in enumerate(chunks):
-        chunk_frame = await manager.send_reply(
-            generate_req_id("upload_chunk"),
-            {
-                "upload_id": upload_id,
-                "chunk_index": index,
-                "base64_data": base64.b64encode(chunk).decode("ascii"),
-            },
-            "aibot_upload_media_chunk",
-        )
-        _response_body(chunk_frame)
-
-    finish_frame = await manager.send_reply(
-        generate_req_id("upload_finish"),
-        {"upload_id": upload_id},
-        "aibot_upload_media_finish",
-    )
-    media_id = _response_body(finish_frame).get("media_id")
-    if not media_id:
-        raise RuntimeError("上传完成但未返回 media_id")
-
-    logger.info("图片已上传 media_id=%s…", str(media_id)[:12])
-    return str(media_id)
-
-
-async def reply_image(ws_client: Any, frame: dict[str, Any], media_id: str) -> None:
-    await ws_client.reply(
-        frame,
-        {
-            "msgtype": "image",
-            "image": {"media_id": media_id},
-        },
-    )