Python SDK

pip install https://pub-69fcb37602174d10b2152f09439de470.r2.dev/sdk/linege_sdk-0.3.21-py3-none-any.whl

python -c "import linege_sdk; print(linege_sdk.__version__)"
# 输出: 0.3.21

登录 → 申请环境(envId) → 分配代理+指纹 → 启动 Chrome → sidecar 连接 → 业务操作 → 自动清理

from linege_sdk import LinegeClient

client = LinegeClient(username="你的用户名", password="你的密码")

# 查看配额
profile = client.get_user_profile()
print(f"配额: {profile['used']}/{profile['total']}")

# 查看可用国家
countries = client.list_available_countries()
print(countries)  # [{"code": "US", "proxies_available": 42, ...}, ...]

# 打开浏览器 — 内部自动: 申请环境→拿envId→分配代理+指纹→启动Chrome
with client.open_cloud_browser(
    start_url="https://www.wikipedia.org",
    preferred_countries=["US", "JP"],  # 优先选这些国家的代理
) as browser:
    browser.find("input#searchInput").wait().input("hello")
    browser.find("button[type='submit']").click()  # 表单提交用按钮点击
    print(browser.find("#firstHeading").text())

# 创建本地环境，指定代理，自动通过 gost 挂载
with client.open_local_browser(
    env_id="shop-account-01",
    proxy="socks5://user:pass@proxy-host:1080",
    start_url="https://example.com",
) as browser:
    print(browser.find("h1").text())

from linege_sdk import list_local_environments

# 列出本地所有环境（和桌面端看到的一样）
envs = list_local_environments()
for env in envs:
    print(f"{env['id']}  proxy={env['proxy']}  country={env['country']}")

# 用已有环境的 envId 启动
target = envs[0]
with client.open_local_browser(
    env_id=target["id"],
    proxy=target["proxy"] or None,
    start_url="https://example.com",
) as browser:
    print(browser.find("h1").text())

# 限定标签
browser.find_by_text("提交", tag="button").click()

# 精确匹配
browser.find_by_text("登录", exact=True).click()

# 多语言候选 — 不管页面显示哪种语言，任一匹配即可
browser.find_by_text(["Login", "登录", "ログイン"], tag="button").click()

browser.find("input#email").wait().input("user@example.com")
browser.find("input#password").click().input("secret123")
browser.find("button[type='submit']").click()  # 表单提交用按钮点击
title = browser.find("h1").text()

# 站点跳转 + 等待 + URL 校验（host 模式适合反爬重定向场景）
browser.goto(
    "https://polymarket.com/event/bitcoin-above-on-march-4",
    timeout=45,
    wait_until="domcontentloaded",
    ensure_url=True,
    match="host",
)

browser.navigate(
    "https://polymarket.com/event/bitcoin-above-on-march-4",
    wait=True,
    ensure_url=True,
    match="host",
)

# 主标签：Polymarket
browser.goto("https://polymarket.com", match="host")

# 新开标签：Outlook
outlook_tab = browser.new_tab("https://outlook.live.com")
browser.switch_tab(outlook_tab)
browser.wait_for_url("https://outlook.live.com", match="host")

# 切回主标签
browser.switch_tab(0)
browser.wait_for_url("https://polymarket.com", match="host")

# 在弹窗内精确点击按钮（限定作用域到 [role=dialog]）
result = browser.eval_js("""
    (function() {
        var dialog = document.querySelector('[role=dialog]');
        if (!dialog) return null;
        var btns = dialog.querySelectorAll('button');
        for (var i = 0; i < btns.length; i++) {
            if (btns[i].textContent.trim() === 'Continue' && !btns[i].disabled) {
                btns[i].click();
                return 'OK';
            }
        }
        return 'not found';
    })()
""")

