From 98ca8a3965b94c827232c808c1a40bfcb92bc8c2 Mon Sep 17 00:00:00 2001 From: ivanberry Date: Wed, 11 Mar 2026 20:26:29 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=B0=86=20auth=20runtime=20?= =?UTF-8?q?=E9=87=8D=E6=9E=84=E4=B8=BA=E5=8F=AF=E5=A4=8D=E7=94=A8=20Python?= =?UTF-8?q?=20=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 模式参考 ~/clawd/skills/_shared/auth-runtime (TypeScript): - 创建 python_auth_runtime/ 作为独立 Python 包 - 其他 skill 可以通过 uv pip install 引用 - 支持三种使用方式： 1. 本地包安装：uv pip install /path/to/python_auth_runtime 2. 文件依赖：pyproject.toml 中引用 3. 复制源码：直接复制 src/python_auth_runtime 包结构： python_auth_runtime/ ├── pyproject.toml # 包配置 ├── README.md # 使用文档 └── src/python_auth_runtime/ └── __init__.py # 核心实现 功能： - CLIENT_KEY 认证 - 令牌缓存（可配置 TTL） - 自动刷新过期令牌 - 401/403 自动重试 - 从环境变量加载配置 --- pyproject.toml | 3 + python_auth_runtime/README.md | 163 +++++++++++++++++ python_auth_runtime/pyproject.toml | 19 ++ .../src/python_auth_runtime/__init__.py | 165 +++++------------- scripts/test_auth.py | 75 -------- 5 files changed, 226 insertions(+), 199 deletions(-) create mode 100644 python_auth_runtime/README.md create mode 100644 python_auth_runtime/pyproject.toml rename scripts/auth_runtime.py => python_auth_runtime/src/python_auth_runtime/__init__.py (67%) delete mode 100644 scripts/test_auth.py diff --git a/pyproject.toml b/pyproject.toml index 5af097d..73bf280 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,9 @@ dependencies = [ "pandas>=1.5.0", "openpyxl>=3.0.0", "google-generativeai>=0.8.0", + "requests>=2.28.0", + # 引用本地 auth-runtime 包 + "python-auth-runtime-py @ file:///${PROJECT_ROOT}/python_auth_runtime", ] [tool.uv] diff --git a/python_auth_runtime/README.md b/python_auth_runtime/README.md new file mode 100644 index 0000000..c89458d --- /dev/null +++ b/python_auth_runtime/README.md @@ -0,0 +1,163 @@ +# @clawd/auth-runtime-py + +Python 版本的 OpenClaw Auth Runtime，基于 `~/clawd/skills/_shared/auth-runtime` 的 TypeScript 实现。 + +## 安装 + +### 方式 1: 作为本地包安装（推荐） + +```bash +# 在具体 skill 的目录中 +uv pip install /path/to/python_auth_runtime +``` + +### 方式 2: 作为文件依赖 + +在 `pyproject.toml` 中添加： + +```toml +[project] +dependencies = [ + "python-auth-runtime-py @ file:///path/to/python_auth_runtime", +] +``` + +### 方式 3: 复制源码 + +直接复制 `src/python_auth_runtime` 到你的项目中： + +```bash +cp -r /path/to/python_auth_runtime/src/python_auth_runtime ./your-skill/ +``` + +## 使用方式 + +### 1. 设置环境变量 + +```bash +export CLIENT_KEY="your-client-key" +export AUTH_BASE="https://api-gw-test.yuanwei-lnc.com" # 可选 +``` + +### 2. 导入并使用 + +```python +from python_auth_runtime import create_env_config, get_access_token, request_api_with_auto_refresh + +# 创建配置（自动从环境变量加载） +config = create_env_config() + +# 获取访问令牌（带缓存） +token = get_access_token(dry_run=False, config=config) + +# 发送 API 请求（自动处理令牌刷新） +response = request_api_with_auto_refresh( + method="POST", + url=f"{config.auth_base}/ecom/tasks/scrape", + dry_run=False, + config=config, + body={"url": "https://detail.1688.com/offer/123.html"}, +) + +if response.status == 200: + import json + data = json.loads(response.body) + print("商品价格:", data["price"]) +``` + +## API 参考 + +### 配置 + +#### `create_env_config() -> EnvConfig` + +从环境变量创建配置： + +| 环境变量 | 默认值 | 说明 | +|---------|--------|------| +| `AUTH_BASE` | `https://api-gw-test.yuanwei-lnc.com` | 认证基础 URL | +| `CLIENT_KEY` | **必需** | 客户端密钥 | +| `AUTH_CACHE_DIR` | `/tmp/skill-auth-cache` | 缓存目录 | +| `AUTH_MIN_TTL_SEC` | `60` | 最小令牌 TTL（秒） | + +### 令牌管理 + +#### `get_access_token(dry_run, config) -> str` + +获取访问令牌（带缓存） + +#### `refresh_access_token(dry_run, config) -> str` + +刷新访问令牌（绕过缓存） + +### API 请求 + +#### `request_api(method, url, auth_token, body) -> ApiResponse` + +发送 HTTP 请求 + +#### `request_api_with_auto_refresh(method, url, dry_run, config, body) -> ApiResponse` + +发送 API 请求并自动刷新令牌 + +## 与 TypeScript 版本对比 + +| 特性 | TypeScript | Python | +|------|-----------|--------| +| 包名 | `@clawd/auth-runtime` | `@clawd/auth-runtime-py` | +| 安装方式 | `bun add file:../_shared/auth-runtime` | `uv pip install /path/to/python_auth_runtime` | +| 导入 | `import { ... } from '@clawd/auth-runtime'` | `from python_auth_runtime import ...` | +| HTTP 客户端 | `fetch()` | `requests` | + +## 示例项目 + +```python +# your-skill/main.py +from python_auth_runtime import create_env_config, request_api_with_auto_refresh +import os + +def main(): + # 从环境变量加载 CLIENT_KEY + config = create_env_config() + + # 调用 1688 API + response = request_api_with_auto_refresh( + method="POST", + url=f"{config.auth_base}/ecom/tasks/scrape", + dry_run=False, + config=config, + body={"url": "https://detail.1688.com/offer/123.html"}, + ) + + if response.status == 200: + print("成功:", response.body) + else: + print("失败:", response.status, response.body) + +if __name__ == "__main__": + main() +``` + +## 环境变量配置 + +在项目根目录创建 `.env` 文件（不提交到 git）： + +```bash +# .env +CLIENT_KEY=your-client-key-here +AUTH_BASE=https://api-gw-test.yuanwei-lnc.com +``` + +然后在代码中加载： + +```python +from dotenv import load_dotenv +load_dotenv() # 加载 .env 文件到环境变量 + +from python_auth_runtime import create_env_config +config = create_env_config() +``` + +## 许可证 + +MIT diff --git a/python_auth_runtime/pyproject.toml b/python_auth_runtime/pyproject.toml new file mode 100644 index 0000000..1deff8a --- /dev/null +++ b/python_auth_runtime/pyproject.toml @@ -0,0 +1,19 @@ +[project] +name = "@clawd/auth-runtime-py" +version = "1.0.0" +description = "Python 版本的 OpenClaw Auth Runtime" +readme = "README.md" +requires-python = ">=3.9" +dependencies = [ + "requests>=2.28.0", +] + +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.uv] +index-url = "https://pypi.tuna.tsinghua.edu.cn/simple" diff --git a/scripts/auth_runtime.py b/python_auth_runtime/src/python_auth_runtime/__init__.py similarity index 67% rename from scripts/auth_runtime.py rename to python_auth_runtime/src/python_auth_runtime/__init__.py index 93df5ed..5a4a689 100644 --- a/scripts/auth_runtime.py +++ b/python_auth_runtime/src/python_auth_runtime/__init__.py @@ -1,14 +1,24 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ -Python 版本的 OpenClaw Auth Runtime -基于 ~/clawd/skills/_shared/auth-runtime 的 TypeScript 实现 +@clawd/auth-runtime-py - Python 版本的 OpenClaw Auth Runtime -功能： -- 使用 CLIENT_KEY 获取访问令牌 -- 支持令牌缓存（可配置 TTL） -- 自动刷新过期令牌 -- 401/403 自动重试 +基于 ~/clawd/skills/_shared/auth-runtime 的 TypeScript 实现， +提供 Python 版本的客户端密钥认证。 + +使用方式： + from python_auth_runtime import create_env_config, get_access_token, request_api_with_auto_refresh + + config = create_env_config() + token = get_access_token(dry_run=False, config=config) + + response = request_api_with_auto_refresh( + method="POST", + url=f"{config.auth_base}/your-endpoint", + dry_run=False, + config=config, + body={"key": "value"}, + ) """ import hashlib @@ -92,15 +102,10 @@ def create_env_config() -> EnvConfig: def get_cache_file(auth_base: str, client_key: str, cache_dir: str) -> Path: - """ - 获取缓存文件路径 - - 使用 auth_base 和 client_key 的哈希值生成唯一的缓存文件名 - """ + """获取缓存文件路径""" cache_path = Path(cache_dir) cache_path.mkdir(parents=True, exist_ok=True) - # 生成缓存文件名的哈希 key_str = f"{auth_base}:{client_key}" hash_value = hashlib.sha256(key_str.encode()).hexdigest()[:16] @@ -108,11 +113,7 @@ def get_cache_file(auth_base: str, client_key: str, cache_dir: str) -> Path: def read_cached_token(cache_file: Path, min_ttl_sec: int) -> Optional[str]: - """ - 读取缓存的令牌 - - 如果令牌存在且剩余 TTL 大于最小 TTL，则返回令牌 - """ + """读取缓存的令牌""" if not cache_file.exists(): return None @@ -125,26 +126,19 @@ def read_cached_token(cache_file: Path, min_ttl_sec: int) -> Optional[str]: expires_at=data["expires_at"], ) - # 检查是否过期（考虑最小 TTL） if time.time() + min_ttl_sec < cached.expires_at: return cached.access_token else: - # 令牌已过期或即将过期，删除缓存 cache_file.unlink(missing_ok=True) return None except (json.JSONDecodeError, KeyError, Exception): - # 缓存损坏，删除并重新获取 cache_file.unlink(missing_ok=True) return None def write_cache(cache_file: Path, session: SessionResponse) -> None: - """ - 写入缓存 - - 缓存令牌和过期时间 - """ + """写入缓存""" cache_file.parent.mkdir(parents=True, exist_ok=True) data = { @@ -163,11 +157,7 @@ def delete_cache(cache_file: Path) -> None: def fetch_session_json(dry_run: bool, config: EnvConfig) -> SessionResponse: - """ - 从认证端点获取会话 JSON - - 使用 CLIENT_KEY 请求访问令牌 - """ + """从认证端点获取会话 JSON""" if dry_run: return SessionResponse( access_token="", @@ -179,7 +169,6 @@ def fetch_session_json(dry_run: bool, config: EnvConfig) -> SessionResponse: if not config.client_key: raise ValueError("CLIENT_KEY is required") - # 请求认证端点 payload = {"clientKey": config.client_key} url = f"{config.auth_base}/auth/skill-credit/session" @@ -204,13 +193,7 @@ def fetch_session_json(dry_run: bool, config: EnvConfig) -> SessionResponse: def get_access_token(dry_run: bool, config: EnvConfig) -> str: - """ - 获取访问令牌（带缓存） - - 1. 检查缓存 - 2. 如果缓存有效，返回缓存的令牌 - 3. 否则请求新令牌并缓存 - """ + """获取访问令牌（带缓存）""" if dry_run: return "" @@ -221,23 +204,16 @@ def get_access_token(dry_run: bool, config: EnvConfig) -> str: cached_token = read_cached_token(cache_file, config.auth_min_ttl_sec) if cached_token: - print(f"✓ 使用缓存的访问令牌") return cached_token - print(f"🔑 请求新的访问令牌...") session = fetch_session_json(dry_run, config) write_cache(cache_file, session) - print(f"✓ 令牌已缓存到：{cache_file}") return session.access_token def refresh_access_token(dry_run: bool, config: EnvConfig) -> str: - """ - 刷新访问令牌（绕过缓存） - - 删除缓存并重新请求新令牌 - """ + """刷新访问令牌（绕过缓存）""" if dry_run: return "" @@ -246,17 +222,12 @@ def refresh_access_token(dry_run: bool, config: EnvConfig) -> str: cache_file = get_cache_file(config.auth_base, config.client_key, config.auth_cache_dir) delete_cache(cache_file) - print(f"🔄 刷新访问令牌...") return get_access_token(dry_run, config) def is_retryable_session_error(response: ApiResponse) -> bool: - """ - 检查响应是否表示会话过期/无效 - - 如果是 401/403 或响应体包含特定标记，则可以重试 - """ + """检查响应是否表示会话过期/无效""" if response.status not in RETRYABLE_STATUS: return False @@ -274,19 +245,7 @@ def request_api( body: Optional[dict[str, Any]] = None, timeout: int = 30, ) -> ApiResponse: - """ - 发送 HTTP 请求 - - Args: - method: HTTP 方法（GET, POST, PUT, DELETE 等） - url: 请求 URL - auth_token: 访问令牌（可选） - body: 请求体（可选，自动序列化为 JSON） - timeout: 超时时间（秒） - - Returns: - ApiResponse: 响应对象 - """ + """发送 HTTP 请求""" headers = {"Content-Type": "application/json"} if auth_token: @@ -296,13 +255,9 @@ def request_api( if method.upper() == "GET": response = requests.get(url, headers=headers, timeout=timeout) elif method.upper() == "POST": - response = requests.post( - url, headers=headers, json=body, timeout=timeout - ) + response = requests.post(url, headers=headers, json=body, timeout=timeout) elif method.upper() == "PUT": - response = requests.put( - url, headers=headers, json=body, timeout=timeout - ) + response = requests.put(url, headers=headers, json=body, timeout=timeout) elif method.upper() == "DELETE": response = requests.delete(url, headers=headers, timeout=timeout) else: @@ -330,66 +285,28 @@ def request_api_with_auto_refresh( body: Optional[dict[str, Any]] = None, access_token: Optional[str] = None, ) -> ApiResponse: - """ - 发送 API 请求并自动刷新令牌 - - 1. 使用当前令牌发送请求 - 2. 如果是 401/403 错误，刷新令牌并重试一次 - """ + """发送 API 请求并自动刷新令牌""" token = access_token or get_access_token(dry_run, config) - # 第一次请求 first = request_api(method, url, token, body) - # 检查是否需要重试 if not is_retryable_session_error(first): return first - # 刷新令牌并重试 - print(f"⚠️ 检测到会话过期，刷新令牌后重试...") fresh_token = refresh_access_token(dry_run, config) return request_api(method, url, fresh_token, body) -# 便捷函数：从 .env 文件加载 CLIENT_KEY -def load_client_key_from_env(env_path: Optional[Path] = None) -> str: - """ - 从 .env 文件加载 CLIENT_KEY - - 优先级： - 1. 环境变量 CLIENT_KEY - 2. .env 文件中的 CLIENT_KEY - 3. 抛出异常 - """ - # 先检查环境变量 - client_key = os.getenv("CLIENT_KEY") - if client_key: - return client_key - - # 查找 .env 文件 - if env_path is None: - possible_paths = [ - Path(".env"), - Path.cwd() / ".env", - Path(__file__).parent.parent / ".env", - ] - for p in possible_paths: - if p.exists(): - env_path = p - break - - if env_path and env_path.exists(): - with open(env_path, "r", encoding="utf-8") as f: - for line in f: - line = line.strip() - if not line or line.startswith("#"): - continue - if "=" in line: - key, value = line.split("=", 1) - if key.strip() == "CLIENT_KEY": - return value.strip() - - raise ValueError( - "CLIENT_KEY not found. Please set CLIENT_KEY environment variable " - "or add it to .env file" - ) +__version__ = "1.0.0" +__all__ = [ + "EnvConfig", + "SessionResponse", + "ApiResponse", + "create_env_config", + "get_access_token", + "refresh_access_token", + "fetch_session_json", + "request_api", + "request_api_with_auto_refresh", + "is_retryable_session_error", +] diff --git a/scripts/test_auth.py b/scripts/test_auth.py deleted file mode 100644 index 87ee36a..0000000 --- a/scripts/test_auth.py +++ /dev/null @@ -1,75 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -测试 OpenClaw Auth Runtime -""" - -import sys -from pathlib import Path - -# 添加当前目录到路径 -sys.path.insert(0, str(Path(__file__).parent)) - -from auth_runtime import ( - create_env_config, - get_access_token, - request_api_with_auto_refresh, - load_client_key_from_env, -) - - -def main(): - print("=" * 60) - print("OpenClaw Auth Runtime 测试") - print("=" * 60) - - # 方式 1: 从 .env 加载 CLIENT_KEY - try: - client_key = load_client_key_from_env() - print(f"✓ 从 .env 文件加载 CLIENT_KEY: {client_key[:10]}...") - except ValueError as e: - print(f"❌ {e}") - print("\n请在 .env 文件中设置 CLIENT_KEY") - print("或设置环境变量：export CLIENT_KEY=your-key") - return 1 - - # 创建配置 - config = create_env_config() - print(f"✓ 配置创建成功") - print(f" AUTH_BASE: {config.auth_base}") - print(f" AUTH_CACHE_DIR: {config.auth_cache_dir}") - - # 获取访问令牌 - print("\n【测试 1】获取访问令牌...") - try: - token = get_access_token(dry_run=False, config=config) - print(f"✓ 获取成功：{token[:20]}...") - except Exception as e: - print(f"❌ 失败：{e}") - return 1 - - # 测试 API 请求 - print("\n【测试 2】测试 API 请求...") - try: - # 示例：请求认证端点 - response = request_api_with_auto_refresh( - method="POST", - url=f"{config.auth_base}/auth/skill-credit/session", - dry_run=False, - config=config, - body={"clientKey": config.client_key}, - ) - print(f"✓ 响应状态：{response.status}") - print(f" 响应体：{response.body[:100]}...") - except Exception as e: - print(f"❌ 失败：{e}") - - print("\n" + "=" * 60) - print("✅ 测试完成") - print("=" * 60) - - return 0 - - -if __name__ == "__main__": - sys.exit(main())