From 6192dd4e2a5c0a52b44e46e2686f9216fbc429a6 Mon Sep 17 00:00:00 2001 From: yumao Date: Fri, 17 Jul 2026 18:11:11 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E7=A7=BB=E9=99=A4=E7=8B=AC=E7=AB=8B?= =?UTF-8?q?=E4=BC=81=E5=BE=AE=20bot/=E6=B5=8F=E8=A7=88=E5=99=A8=E6=A1=A5?= =?UTF-8?q?=E6=8E=A5=E7=B3=BB=E7=BB=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两套系统并存，bot/（browser_service、preview_service、wecom_media、 router、xiaobao 场景等）与早报生成是独立的一条线，不再维护，整体删除。 同时将 __pycache__/*.pyc/.cursor 加入 .gitignore。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 8 +- bot/.env.example | 29 --- bot/.gitignore | 4 - bot/bot_types.py | 12 - bot/bridge_manager.py | 156 ------------ bot/browser_env.py | 49 ---- bot/browser_executor.py | 269 -------------------- bot/browser_models.py | 26 -- bot/browser_parser.py | 307 ----------------------- bot/browser_service.py | 87 ------- bot/cursor_runner.py | 106 -------- bot/env_config.py | 16 -- bot/image_extract.py | 58 ----- bot/main.py | 156 ------------ bot/preview_service.py | 255 ------------------- bot/requirements.txt | 7 - bot/router.py | 109 -------- bot/scenarios/xiaobao-agent-manage.yaml | 17 -- bot/skills_service.py | 317 ------------------------ bot/wecom_media.py | 96 ------- 20 files changed, 4 insertions(+), 2080 deletions(-) delete mode 100644 bot/.env.example delete mode 100644 bot/.gitignore delete mode 100644 bot/bot_types.py delete mode 100644 bot/bridge_manager.py delete mode 100644 bot/browser_env.py delete mode 100644 bot/browser_executor.py delete mode 100644 bot/browser_models.py delete mode 100644 bot/browser_parser.py delete mode 100644 bot/browser_service.py delete mode 100644 bot/cursor_runner.py delete mode 100644 bot/env_config.py delete mode 100644 bot/image_extract.py delete mode 100644 bot/main.py delete mode 100644 bot/preview_service.py delete mode 100644 bot/requirements.txt delete mode 100644 bot/router.py delete mode 100644 bot/scenarios/xiaobao-agent-manage.yaml delete mode 100644 bot/skills_service.py delete mode 100644 bot/wecom_media.py diff --git a/.gitignore b/.gitignore index 45f5e63..9985448 100644 --- a/.gitignore +++ b/.gitignore @@ -2,7 +2,7 @@ .env.local logs/ .cache/ -bot/.env -bot/.venv/ -bot/.cache/ -output \ No newline at end of file +output +__pycache__/ +*.pyc +.cursor/ \ No newline at end of file diff --git a/bot/.env.example b/bot/.env.example deleted file mode 100644 index 3cdc3a7..0000000 --- a/bot/.env.example +++ /dev/null @@ -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 diff --git a/bot/.gitignore b/bot/.gitignore deleted file mode 100644 index fcb6394..0000000 --- a/bot/.gitignore +++ /dev/null @@ -1,4 +0,0 @@ -.cache/ -.env -.venv/ -.cache/screenshots/ diff --git a/bot/bot_types.py b/bot/bot_types.py deleted file mode 100644 index 593242e..0000000 --- a/bot/bot_types.py +++ /dev/null @@ -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 diff --git a/bot/bridge_manager.py b/bot/bridge_manager.py deleted file mode 100644 index 89668d3..0000000 --- a/bot/bridge_manager.py +++ /dev/null @@ -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 diff --git a/bot/browser_env.py b/bot/browser_env.py deleted file mode 100644 index 26ccaa0..0000000 --- a/bot/browser_env.py +++ /dev/null @@ -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 diff --git a/bot/browser_executor.py b/bot/browser_executor.py deleted file mode 100644 index e340c16..0000000 --- a/bot/browser_executor.py +++ /dev/null @@ -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, - ) diff --git a/bot/browser_models.py b/bot/browser_models.py deleted file mode 100644 index d9326b7..0000000 --- a/bot/browser_models.py +++ /dev/null @@ -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) diff --git a/bot/browser_parser.py b/bot/browser_parser.py deleted file mode 100644 index b79fa8c..0000000 --- a/bot/browser_parser.py +++ /dev/null @@ -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 diff --git a/bot/browser_service.py b/bot/browser_service.py deleted file mode 100644 index a7d98a6..0000000 --- a/bot/browser_service.py +++ /dev/null @@ -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) diff --git a/bot/cursor_runner.py b/bot/cursor_runner.py deleted file mode 100644 index 03bfd7e..0000000 --- a/bot/cursor_runner.py +++ /dev/null @@ -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 diff --git a/bot/env_config.py b/bot/env_config.py deleted file mode 100644 index 703c286..0000000 --- a/bot/env_config.py +++ /dev/null @@ -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) diff --git a/bot/image_extract.py b/bot/image_extract.py deleted file mode 100644 index e23ac3d..0000000 --- a/bot/image_extract.py +++ /dev/null @@ -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() diff --git a/bot/main.py b/bot/main.py deleted file mode 100644 index d6fc114..0000000 --- a/bot/main.py +++ /dev/null @@ -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() diff --git a/bot/preview_service.py b/bot/preview_service.py deleted file mode 100644 index 5494a91..0000000 --- a/bot/preview_service.py +++ /dev/null @@ -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) diff --git a/bot/requirements.txt b/bot/requirements.txt deleted file mode 100644 index 5dd6647..0000000 --- a/bot/requirements.txt +++ /dev/null @@ -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 diff --git a/bot/router.py b/bot/router.py deleted file mode 100644 index 524e6b2..0000000 --- a/bot/router.py +++ /dev/null @@ -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)) diff --git a/bot/scenarios/xiaobao-agent-manage.yaml b/bot/scenarios/xiaobao-agent-manage.yaml deleted file mode 100644 index e79ccc2..0000000 --- a/bot/scenarios/xiaobao-agent-manage.yaml +++ /dev/null @@ -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 diff --git a/bot/skills_service.py b/bot/skills_service.py deleted file mode 100644 index dc7e821..0000000 --- a/bot/skills_service.py +++ /dev/null @@ -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() diff --git a/bot/wecom_media.py b/bot/wecom_media.py deleted file mode 100644 index 3f57112..0000000 --- a/bot/wecom_media.py +++ /dev/null @@ -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}, - }, - )