"""
Phantomshield SDK 完整示例（面向对象 + 多实例真实业务流）
=================================================

定位:
  - 面向业务开发者：账号密码登录后，直接跑一个完整业务编排
  - 结构是 OOP：一个编排类管理预检、业务流、多实例与异常护栏
  - 不包含邮箱验证码等条件分支，默认只要求账号密码

运行:
  python examples/example_full.py --username alice --password secret
  LINEGE_USERNAME=alice LINEGE_PASSWORD=secret python examples/example_full.py
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
import uuid
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Any, Callable, Dict, List, Optional

from linege_sdk import LinegeClient, LinegeSDKError, list_local_environments

def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")

def jlog(trace_id: str, event: str, status: str = "OK", **payload: Any) -> None:
    print(
        json.dumps(
            {
                "ts": now_iso(),
                "trace_id": trace_id,
                "event": event,
                "status": status,
                **payload,
            },
            ensure_ascii=False,
        )
    )

@dataclass
class MultiInstanceTask:
    env_id: str
    start_url: str
    next_url: str

@dataclass
class MultiInstanceResult:
    env_id: str
    start_url: str
    title_before: str
    url_before: str
    title_after: str
    url_after: str

class FullBusinessWorkflow:
    def __init__(self, username: str, password: str) -> None:
        self.trace_id = f"full-{uuid.uuid4().hex[:10]}"
        self.client = LinegeClient(username=username, password=password)
        self.report: Dict[str, Any] = {
            "trace_id": self.trace_id,
            "status": "running",
            "steps": [],
            "artifacts": {},
        }

    def run(self) -> int:
        jlog(self.trace_id, "flow.start")
        try:
            self._run_step("preflight.profile", self._preflight_profile, required=True)
            self._run_step("preflight.core", self._preflight_core, required=True)
            self._run_step("preflight.countries", self._preflight_countries, required=True)
            self._run_step("preflight.local_envs", self._preflight_local_envs, required=False)

            self._run_step("biz.cloud_research", self._biz_cloud_research, required=True)
            self._run_step("biz.local_workspace", self._biz_local_workspace, required=True)
            self._run_step("biz.multi_instance_watch", self._biz_multi_instance_watch, required=True)
            self._run_step("biz.error_guardrail", self._biz_error_guardrail, required=True)

            self.report["status"] = "ok"
            jlog(self.trace_id, "flow.success", step_count=len(self.report["steps"]))
            return 0
        except LinegeSDKError as exc:
            self.report["status"] = "failed"
            self.report["error"] = {
                "type": "LinegeSDKError",
                "code": exc.code,
                "message": exc.message,
                "hint": exc.hint,
                "detail": exc.detail,
            }
            jlog(self.trace_id, "flow.failed", status="ERROR", code=exc.code, message=exc.message)
            return 1
        except Exception as exc:
            self.report["status"] = "failed"
            self.report["error"] = {
                "type": type(exc).__name__,
                "message": str(exc),
            }
            jlog(self.trace_id, "flow.failed", status="ERROR", error_type=type(exc).__name__, message=str(exc))
            return 1
        finally:
            try:
                self.client.close_all()
            except Exception:
                pass
            jlog(self.trace_id, "flow.finish", status=self.report["status"])
            print(json.dumps(self.report, ensure_ascii=False, indent=2))

    def _run_step(self, name: str, fn: Callable[[], Any], required: bool) -> Any:
        started = time.perf_counter()
        try:
            value = fn()
            elapsed_ms = round((time.perf_counter() - started) * 1000)
            step_record = {
                "name": name,
                "required": required,
                "ok": True,
                "elapsed_ms": elapsed_ms,
                "value": value,
            }
            self.report["steps"].append(step_record)
            jlog(self.trace_id, "step.ok", step=name, elapsed_ms=elapsed_ms)
            return value
        except Exception as exc:
            elapsed_ms = round((time.perf_counter() - started) * 1000)
            step_record = {
                "name": name,
                "required": required,
                "ok": False,
                "elapsed_ms": elapsed_ms,
                "error": str(exc),
            }
            self.report["steps"].append(step_record)
            jlog(self.trace_id, "step.fail", status="ERROR", step=name, elapsed_ms=elapsed_ms, error=str(exc))
            if required:
                raise
            return None

    def _preflight_profile(self) -> Dict[str, Any]:
        profile = self.client.get_user_profile()
        return {
            "username": profile["username"],
            "quota": {
                "used": profile["used"],
                "total": profile["total"],
                "remaining": profile["remaining"],
            },
        }

    def _preflight_core(self) -> Dict[str, Any]:
        core = self.client.ensure_core_latest()
        return {
            "updated": core["updated"],
            "local_version": core["local_version"],
            "remote_version": core["remote_version"],
        }

    def _preflight_countries(self) -> Dict[str, Any]:
        countries = self.client.list_available_countries(preferred=["US", "JP"])
        top = [
            {
                "code": c["code"],
                "name": c["name"],
                "proxies_available": c["proxies_available"],
                "preferred": c["preferred"],
            }
            for c in countries[:8]
        ]
        return {
            "count": len(countries),
            "top": top,
        }

    def _preflight_local_envs(self) -> Dict[str, Any]:
        envs = list_local_environments()
        return {
            "count": len(envs),
            "sample": [
                {
                    "id": row.get("id"),
                    "proxy": bool(row.get("proxy")),
                    "country": row.get("country") or "-",
                }
                for row in envs[:8]
            ],
        }

    def _biz_cloud_research(self) -> Dict[str, Any]:
        with self.client.open_cloud_browser(
            start_url="https://www.wikipedia.org",
            preferred_countries=["US", "JP"],
        ) as browser:
            query = "Browser automation"
            browser.find("input#searchInput").wait(timeout=30).input(query)
            browser.find("button[type='submit'], input.searchButton, button.pure-button").click()

            heading = browser.find("#firstHeading").wait(timeout=30).text() or ""
            summary = (
                browser.find_by_xpath(
                    "//div[contains(@class,'mw-parser-output')]//p[string-length(normalize-space()) > 30][1]"
                ).text()
                or ""
            )
            refs_visible = browser.find_by_text(["References", "参考文献", "脚注"], tag="span").is_visible()
            browser.find("body").scroll(delta_y=500)

            result = {
                "env_id": browser.env.env_id,
                "mode": browser.env.mode,
                "country": getattr(browser.env, "country", None),
                "title": browser.eval_js("document.title"),
                "url": browser.eval_js("window.location.href"),
                "heading": heading,
                "summary_preview": summary[:160],
                "references_visible": refs_visible,
            }
            self.report["artifacts"]["cloud_env_id"] = browser.env.env_id
            return result

    def _biz_local_workspace(self) -> Dict[str, Any]:
        with self.client.open_local_browser(
            env_id="biz-workspace-main",
            start_url="https://example.com",
        ) as browser:
            homepage_h1 = browser.find("h1").wait(timeout=20).text() or ""
            homepage_link = browser.find("a").wait(timeout=10).attribute("href")
            homepage_visible = browser.find("h1").is_visible()

            browser.goto("https://www.wikipedia.org", match="host")
            input_box = browser.find("input#searchInput").wait(timeout=30)
            input_box.click()
            input_box.press_key("L")
            browser.input("input#searchInput", "inege SDK", human=True)
            browser.find("button[type='submit'], input.searchButton, button.pure-button").click()
            browser.find("#firstHeading").wait(timeout=30)

            # 多标签业务流：同一环境切到 Outlook 再切回主业务站点
            outlook_tab = browser.new_tab("https://outlook.live.com", activate=True)
            browser.switch_tab(outlook_tab)
            outlook_hosts = [
                "https://outlook.live.com",
                "https://login.live.com",
                "https://www.microsoft.com",
            ]
            outlook_ready = browser.wait_for_url(
                outlook_hosts,
                match="host",
                timeout=30,
                required=False,
            )
            tabs_after_open = browser.list_tabs()
            browser.switch_tab(0)
            wiki_ready = browser.wait_for_url("https://wikipedia.org", match="host", timeout=30, required=False)

            browser.find("body").scroll(delta_y=600)
            result_title = browser.find("#firstHeading").text() or ""
            page_title = browser.eval_js("document.title")
            page_url = browser.eval_js("window.location.href")

            return {
                "env_id": browser.env.env_id,
                "mode": browser.env.mode,
                "homepage_h1": homepage_h1,
                "homepage_link": homepage_link,
                "homepage_visible": homepage_visible,
                "result_title": result_title,
                "page_title": page_title,
                "page_url": page_url,
                "outlook_tab_index": outlook_tab,
                "outlook_ready": outlook_ready,
                "wiki_ready_after_switch": wiki_ready,
                "tabs_after_open": tabs_after_open,
            }

    def _biz_multi_instance_watch(self) -> List[Dict[str, Any]]:
        tasks = [
            MultiInstanceTask(env_id="watch-us", start_url="https://example.com", next_url="https://www.wikipedia.org"),
            MultiInstanceTask(env_id="watch-jp", start_url="https://www.wikipedia.org", next_url="https://httpbin.org/html"),
            MultiInstanceTask(env_id="watch-eu", start_url="https://httpbin.org/html", next_url="https://example.com"),
        ]

        sessions: List[tuple[MultiInstanceTask, Any]] = []
        results: List[MultiInstanceResult] = []
        try:
            for task in tasks:
                session = self.client.open_local_browser(env_id=task.env_id, start_url=task.start_url)
                sessions.append((task, session))

            for task, session in sessions:
                session.wait("h1, h2, body", timeout=30, required=False)
                title_before = str(session.eval_js("document.title", required=False) or "")
                url_before = str(session.eval_js("window.location.href", required=False) or "")
                session.goto(task.next_url, timeout=30, ensure_url=True, match="host", required=False)

                session.wait("h1, h2, body", timeout=30, required=False)
                title_after = str(session.eval_js("document.title", required=False) or "")
                url_after = str(session.eval_js("window.location.href", required=False) or "")

                results.append(
                    MultiInstanceResult(
                        env_id=task.env_id,
                        start_url=task.start_url,
                        title_before=title_before,
                        url_before=url_before,
                        title_after=title_after,
                        url_after=url_after,
                    )
                )
        finally:
            for _, session in sessions:
                try:
                    session.close()
                except Exception:
                    pass

        return [asdict(row) for row in results]

    def _biz_error_guardrail(self) -> Dict[str, Any]:
        bad_auth: Dict[str, Any] = {}
        try:
            LinegeClient(username="wrong_user", password="bad_password").get_user_profile()
        except LinegeSDKError as exc:
            bad_auth = {
                "code": exc.code,
                "message": exc.message,
                "hint": exc.hint,
            }

        with self.client.open_local_browser(
            env_id="guardrail-silent",
            start_url="https://example.com",
        ) as browser:
            browser.find("h1").wait(timeout=10)
            silent_click = browser.click("#nonexistent-element", required=False, timeout=3)
            silent_text = browser.text("#nonexistent-element", required=False, timeout=3)

        strict_error: Dict[str, Any] = {}
        try:
            with self.client.open_local_browser(
                env_id="guardrail-strict",
                start_url="https://example.com",
            ) as browser:
                browser.find("h1").wait(timeout=10)
                browser.click("#nonexistent-element", required=True, timeout=3)
        except LinegeSDKError as exc:
            strict_error = {
                "code": exc.code,
                "message": exc.message,
                "hint": exc.hint,
            }

        return {
            "bad_auth": bad_auth,
            "silent_mode": {
                "click_result": silent_click,
                "text_result": silent_text,
            },
            "strict_mode": strict_error,
        }

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Phantomshield SDK 面向对象多实例完整业务示例")
    parser.add_argument("--username", default=os.environ.get("LINEGE_USERNAME", ""))
    parser.add_argument("--password", default=os.environ.get("LINEGE_PASSWORD", ""))
    return parser.parse_args()

def main() -> int:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    if hasattr(sys.stderr, "reconfigure"):
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")

    args = parse_args()
    if not args.username or not args.password:
        print("缺少凭据：请设置 --username/--password 或 LINEGE_USERNAME/LINEGE_PASSWORD")
        return 2

    workflow = FullBusinessWorkflow(username=args.username, password=args.password)
    return workflow.run()

if __name__ == "__main__":
    raise SystemExit(main())

# 1) 创建隔离环境
python -m venv .venv
# Windows
.venv\Scripts\python -m pip install --upgrade pip
.venv\Scripts\python -m pip install "https://pub-69fcb37602174d10b2152f09439de470.r2.dev/sdk/linege_sdk-0.3.21-py3-none-any.whl"

# 2) 下载完整示例脚本（站内）
powershell -Command "Invoke-WebRequest -Uri https://docs.sybilslayer.com/examples/example_full.py -OutFile example_full.py"

# 3) 准备凭据并运行 full example（账号密码模式）
set LINEGE_USERNAME=你的账号
set LINEGE_PASSWORD=你的密码
.venv\Scripts\python example_full.py

client = LinegeClient(
    username="...",                  # 或 token="..."
    password="...",
    auto_cleanup_cloud_env=True,     # with 结束自动删除云端环境
    enable_logging=True,             # 结构化 JSON 日志
    sidecar_base_port=7788,          # 端口被占自动递增
    max_startup_retries=1,           # 启动失败后重试次数
    desktop_compatible=False,        # True 时使用桌面端兼容预设
    chrome_path=None,                # 自定义 Chrome 路径
    render_mode="software",          # "software" 或 "gpu"
    http_cache_mode="isolated",      # "isolated" / "global" / "off"
    shared_http_cache_root=None,     # 共享 HTTP 缓存目录
    shared_http_cache_size_bytes=None, # 共享缓存大小（字节）
    restore_last_session=None,       # 恢复上次会话（默认跟随 desktop_compatible）
)

{
    "username": "your_username",
    "used": 5,
    "total": 50,
    "remaining": 45,
}