From 8a4a3c8f5db9ebd2138473234043c70d991454d9 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Sun, 31 May 2026 10:34:42 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E6=A0=87=E5=87=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/jiangchang_skill_core/__init__.py | 14 ++ src/jiangchang_skill_core/config.py | 125 +++++++++++ src/jiangchang_skill_core/rpa/__init__.py | 59 +++++ src/jiangchang_skill_core/rpa/anti_detect.py | 215 +++++++++++++++++++ src/jiangchang_skill_core/rpa/artifacts.py | 28 +++ src/jiangchang_skill_core/rpa/browser.py | 61 ++++++ src/jiangchang_skill_core/rpa/errors.py | 19 ++ src/jiangchang_skill_core/rpa/human_login.py | 96 +++++++++ src/jiangchang_skill_core/rpa/stealth.py | 66 ++++++ tests/test_anti_detect_import.py | 26 +++ tests/test_config.py | 86 ++++++++ tests/test_stealth.py | 54 +++++ 12 files changed, 849 insertions(+) create mode 100644 src/jiangchang_skill_core/config.py create mode 100644 src/jiangchang_skill_core/rpa/__init__.py create mode 100644 src/jiangchang_skill_core/rpa/anti_detect.py create mode 100644 src/jiangchang_skill_core/rpa/artifacts.py create mode 100644 src/jiangchang_skill_core/rpa/browser.py create mode 100644 src/jiangchang_skill_core/rpa/errors.py create mode 100644 src/jiangchang_skill_core/rpa/human_login.py create mode 100644 src/jiangchang_skill_core/rpa/stealth.py create mode 100644 tests/test_anti_detect_import.py create mode 100644 tests/test_config.py create mode 100644 tests/test_stealth.py diff --git a/src/jiangchang_skill_core/__init__.py b/src/jiangchang_skill_core/__init__.py index 66266ce..df9c549 100644 --- a/src/jiangchang_skill_core/__init__.py +++ b/src/jiangchang_skill_core/__init__.py @@ -1,9 +1,11 @@ from .client import EntitlementClient +from .config import ensure_env_file, get, get_bool, get_float, get_int from .errors import EntitlementDeniedError, EntitlementError, EntitlementServiceError from .guard import enforce_entitlement from .models import EntitlementResult from .runtime_env import ( apply_cli_local_defaults, + find_chrome_executable, get_data_root, get_sibling_skills_root, get_skills_root, @@ -20,6 +22,11 @@ from .unified_logging import ( subprocess_env_with_trace, ) +try: + from . import rpa +except ImportError: + rpa = None # type: ignore[assignment,misc] + __all__ = [ "EntitlementClient", "EntitlementDeniedError", @@ -28,9 +35,15 @@ __all__ = [ "EntitlementServiceError", "apply_cli_local_defaults", "attach_unified_file_handler", + "ensure_env_file", "enforce_entitlement", "ensure_trace_for_process", + "find_chrome_executable", + "get", + "get_bool", "get_data_root", + "get_float", + "get_int", "get_sibling_skills_root", "get_skills_root", "get_skill_log_file_path", @@ -38,6 +51,7 @@ __all__ = [ "get_unified_logs_dir", "get_user_id", "platform_default_data_root", + "rpa", "setup_skill_logging", "subprocess_env_with_trace", ] diff --git a/src/jiangchang_skill_core/config.py b/src/jiangchang_skill_core/config.py new file mode 100644 index 0000000..1a0a45a --- /dev/null +++ b/src/jiangchang_skill_core/config.py @@ -0,0 +1,125 @@ +"""三级优先级配置读取 + 首次 .env 落盘（进程 env > 数据目录 .env > .env.example）。""" + +from __future__ import annotations + +import os +import shutil +from typing import Any + +from .runtime_env import get_data_root, get_user_id + +_skill_slug: str | None = None +_example_path: str | None = None +_env_file_path: str | None = None +_user_env_cache: dict[str, str] | None = None +_example_defaults_cache: dict[str, str] | None = None + + +def _parse_env_file(path: str) -> dict[str, str]: + """标准库手写 .env 解析（不引 python-dotenv）。""" + result: dict[str, str] = {} + if not path or not os.path.isfile(path): + return result + with open(path, encoding="utf-8") as f: + for raw_line in f: + line = raw_line.strip() + if not line or line.startswith("#"): + continue + if "=" not in line: + continue + key, _, value = line.partition("=") + key = key.strip() + value = value.strip() + if not key: + continue + if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"): + value = value[1:-1] + elif " #" in value: + value = value.split(" #", 1)[0].strip() + result[key] = value + return result + + +def _get_user_env() -> dict[str, str]: + global _user_env_cache + if _user_env_cache is not None: + return _user_env_cache + if _env_file_path and os.path.isfile(_env_file_path): + _user_env_cache = _parse_env_file(_env_file_path) + else: + _user_env_cache = {} + return _user_env_cache + + +def _get_example_defaults() -> dict[str, str]: + global _example_defaults_cache + if _example_defaults_cache is not None: + return _example_defaults_cache + if _example_path and os.path.isfile(_example_path): + _example_defaults_cache = _parse_env_file(_example_path) + else: + _example_defaults_cache = {} + return _example_defaults_cache + + +def reset_cache() -> None: + """测试用：清空解析缓存。""" + global _user_env_cache, _example_defaults_cache + _user_env_cache = None + _example_defaults_cache = None + + +def ensure_env_file(skill_slug: str, example_path: str) -> str: + """首次把 .env.example copy 到 {data_root}/{user}/{slug}/.env（已存在不覆盖），返回路径。""" + global _skill_slug, _example_path, _env_file_path + _skill_slug = skill_slug + _example_path = os.path.abspath(example_path) + dest_dir = os.path.join(get_data_root(), get_user_id(), skill_slug) + os.makedirs(dest_dir, exist_ok=True) + dest = os.path.join(dest_dir, ".env") + _env_file_path = dest + if not os.path.isfile(dest) and os.path.isfile(_example_path): + shutil.copy2(_example_path, dest) + reset_cache() + return dest + + +def get(key: str, default: Any = None) -> str | None: + """进程环境变量 > 数据目录 .env > .env.example 默认值。""" + env_val = os.environ.get(key) + if env_val is not None and env_val != "": + return env_val + user_val = _get_user_env().get(key) + if user_val is not None and user_val != "": + return user_val + example_val = _get_example_defaults().get(key) + if example_val is not None and example_val != "": + return example_val + return default + + +def get_bool(key: str, default: bool = False) -> bool: + val = get(key) + if val is None: + return default + return str(val).strip().lower() in ("1", "true", "yes", "on") + + +def get_float(key: str, default: float) -> float: + val = get(key) + if val is None: + return default + try: + return float(val) + except (TypeError, ValueError): + return default + + +def get_int(key: str, default: int) -> int: + val = get(key) + if val is None: + return default + try: + return int(val) + except (TypeError, ValueError): + return default diff --git a/src/jiangchang_skill_core/rpa/__init__.py b/src/jiangchang_skill_core/rpa/__init__.py new file mode 100644 index 0000000..284b12c --- /dev/null +++ b/src/jiangchang_skill_core/rpa/__init__.py @@ -0,0 +1,59 @@ +"""浏览器 RPA 共享库：stealth、拟人操作、启动封装、HITL 等待、失败存证。""" + +from __future__ import annotations + +from . import anti_detect, artifacts, errors, human_login, stealth + +try: + from .browser import launch_persistent_browser +except ImportError: + launch_persistent_browser = None # type: ignore[misc, assignment] + +__all__ = [ + "anti_detect", + "artifacts", + "errors", + "human_login", + "stealth", + "launch_persistent_browser", + # re-export common anti_detect helpers + "random_delay", + "human_delay_short", + "human_delay_page", + "human_delay_batch", + "human_mouse_move", + "human_mouse_wiggle", + "human_click", + "human_type", + "human_scroll", + # human_login + "wait_for_captcha_pass", + "wait_for_login", + "DEFAULT_CAPTCHA_MARKERS", + # artifacts + "artifact_path", + "capture_failure", + # stealth + "STEALTH_INIT_SCRIPT", + "stealth_enabled", + "persistent_context_launch_parts", +] + +from .anti_detect import ( + human_click, + human_delay_batch, + human_delay_page, + human_delay_short, + human_mouse_move, + human_mouse_wiggle, + human_scroll, + human_type, + random_delay, +) +from .artifacts import artifact_path, capture_failure +from .human_login import DEFAULT_CAPTCHA_MARKERS, wait_for_captcha_pass, wait_for_login +from .stealth import ( + STEALTH_INIT_SCRIPT, + persistent_context_launch_parts, + stealth_enabled, +) diff --git a/src/jiangchang_skill_core/rpa/anti_detect.py b/src/jiangchang_skill_core/rpa/anti_detect.py new file mode 100644 index 0000000..88e644b --- /dev/null +++ b/src/jiangchang_skill_core/rpa/anti_detect.py @@ -0,0 +1,215 @@ +"""防反爬：随机延迟、人工鼠标轨迹模拟、拟人滚动。 + +所有延迟使用 random.uniform() + asyncio.sleep()。 +鼠标移动使用三次贝塞尔曲线 + 随机抖动，避免机械直线轨迹。 +""" + +from __future__ import annotations + +import asyncio +import math +import random +from typing import Optional + + +# ── 随机延迟 ──────────────────────────────────────────────── + + +async def random_delay(min_s: float, max_s: float) -> float: + """随机延迟 min_s ~ max_s 秒，返回实际延迟秒数。""" + delay = random.uniform(min_s, max_s) + await asyncio.sleep(delay) + return delay + + +async def human_delay_short() -> float: + """页面内操作间短延迟 0.5~2 秒（点击/滚动）。""" + return await random_delay(0.5, 2.0) + + +async def human_delay_page(min_s: float = 2, max_s: float = 8) -> float: + """页面间/店铺间切换延迟（可配置范围）。""" + return await random_delay(min_s, max_s) + + +async def human_delay_batch(min_s: float = 5, max_s: float = 15) -> float: + """批次间/翻页间长延迟。""" + return await random_delay(min_s, max_s) + + +# ── 鼠标轨迹 ──────────────────────────────────────────────── + + +async def human_mouse_move( + page, + target_x: float, + target_y: float, + steps: Optional[int] = None, +) -> None: + """模拟人工鼠标移动到目标坐标。 + + 使用三次贝塞尔曲线生成 S/C 形路径，中间控制点随机偏移， + 每步添加微小随机抖动，避免完美直线移动。 + + Args: + page: Playwright Page 对象。 + target_x: 目标 X 坐标。 + target_y: 目标 Y 坐标。 + steps: 移动步数（None 时按距离自动计算）。 + """ + if steps is None: + dist = math.hypot(target_x, target_y) + steps = max(8, int(dist / 15)) + + # 随机起始点（模拟鼠标从页面其他位置移动过来） + start_x = target_x + random.uniform(-300, 300) + start_y = target_y + random.uniform(-200, 200) + + # 两个贝塞尔控制点（制造弯曲路径） + dx = target_x - start_x + dy = target_y - start_y + + cp1_x = start_x + dx * random.uniform(0.2, 0.5) + random.uniform(-120, 120) + cp1_y = start_y + dy * random.uniform(0.1, 0.4) + random.uniform(-100, 100) + cp2_x = start_x + dx * random.uniform(0.5, 0.8) + random.uniform(-100, 100) + cp2_y = start_y + dy * random.uniform(0.4, 0.7) + random.uniform(-80, 80) + + for i in range(steps + 1): + t = i / steps + # 三次贝塞尔: B(t) = (1-t)³·P0 + 3(1-t)²·t·P1 + 3(1-t)·t²·P2 + t³·P3 + u = 1 - t + x = ( + (u ** 3) * start_x + + 3 * (u ** 2) * t * cp1_x + + 3 * u * (t ** 2) * cp2_x + + (t ** 3) * target_x + ) + y = ( + (u ** 3) * start_y + + 3 * (u ** 2) * t * cp1_y + + 3 * u * (t ** 2) * cp2_y + + (t ** 3) * target_y + ) + + # 微小随机抖动 + x += random.uniform(-1.5, 1.5) + y += random.uniform(-1.5, 1.5) + + await page.mouse.move(x, y) + await asyncio.sleep(random.uniform(0.003, 0.015)) + + # 最终精确到位 + await page.mouse.move(target_x, target_y) + + +async def human_click( + page, + selector: Optional[str] = None, + *, + x: Optional[float] = None, + y: Optional[float] = None, +) -> None: + """拟人点击：贝塞尔轨迹移动 → 微延迟 → 点击。 + + 支持两种定位方式： + - selector: CSS 选择器，自动取元素中心附近随机偏移 + - x, y: 直接指定坐标（自动加微小随机偏移） + + Args: + page: Playwright Page 对象。 + selector: CSS 选择器（与 x/y 二选一）。 + x: 目标 X 坐标。 + y: 目标 Y 坐标。 + """ + if selector: + locator = page.locator(selector).first + box = await locator.bounding_box() + if box is None: + raise ValueError(f"无法获取元素边界: {selector}") + target_x = box["x"] + box["width"] * random.uniform(0.3, 0.7) + target_y = box["y"] + box["height"] * random.uniform(0.3, 0.7) + elif x is not None and y is not None: + target_x = x + random.uniform(-3, 3) + target_y = y + random.uniform(-3, 3) + else: + raise ValueError("必须提供 selector 或 (x, y) 坐标") + + await human_mouse_move(page, target_x, target_y) + await human_delay_short() + await page.mouse.click(target_x, target_y) + + +# ── 滚动模拟 ──────────────────────────────────────────────── + + +async def human_scroll( + page, direction: str = "down", amount: Optional[int] = None +) -> None: + """模拟人工滚动（分段 + 变速 + 微延迟）。 + + Args: + page: Playwright Page 对象。 + direction: "up" 或 "down"。 + amount: 滚动像素总量，默认随机 300~800。 + """ + if amount is None: + amount = random.randint(300, 800) + + sign = -1 if direction == "up" else 1 + remaining = amount + step_count = random.randint(3, 7) + + for i in range(step_count): + step = remaining // (step_count - i) + random.randint(-30, 30) + step = max(10, min(step, remaining)) + remaining -= step + + await page.mouse.wheel(0, sign * step) + await asyncio.sleep(random.uniform(0.05, 0.2)) + + if remaining > 0: + await page.mouse.wheel(0, sign * remaining) + + await human_delay_short() + + +# ── 进场鼠标随机晃动 ──────────────────────────────────────── + + +async def human_mouse_wiggle(page) -> None: + """页面进入后随机移动鼠标 2~4 次（贝塞尔轨迹），模拟真人到场环顾。""" + try: + vp = page.viewport_size or {"width": 1280, "height": 800} + except Exception: + vp = {"width": 1280, "height": 800} + w, h = vp.get("width", 1280), vp.get("height", 800) + for _ in range(random.randint(2, 4)): + tx = random.uniform(w * 0.2, w * 0.8) + ty = random.uniform(h * 0.2, h * 0.7) + await human_mouse_move(page, tx, ty) + await asyncio.sleep(random.uniform(0.2, 0.8)) + + +# ── 拟人逐字输入 ──────────────────────────────────────────── + + +async def human_type(page, locator, text: str) -> None: + """真人式输入：贝塞尔移动到输入框 → 真实点击聚焦 → 逐字符 keyboard.type。 + + 全程使用 isTrusted=true 的可信事件，绝不用 JS 设置 value。 + """ + box = await locator.bounding_box() + if box is None: + raise ValueError("human_type: 无法获取输入框边界") + tx = box["x"] + box["width"] * random.uniform(0.3, 0.7) + ty = box["y"] + box["height"] * random.uniform(0.3, 0.7) + await human_mouse_move(page, tx, ty) + await asyncio.sleep(random.uniform(0.15, 0.5)) + await page.mouse.click(tx, ty) + await asyncio.sleep(random.uniform(0.2, 0.6)) + # 清空已有内容（用键盘，不用 JS） + await page.keyboard.press("Control+A") + await page.keyboard.press("Delete") + await asyncio.sleep(random.uniform(0.2, 0.5)) + for ch in text: + await page.keyboard.type(ch, delay=random.uniform(90, 240)) diff --git a/src/jiangchang_skill_core/rpa/artifacts.py b/src/jiangchang_skill_core/rpa/artifacts.py new file mode 100644 index 0000000..92482b3 --- /dev/null +++ b/src/jiangchang_skill_core/rpa/artifacts.py @@ -0,0 +1,28 @@ +"""RPA 失败存证路径与截图。""" + +from __future__ import annotations + +import os +from datetime import datetime + + +def _artifacts_enabled() -> bool: + v = (os.environ.get("OPENCLAW_ARTIFACTS_ON_FAILURE") or "1").strip().lower() + return v not in ("0", "false", "no", "off") + + +def artifact_path(data_dir: str, batch_id: str, tag: str, ext: str = "png") -> str: + """返回 {data_dir}/rpa-artifacts/{batch_id}/{tag}_{timestamp}.{ext}""" + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + directory = os.path.join(data_dir, "rpa-artifacts", batch_id) + os.makedirs(directory, exist_ok=True) + return os.path.join(directory, f"{tag}_{ts}.{ext}") + + +async def capture_failure(page, data_dir: str, batch_id: str, tag: str) -> str: + """受 OPENCLAW_ARTIFACTS_ON_FAILURE 控制，截图并返回路径；禁用时返回空字符串。""" + if not _artifacts_enabled(): + return "" + path = artifact_path(data_dir, batch_id, tag, ext="png") + await page.screenshot(path=path, full_page=True) + return path diff --git a/src/jiangchang_skill_core/rpa/browser.py b/src/jiangchang_skill_core/rpa/browser.py new file mode 100644 index 0000000..86cb8a0 --- /dev/null +++ b/src/jiangchang_skill_core/rpa/browser.py @@ -0,0 +1,61 @@ +"""统一 persistent context 启动封装。""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING + +from ..runtime_env import find_chrome_executable +from .stealth import ( + STEALTH_INIT_SCRIPT, + persistent_context_launch_parts, + stealth_enabled, +) + +if TYPE_CHECKING: + from playwright.async_api import BrowserContext + + +async def launch_persistent_browser( + playwright, + *, + profile_dir: str, + channel: str = "chrome", + executable_path: str | None = None, + headless: bool | None = None, + extra_args: list[str] | None = None, + record_video_dir: str | None = None, +) -> "BrowserContext": + """用 stealth 参数启动 persistent context + 注入 STEALTH_INIT_SCRIPT。 + + headless=None 时读 OPENCLAW_BROWSER_HEADLESS。 + record_video_dir 非空则开录屏。 + """ + if headless is None: + v = (os.environ.get("OPENCLAW_BROWSER_HEADLESS") or "0").strip().lower() + headless = v in ("1", "true", "yes", "on") + + chrome = executable_path or find_chrome_executable() + if not chrome: + raise RuntimeError( + "ERROR:MISSING_BROWSER 未检测到 Chrome 或 Edge 浏览器，请安装后重试。" + ) + + args, ignore = persistent_context_launch_parts(extra_args=extra_args) + launch_kwargs: dict = dict( + user_data_dir=profile_dir, + headless=headless, + executable_path=chrome, + locale="zh-CN", + no_viewport=True, + args=args, + ) + if ignore is not None: + launch_kwargs["ignore_default_args"] = ignore + if record_video_dir: + launch_kwargs["record_video_dir"] = record_video_dir + + context = await playwright.chromium.launch_persistent_context(**launch_kwargs) + if stealth_enabled(): + await context.add_init_script(STEALTH_INIT_SCRIPT) + return context diff --git a/src/jiangchang_skill_core/rpa/errors.py b/src/jiangchang_skill_core/rpa/errors.py new file mode 100644 index 0000000..151bbff --- /dev/null +++ b/src/jiangchang_skill_core/rpa/errors.py @@ -0,0 +1,19 @@ +"""RPA 场景统一错误码（stdout / 异常消息前缀）。""" + +ERR_REQUIRE_LOGIN = "ERROR:REQUIRE_LOGIN" +ERR_LOGIN_TIMEOUT = "ERROR:LOGIN_TIMEOUT" +ERR_CAPTCHA_NEED_HUMAN = "ERROR:CAPTCHA_NEED_HUMAN" +ERR_RATE_LIMITED = "ERROR:RATE_LIMITED" +ERR_MISSING_BROWSER = "ERROR:MISSING_BROWSER" +ERR_DEVICE_NOT_READY = "ERROR:DEVICE_NOT_READY" +ERR_WINDOW_NOT_FOUND = "ERROR:WINDOW_NOT_FOUND" + +__all__ = [ + "ERR_REQUIRE_LOGIN", + "ERR_LOGIN_TIMEOUT", + "ERR_CAPTCHA_NEED_HUMAN", + "ERR_RATE_LIMITED", + "ERR_MISSING_BROWSER", + "ERR_DEVICE_NOT_READY", + "ERR_WINDOW_NOT_FOUND", +] diff --git a/src/jiangchang_skill_core/rpa/human_login.py b/src/jiangchang_skill_core/rpa/human_login.py new file mode 100644 index 0000000..7eabb8a --- /dev/null +++ b/src/jiangchang_skill_core/rpa/human_login.py @@ -0,0 +1,96 @@ +"""验证码/登录 HITL 等待（站点无关通用部分）。""" + +from __future__ import annotations + +import asyncio + +DEFAULT_CAPTCHA_MARKERS = ( + "punish", + "baxia", + "nc_login", + "tmd", + "x5sec", + "安全验证", + "请按住滑块", + "拖动到最右边", +) + + +async def wait_for_captcha_pass( + page, + captcha_markers: tuple[str, ...] = DEFAULT_CAPTCHA_MARKERS, + timeout_sec: int = 120, +) -> bool: + """ + 检测到验证码/拦截页时，暂停等待用户手工完成验证。 + 每 3 秒轮询一次页面状态，直到页面不再是拦截状态或超时。 + + Returns: + True — 验证码已通过，可以继续 + False — 超时，用户未在规定时间内完成验证 + """ + print( + f"[验证码] 检测到安全拦截，请在浏览器中手工完成验证。" + f"等待最多 {timeout_sec} 秒..." + ) + + elapsed = 0 + while elapsed < timeout_sec: + await asyncio.sleep(3) + elapsed += 3 + try: + url = page.url or "" + text = "" + try: + text = await page.locator("body").inner_text(timeout=2_000) + except Exception: + pass + haystack = (url + " " + text[:1000]).lower() + still_blocked = any(m.lower() in haystack for m in captcha_markers) + if not still_blocked: + print(f"[验证码] 已通过！({elapsed}s 后检测到正常页面)") + return True + except Exception: + pass + + print(f"[验证码] 超时 {timeout_sec}s，用户未完成验证") + return False + + +async def wait_for_login( + page, + success_selectors: list[str], + timeout_sec: int = 180, +) -> bool: + """ + 轮询等待登录成功标志出现（站点专属选择器由调用方传入）。 + + Args: + page: Playwright Page 对象。 + success_selectors: 登录成功后应出现的 CSS 选择器列表。 + timeout_sec: 最长等待秒数。 + + Returns: + True — 检测到登录成功标志 + False — 超时 + """ + print( + f"[登录] 请在浏览器中完成登录。等待最多 {timeout_sec} 秒，" + "期间程序不会操作页面..." + ) + + elapsed = 0 + while elapsed < timeout_sec: + await asyncio.sleep(2) + elapsed += 2 + for selector in success_selectors: + try: + el = await page.query_selector(selector) + if el is not None: + print("[登录] 检测到登录成功！") + return True + except Exception: + pass + + print(f"[登录] 超时 {timeout_sec}s，未完成登录") + return False diff --git a/src/jiangchang_skill_core/rpa/stealth.py b/src/jiangchang_skill_core/rpa/stealth.py new file mode 100644 index 0000000..104798a --- /dev/null +++ b/src/jiangchang_skill_core/rpa/stealth.py @@ -0,0 +1,66 @@ +"""淡化 Playwright 自动化指纹，降低 baxia/x5sec 风控命中率。 + +关闭：环境变量 OPENCLAW_PLAYWRIGHT_STEALTH=0（或 false/off/no）。 +""" + +from __future__ import annotations + +import os + +STEALTH_INIT_SCRIPT = """ +(() => { + try { Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); } catch (e) {} + try { + window.chrome = window.chrome || {}; + window.chrome.runtime = window.chrome.runtime || {}; + } catch (e) {} + try { + const orig = navigator.permissions && navigator.permissions.query; + if (orig) { + navigator.permissions.query = (p) => + p && p.name === 'notifications' + ? Promise.resolve({ state: Notification.permission }) + : orig(p); + } + } catch (e) {} + try { + Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); + } catch (e) {} + try { + Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh'] }); + } catch (e) {} + try { + Object.defineProperty(navigator, 'hardwareConcurrency', { get: () => 8 }); + } catch (e) {} + try { + Object.defineProperty(navigator, 'deviceMemory', { get: () => 8 }); + } catch (e) {} + try { + const getParameter = WebGLRenderingContext.prototype.getParameter; + WebGLRenderingContext.prototype.getParameter = function (p) { + if (p === 37445) return 'Intel Inc.'; + if (p === 37446) return 'Intel Iris OpenGL Engine'; + return getParameter.call(this, p); + }; + } catch (e) {} +})(); +""" + + +def stealth_enabled() -> bool: + v = (os.environ.get("OPENCLAW_PLAYWRIGHT_STEALTH") or "1").strip().lower() + return v not in ("0", "false", "no", "off") + + +def persistent_context_launch_parts( + *, extra_args: list[str] | None = None +) -> tuple[list[str], list[str] | None]: + """返回 (args, ignore_default_args)。ignore_default_args 为 None 时不要传给 launch。""" + args = ["--start-maximized"] + if extra_args: + args = args + list(extra_args) + if not stealth_enabled(): + return args, None + if "--disable-blink-features=AutomationControlled" not in args: + args.append("--disable-blink-features=AutomationControlled") + return args, ["--enable-automation"] diff --git a/tests/test_anti_detect_import.py b/tests/test_anti_detect_import.py new file mode 100644 index 0000000..13fd8f4 --- /dev/null +++ b/tests/test_anti_detect_import.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +"""anti_detect / rpa 子包导入冒烟。""" +from __future__ import annotations + +import importlib +import unittest + + +class TestAntiDetectImport(unittest.TestCase): + def test_import_anti_detect(self) -> None: + mod = importlib.import_module("jiangchang_skill_core.rpa.anti_detect") + self.assertTrue(callable(mod.human_click)) + self.assertTrue(callable(mod.random_delay)) + + def test_import_rpa_package(self) -> None: + mod = importlib.import_module("jiangchang_skill_core.rpa") + self.assertTrue(callable(mod.wait_for_captcha_pass)) + self.assertIsNotNone(mod.stealth_enabled) + + def test_top_level_import_with_rpa(self) -> None: + pkg = importlib.import_module("jiangchang_skill_core") + self.assertIsNotNone(getattr(pkg, "config", None)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..4abc86b --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,86 @@ +# -*- coding: utf-8 -*- +"""config 三级优先级与 ensure_env_file 测试。""" +from __future__ import annotations + +import os +import tempfile +import unittest + +from jiangchang_skill_core import config + + +class TestConfig(unittest.TestCase): + def setUp(self) -> None: + config.reset_cache() + self._saved_env = dict(os.environ) + + def tearDown(self) -> None: + os.environ.clear() + os.environ.update(self._saved_env) + config.reset_cache() + + def test_ensure_env_file_copies_once(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + example = os.path.join(tmp, ".env.example") + with open(example, "w", encoding="utf-8") as f: + f.write("FOO=from_example\nBAR=2\n") + + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_USER_ID"] = "testuser" + + dest = config.ensure_env_file("demo-skill", example) + self.assertTrue(os.path.isfile(dest)) + with open(dest, encoding="utf-8") as f: + content = f.read() + self.assertIn("FOO=from_example", content) + + with open(dest, "w", encoding="utf-8") as f: + f.write("FOO=user_edited\n") + + dest2 = config.ensure_env_file("demo-skill", example) + self.assertEqual(dest, dest2) + with open(dest2, encoding="utf-8") as f: + self.assertIn("user_edited", f.read()) + + def test_three_level_priority(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + example = os.path.join(tmp, ".env.example") + with open(example, "w", encoding="utf-8") as f: + f.write("KEY_A=example\nKEY_B=example\nKEY_C=example\n") + + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_USER_ID"] = "u1" + dest = config.ensure_env_file("sk", example) + + with open(dest, "w", encoding="utf-8") as f: + f.write("KEY_B=user_env\n") + + config.reset_cache() + self.assertEqual(config.get("KEY_A"), "example") + self.assertEqual(config.get("KEY_B"), "user_env") + self.assertEqual(config.get("KEY_C"), "example") + self.assertIsNone(config.get("MISSING")) + + os.environ["KEY_C"] = "process_env" + self.assertEqual(config.get("KEY_C"), "process_env") + + def test_get_bool_float_int(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + example = os.path.join(tmp, ".env.example") + with open(example, "w", encoding="utf-8") as f: + f.write("FLAG=1\nRATE=2.5\nCOUNT=42\nBAD=x\n") + + os.environ["JIANGCHANG_DATA_ROOT"] = tmp + os.environ["JIANGCHANG_USER_ID"] = "u1" + config.ensure_env_file("sk", example) + + self.assertTrue(config.get_bool("FLAG")) + self.assertFalse(config.get_bool("MISSING", default=False)) + self.assertAlmostEqual(config.get_float("RATE", 0.0), 2.5) + self.assertEqual(config.get_int("COUNT", 0), 42) + self.assertEqual(config.get_float("BAD", 9.9), 9.9) + self.assertEqual(config.get_int("BAD", 7), 7) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_stealth.py b/tests/test_stealth.py new file mode 100644 index 0000000..588b065 --- /dev/null +++ b/tests/test_stealth.py @@ -0,0 +1,54 @@ +# -*- coding: utf-8 -*- +"""stealth 启动参数与开关测试。""" +from __future__ import annotations + +import os +import unittest + +from jiangchang_skill_core.rpa.stealth import ( + persistent_context_launch_parts, + stealth_enabled, +) + + +class TestStealth(unittest.TestCase): + def setUp(self) -> None: + self._saved = os.environ.get("OPENCLAW_PLAYWRIGHT_STEALTH") + + def tearDown(self) -> None: + if self._saved is None: + os.environ.pop("OPENCLAW_PLAYWRIGHT_STEALTH", None) + else: + os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = self._saved + + def test_stealth_enabled_default_on(self) -> None: + os.environ.pop("OPENCLAW_PLAYWRIGHT_STEALTH", None) + self.assertTrue(stealth_enabled()) + + def test_stealth_enabled_off_values(self) -> None: + for off in ("0", "false", "no", "off", "FALSE"): + os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = off + self.assertFalse(stealth_enabled(), msg=off) + + def test_launch_parts_stealth_on(self) -> None: + os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = "1" + args, ignore = persistent_context_launch_parts() + self.assertIn("--start-maximized", args) + self.assertIn("--disable-blink-features=AutomationControlled", args) + self.assertEqual(ignore, ["--enable-automation"]) + + def test_launch_parts_stealth_off(self) -> None: + os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = "0" + args, ignore = persistent_context_launch_parts() + self.assertIn("--start-maximized", args) + self.assertNotIn("--disable-blink-features=AutomationControlled", args) + self.assertIsNone(ignore) + + def test_launch_parts_extra_args(self) -> None: + os.environ["OPENCLAW_PLAYWRIGHT_STEALTH"] = "1" + args, _ = persistent_context_launch_parts(extra_args=["--foo"]) + self.assertIn("--foo", args) + + +if __name__ == "__main__": + unittest.main()