Compare commits

...

38 Commits

Author SHA1 Message Date
75a332e7b8 release: docs, browser_service, runtime_env, SKILL v1.0.52
All checks were successful
技能自动化发布 / release (push) Successful in 28s
2026-04-08 14:00:42 +08:00
4605d3d0e4 release: runtime_env marker, SKILL v1.0.51
All checks were successful
技能自动化发布 / release (push) Successful in 25s
2026-04-08 12:05:14 +08:00
beb95c1415 release: v1.0.50 playwright stealth gemini
All checks were successful
技能自动化发布 / release (push) Successful in 34s
2026-04-07 15:18:14 +08:00
e2eafb7407 feat: remove wait-login; pick-web only; docs sync
All checks were successful
技能自动化发布 / release (push) Successful in 33s
2026-04-07 11:58:27 +08:00
96e5b9ea14 fix: spawn login/open child with python -m for PyArmor runtime path
All checks were successful
技能自动化发布 / release (push) Successful in 31s
2026-04-07 10:58:32 +08:00
97d970dd1b chore: account-manager — runtime_env sibling root, wait-login and browser fixes
All checks were successful
技能自动化发布 / release (push) Successful in 43s
2026-04-07 10:36:40 +08:00
e3da478116 feat: release ZIP keeps scripts/ tree; JIANGCHANG_SKILLS_ROOT and sibling skills root
All checks were successful
技能自动化发布 / release (push) Successful in 31s
2026-04-06 18:38:09 +08:00
ac76f89dc8 feat: wait-login 支持 --force,供搜狐发布等 Cookie 失效仍开窗检测
All checks were successful
技能自动化发布 / release (push) Successful in 31s
Made-with: Cursor
2026-04-06 17:30:50 +08:00
c83841cf90 chore: auto release commit (2026-04-06 16:44:58)
All checks were successful
技能自动化发布 / release (push) Successful in 39s
2026-04-06 16:44:59 +08:00
d938885a49 chore: auto release commit (2026-04-06 14:32:01)
All checks were successful
技能自动化发布 / release (push) Successful in 27s
2026-04-06 14:32:01 +08:00
5afad3285a chore: auto release commit (2026-04-06 12:57:26)
Some checks failed
技能自动化发布 / release (push) Failing after 21s
2026-04-06 12:57:26 +08:00
39daeaca59 chore: auto release commit (2026-04-06 12:10:31)
All checks were successful
技能自动化发布 / release (push) Successful in 13s
2026-04-06 12:10:31 +08:00
c428bedfd5 chore: auto release commit (2026-04-06 10:37:07)
All checks were successful
技能自动化发布 / release (push) Successful in 34s
2026-04-06 10:37:08 +08:00
5082da78ab chore: auto release commit (2026-04-06 10:16:00)
Some checks failed
技能自动化发布 / release (push) Failing after 13s
2026-04-06 10:16:00 +08:00
bfb9c15f13 ci: use kit @main + pass PYARMOR_REG_B64 for licensed CI obfuscation
Some checks failed
技能自动化发布 / release (push) Failing after 12s
Made-with: Cursor
2026-04-05 09:28:58 +08:00
6b4cdec1ec ci: pin reusable workflow to jiangchang-platform-kit@3ee09bb (avoid stale @main)
Some checks failed
技能自动化发布 / release (push) Failing after 12s
Made-with: Cursor
2026-04-05 09:25:05 +08:00
52cef5a7c0 chore: sync from OpenClaw workspace (2026-04-05 09:17)
Some checks failed
技能自动化发布 / release (push) Failing after 11s
2026-04-05 09:17:20 +08:00
80e7de6183 chore: merge origin/main; resolve conflicts
Some checks failed
技能自动化发布 / release (push) Failing after 11s
2026-04-05 08:53:50 +08:00
27cfe1018e chore: initialize account-manager skill repository 2026-04-05 08:36:41 +08:00
ba1c21c0b6 chore: auto release commit (2026-03-31 18:37:04)
All checks were successful
技能自动化发布 / release (push) Successful in 20s
2026-03-31 18:37:04 +08:00
58dc81449d chore: auto release commit (2026-03-31 18:19:36)
All checks were successful
技能自动化发布 / release (push) Successful in 14s
2026-03-31 18:19:37 +08:00
ae1da709ef chore: auto release commit (2026-03-31 17:00:42)
All checks were successful
技能自动化发布 / release (push) Successful in 15s
2026-03-31 17:00:42 +08:00
469b9c8bc7 chore: auto release commit (2026-03-31 15:24:21)
All checks were successful
技能自动化发布 / release (push) Successful in 15s
2026-03-31 15:24:22 +08:00
9fcf0aaf78 chore: auto release commit (2026-03-31 14:18:51)
All checks were successful
技能自动化发布 / release (push) Successful in 50s
2026-03-31 14:18:51 +08:00
01d9c82f80 chore: auto release commit (2026-03-31 11:50:55)
All checks were successful
技能自动化发布 / release (push) Successful in 14s
2026-03-31 11:50:56 +08:00
66d3801e43 chore: auto release commit (2026-03-30 19:25:21)
All checks were successful
技能自动化发布 / release (push) Successful in 17s
2026-03-30 19:25:21 +08:00
973547b6fb chore: auto release commit (2026-03-30 19:10:46)
Some checks failed
技能自动化发布 / release (push) Failing after 0s
2026-03-30 19:10:46 +08:00
edaac8aa45 chore: auto release commit (2026-03-24 18:22:46)
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 9s
2026-03-24 18:22:46 +08:00
157f51c765 chore: auto release commit (2026-03-24 18:07:44)
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 9s
2026-03-24 18:07:44 +08:00
d54a69cba8 chore: auto release commit (2026-03-24 18:06:10) 2026-03-24 18:06:10 +08:00
3b6f15ecbb chore: auto release commit (2026-03-24 18:05:25) 2026-03-24 18:05:25 +08:00
cd131697d5 修改版本
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 14s
2026-03-24 16:42:34 +08:00
2e3ac2cfe7 feat: separate skill display name from slug
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 11s
Set SKILL name to Chinese display text and add openclaw.slug for stable machine identifier, so database records can keep readable names while preserving stable install path keys.

Made-with: Cursor
2026-03-24 16:24:09 +08:00
bea22d5e9e ci: enforce tag-based version and strict sync checks
Use Git tag as release version, validate SKILL.md version consistency, and fail workflow when API business code is not success.
2026-03-24 16:14:06 +08:00
7759e0b376 fix: add required openclaw metadata for skill sync
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 12s
Include category and pricing-related openclaw metadata in SKILL frontmatter so /api/skill/update validation can pass during CI sync.

Made-with: Cursor
2026-03-24 15:55:39 +08:00
92a9d61007 ci: use mirrored checkout action
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 11s
Switch workflow checkout step to internal Gitea mirrored actions-checkout to reduce external GitHub dependency and speed up setup.

Made-with: Cursor
2026-03-24 15:09:02 +08:00
16b3f50dac ci: set artifact_type=skill for release upload
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 14s
2026-03-24 14:20:34 +08:00
48c6bec254 chore: 使用清华源加速 pip 安装
All checks were successful
技能自动化发布 (测试版-无鉴权) / build-and-deploy (push) Successful in 11s
2026-03-24 11:47:12 +08:00
39 changed files with 2617 additions and 305 deletions

View File

@@ -1,76 +1,13 @@
name: 技能自动化发布 (测试版-无鉴权)
name: 技能自动化发布
on:
push:
tags: ["v*"]
jobs:
build-and-deploy:
runs-on: ubuntu-latest
steps:
# 1. 下载源码
- uses: actions/checkout@v3
# 2. 环境及工具准备
- name: Setup Tools
run: pip install pyarmor requests python-frontmatter --break-system-packages
# 3. 【核心步骤】PyArmor 源代码混淆加密
- name: Encrypt Source Code
run: |
mkdir -p dist/package
pyarmor gen -O dist/package scripts/*.py
cp SKILL.md dist/package/
# 4. 提取元数据并物理打包
- name: Parse Metadata and Pack
id: build_task
run: |
python -c "
import frontmatter, os, json, shutil
post = frontmatter.load('SKILL.md')
metadata = post.metadata
slug = metadata['name']
version = metadata['version']
# 安全写入输出
with open(os.environ['GITHUB_OUTPUT'], 'a') as f:
f.write(f'slug={slug}\n')
f.write(f'version={version}\n')
f.write(f'metadata={json.dumps(metadata)}\n')
# 制作最终的发布 ZIP 包
shutil.make_archive(slug, 'zip', 'dist/package')
"
# 5. 同步数据库记录
- name: Sync Database (V2.0)
# 核心修复:通过 env 传递复杂 JSON 字符串
env:
METADATA_JSON: ${{ steps.build_task.outputs.metadata }}
run: |
python -c "
import requests, json, os
# 从环境变量读取,不涉及脚本注入风险
metadata = json.loads(os.environ['METADATA_JSON'])
url = 'https://jc2009.com/api/skill/update'
res = requests.post(url, json=metadata)
print(f'DB Sync Result: {res.text}')
if res.status_code != 200: exit(1)
"
# 6. 上传物理包
- name: Upload Encrypted ZIP
env:
SLUG: ${{ steps.build_task.outputs.slug }}
VERSION: ${{ steps.build_task.outputs.version }}
run: |
python -c "
import requests, os
url = 'https://jc2009.com/api/upload'
slug = os.environ['SLUG']
version = os.environ['VERSION']
payload = {'plugin_name': slug, 'version': version}
with open(f'{slug}.zip', 'rb') as f:
res = requests.post(url, data=payload, files={'file': f})
print(f'Upload Result: {res.text}')
if res.status_code != 200: exit(1)
"
release:
uses: admin/jiangchang-platform-kit/.github/workflows/reusable-release-skill.yaml@main
secrets:
PYARMOR_REG_B64: ${{ secrets.PYARMOR_REG_B64 }}
with:
artifact_platform: windows
pyarmor_platform: windows.x86_64

3
.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
__pycache__/
*.py[cod]
.env

View File

@@ -1,16 +1,22 @@
---
name: account-manager
name: 账号管理
description: 多平台多账号管理。管理各平台账号与Chrome Profile的对应关系供publisher类Skill调用获取账号信息。
version: 1.0.1
version: 1.0.52
author: 深圳匠厂科技有限公司
metadata:
openclaw:
slug: account-manager
emoji: "👤"
category: "通用"
allowed-tools:
- bash
---
# 账号管理
# 账号管理
## JIANGCHANG 共享 Python匠厂桌面
依赖由宿主安装到 `{JIANGCHANG_DATA_ROOT}/python-runtime/.venv`(技能市场安装时自动 `uv sync`)。勿在技能目录内创建 `.venv`。Playwright 使用本机 **Chrome / Edge**`channel=chrome|msedge`**不**下载自带 Chromium。
## 使用时机
@@ -21,34 +27,45 @@ allowed-tools:
## 执行步骤
### 列出某平台所有账号
### 添加账号(平台用中文名;须填合法中国大陆 11 位手机号同平台下同手机号不可重复ID 自增名称如「搜狐1号」
```bash
python3 {baseDir}/scripts/account.py list <platform>
python3 {baseDir}/scripts/main.py add 搜狐号 189xxxxxxxx
python3 {baseDir}/scripts/main.py add 知乎 13800138000
```
支持的平台称呼示例搜狐号、头条号、知乎、微信公众号、Kimi、DeepSeek、豆包、通义千问、文心一言、腾讯元宝亦支持英文键如 sohu、toutiao
### 仅打开浏览器核对是否已登录(不写数据库)
```bash
python3 {baseDir}/scripts/main.py open <id>
```
### 获取某个账号详情
### 登录并自动检测(成功则刷新该账号 `updated_at`,不在库中保存「是否已登录」)
```bash
python3 {baseDir}/scripts/account.py get <account_id>
python3 {baseDir}/scripts/main.py login <id>
```
### 初始化账号登录(首次使用,打开浏览器手动登录)
### 删除账号(同时删库里的记录与 profile 用户数据目录)
```bash
python3 {baseDir}/scripts/account.py login <account_id>
python3 {baseDir}/scripts/main.py delete id <id>
python3 {baseDir}/scripts/main.py delete platform <平台>
python3 {baseDir}/scripts/main.py delete platform <平台> <手机号>
```
## 输出格式
### 列出某平台所有账号platform 可为中文名、all 或 全部)
```bash
python3 {baseDir}/scripts/main.py list all
python3 {baseDir}/scripts/main.py list 搜狐号
```
- list逐行返回账号ID和名称
- get返回账号完整信息
- login打开浏览器后提示用户手动完成登录
### 跨技能:单行 JSONget / list-json / pick-web
```bash
python3 {baseDir}/scripts/main.py get <id>
python3 {baseDir}/scripts/main.py list-json all --limit 50
python3 {baseDir}/scripts/main.py pick-web 搜狐号
```
约定与字段说明见 `references/INTEGRATION.md`;错误前缀见 `references/ERRORS.md`。是否已登录由调用方网页会话处理,本 skill 不做编排用预检。
## 错误处理
## 扩展文档(便于 AI 精确调用)
- 账号不存在:提示"未找到该账号请检查账号ID是否正确"
- 平台无账号:提示"该平台暂无账号请先在accounts.json中添加"
## 注意事项
- 账号档案存储在 `accounts.json`,新增账号需手动编辑该文件
- Chrome Profile目录首次登录后自动创建登录态永久保存
- 每个账号只需初始化登录一次
- **references/**:完整 CLI、平台表、运行时环境、错误码、集成协议 — 见 [references/README.md](references/README.md)。
- **assets/**:示例 JSON 与单条账号 Schema — 见 [assets/README.md](assets/README.md)。

View File

@@ -1,81 +0,0 @@
{
"sohu": [
{
"id": "sohu_account1",
"name": "搜狐账号1",
"phone": "13800000000",
"platform": "sohu",
"profile_dir": "D:/OpenClaw/chrome-profiles/sohu_account1"
},
{
"id": "sohu_account2",
"name": "搜狐账号2",
"phone": "13900000000",
"platform": "sohu",
"profile_dir": "D:/OpenClaw/chrome-profiles/sohu_account2"
}
],
"zhihu": [
{
"id": "zhihu_account1",
"name": "知乎账号1",
"phone": "",
"platform": "zhihu",
"profile_dir": "D:/OpenClaw/chrome-profiles/zhihu_account1"
}
],
"kimi": [
{
"id": "kimi_account1",
"name": "Kimi账号1",
"phone": "13800000000",
"platform": "kimi",
"profile_dir": "D:/OpenClaw/chrome-profiles/kimi_account1"
}
],
"deepseek": [
{
"id": "deepseek_account1",
"name": "DeepSeek账号1",
"phone": "",
"platform": "deepseek",
"profile_dir": "D:/OpenClaw/chrome-profiles/deepseek_account1"
}
],
"doubao": [
{
"id": "doubao_account1",
"name": "豆包账号1",
"phone": "",
"platform": "doubao",
"profile_dir": "D:/OpenClaw/chrome-profiles/doubao_account1"
}
],
"qianwen": [
{
"id": "qianwen_account1",
"name": "千问账号1",
"phone": "",
"platform": "qianwen",
"profile_dir": "D:/OpenClaw/chrome-profiles/qianwen_account1"
}
],
"yiyan": [
{
"id": "yiyan_account1",
"name": "一言账号1",
"phone": "",
"platform": "yiyan",
"profile_dir": "D:/OpenClaw/chrome-profiles/yiyan_account1"
}
],
"yuanbao": [
{
"id": "yuanbao_account1",
"name": "元宝账号1",
"phone": "",
"platform": "yuanbao",
"profile_dir": "D:/OpenClaw/chrome-profiles/yuanbao_account1"
}
]
}

6
assets/README.md Normal file
View File

@@ -0,0 +1,6 @@
# assets
- `examples/`CLI 成功输出形状示例(虚构路径与数据)。
- `schemas/``get` / `list-json` / `pick-web` 单条对象的轻量 JSON Schema`additionalProperties: true` 允许扩展字段)。
详细命令与错误码见仓库根目录 `references/`

View File

@@ -0,0 +1,4 @@
# 示例 JSON
- `get-response.json``python main.py get <id>` 成功时 **stdout 单行** 对象的形状参考(示例路径为虚构)。
- `list-json-array.json``list-json` 成功时 **stdout 单行** 数组的形状参考(本文件为多行排版便于阅读)。

View File

@@ -0,0 +1,10 @@
{
"id": 1,
"name": "搜狐1号",
"platform": "sohu",
"phone": "13800138000",
"profile_dir": "D:/jiangchang-data/_anon/account-manager/profiles/搜狐号/13800138000",
"url": "https://mp.sohu.com",
"created_at": "2026-03-01T10:00:00",
"updated_at": "2026-04-01T12:00:00"
}

View File

@@ -0,0 +1,22 @@
[
{
"id": 2,
"name": "知乎1号",
"platform": "zhihu",
"phone": "13900001111",
"profile_dir": "D:/jiangchang-data/_anon/account-manager/profiles/知乎/13900001111",
"url": "https://www.zhihu.com",
"created_at": "2026-04-02T09:00:00",
"updated_at": "2026-04-02T09:00:00"
},
{
"id": 1,
"name": "搜狐1号",
"platform": "sohu",
"phone": "13800138000",
"profile_dir": "D:/jiangchang-data/_anon/account-manager/profiles/搜狐号/13800138000",
"url": "https://mp.sohu.com",
"created_at": "2026-03-01T10:00:00",
"updated_at": "2026-04-01T12:00:00"
}
]

View File

@@ -0,0 +1,19 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://openclaw.local/account-manager/account-record.schema.json",
"title": "AccountRecord",
"description": "account-manager get / list-json / pick-web 返回的单条账号对象核心字段extra_json 合并键不枚举)",
"type": "object",
"required": ["id", "name", "platform", "phone", "profile_dir", "url", "created_at", "updated_at"],
"properties": {
"id": { "type": "integer" },
"name": { "type": "string" },
"platform": { "type": "string", "description": "内部英文键,如 sohu、zhihu" },
"phone": { "type": "string", "description": "归一后的数字串,可能为空字符串" },
"profile_dir": { "type": "string", "description": "Playwright 用户数据目录绝对路径" },
"url": { "type": "string", "format": "uri" },
"created_at": { "type": ["string", "null"] },
"updated_at": { "type": ["string", "null"] }
},
"additionalProperties": true
}

100
references/CLI.md Normal file
View File

@@ -0,0 +1,100 @@
# account-manager CLI 参考
入口:`{skillRoot}/scripts/main.py`(宿主可将 `{baseDir}` 或技能根目录代入)。下文用 `python main.py` 表示在 `scripts/` 目录执行,或写成 `python {skillRoot}/scripts/main.py`
## 通用说明
- **平台参数**:可为 `scripts/util/platforms.py` 中配置的**中文展示名、别名或英文键**(如 `sohu``搜狐号`)。详见 [PLATFORMS.md](PLATFORMS.md)。
- **数据与路径**库文件、profile 目录由环境变量决定,网关与本地终端不一致时会「看不到同一份数据」。详见 [RUNTIME.md](RUNTIME.md)。
- **机器可读子命令**`get``list-json``pick-web` 的成功/失败约定见 [INTEGRATION.md](INTEGRATION.md)。
- **错误码前缀**:见 [ERRORS.md](ERRORS.md)。
> 说明:当前版本 **未实现** `health` / `version` 子命令;与工作区其它技能的统一约定可对齐后另补。
---
## 子命令一览
### `add` — 添加账号
```bash
python main.py add <平台> <手机号>
```
- 手机号为**中国大陆 11 位**`1` 开头,第二位 `39`),可含空格/分隔符,入库前会归一成数字。
- 同一 `platform` 下同一手机号不可重复。
### `list` — 列出账号(人类可读多行块)
```bash
python main.py list [平台|all|全部]
```
- 省略第二参数时等价于 `all`
- 默认最多 **10** 条(按创建时间倒序);内部实现见 `service/account_service.cmd_list`
### 简写:`python main.py <平台>` / `python main.py <平台> list`
- 若第一个参数能解析为平台,则等价于 `list <平台>`
### `get` — 按 id 输出**单行 JSON**
```bash
python main.py get <id>
```
- 成功:`stdout` **仅一行** JSON 对象UTF-8
- 失败:首行 `ERROR:ACCOUNT_NOT_FOUND`,路径提示在 stderr。
### `list-json` — 列表的**单行 JSON 数组**(跨技能)
```bash
python main.py list-json <平台|all|全部> [--limit N]
```
- 成功:`stdout` **仅一行** JSON 数组;元素结构与 `get` 单条一致。
- 默认 `limit=200`,实际 clamp 为 `1..500`
- 排序:按 `updated_at` 倒序。
### `pick-web` — 网页自动化候选账号(单行 JSON
```bash
python main.py pick-web <平台>
```
- 按该平台 `updated_at``created_at` 倒序选一条;**不**表示当前浏览器已登录。
- 失败:首行 `ERROR:`
### `open` — 仅打开浏览器(**不写库**
```bash
python main.py open <id>
```
- 需本机 Chrome/Edge + Playwright详见运行环境说明。
### `login` — 打开浏览器并**自动检测登录**(成功则刷新 `updated_at`
```bash
python main.py login <id>
```
- 依赖 Playwright超时与轮询间隔等见 [RUNTIME.md](RUNTIME.md)。
- **不**在库中持久化「是否已登录」;仅刷新 `updated_at``pick-web` 等排序参考。
- 跨技能编排(如 llm-manager 网页生成)**不再**通过本 skill 单独做「登录预检」;由业务侧在同一浏览器会话内处理未登录态。
### `delete` — 删除库记录并尽量删除 profile 目录
```bash
python main.py delete id <id>
python main.py delete platform <平台>
python main.py delete platform <平台> <手机号>
```
- 第二段必须为 `id``platform`(小写不敏感)。
---
## 无参数 / 未知子命令
- 无参数或无法解析:打印用法概览;未知子命令时首行 `ERROR:CLI_UNKNOWN_OR_BAD_ARGS`,退出码 `1`

44
references/ERRORS.md Normal file
View File

@@ -0,0 +1,44 @@
# 终端前缀约定account-manager
约定:**机器解析只依赖 `stdout` 首行**时,应先判断首行是否以 `ERROR:` 开头;多行说明可忽略或仅作展示。
## `ERROR:` — CLI 参数 / 路由
| 前缀 | 含义 |
|------|------|
| `ERROR:CLI_ADD_MISSING_ARGS` | `add` 参数不足 |
| `ERROR:CLI_GET_MISSING_ID` | `get` 缺少 id |
| `ERROR:CLI_OPEN_MISSING_ID` | `open` 缺少 id |
| `ERROR:CLI_LOGIN_MISSING_ID` | `login` 缺少 id |
| `ERROR:CLI_DELETE_MISSING_ARGS` | `delete` 参数不足 |
| `ERROR:CLI_DELETE_BAD_MODE` | `delete` 第二段不是 `id`/`platform` |
| `ERROR:CLI_LIST_JSON_MISSING_PLATFORM` | `list-json` 缺少平台参数 |
| `ERROR:CLI_PICK_WEB_MISSING_ARGS` | `pick-web` 缺少平台 |
| `ERROR:CLI_UNKNOWN_OR_BAD_ARGS` | 无法识别的子命令或参数组合 |
## `ERROR:` — 业务与数据
| 前缀 | 含义 |
|------|------|
| `ERROR:INVALID_PLATFORM_LIST` | `list` 的平台名无法识别 |
| `ERROR:INVALID_PLATFORM_LIST_JSON` | `list-json` 的平台名无法识别 |
| `ERROR:INVALID_PLATFORM` | 其它子命令中平台无法识别 |
| `ERROR:NO_ACCOUNTS_FOUND` | 列表无结果(或 `delete platform` 下无记录) |
| `ERROR:ACCOUNT_NOT_FOUND` | 指定 id/记录不存在 |
| `ERROR:PHONE_REQUIRED` | 缺少手机号 |
| `ERROR:PHONE_INVALID` | 手机号格式不符合大陆 11 位规则 |
| `ERROR:DUPLICATE_PHONE_PLATFORM` | 同平台下山手机号已存在 |
| `ERROR:DELETE_INVALID_ID` | 删除 id 非法 |
| `ERROR:NO_ACCOUNT` | `pick-web` 该平台无任何记录 |
## `ERROR:` — 浏览器 / 环境
| 前缀 | 含义 |
|------|------|
| `ERROR:需要 playwright Python 包:…` | 未安装 `playwright` 包,或本机无 Chrome/Edgechannel 模式,**不**需 `playwright install chromium` |
| `ERROR:PROFILE_DIR_MISSING` | 账号记录中 `profile_dir` 为空 |
## 非前缀输出
- **成功添加账号**、**删除成功**等可能包含 `✅` 等人类可读行,**不宜**作为稳定协议解析。
- **`get` / `list-json` / `pick-web` 成功**`stdout` 应为**单行 JSON**(失败时首行 `ERROR:`,勿当 JSON

36
references/INTEGRATION.md Normal file
View File

@@ -0,0 +1,36 @@
# 与其它 Skill 集成account-manager
## 推荐调用方式
使用 **`subprocess`** 调用技能根目录下的 `scripts/main.py`,工作目录可为技能根或 `scripts/`(只要命令中的 `python` 与路径一致)。宿主注入的 **`JIANGCHANG_DATA_ROOT` / `JIANGCHANG_USER_ID`** 必须与最终用户会话一致,否则读写到错误库。
## 机器可读命令摘要
| 目的 | 命令 | 成功时 stdout |
|------|------|----------------|
| 取单条账号 JSON | `get <id>` | 单行 JSON 对象 |
| 批量账号 JSON | `list-json <平台或all> [--limit N]` | 单行 JSON 数组 |
| 网页自动化候选账号 | `pick-web <平台>` | 单行 JSON 对象(按 `updated_at` 选一条,**不**表示当前已登录) |
## 解析协议(强建议)
1. 读取子进程 **stdout** 全文或首行。
2. 若首行以 **`ERROR:`** 开头 → 失败,不要 `json.loads` 整段 stdout。
3. **`get` / `list-json` / `pick-web`**:成功时应对**第一行**做 `json.loads`;后续行应不存在,若有则视为非协议内容。
## JSON 对象字段(与 `get` 一致)
典型字段见仓库 `assets/examples/get-response.json``assets/schemas/account-record.schema.json`
常见键名包括:`id`, `name`, `platform`, `phone`, `profile_dir`, `url`, `created_at`, `updated_at``extra_json` 列中的扁平扩展可能并入同一对象。
**库中不再持久化「是否已登录」**;编排侧在网页自动化中自行处理登录(与 `pick-web` / `get` 返回的 `profile_dir` 同一浏览器会话)。
## Publisher / 自动化流水线示例
1. `pick-web 搜狐号`(或目标平台)→ 解析 JSON → 取 `id``profile_dir``url`
2. 使用 `profile_dir` 启动持久化浏览器上下文,在业务页面内完成或等待登录。
## 退出码
请以 **stdout 首行前缀****JSON 可解析性** 为准;不要仅依赖退出码。

23
references/PLATFORMS.md Normal file
View File

@@ -0,0 +1,23 @@
# 平台与别名account-manager
**权威来源**`scripts/util/platforms.py` 中的 `PLATFORMS`。下表便于 AI/文档阅读;若与代码不一致,**以代码为准**。
入库字段 `accounts.platform` 使用下表 **英文键**小写。CLI 与用户话术可使用 **展示名****别名**
| 英文键 (`platform`) | 展示名 (`label`) | 默认账号名前缀 (`prefix`) | 别名 (`aliases`) | 默认 URL |
|---------------------|------------------|---------------------------|------------------|----------|
| `sohu` | 搜狐号 | 搜狐 | 搜狐 | https://mp.sohu.com |
| `toutiao` | 头条号 | 头条 | 头条 | https://mp.toutiao.com/ |
| `zhihu` | 知乎 | (同 label | 知乎号 | https://www.zhihu.com |
| `wechat` | 微信公众号 | 微信 | 公众号、微信 | https://mp.weixin.qq.com |
| `kimi` | Kimi | (同 label | 月之暗面 | https://kimi.moonshot.cn |
| `deepseek` | DeepSeek | (同 label | — | https://chat.deepseek.com |
| `doubao` | 豆包 | (同 label | — | https://www.doubao.com |
| `qianwen` | 通义千问 | 通义 | 通义、千问 | https://tongyi.aliyun.com |
| `yiyan` | 文心一言 | 文心 | 文心、一言 | https://yiyan.baidu.com |
| `yuanbao` | 腾讯元宝 | 元宝 | 元宝 | https://yuanbao.tencent.com |
## 解析规则(摘要)
- 英文键本身、各 `label`、各 `aliases` 条目(及小写形式)均可解析到对应键。
- 无法识别时,相关子命令会输出 `ERROR:INVALID_PLATFORM``ERROR:INVALID_PLATFORM_LIST` / `INVALID_PLATFORM_LIST_JSON` 等,并附带「支持的平台」提示。

13
references/README.md Normal file
View File

@@ -0,0 +1,13 @@
# referencesaccount-manager
供 Agent 按需加载的说明文档(与 `SKILL.md` 配合,渐进式披露)。
| 文件 | 内容 |
|------|------|
| [CLI.md](CLI.md) | 子命令、参数、行为说明 |
| [PLATFORMS.md](PLATFORMS.md) | 平台英文键、中文名、别名 |
| [RUNTIME.md](RUNTIME.md) | 环境变量、数据路径、日志、登录调参 |
| [ERRORS.md](ERRORS.md) | `ERROR:` 前缀一览 |
| [INTEGRATION.md](INTEGRATION.md) | 跨 Skill 调用与 JSON 解析约定 |
示例 JSON 与 Schema 见仓库 `assets/`

55
references/RUNTIME.md Normal file
View File

@@ -0,0 +1,55 @@
# 运行环境与数据路径account-manager
## 数据落盘位置(当前实现)
`get_data_root()` / `get_user_id()` 来自 **`scripts/jiangchang_skill_core/runtime_env.py`**(与 `jiangchang-platform-kit` 同源);`scripts/util/runtime_paths.py` 在此基础上拼技能子目录。
- **数据根目录**:优先 **`JIANGCHANG_DATA_ROOT`**未设置且未走本地注入时Windows 默认 `D:\jiangchang-data`,其它系统 `~/.jiangchang-data`
- **用户/工作空间 id**:优先 **`JIANGCHANG_USER_ID`**;未设置且未走本地注入时为 **`_anon`**。
- **技能数据目录**`{数据根}/{用户id}/account-manager/`
- **SQLite**`account-manager.db`
- **Profile**`profiles/<平台展示名>/<手机号>/`(由 `get_default_profile_dir` 生成)
- **日志**:与其它技能共用**用户级**目录 `{数据根}/{用户id}/logs/jiangchang.log`(按日轮转)
## 调试为什么「list 没数据」
- 若终端与网关注入的 `JIANGCHANG_*` 不一致,会写到**不同用户目录**,库文件不是同一份。
- 设置 **`JIANGCHANG_ACCOUNT_DEBUG_PATHS=1`**`1`/`true`/`yes`/`on`)后,每次 CLI 子命令前会在 **stderr** 打印实际根目录、用户 id 与数据库路径(`_maybe_print_paths_debug_cli`)。
## 本地 CLI 默认值(`runtime_env`
各技能 **`main.py`** 在 `sys.path` 就绪后、import 业务包之前调用 **`apply_cli_local_defaults()`**(实现见 `jiangchang_skill_core.runtime_env`
- 代码内 **`CLI_LOCAL_DEV_ENABLED = True`** 时,若当前未设置 `JIANGCHANG_DATA_ROOT` / `JIANGCHANG_USER_ID`,会注入平台默认数据根与 **`DEFAULT_LOCAL_USER_ID`**(与 kit 中常量一致)。
- 代码内为 **`False`** 时,可设环境变量 **`JIANGCHANG_CLI_LOCAL_DEV=1`** 达到同样效果。
- **发版/嵌入宿主前**:将各技能 vendored 的 `runtime_env.py``CLI_LOCAL_DEV_ENABLED` 改为 **`False`**,或完全依赖宿主注入。
## 日志
实现见 `scripts/jiangchang_skill_core/unified_logging.py``util/logging_config.py` 仅 re-export。
| 变量 | 作用 |
|------|------|
| `JIANGCHANG_LOG_LEVEL` | 日志级别,默认 `INFO` |
| `JIANGCHANG_LOG_TO_STDERR` | 设为 `1`/`true` 等时WARNING 及以上同步 stderr |
| `JIANGCHANG_LOG_FILE` | 覆盖主日志文件绝对路径(默认 `{数据根}/{用户id}/logs/jiangchang.log` |
| `JIANGCHANG_LOG_BACKUP_COUNT` | 轮转保留份数,默认 `30` |
| `JIANGCHANG_TRACE_ID` | 可选;跨技能子进程由父进程注入以串联同一调用链 |
## 登录 / 浏览器子进程login
`service/browser_service.py` 与子进程 `service/login_child_runner.py`
| 变量 | 作用 |
|------|------|
| `JIANGCHANG_LOGIN_TIMEOUT_SECONDS` | 登录检测最长秒数,默认 `300`,不少于 `60` |
| `JIANGCHANG_LOGIN_POLL_INTERVAL_SECONDS` | DOM 轮询间隔,默认 `1.5`,限制在 `0.5..10` |
| `JIANGCHANG_LOGIN_DOM_GRACE_SECONDS` | 打开 URL 后等待再检测,默认 `4`,减轻跳转误判 |
依赖:**本机已安装 Chrome 或 Edge**Python 环境需 **`playwright` 包**(匠厂共享 `python-runtime` 已包含)。运行时使用 **Playwright `channel=chrome|msedge`** 调系统浏览器,**不要**执行 `playwright install chromium`(避免下载自带 Chromium
## 常见问题(简要)
- **`ERROR:需要 playwright Python 包`**:确认共享 venv 已 `uv sync`;本机安装 Chrome 或 Edge。
- **`ERROR:PROFILE_DIR_MISSING`**:库中该账号 `profile_dir` 为空,需检查数据或重新添加账号流程。
- **登录检测一直失败**:关闭占用同一 profile 目录的其它浏览器;可调高超时与环境变量;查看统一日志文件。

23
release.ps1 Normal file
View File

@@ -0,0 +1,23 @@
[CmdletBinding()]
param(
[string]$Prefix = "v",
[string]$Message = "正式发布",
[switch]$AutoCommit,
[switch]$RequireClean,
[string]$CommitMessage,
[switch]$DryRun
)
Set-StrictMode -Version Latest
$ErrorActionPreference = "Stop"
$scriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
$sharedScript = Join-Path $scriptDir "..\jiangchang-platform-kit\tools\release.ps1"
$sharedScript = [System.IO.Path]::GetFullPath($sharedScript)
if (-not (Test-Path $sharedScript)) {
throw "Shared release script not found: $sharedScript"
}
& $sharedScript @PSBoundParameters
exit $LASTEXITCODE

View File

@@ -1,131 +0,0 @@
import sys
import json
import os
import subprocess
# Windows GBK 编码兼容修复
if sys.platform == "win32":
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
ACCOUNTS_FILE = os.path.join(BASE_DIR, "accounts.json")
PLATFORM_URLS = {
# 自媒体/图文平台
"sohu": "https://mp.sohu.com",
"zhihu": "https://www.zhihu.com",
"wechat": "https://mp.weixin.qq.com",
# 大模型平台
"kimi": "https://kimi.moonshot.cn",
"deepseek": "https://chat.deepseek.com",
"doubao": "https://www.doubao.com",
"qianwen": "https://tongyi.aliyun.com",
"yiyan": "https://yiyan.baidu.com",
"yuanbao": "https://yuanbao.tencent.com"
}
def load_accounts():
with open(ACCOUNTS_FILE, "r", encoding="utf-8") as f:
return json.load(f)
def cmd_list(platform="all"):
"""
修改后的 list 功能能直观打出带着【手机号】的详细台账供AI查询匹配
支持查全部(all),或指定(sohu)
"""
accounts = load_accounts()
if platform == "all":
platforms_to_check = accounts.keys()
else:
platforms_to_check = [platform]
found = False
for plat in platforms_to_check:
platform_accounts = accounts.get(plat, [])
for acc in platform_accounts:
phone = acc.get('phone', '未绑定手机')
print(f"账号ID:{acc['id']} | 名称:{acc['name']} | 手机号:{phone} | 平台:{acc['platform']}")
found = True
if not found:
print(f"ERROR:NO_ACCOUNTS_FOUND")
def cmd_get(account_id):
accounts = load_accounts()
for platform_accounts in accounts.values():
for acc in platform_accounts:
if acc["id"] == account_id:
print(json.dumps(acc, ensure_ascii=False))
return
print("ERROR:ACCOUNT_NOT_FOUND")
def cmd_login(account_id):
"""首次登录打开浏览器用户手动登录登录态自动保存到Profile目录"""
accounts = load_accounts()
target = None
for platform_accounts in accounts.values():
for acc in platform_accounts:
if acc["id"] == account_id:
target = acc
break
if not target:
print("ERROR:ACCOUNT_NOT_FOUND")
return
profile_dir = target["profile_dir"]
os.makedirs(profile_dir, exist_ok=True)
url = PLATFORM_URLS.get(target["platform"], "https://www.google.com")
print(f"正在为账号 [{target['name']}] 打开浏览器...")
print(f"请在浏览器中完成登录,登录后直接关闭浏览器窗口即可。")
print(f"登录态将自动保存到:{profile_dir}")
# 用Playwright打开持久化浏览器等待用户手动登录
login_script = f"""
import asyncio
from playwright.async_api import async_playwright
async def main():
async with async_playwright() as p:
browser = await p.chromium.launch_persistent_context(
user_data_dir=r"{profile_dir}",
headless=False,
channel="chrome", # 【修改处】也加上这行
no_viewport=True, # 确保浏览器内容视口也最大化
args=["--start-maximized"]
)
page = browser.pages[0] if browser.pages else await browser.new_page()
await page.goto("{url}")
print("浏览器已打开,请完成登录后关闭窗口...")
# 移除自动超时,无限等待直到用户扫码完手动关闭
await browser.wait_for_event("close", timeout=0)
print("登录态已保存!")
asyncio.run(main())
"""
import tempfile
with tempfile.NamedTemporaryFile(mode='w', suffix='.py',
delete=False, encoding='utf-8') as f:
f.write(login_script)
tmp_path = f.name
os.system(f'python "{tmp_path}"')
os.unlink(tmp_path)
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法python account.py list [platform] | get <id> | login <id>")
sys.exit(1)
cmd = sys.argv[1]
if cmd == "list":
# 默认不传 platform 时打出 all
cmd_list(sys.argv[2] if len(sys.argv) >= 3 else "all")
elif cmd == "get" and len(sys.argv) >= 3:
cmd_get(sys.argv[2])
elif cmd == "login" and len(sys.argv) >= 3:
cmd_login(sys.argv[2])
else:
print("参数错误")
sys.exit(1)

1
scripts/cli/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Command-line entry and argv dispatch."""

167
scripts/cli/app.py Normal file
View File

@@ -0,0 +1,167 @@
"""CLI 入口:用法说明与 argv 分发。"""
import sys
from service.account_service import (
cmd_add,
cmd_delete_by_id,
cmd_delete_by_platform,
cmd_delete_by_platform_phone,
cmd_get,
cmd_list,
cmd_list_json,
cmd_pick_web,
)
from service.browser_service import cmd_login, cmd_open
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
from util.logging_config import get_skill_logger, setup_skill_logging
from util.platforms import _platform_list_cn_for_help, resolve_platform_key
from util.runtime_paths import _maybe_print_paths_debug_cli
def _cli_print_full_usage() -> None:
"""总览(未带子命令或需要完整说明时)。"""
print("用法概览(将 main.py 换为你的路径):")
print(" python main.py add <平台中文名> <手机号>")
print(" python main.py list [平台|all|全部]")
print(" python main.py pick-web <平台> # 跨技能用:按最近更新选一条账号,一行 JSON")
print(" python main.py list-json <平台|all> [--limit N] # 跨技能用:一行 JSON 数组,元素同 get")
print(" python main.py get <id>")
print(" python main.py open <id>")
print(" python main.py login <id>")
print(" python main.py delete id <id>")
print(" python main.py delete platform <平台>")
print(" python main.py delete platform <平台> <手机号>")
print(" python main.py <平台> # 等价于只列该平台")
print(" python main.py <平台> list # 同上")
print()
print("支持的平台(可用下列中文称呼,亦支持英文键如 sohu")
print(" " + _platform_list_cn_for_help())
def _cli_fail_add() -> None:
print("ERROR:CLI_ADD_MISSING_ARGS")
print()
print("【add 添加账号】参数不足。")
print(" 必填:平台名称(中文即可,如 搜狐号、知乎、微信公众号)")
print(" 必填:中国大陆 11 位手机号1 开头,第二位 39同平台下不可重复")
print()
print("示例:")
print(" python main.py add 搜狐号 13800138000")
print(" python main.py add 知乎 13900001111")
print()
print("支持的平台:" + _platform_list_cn_for_help())
def _cli_fail_need_account_id(subcmd: str) -> None:
print(f"ERROR:CLI_{subcmd.upper()}_MISSING_ID")
print()
print(f"{subcmd}】缺少必填参数:账号 id数据库自增整数可先 list 查看)。")
print("示例:")
print(f" python main.py {subcmd} 1")
def _cli_fail_delete() -> None:
print("ERROR:CLI_DELETE_MISSING_ARGS")
print()
print("【delete 删除账号】参数不足。支持三种方式:")
print(" 按 idpython main.py delete id <id>")
print(" 按平台删该平台下全部python main.py delete platform <平台中文名或英文键>")
print(" 按平台+手机号python main.py delete platform <平台> <手机号>")
print()
print("示例:")
print(" python main.py delete id 3")
print(" python main.py delete platform 搜狐号")
print(" python main.py delete platform 头条号 18925203701")
print()
print("说明:会同时删除数据库记录与 profile 用户数据目录(若存在)。")
def main() -> None:
if len(sys.argv) < 2:
_cli_print_full_usage()
sys.exit(1)
setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
get_skill_logger().info("cli_start argv=%s", sys.argv)
cmd = sys.argv[1]
_maybe_print_paths_debug_cli()
plat_from_first = resolve_platform_key(cmd)
if len(sys.argv) >= 3 and sys.argv[2] == "list" and plat_from_first:
cmd_list(plat_from_first)
elif plat_from_first and len(sys.argv) == 2:
cmd_list(plat_from_first)
elif cmd == "add":
if len(sys.argv) < 4:
_cli_fail_add()
sys.exit(1)
cmd_add(sys.argv[2], sys.argv[3])
elif cmd == "list":
cmd_list(sys.argv[2] if len(sys.argv) >= 3 else "all")
elif cmd == "get":
if len(sys.argv) < 3:
_cli_fail_need_account_id("get")
sys.exit(1)
cmd_get(sys.argv[2])
elif cmd == "list-json":
if len(sys.argv) < 3:
print("ERROR:CLI_LIST_JSON_MISSING_PLATFORM")
print("用法python main.py list-json <平台|all|全部> [--limit N]")
print("说明:跨技能编排;成功时 stdout 仅一行 JSON 数组(元素与 get 一致)。")
sys.exit(1)
av = sys.argv[2:]
platform_arg = av[0]
lim = 200
if "--limit" in av:
i = av.index("--limit")
if i + 1 < len(av):
try:
lim = int(av[i + 1])
except ValueError:
pass
cmd_list_json(platform_arg, limit=lim)
elif cmd == "pick-web":
if len(sys.argv) < 3:
print("ERROR:CLI_PICK_WEB_MISSING_ARGS")
print("用法python main.py pick-web <平台中文名或英文键>")
print("说明:按 updated_at 选一条账号;是否已登录由调用方网页会话自行处理。")
sys.exit(1)
cmd_pick_web(sys.argv[2])
elif cmd == "open":
if len(sys.argv) < 3:
_cli_fail_need_account_id("open")
sys.exit(1)
cmd_open(sys.argv[2])
elif cmd == "login":
if len(sys.argv) < 3:
_cli_fail_need_account_id("login")
sys.exit(1)
cmd_login(sys.argv[2])
elif cmd == "delete":
if len(sys.argv) < 4:
_cli_fail_delete()
sys.exit(1)
mode = (sys.argv[2] or "").strip().lower()
if mode == "id":
cmd_delete_by_id(sys.argv[3])
elif mode == "platform":
if len(sys.argv) >= 5:
cmd_delete_by_platform_phone(sys.argv[3], sys.argv[4])
else:
cmd_delete_by_platform(sys.argv[3])
else:
print(f"ERROR:CLI_DELETE_BAD_MODE 不支持的删除方式「{sys.argv[2]}」,请使用 id 或 platform。")
_cli_fail_delete()
sys.exit(1)
else:
print(f"ERROR:CLI_UNKNOWN_OR_BAD_ARGS 子命令「{cmd}」无法识别,或参数组合不合法。")
print()
_cli_print_full_usage()
print()
print("说明list 的筛选可为 all/全部,或上面列出的任一平台中文名。")
sys.exit(1)
if __name__ == "__main__":
main()

5
scripts/db/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Persistence layer (SQLite)."""
from db.connection import get_conn, init_db
__all__ = ["get_conn", "init_db"]

230
scripts/db/accounts_repo.py Normal file
View File

@@ -0,0 +1,230 @@
"""Accounts table: CRUD and queries (no stdout)."""
import json
import time
from datetime import datetime
from typing import Any, Optional
from db.connection import get_conn, init_db
from util.platforms import PLATFORM_URLS, _PLATFORM_NAME_CN
def _now_unix() -> int:
return int(time.time())
def _unix_to_iso_local(ts: Optional[int]) -> Optional[str]:
"""对外 JSON本地时区 ISO8601 字符串,便于人读。"""
if ts is None:
return None
try:
return datetime.fromtimestamp(int(ts)).isoformat(timespec="seconds")
except (ValueError, OSError, OverflowError):
return None
def normalize_account_id(account_id):
if account_id is None:
return None
s = str(account_id).strip()
if s.isdigit():
return int(s)
return s
def default_name_for_platform(platform: str, index: int) -> str:
cn = _PLATFORM_NAME_CN.get(platform)
if cn:
return f"{cn}{index}"
return f"{platform}_{index}"
def get_account_by_id(account_id):
init_db()
aid = normalize_account_id(account_id)
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT id, name, platform, phone, profile_dir, url, extra_json,
created_at, updated_at
FROM accounts WHERE id = ?
""",
(aid,),
)
row = cur.fetchone()
if not row:
return None
acc: dict[str, Any] = {
"id": row[0],
"name": row[1],
"platform": row[2],
"phone": row[3] or "",
"profile_dir": (row[4] or "").strip(),
"url": row[5] or PLATFORM_URLS.get(row[2], ""),
"created_at": _unix_to_iso_local(row[7]),
"updated_at": _unix_to_iso_local(row[8]),
}
if row[6]:
try:
extra = json.loads(row[6])
if isinstance(extra, dict):
acc.update(extra)
except Exception:
pass
return acc
finally:
conn.close()
def touch_account_updated_at(account_id) -> None:
"""登录 DOM 检测成功等场景下刷新 updated_at供 pick-web 等按最近活跃选号。"""
now = _now_unix()
conn = get_conn()
try:
cur = conn.cursor()
cur.execute("UPDATE accounts SET updated_at = ? WHERE id = ?", (now, account_id))
conn.commit()
finally:
conn.close()
def fetch_list_rows(platform_key: str, limit: int):
"""platform_key 'all' or platform slug; returns list of row tuples (full row)."""
conn = get_conn()
try:
cur = conn.cursor()
sql = (
"SELECT id, name, platform, phone, profile_dir, url, "
"extra_json, created_at, updated_at FROM accounts "
)
if platform_key == "all":
cur.execute(sql + "ORDER BY created_at DESC, id DESC LIMIT ?", (int(limit),))
else:
cur.execute(
sql + "WHERE platform = ? ORDER BY created_at DESC, id DESC LIMIT ?",
(platform_key, int(limit)),
)
return cur.fetchall()
finally:
conn.close()
def fetch_ids_for_list_json(platform_key: str, lim: int):
conn = get_conn()
try:
cur = conn.cursor()
if platform_key == "all":
cur.execute(
"SELECT id FROM accounts ORDER BY updated_at DESC, id DESC LIMIT ?",
(lim,),
)
else:
cur.execute(
"SELECT id FROM accounts WHERE platform = ? "
"ORDER BY updated_at DESC, id DESC LIMIT ?",
(platform_key, lim),
)
return [r[0] for r in cur.fetchall()]
finally:
conn.close()
def fetch_pick_web_candidate_id(platform_key: str) -> Optional[int]:
"""该平台一条账号:按 updated_at、created_at 倒序,与是否登录无关。"""
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ?
ORDER BY updated_at DESC, created_at DESC, id DESC
LIMIT 1
""",
(platform_key,),
)
row = cur.fetchone()
return int(row[0]) if row else None
finally:
conn.close()
def has_duplicate_phone_conn(conn, platform_key: str, phone_store: str) -> bool:
cur = conn.cursor()
cur.execute(
"SELECT id FROM accounts WHERE platform = ? AND phone = ?",
(platform_key, phone_store),
)
return cur.fetchone() is not None
def count_platform_accounts_conn(conn, platform_key: str) -> int:
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM accounts WHERE platform = ?", (platform_key,))
return int(cur.fetchone()[0])
def insert_account_row(
conn,
name: str,
platform_key: str,
phone_store: str,
profile_dir: str,
url: str,
now: int,
) -> int:
cur = conn.cursor()
cur.execute(
"""
INSERT INTO accounts (name, platform, phone, profile_dir, url, extra_json, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, NULL, ?, ?)
""",
(name, platform_key, phone_store, profile_dir, url, now, now),
)
return int(cur.lastrowid)
def fetch_delete_row_by_id_conn(conn, aid: int):
cur = conn.cursor()
cur.execute(
"SELECT id, name, platform, phone, profile_dir FROM accounts WHERE id = ?",
(aid,),
)
return cur.fetchone()
def delete_account_by_id_conn(conn, rid: int) -> None:
cur = conn.cursor()
cur.execute("DELETE FROM accounts WHERE id = ?", (rid,))
def fetch_ids_profile_for_platform_conn(conn, platform_key: str):
cur = conn.cursor()
cur.execute(
"SELECT id, profile_dir FROM accounts WHERE platform = ?",
(platform_key,),
)
return cur.fetchall()
def delete_accounts_by_platform_conn(conn, platform_key: str) -> None:
cur = conn.cursor()
cur.execute("DELETE FROM accounts WHERE platform = ?", (platform_key,))
def fetch_row_platform_phone_conn(conn, platform_key: str, phone_norm: str):
cur = conn.cursor()
cur.execute(
"SELECT id, name, profile_dir FROM accounts WHERE platform = ? AND phone = ?",
(platform_key, phone_norm),
)
return cur.fetchone()
def delete_account_platform_phone_conn(conn, platform_key: str, phone_norm: str) -> None:
cur = conn.cursor()
cur.execute(
"DELETE FROM accounts WHERE platform = ? AND phone = ?",
(platform_key, phone_norm),
)

21
scripts/db/connection.py Normal file
View File

@@ -0,0 +1,21 @@
"""SQLite connection and schema bootstrap."""
import sqlite3
from db.schema import ACCOUNTS_TABLE_SQL
from util.runtime_paths import get_db_path
def get_conn():
return sqlite3.connect(get_db_path())
def init_db():
conn = get_conn()
try:
cur = conn.cursor()
cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='accounts'")
if not cur.fetchone():
cur.executescript(ACCOUNTS_TABLE_SQL)
conn.commit()
finally:
conn.close()

16
scripts/db/schema.py Normal file
View File

@@ -0,0 +1,16 @@
# SQLite 无独立 DATETIME 类型: 时间统一存 INTEGER Unix 秒 (UTC), 查询/JSON 再转 ISO8601.
# 下列 DDL 含注释, 会原样写入 sqlite_master, 便于 Navicat / DBeaver 等查看建表语句.
ACCOUNTS_TABLE_SQL = """
/* 多平台账号表: 与本地 Chromium/Edge 用户数据目录 (profile) 绑定 */
CREATE TABLE accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT, -- 账号主键 (自增)
name TEXT NOT NULL, -- 展示名称, 如「搜狐1号」
platform TEXT NOT NULL, -- 平台标识, 与内置 PLATFORMS 键一致, 如 sohu
phone TEXT, -- 可选绑定手机号
profile_dir TEXT, -- Playwright 用户数据目录 (绝对路径); 默认可读结构 profiles/<平台展示名>/<手机号>/
url TEXT, -- 平台入口或登录页 URL
extra_json TEXT, -- 扩展字段 JSON
created_at INTEGER NOT NULL, -- 记录创建时间, Unix 秒 UTC
updated_at INTEGER NOT NULL -- 记录最后更新时间, Unix 秒 UTC
);
"""

View File

@@ -0,0 +1 @@
# Vendored from jiangchang-platform-kit/sdk/jiangchang_skill_core/ — keep runtime_env + unified_logging in sync.

View File

@@ -0,0 +1,137 @@
"""
JIANGCHANG_* 数据根与用户目录:解析规则 + 可选本地 CLI 默认值。
各技能 scripts/jiangchang_skill_core/ 为发布用副本,与 kit 保持同步。
宿主在 skills.entries 与 Gateway 中注入:
- PATH 前缀:{JIANGCHANG_DATA_ROOT}/python-runtime/.venv/(Scripts|bin)
- VIRTUAL_ENV、JIANGCHANG_PYTHON_EXE
Playwright 使用本机 Chrome/Edgelaunch 时 channel=chrome|msedge不依赖宿主下载的 Chromium 包。
技能勿在仓库内维护独立 .venv依赖以 jiangchang-platform-kit/python-runtime 锁文件为准。
"""
from __future__ import annotations
import os
import sys
# 发版/嵌入宿主前改为 False或仅通过环境变量 JIANGCHANG_CLI_LOCAL_DEV=1 开启
CLI_LOCAL_DEV_ENABLED = True
# 本地开发且未设置 JIANGCHANG_USER_ID 时注入(与宿主约定一致即可修改)
DEFAULT_LOCAL_USER_ID = "10032"
# Windows 下未设置 JIANGCHANG_DATA_ROOT 时的默认盘路径
WIN_DEFAULT_DATA_ROOT = r"D:\jiangchang-data"
# 匠厂桌面宿主应用根(未设置 JIANGCHANG_APP_ROOT 时 Windows 兜底;技能一般在 {APP}/skills/ 下并列)
WIN_DEFAULT_JIANGCHANG_APP_ROOT = r"D:\AI\jiangchang"
def platform_default_data_root() -> str:
if sys.platform == "win32":
return WIN_DEFAULT_DATA_ROOT
return os.path.join(os.path.expanduser("~"), ".jiangchang-data")
def get_data_root() -> str:
env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip()
if env:
return env
return platform_default_data_root()
def get_user_id() -> str:
return (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "_anon"
def _looks_like_skills_root(path: str) -> bool:
if not path or not os.path.isdir(path):
return False
for marker in (
"llm-manager",
"content-manager",
"account-manager",
"sohu-publisher",
"toutiao-publisher",
"logistics-tracker",
"api-key-vault",
):
if os.path.isdir(os.path.join(path, marker)):
return True
return False
def get_skills_root() -> str:
"""
并列技能安装目录(其下为「技能 slug」子目录
优先级:
1) JIANGCHANG_SKILLS_ROOT
2) CLAW_SKILLS_ROOT
3) JIANGCHANG_APP_ROOT{APP}/skills 存在且像技能根则用之;否则若 APP 下已有技能子目录则 APP 即根
4) Windows默认 D:\\AI\\jiangchang 同上规则
5) 其他平台:~/.openclaw/skills
"""
for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
v = (os.getenv(key) or "").strip()
if v:
return os.path.normpath(v)
app = (os.getenv("JIANGCHANG_APP_ROOT") or "").strip()
if sys.platform == "win32" and not app:
app = WIN_DEFAULT_JIANGCHANG_APP_ROOT
if app:
nested = os.path.join(app, "skills")
if _looks_like_skills_root(nested):
return os.path.normpath(nested)
if _looks_like_skills_root(app):
return os.path.normpath(app)
if sys.platform == "win32":
nested = os.path.join(WIN_DEFAULT_JIANGCHANG_APP_ROOT, "skills")
if _looks_like_skills_root(nested):
return os.path.normpath(nested)
if _looks_like_skills_root(WIN_DEFAULT_JIANGCHANG_APP_ROOT):
return os.path.normpath(WIN_DEFAULT_JIANGCHANG_APP_ROOT)
return os.path.normpath(os.path.join(os.path.expanduser("~"), ".openclaw", "skills"))
def get_sibling_skills_root(skill_scripts_dir: str | None = None) -> str:
"""
编排子进程调用兄弟技能时用:先根据本技能 scripts 目录推断并列根;若目录下能识别出兄弟技能
则用之(开发仓 D:\\OpenClaw 与安装目录 ~/.openclaw/skills 均满足「技能根之上一级 = 并列根」),
避免全局 JIANGCHANG_SKILLS_ROOT / CLAW_SKILLS_ROOT 指向与当前检出不一致时错调兄弟进程。
推断失败时再读上述环境变量,最后回落 get_skills_root()。
"""
if skill_scripts_dir:
scripts = os.path.abspath(skill_scripts_dir)
skill_root = os.path.dirname(scripts)
inferred = os.path.dirname(skill_root)
if _looks_like_skills_root(inferred):
return os.path.normpath(inferred)
for key in ("JIANGCHANG_SKILLS_ROOT", "CLAW_SKILLS_ROOT"):
v = (os.getenv(key) or "").strip()
if v:
return os.path.normpath(v)
return get_skills_root()
def apply_cli_local_defaults() -> None:
"""
在 CLI 最早阶段调用main.py 在 import 业务包之前)。
宿主已设置 JIANGCHANG_* 时不会覆盖。
"""
enabled = CLI_LOCAL_DEV_ENABLED
if not enabled:
v = (os.getenv("JIANGCHANG_CLI_LOCAL_DEV") or "").strip().lower()
enabled = v in ("1", "true", "yes", "on")
if not enabled:
return
if not (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip():
os.environ["JIANGCHANG_DATA_ROOT"] = platform_default_data_root()
if not (os.getenv("JIANGCHANG_USER_ID") or "").strip():
os.environ["JIANGCHANG_USER_ID"] = DEFAULT_LOCAL_USER_ID.strip()

View File

@@ -0,0 +1,158 @@
"""
统一文件日志:{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/logs/jiangchang.log
按日轮转;行内带 trace_id 与 skill_slug便于跨技能排查。
实现为各技能 scripts/jiangchang_skill_core/ 下的同名副本之源,修改后请同步到各技能。
"""
from __future__ import annotations
import logging
import os
import sys
import uuid
from logging.handlers import TimedRotatingFileHandler
from typing import Optional
from .runtime_env import get_data_root, get_user_id
_skill_slug: str = ""
_logger_name: str = ""
def get_unified_logs_dir() -> str:
path = os.path.join(get_data_root(), get_user_id(), "logs")
os.makedirs(path, exist_ok=True)
return path
def get_skill_log_file_path() -> str:
"""与 setup 写入的主日志文件一致(供终端提示等)。"""
override = (os.getenv("JIANGCHANG_LOG_FILE") or "").strip()
if override:
parent = os.path.dirname(os.path.abspath(override))
if parent:
os.makedirs(parent, exist_ok=True)
return os.path.abspath(override)
return os.path.join(get_unified_logs_dir(), "jiangchang.log")
def ensure_trace_for_process() -> str:
"""本进程调用链 ID沿用 JIANGCHANG_TRACE_ID否则生成并写入环境子进程可继承"""
existing = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
if existing:
return existing
tid = uuid.uuid4().hex[:12]
os.environ["JIANGCHANG_TRACE_ID"] = tid
return tid
def subprocess_env_with_trace(environ: Optional[dict] = None) -> dict:
"""subprocess.run(..., env=...) 时使用,保证带上当前 trace。"""
ensure_trace_for_process()
tid = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
if not tid:
tid = ensure_trace_for_process()
base = os.environ if environ is None else environ
return {**base, "JIANGCHANG_TRACE_ID": tid}
class _SkillContextFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
record.trace_id = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip() or "-"
record.skill_slug = _skill_slug or "-"
return True
_FORMAT = (
"%(asctime)s | %(levelname)-8s | %(trace_id)s | %(skill_slug)s | %(name)s | %(message)s"
)
_DATEFMT = "%Y-%m-%dT%H:%M:%S"
def _log_level_from_env() -> int:
v = (os.getenv("JIANGCHANG_LOG_LEVEL") or "INFO").strip().upper()
return getattr(logging, v, None) or logging.INFO
def _backup_count() -> int:
try:
n = int((os.getenv("JIANGCHANG_LOG_BACKUP_COUNT") or "30").strip())
return max(1, min(n, 365))
except ValueError:
return 30
def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
"""
幂等:为指定 logger 挂载统一文件日志与可选 stderr。
须在进程早期调用(如 CLI main 在业务日志之前)。
"""
global _skill_slug, _logger_name
ensure_trace_for_process()
_skill_slug = skill_slug
_logger_name = logger_name
log = logging.getLogger(logger_name)
if log.handlers:
return
log.setLevel(_log_level_from_env())
path = get_skill_log_file_path()
fh = TimedRotatingFileHandler(
path,
when="midnight",
interval=1,
backupCount=_backup_count(),
encoding="utf-8",
delay=True,
)
fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
fh.setFormatter(fmt)
fh.addFilter(_SkillContextFilter())
log.addHandler(fh)
if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in (
"1",
"true",
"yes",
"on",
):
sh = logging.StreamHandler(sys.stderr)
sh.setLevel(logging.WARNING)
sh.setFormatter(fmt)
sh.addFilter(_SkillContextFilter())
log.addHandler(sh)
log.propagate = False
def get_skill_logger() -> logging.Logger:
if not _logger_name:
raise RuntimeError("get_skill_logger: call setup_skill_logging first")
return logging.getLogger(_logger_name)
def attach_unified_file_handler(
log_path: str,
*,
skill_slug: str,
logger_name: str,
level: int = logging.DEBUG,
) -> logging.Logger:
"""
独立子进程内追加同一日志文件(如 login 检测子进程),格式与主进程一致。
"""
global _skill_slug
ensure_trace_for_process()
_skill_slug = skill_slug
lg = logging.getLogger(logger_name)
lg.handlers.clear()
lg.setLevel(level)
parent = os.path.dirname(os.path.abspath(log_path))
if parent:
os.makedirs(parent, exist_ok=True)
fh = logging.FileHandler(log_path, encoding="utf-8")
fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
fh.setFormatter(fmt)
fh.addFilter(_SkillContextFilter())
lg.addHandler(fh)
lg.propagate = False
return lg

27
scripts/main.py Normal file
View File

@@ -0,0 +1,27 @@
"""
account-manager CLI 入口。
分层cliargv→ service用例→ dbSQLite共用工具在 util/。
"""
import os
import sys
_scripts_dir = os.path.dirname(os.path.abspath(__file__))
if _scripts_dir not in sys.path:
sys.path.insert(0, _scripts_dir)
from jiangchang_skill_core.runtime_env import apply_cli_local_defaults
apply_cli_local_defaults()
# Windows GBK 编码兼容修复(须在任何业务 import 与 print 之前)
if sys.platform == "win32":
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
from cli.app import main
if __name__ == "__main__":
main()

View File

@@ -0,0 +1 @@
"""Application services (use cases)."""

View File

@@ -0,0 +1,318 @@
"""账号用例list / get / add / delete / pick-web含终端输出供 CLI 调用)。"""
import json
import os
import shutil
import sys
import time
from db.accounts_repo import (
count_platform_accounts_conn,
default_name_for_platform,
delete_account_by_id_conn,
delete_account_platform_phone_conn,
delete_accounts_by_platform_conn,
fetch_delete_row_by_id_conn,
fetch_ids_for_list_json,
fetch_ids_profile_for_platform_conn,
fetch_list_rows,
fetch_pick_web_candidate_id,
fetch_row_platform_phone_conn,
get_account_by_id,
has_duplicate_phone_conn,
insert_account_row,
normalize_account_id,
)
from db.connection import get_conn, init_db
from util.logging_config import get_skill_logger
from util.platforms import (
PLATFORM_URLS,
_PLATFORM_PRIMARY_CN,
_is_valid_cn_mobile11,
_normalize_phone_digits,
_platform_list_cn_for_help,
resolve_platform_key,
)
from util.runtime_paths import _runtime_paths_debug_text, get_default_profile_dir
def _remove_profile_dir(path: str) -> None:
p = (path or "").strip()
if not p or not os.path.isdir(p):
return
try:
shutil.rmtree(p)
except OSError as e:
print(f"⚠️ 未能删除用户数据目录(可手工删):{p}\n {e}", file=sys.stderr)
def cmd_list(platform="all", limit: int = 10):
get_skill_logger().info("list filter=%r", platform)
init_db()
raw = (platform or "all").strip()
if not raw or raw.lower() == "all" or raw == "全部":
key = "all"
else:
key = resolve_platform_key(raw)
if not key:
get_skill_logger().warning("list_invalid_platform raw=%r", raw)
print(f"ERROR:INVALID_PLATFORM_LIST 无法识别的平台「{raw}")
print("支持:" + _platform_list_cn_for_help())
return
if limit <= 0:
limit = 10
rows = fetch_list_rows(key, int(limit))
found = bool(rows)
if found:
get_skill_logger().info("list_ok rows=%s key=%s", len(rows), key)
sep_line = "_" * 39
for idx, row in enumerate(rows):
(
aid,
name,
plat,
phone,
profile_dir,
url,
extra_json,
created_at,
updated_at,
) = row
print(f"id{aid}")
print(f"name{name or ''}")
print(f"platform{plat or ''}")
print(f"phone{phone or ''}")
print(f"profile_dir{profile_dir or ''}")
print(f"url{url or ''}")
print(f"extra_json{extra_json or ''}")
print(f"created_at{int(created_at) if created_at is not None else ''}")
print(f"updated_at{int(updated_at) if updated_at is not None else ''}")
if idx != len(rows) - 1:
print(sep_line)
print()
if not found:
print("ERROR:NO_ACCOUNTS_FOUND")
print(_runtime_paths_debug_text(), file=sys.stderr)
def cmd_get(account_id):
get_skill_logger().info("get account_id=%r", account_id)
acc = get_account_by_id(account_id)
if acc:
print(json.dumps(acc, ensure_ascii=False))
return
get_skill_logger().warning("get_not_found account_id=%r", account_id)
print("ERROR:ACCOUNT_NOT_FOUND")
print(_runtime_paths_debug_text(), file=sys.stderr)
def cmd_list_json(platform_input: str, limit: int = 200) -> None:
"""
跨技能接口stdout 仅输出一行 JSON 数组,元素结构与 get 单条一致。
平台可为 all/全部 或各平台中文名/英文键;按 updated_at 倒序,最多 limit 条(上限 500
"""
get_skill_logger().info("list_json filter=%r limit=%s", platform_input, limit)
init_db()
raw = (platform_input or "all").strip()
if not raw or raw.lower() == "all" or raw == "全部":
key = "all"
else:
key = resolve_platform_key(raw)
if not key:
print("ERROR:INVALID_PLATFORM_LIST_JSON 无法识别的平台名称。")
print("支持:" + _platform_list_cn_for_help())
return
lim = max(1, min(int(limit), 500))
ids = fetch_ids_for_list_json(key, lim)
out = []
for aid in ids:
acc = get_account_by_id(aid)
if acc:
out.append(acc)
print(json.dumps(out, ensure_ascii=False))
def cmd_pick_web(platform_input: str):
"""
供 llm-manager 等:取该平台用于网页自动化的账号候选。
按 updated_at、created_at 倒序选一条;是否已登录由调用方(如网页自动化)在同一会话内处理。
成功stdout 仅一行 JSON与 get 一致);失败:首行 ERROR:。
"""
get_skill_logger().info("pick_web platform_input=%r", platform_input)
key = resolve_platform_key((platform_input or "").strip())
if not key:
print("ERROR:INVALID_PLATFORM 无法识别的平台名称。")
print("支持:" + _platform_list_cn_for_help())
return
init_db()
picked = fetch_pick_web_candidate_id(key)
if picked is None:
print("ERROR:NO_ACCOUNT 该平台在账号库中没有任何记录,请先执行 add。")
return
acc = get_account_by_id(picked)
if not acc:
print("ERROR:ACCOUNT_NOT_FOUND")
print(_runtime_paths_debug_text(), file=sys.stderr)
return
print(json.dumps(acc, ensure_ascii=False))
def cmd_add(platform_input: str, phone: str):
"""添加账号platform 可为中文展示名或英文键;手机号必填;同平台下同手机号不可重复。"""
log = get_skill_logger()
log.info("add_attempt platform_input=%r", platform_input)
init_db()
key = resolve_platform_key((platform_input or "").strip())
if not key:
log.warning("add_invalid_platform input=%r", platform_input)
print("ERROR:INVALID_PLATFORM 无法识别的平台名称。")
print("支持:" + _platform_list_cn_for_help())
return
phone_raw = (phone or "").strip()
if not phone_raw:
log.warning("add_phone_missing")
print("ERROR:PHONE_REQUIRED 手机号为必填。")
return
phone_norm = _normalize_phone_digits(phone_raw)
if not _is_valid_cn_mobile11(phone_norm):
log.warning("add_phone_invalid digits_len=%s", len(phone_norm or ""))
print(
"ERROR:PHONE_INVALID 手机号格式不正确:须为中国大陆 11 位号码,"
"以 1 开头、第二位为 39示例 13800138000"
)
return
url = PLATFORM_URLS[key]
now = int(time.time())
phone_store = phone_norm
conn = get_conn()
try:
if has_duplicate_phone_conn(conn, key, phone_store):
log.warning("add_duplicate platform=%s phone_suffix=%s", key, phone_store[-4:])
print(
f"ERROR:DUPLICATE_PHONE_PLATFORM 该平台下已存在手机号 {phone_store},请勿重复添加。"
)
return
next_idx = count_platform_accounts_conn(conn, key) + 1
name = default_name_for_platform(key, next_idx)
profile_dir = get_default_profile_dir(key, phone_store, None)
new_id = insert_account_row(conn, name, key, phone_store, profile_dir, url, now)
conn.commit()
finally:
conn.close()
print(
f"✅ 已保存账号ID {new_id} | {name} | {_PLATFORM_PRIMARY_CN.get(key, key)} | 手机 {phone_store}"
)
print(f" 登录请执行python main.py login {new_id}")
log.info(
"add_success id=%s platform=%s profile_dir=%s",
new_id,
key,
profile_dir,
)
def cmd_delete_by_id(account_id) -> None:
log = get_skill_logger()
log.info("delete_by_id account_id=%r", account_id)
init_db()
aid = normalize_account_id(account_id)
if aid is None or (isinstance(aid, str) and not str(aid).isdigit()):
log.warning("delete_invalid_id")
print("ERROR:DELETE_INVALID_ID 账号 id 须为正整数。")
return
conn = get_conn()
try:
row = fetch_delete_row_by_id_conn(conn, int(aid))
if not row:
log.warning("delete_by_id_not_found id=%s", aid)
print("ERROR:ACCOUNT_NOT_FOUND")
print(_runtime_paths_debug_text(), file=sys.stderr)
return
rid, name, plat, phone, profile_dir = row
delete_account_by_id_conn(conn, rid)
conn.commit()
finally:
conn.close()
_remove_profile_dir(profile_dir)
print(
f"✅ 已删除账号ID {rid} | {name} | {_PLATFORM_PRIMARY_CN.get(plat, plat)} | 手机 {phone or '(无)'}"
)
log.info("delete_by_id_done id=%s platform=%s", rid, plat)
def cmd_delete_by_platform(platform_input: str) -> None:
log = get_skill_logger()
log.info("delete_by_platform platform_input=%r", platform_input)
init_db()
key = resolve_platform_key((platform_input or "").strip())
if not key:
log.warning("delete_by_platform_invalid_platform")
print("ERROR:INVALID_PLATFORM 无法识别的平台名称。")
print("支持:" + _platform_list_cn_for_help())
return
conn = get_conn()
try:
rows = fetch_ids_profile_for_platform_conn(conn, key)
if not rows:
log.warning("delete_by_platform_empty platform=%s", key)
print("ERROR:NO_ACCOUNTS_FOUND 该平台下没有账号记录。")
return
delete_accounts_by_platform_conn(conn, key)
conn.commit()
finally:
conn.close()
for _rid, pdir in rows:
_remove_profile_dir(pdir)
print(
f"✅ 已删除 {_PLATFORM_PRIMARY_CN.get(key, key)} 下共 {len(rows)} 条账号及对应用户数据目录。"
)
log.info("delete_by_platform_done platform=%s count=%s", key, len(rows))
def cmd_delete_by_platform_phone(platform_input: str, phone: str) -> None:
log = get_skill_logger()
log.info("delete_by_platform_phone platform_input=%r", platform_input)
init_db()
key = resolve_platform_key((platform_input or "").strip())
if not key:
print("ERROR:INVALID_PLATFORM 无法识别的平台名称。")
print("支持:" + _platform_list_cn_for_help())
return
phone_raw = (phone or "").strip()
if not phone_raw:
print("ERROR:PHONE_REQUIRED 手机号为必填。")
return
phone_norm = _normalize_phone_digits(phone_raw)
if not _is_valid_cn_mobile11(phone_norm):
print(
"ERROR:PHONE_INVALID 手机号格式不正确:须为中国大陆 11 位号码,"
"以 1 开头、第二位为 39。"
)
return
conn = get_conn()
try:
row = fetch_row_platform_phone_conn(conn, key, phone_norm)
if not row:
print("ERROR:ACCOUNT_NOT_FOUND 该平台下无此手机号。")
return
rid, name, profile_dir = row
delete_account_platform_phone_conn(conn, key, phone_norm)
conn.commit()
finally:
conn.close()
_remove_profile_dir(profile_dir)
print(
f"✅ 已删除账号ID {rid} | {name} | {_PLATFORM_PRIMARY_CN.get(key, key)} | 手机 {phone_norm}"
)
log.info("delete_by_platform_phone_done id=%s platform=%s", rid, key)

View File

@@ -0,0 +1,398 @@
"""Chromium/Edge 检测与 Playwright 登录、仅打开浏览器。"""
import json
import os
import subprocess
import sys
import tempfile
import time
from urllib.parse import urlparse
from db.accounts_repo import get_account_by_id, touch_account_updated_at
from util.logging_config import (
get_skill_logger,
get_skill_log_file_path,
subprocess_env_with_trace,
)
from util.platforms import (
PLATFORMS,
PLATFORM_URLS,
_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS,
)
# 子进程入口脚本与本模块同目录(独立解释器执行,非 import
_SERVICE_DIR = os.path.dirname(os.path.abspath(__file__))
# PyArmor 在模块首行注入对 pyarmor_runtime_* 的 import若以「脚本路径」启动子进程
# sys.path[0] 仅为 service/,找不到 scripts/ 下的运行时包。必须用 -m 且 cwd=scripts/(标准做法)。
_SCRIPTS_ROOT = os.path.dirname(_SERVICE_DIR)
def _win_find_exe(candidates):
for p in candidates:
if p and os.path.isfile(p):
return p
return None
def resolve_chromium_channel():
"""
Playwright channel: 'chrome' | 'msedge' | None
Windows 优先 Chrome其次 Edge其它系统默认 chrome由 Playwright 解析 PATH
"""
if sys.platform != "win32":
return "chrome"
pf = os.environ.get("ProgramFiles", r"C:\Program Files")
pfx86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)")
local = os.environ.get("LocalAppData", "")
chrome = _win_find_exe(
[
os.path.join(pf, "Google", "Chrome", "Application", "chrome.exe"),
os.path.join(pfx86, "Google", "Chrome", "Application", "chrome.exe"),
os.path.join(local, "Google", "Chrome", "Application", "chrome.exe") if local else "",
]
)
if chrome:
return "chrome"
edge = _win_find_exe(
[
os.path.join(pfx86, "Microsoft", "Edge", "Application", "msedge.exe"),
os.path.join(pf, "Microsoft", "Edge", "Application", "msedge.exe"),
os.path.join(local, "Microsoft", "Edge", "Application", "msedge.exe") if local else "",
]
)
if edge:
return "msedge"
return None
def _print_browser_install_hint():
print("❌ 未检测到 Google Chrome 或 Microsoft Edge请先安装后再登录")
print(" • Chrome: https://www.google.com/chrome/")
print(" • Edge: https://www.microsoft.com/zh-cn/edge")
def _login_timeout_seconds():
try:
return max(60, int((os.getenv("JIANGCHANG_LOGIN_TIMEOUT_SECONDS") or "300").strip()))
except ValueError:
return 300
def _login_poll_interval_seconds():
"""轮询间隔,默认 1.5s。"""
try:
v = float((os.getenv("JIANGCHANG_LOGIN_POLL_INTERVAL_SECONDS") or "1.5").strip())
return max(0.5, min(v, 10.0))
except ValueError:
return 1.5
def _login_dom_grace_seconds() -> float:
"""goto 后等待再开始 DOM 判定,减轻跳转中途误判。默认 4sJIANGCHANG_LOGIN_DOM_GRACE_SECONDS。"""
try:
v = float((os.getenv("JIANGCHANG_LOGIN_DOM_GRACE_SECONDS") or "4").strip())
return max(0.0, min(v, 60.0))
except ValueError:
return 4.0
def _login_out_bundle_for(platform_key: str) -> dict:
"""传给子进程anchor + 未登录可见的选择器列表。"""
spec = PLATFORMS.get(platform_key, {}) if platform_key else {}
anchor = (urlparse(PLATFORM_URLS.get(platform_key, "") or "").netloc or "").lower()
spec_lo_dom = [
str(s).strip()
for s in (spec.get("login_logged_out_selectors") or [])
if s is not None and str(s).strip()
]
if bool(spec.get("login_skip_generic_logged_out_dom")):
logged_out_dom_selectors = list(spec_lo_dom)
else:
logged_out_dom_selectors = list(_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS) + spec_lo_dom
return {
"logged_out_dom_selectors": logged_out_dom_selectors,
"anchor_host": anchor,
}
def _run_login_browser_session(
target: dict,
timeout_sec: int,
*,
verbose_ui: bool = True,
) -> tuple[bool, str]:
"""
启动 Playwright 登录检测子进程;**是否已登录以子进程 DOM 判定为准**。
若 DOM 判定成功则刷新该账号 `updated_at`,便于 `pick-web` 等按最近活跃选号;**是否已登录仅以本次检测为准**。
返回 (是否已登录成功, end_reason)end_reason 为 login_child 写入的 reason失败时可能为
timeout / user_closed / subprocess_error。
"""
log = get_skill_logger()
account_id = target.get("id")
channel = resolve_chromium_channel()
if not channel:
log.warning("login_aborted no_chromium_channel")
if verbose_ui:
_print_browser_install_hint()
return False, "no_channel"
try:
import playwright # noqa: F401
except ImportError:
log.error("login_aborted playwright_missing")
if verbose_ui:
print("ERROR:需要 playwright Python 包:请在匠厂共享环境中已含 playwright本机需安装 Google Chrome 或 Microsoft Edgechannel 模式,无需 playwright install chromium")
return False, "playwright_missing"
profile_dir = (target.get("profile_dir") or "").strip()
if not profile_dir:
log.warning("login_aborted profile_dir_empty account_id=%s", account_id)
if verbose_ui:
print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。")
return False, "profile_missing"
os.makedirs(profile_dir, exist_ok=True)
url = (target.get("url") or "").strip() or PLATFORM_URLS.get(
target["platform"], "https://www.google.com"
)
platform = (target.get("platform") or "").strip().lower()
poll_sec = _login_poll_interval_seconds()
dom_grace_sec = _login_dom_grace_seconds()
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
if verbose_ui:
print(f"正在为账号 [{target['name']}] 打开 {browser_name}")
print(f"访问地址:{url}")
if platform == "gemini":
print(
"Gemini / Google 登录:若提示「浏览器不安全」或「无法登录」,请关闭本窗口后,"
"用普通 Chrome 启动参数 --user-data-dir= 指向下面 profile 目录,在普通浏览器里完成"
" Google 登录后完全退出 Chrome再执行本 login详见 llm-manager Gemini 引擎说明。"
)
print(f" (本账号 profile 目录){profile_dir}")
print(
"请在窗口内完成登录。系统根据页面是否仍出现「登录」等未登录控件判断;"
"成功后自动关闭窗口并写入状态(无需手动关浏览器)。"
)
print(
f"页面加载后先等待约 {dom_grace_sec:g} 秒再开始检测(减轻跳转误判,可用 JIANGCHANG_LOGIN_DOM_GRACE_SECONDS 调整)。"
)
print(f"轮询间隔约 {poll_sec} 秒;全程最长 {timeout_sec} 秒。")
log_file = get_skill_log_file_path()
log.info(
"login_browser_start account_id=%s platform=%s channel=%s timeout_sec=%s poll_sec=%s "
"dom_grace_sec=%s profile_dir=%s start_url=%s log_file=%s",
account_id,
platform,
channel,
timeout_sec,
poll_sec,
dom_grace_sec,
profile_dir,
url,
log_file,
)
cfg = {
"channel": channel,
"profile_dir": profile_dir,
"url": url,
"timeout_sec": timeout_sec,
"poll_interval": poll_sec,
"dom_grace_sec": dom_grace_sec,
"login_detect_bundle": _login_out_bundle_for(platform),
"log_file": log_file,
}
with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False, encoding="utf-8"
) as rf:
result_path = rf.name
cfg["result_path"] = result_path
with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False, encoding="utf-8"
) as jf:
json.dump(cfg, jf, ensure_ascii=False)
cfg_path = jf.name
proc_rc = None
child_hard_timeout = False
child_stderr = ""
child_stdout = ""
try:
r = subprocess.run(
[sys.executable, "-m", "service.login_child_runner", cfg_path],
cwd=_SCRIPTS_ROOT,
timeout=timeout_sec + 180,
env=subprocess_env_with_trace(),
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
)
proc_rc = r.returncode
child_stderr = (r.stderr or "").strip()
child_stdout = (r.stdout or "").strip()
if child_stderr:
log.warning(
"login_child_stderr account_id=%s len=%s tail=%s",
account_id,
len(child_stderr),
child_stderr[-6000:],
)
if child_stdout and proc_rc != 0:
log.warning(
"login_child_stdout account_id=%s len=%s tail=%s",
account_id,
len(child_stdout),
child_stdout[-2000:],
)
if proc_rc != 0:
if verbose_ui:
print("⚠️ 浏览器子进程非零退出,以结果 JSON 为准(若已写入)")
log.info(
"login_subprocess_nonzero_return code=%s (exit code 不单独作为成败依据,见 result JSON)",
proc_rc,
)
except subprocess.TimeoutExpired as ex:
child_hard_timeout = True
if verbose_ui:
print("⚠️ 浏览器子进程等待超时,按「限时内未完成登录」处理")
log.warning("login_subprocess_hard_timeout err=%s", ex)
finally:
try:
os.unlink(cfg_path)
except OSError:
pass
interactive_ok = False
end_reason = "timeout" if child_hard_timeout else "subprocess_error"
child_detail = ""
try:
with open(result_path, encoding="utf-8") as rf:
data = json.load(rf)
interactive_ok = bool(data.get("interactive_ok"))
er = str(data.get("end_reason") or "").strip()
if er:
end_reason = er
elif interactive_ok:
end_reason = "login_ok"
elif not child_hard_timeout:
end_reason = "timeout"
child_detail = str(data.get("detail") or "").strip()
if child_detail:
log.warning("login_child_result_detail account_id=%s detail=%s", account_id, child_detail[:8000])
except Exception as read_ex:
log.error(
"login_child_result_read_failed account_id=%s err=%s proc_rc=%s stderr_tail=%s",
account_id,
read_ex,
proc_rc,
(child_stderr or "")[-3000:],
)
if child_hard_timeout:
end_reason = "timeout"
interactive_ok = False
elif proc_rc is not None and proc_rc != 0:
end_reason = "child_crash"
else:
end_reason = "subprocess_error"
try:
os.unlink(result_path)
except OSError:
pass
time.sleep(0.5)
ok = interactive_ok
log.info(
"login_finished account_id=%s interactive_ok=%s end_reason=%s subprocess_rc=%s marking_db=%s",
account_id,
interactive_ok,
end_reason,
proc_rc,
ok,
)
if ok:
touch_account_updated_at(account_id)
return ok, end_reason if not ok else "login_ok"
def cmd_login(account_id):
log = get_skill_logger()
log.info("login_command account_id=%r", account_id)
target = get_account_by_id(account_id)
if not target:
log.warning("login_aborted account_not_found account_id=%r", account_id)
print("ERROR:ACCOUNT_NOT_FOUND")
return
timeout_sec = _login_timeout_seconds()
ok, _end_reason = _run_login_browser_session(target, timeout_sec, verbose_ui=True)
log_file = get_skill_log_file_path()
if ok:
print("✅ 已判定页面登录成功")
else:
log.warning(
"login_not_detected account_id=%s platform=%s see_log=%s",
target.get("id"),
(target.get("platform") or "").strip().lower(),
log_file,
)
print("⚠️ 未检测到有效登录。请关闭其他占用该用户目录的浏览器后重试,或延长 JIANGCHANG_LOGIN_TIMEOUT_SECONDS 后再登录。")
print(f" 详细轮询日志见:{log_file}(可将 JIANGCHANG_LOG_LEVEL=DEBUG 打开更细粒度)")
def cmd_open(account_id):
"""打开该账号的持久化浏览器,仅用于肉眼确认是否已登录;不写数据库。"""
get_skill_logger().info("open account_id=%r", account_id)
target = get_account_by_id(account_id)
if not target:
get_skill_logger().warning("open_not_found account_id=%r", account_id)
print("ERROR:ACCOUNT_NOT_FOUND")
return
channel = resolve_chromium_channel()
if not channel:
_print_browser_install_hint()
return
try:
import playwright # noqa: F401
except ImportError:
print("ERROR:需要 playwright Python 包:请在匠厂共享环境中已含 playwright本机需安装 Google Chrome 或 Microsoft Edgechannel 模式,无需 playwright install chromium")
return
profile_dir = (target.get("profile_dir") or "").strip()
if not profile_dir:
print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。")
return
os.makedirs(profile_dir, exist_ok=True)
url = (target.get("url") or "").strip() or PLATFORM_URLS.get(
target["platform"], "https://www.google.com"
)
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
print(f"正在打开 [{target['name']}] 的 {browser_name}(仅查看,不写入数据库)")
print(f"地址:{url}")
print("请在本窗口中自行确认登录态。关闭浏览器后命令结束。")
print("需要自动检测登录时请执行python main.py login <id>")
cfg = {"channel": channel, "profile_dir": profile_dir, "url": url}
with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False, encoding="utf-8"
) as jf:
json.dump(cfg, jf, ensure_ascii=False)
cfg_path = jf.name
try:
subprocess.run(
[sys.executable, "-m", "service.open_child_runner", cfg_path],
cwd=_SCRIPTS_ROOT,
env=subprocess_env_with_trace(),
)
finally:
try:
os.unlink(cfg_path)
except OSError:
pass

View File

@@ -0,0 +1,390 @@
"""Playwright 登录检测子进程:由 browser_service 以独立解释器启动argv[1] 为 JSON 配置路径。
注意:不得在模块顶层 import playwright / jiangchang_skill_core —— 若导入失败,进程会在 main() 之前以退出码 1 退出,
且写不了 result_path父进程只能得到 subprocess_error。所有重依赖均在 _run_login_child 内延迟加载。
"""
from __future__ import annotations
import json
import os
import sys
import time
import traceback
from urllib.parse import urlparse
_scripts_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _scripts_dir not in sys.path:
sys.path.insert(0, _scripts_dir)
from util.playwright_stealth import (
STEALTH_INIT_SCRIPT,
persistent_context_launch_parts,
stealth_enabled,
)
def _is_context_closed_error(ex: BaseException) -> bool:
s = str(ex).lower()
if "target closed" in s or "browser has been closed" in s or "context was destroyed" in s:
return True
if "closed" in s or "destroyed" in s:
return True
return False
def _write_result_file(
result_path: str,
interactive_ok: bool,
end_reason: str,
detail: str = "",
) -> None:
if not (result_path or "").strip():
return
payload: dict = {"interactive_ok": interactive_ok, "end_reason": end_reason}
d = (detail or "").strip()
if d:
payload["detail"] = d[:8000]
try:
with open(result_path, "w", encoding="utf-8") as rf:
json.dump(payload, rf, ensure_ascii=False)
except Exception as write_ex:
print(f"login_child: failed to write result file: {write_ex}", file=sys.stderr)
def page_location_href(p):
try:
href = p.evaluate("() => location.href")
if href and str(href).strip():
return str(href).strip()
except Exception:
pass
try:
return (p.url or "").strip()
except Exception:
return ""
def host_matches_anchor(netloc, anchor):
h = (netloc or "").lower().strip()
a = (anchor or "").lower().strip()
if not a:
return True
if not h:
return False
return h == a or h.endswith("." + a) or a.endswith("." + h)
def loc_first_visible(pg, selectors):
for sel in selectors:
s = str(sel).strip()
if not s:
continue
try:
loc = pg.locator(s).first
if loc.is_visible(timeout=500):
return True, s
except Exception:
pass
return False, None
def list_pages_on_anchor(ctx, main_page, bundle):
anchor = (bundle.get("anchor_host") or "").strip().lower()
seen_ids = set()
pages = []
for pg in list(ctx.pages or []):
try:
i = id(pg)
if i not in seen_ids:
seen_ids.add(i)
pages.append(pg)
except Exception:
pages.append(pg)
if main_page is not None:
try:
i = id(main_page)
if i not in seen_ids:
seen_ids.add(i)
pages.append(main_page)
except Exception:
pages.append(main_page)
out = []
for pg in pages:
try:
href = page_location_href(pg)
except Exception:
continue
if not href or str(href).lower().startswith("about:"):
continue
try:
host = urlparse(href).netloc.lower()
except Exception:
continue
if anchor and not host_matches_anchor(host, anchor):
continue
out.append(pg)
return out
def evaluate_logged_out_dom(ctx, main_page, bundle):
selectors = [
str(x).strip()
for x in (bundle.get("logged_out_dom_selectors") or [])
if x is not None and str(x).strip()
]
if not selectors:
return True, "no_logged_out_dom_selectors_configured"
for pg in list_pages_on_anchor(ctx, main_page, bundle):
try:
href = page_location_href(pg)
except Exception:
href = ""
hit, which = loc_first_visible(pg, selectors)
if hit:
return True, "logged_out_dom=%r url=%r" % (which, href)
return False, ""
def _run_login_child(c: dict) -> None:
"""单次登录检测;结果一律写入 c['result_path'](由 finally 保证)。"""
result_path = (c.get("result_path") or "").strip()
try:
from playwright.sync_api import sync_playwright
except ImportError:
_write_result_file(
result_path,
False,
"playwright_missing",
traceback.format_exc(),
)
return
interactive_ok = False
end_reason = "timeout"
ctx = None
lg = None
last_eval_detail = ""
detail_msg = ""
try:
try:
from jiangchang_skill_core.unified_logging import attach_unified_file_handler
except Exception as e:
end_reason = "logging_import_failed"
detail_msg = traceback.format_exc()
print(detail_msg, file=sys.stderr)
return
try:
profile_dir = (c.get("profile_dir") or "").strip()
if not profile_dir:
end_reason = "profile_missing"
detail_msg = "profile_dir empty in config"
return
os.makedirs(profile_dir, exist_ok=True)
try:
n_entries = len(os.listdir(profile_dir))
except OSError as e:
end_reason = "profile_dir_unusable"
detail_msg = f"{type(e).__name__}: {e}"
return
except Exception as e:
end_reason = "profile_dir_unusable"
detail_msg = traceback.format_exc()
print(detail_msg, file=sys.stderr)
return
log_path = (c.get("log_file") or "").strip()
if log_path:
try:
lg = attach_unified_file_handler(
log_path,
skill_slug="account-manager",
logger_name="openclaw.skill.account_manager.login_child",
)
except Exception as e:
lg = None
print(f"login_child: attach_unified_file_handler failed: {e}", file=sys.stderr)
if lg is not None:
lg.info(
"login_child_start profile_dir=%r profile_entry_count=%s channel=%r url=%r timeout_sec=%r",
profile_dir,
n_entries,
c.get("channel"),
c.get("url"),
c.get("timeout_sec"),
)
bundle = c.get("login_detect_bundle") or {}
poll = float(c.get("poll_interval", 1.5))
try:
dom_grace = float(c.get("dom_grace_sec", 4.0))
except (TypeError, ValueError):
dom_grace = 4.0
if dom_grace < 0:
dom_grace = 0.0
t0 = time.time()
deadline = t0 + float(c["timeout_sec"])
with sync_playwright() as p:
try:
args, ignore_automation = persistent_context_launch_parts()
lc_kw = dict(
user_data_dir=profile_dir,
headless=False,
channel=c["channel"],
no_viewport=True,
args=args,
)
if ignore_automation is not None:
lc_kw["ignore_default_args"] = ignore_automation
ctx = p.chromium.launch_persistent_context(**lc_kw)
if stealth_enabled():
ctx.add_init_script(STEALTH_INIT_SCRIPT)
except Exception as e:
if _is_context_closed_error(e):
end_reason = "user_closed"
else:
end_reason = "launch_failed"
detail_msg = f"{type(e).__name__}: {e}"
if lg is not None:
lg.exception("launch_persistent_context failed: %s", e)
print(detail_msg, file=sys.stderr)
return
try:
page = ctx.pages[0] if ctx.pages else ctx.new_page()
page.goto(c["url"], wait_until="domcontentloaded", timeout=60000)
if dom_grace > 0:
time.sleep(dom_grace)
if lg is not None:
lg.info(
"goto_done initial_pages=%s poll_interval_sec=%s dom_grace_sec=%s",
len(ctx.pages or []),
poll,
dom_grace,
)
while time.time() < deadline:
iter_start = time.time()
try:
still_logged_out, dom_detail = evaluate_logged_out_dom(ctx, page, bundle)
ok = not still_logged_out
last_eval_detail = dom_detail
if lg is not None:
parts = []
for tab in list(ctx.pages or []):
try:
href = page_location_href(tab)
pu = (tab.url or "").strip()
parts.append("href=%r playwright_url=%r" % (href, pu))
except Exception:
parts.append("(tab_read_error)")
lg.debug(
"poll elapsed=%.1fs left=%.1fs tabs=%s logged_out=%s ok=%s detail=%s | %s",
iter_start - t0,
deadline - iter_start,
len(ctx.pages or []),
still_logged_out,
ok,
dom_detail,
" ; ".join(parts) if parts else "no_tabs",
)
if ok:
interactive_ok = True
end_reason = "login_ok"
if lg is not None:
lg.info("login_detected_ok %s", dom_detail)
break
except Exception as ex:
if _is_context_closed_error(ex):
end_reason = "user_closed"
if lg is not None:
lg.warning("login_poll_browser_closed err=%s", ex)
break
if lg is not None:
lg.warning("poll_exception err=%s", ex, exc_info=True)
spent = time.time() - iter_start
time.sleep(max(0.0, poll - spent))
if not interactive_ok and end_reason != "user_closed":
if lg is not None:
lg.warning("login_poll_exhausted detail=%s", last_eval_detail)
rem = max(0.0, deadline - time.time())
if rem > 0:
try:
ctx.wait_for_event("close", timeout=rem * 1000)
except Exception:
pass
elif interactive_ok:
if lg is not None:
lg.info("login_dom_ok_preparing_to_release_profile_dir")
print(
"INFO:LOGIN_DOM_OK 页面检测已判定登录成功;约 2 秒后关闭本窗口以释放用户目录,供后续大模型网页会话使用。",
flush=True,
)
time.sleep(2.0)
except Exception as e:
interactive_ok = False
if _is_context_closed_error(e):
end_reason = "user_closed"
if lg is not None:
lg.warning("login_runner_browser_closed err=%s", e)
else:
end_reason = "page_error"
detail_msg = traceback.format_exc()
if lg is not None:
lg.exception("login_runner_page_err %s", e)
print(detail_msg, file=sys.stderr)
finally:
try:
ctx.close()
except Exception:
pass
ctx = None
except Exception as e:
interactive_ok = False
end_reason = "child_fatal"
detail_msg = traceback.format_exc()
if lg is not None:
lg.exception("login_child_unhandled %s", e)
print(detail_msg, file=sys.stderr)
finally:
if ctx is not None:
try:
ctx.close()
except Exception:
pass
_write_result_file(result_path, interactive_ok, end_reason, detail_msg)
def _entrypoint() -> None:
"""读取配置 → 运行检测;任意异常都写入 result JSON进程以 0 退出(由父进程读 JSON 判定成败)。"""
result_path = ""
if len(sys.argv) < 2:
print("login_child: missing cfg path argv", file=sys.stderr)
return
cfg_path = sys.argv[1]
try:
with open(cfg_path, encoding="utf-8") as f:
c = json.load(f)
except Exception:
tb = traceback.format_exc()
print(tb, file=sys.stderr)
return
result_path = (c.get("result_path") or "").strip()
try:
_run_login_child(c)
except Exception:
tb = traceback.format_exc()
print(tb, file=sys.stderr)
_write_result_file(result_path, False, "child_fatal", tb)
if __name__ == "__main__":
_entrypoint()

View File

@@ -0,0 +1,46 @@
"""Playwright 仅打开浏览器子进程:由 browser_service.cmd_open 启动argv[1] 为 JSON 配置路径。"""
import json
import sys
from playwright.sync_api import sync_playwright
from util.playwright_stealth import (
STEALTH_INIT_SCRIPT,
persistent_context_launch_parts,
stealth_enabled,
)
def main():
with open(sys.argv[1], encoding="utf-8") as f:
c = json.load(f)
with sync_playwright() as p:
args, ignore_automation = persistent_context_launch_parts()
lc_kw = dict(
user_data_dir=c["profile_dir"],
headless=False,
channel=c["channel"],
no_viewport=True,
args=args,
)
if ignore_automation is not None:
lc_kw["ignore_default_args"] = ignore_automation
ctx = p.chromium.launch_persistent_context(**lc_kw)
if stealth_enabled():
ctx.add_init_script(STEALTH_INIT_SCRIPT)
try:
page = ctx.pages[0] if ctx.pages else ctx.new_page()
page.goto(c["url"], wait_until="domcontentloaded", timeout=60000)
ctx.wait_for_event("close", timeout=86400000)
except Exception as e:
print(e, file=sys.stderr)
raise
finally:
try:
ctx.close()
except Exception:
pass
if __name__ == "__main__":
main()

1
scripts/util/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Shared helpers for account-manager (paths, platforms, logging, constants)."""

View File

@@ -0,0 +1,9 @@
"""技能级常量。本地 CLI 默认值见 jiangchang_skill_core.runtime_envmain.py 最早调用 apply_cli_local_defaults"""
import os
# scripts/util/constants.py -> skill root = parents[2]
_BASE = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
BASE_DIR = _BASE
SKILL_SLUG = "account-manager"
# 与其它 OpenClaw 技能对齐openclaw.skill.<slug_下划线>
LOG_LOGGER_NAME = "openclaw.skill.account_manager"

View File

@@ -0,0 +1,21 @@
"""Re-export unified logging (implementation: jiangchang_skill_core.unified_logging)."""
from jiangchang_skill_core.unified_logging import (
attach_unified_file_handler,
ensure_trace_for_process,
get_skill_log_file_path,
get_skill_logger,
get_unified_logs_dir,
setup_skill_logging,
subprocess_env_with_trace,
)
__all__ = [
"attach_unified_file_handler",
"ensure_trace_for_process",
"get_skill_log_file_path",
"get_skill_logger",
"get_unified_logs_dir",
"setup_skill_logging",
"subprocess_env_with_trace",
]

145
scripts/util/platforms.py Normal file
View File

@@ -0,0 +1,145 @@
"""平台配置、别名解析、手机号校验。"""
import re
# 平台唯一配置表:只改这里即可。键 = 入库的 platformlabel = 帮助/提示里的展示名;
# prefix = 默认账号名「{prefix}{序号}号」,省略时等于 label
# aliases = 额外 CLI 称呼(英文键与 label 已自动参与解析,不必重复写)。
# 登录检测:仅看页面 DOM。匹配 anchor 的标签页上出现「未登录」选择器则未登录;否则视为已登录。
# - _LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS通用「登录」按钮/链接等(可被 login_skip_generic_logged_out_dom 关闭)。
# - login_logged_out_selectors按平台追加误判时也可用 login_skip_generic_logged_out_dom + 仅平台自选。
PLATFORMS = {
"sohu": {
"url": "https://mp.sohu.com",
"label": "搜狐号",
"prefix": "搜狐",
"aliases": ["搜狐"],
},
"toutiao": {
"url": "https://mp.toutiao.com/",
"label": "头条号",
"prefix": "头条",
"aliases": ["头条"],
},
"zhihu": {
"url": "https://www.zhihu.com",
"label": "知乎",
"aliases": ["知乎号"],
},
"wechat": {
"url": "https://mp.weixin.qq.com",
"label": "微信公众号",
"prefix": "微信",
"aliases": ["公众号", "微信"],
},
"kimi": {
"url": "https://kimi.moonshot.cn",
"label": "Kimi",
"aliases": ["月之暗面"],
},
"deepseek": {
"url": "https://chat.deepseek.com",
"label": "DeepSeek",
},
"doubao": {
"url": "https://www.doubao.com",
"label": "豆包",
},
"qianwen": {
"url": "https://tongyi.aliyun.com",
"label": "通义千问",
"prefix": "通义",
"aliases": ["通义", "千问"],
},
"yiyan": {
"url": "https://yiyan.baidu.com",
"label": "文心一言",
"prefix": "文心",
"aliases": ["文心", "一言"],
},
"yuanbao": {
"url": "https://yuanbao.tencent.com",
"label": "腾讯元宝",
"prefix": "元宝",
"aliases": ["元宝"],
},
"gemini": {
"url": "https://gemini.google.com",
"label": "Gemini",
"prefix": "Gemini",
"aliases": ["谷歌Gemini", "Google Gemini", "google gemini", "Bard"],
},
}
# 未登录页共性主行动点含「登录」——submit 按钮、Element 主按钮、Semi 文案、标题、通栏主按钮等Playwright :has-text 可覆盖多数站点。
_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS = (
'button:has-text("登录")',
'role=button[name="登录"]',
'a:has-text("登录")',
'h4:has-text("登录")',
'span.semi-button-content:has-text("登录")',
)
def _build_platform_derived():
urls = {}
name_prefix = {}
primary_cn = {}
alias_to_key = {}
def _register_alias(alias: str, key: str) -> None:
if not alias or not str(alias).strip():
return
a = str(alias).strip()
if a not in alias_to_key:
alias_to_key[a] = key
lo = a.lower()
if lo not in alias_to_key:
alias_to_key[lo] = key
for key, spec in PLATFORMS.items():
urls[key] = spec["url"]
primary_cn[key] = spec["label"]
name_prefix[key] = (spec.get("prefix") or spec["label"]).strip() or key
_register_alias(key, key)
_register_alias(spec["label"], key)
for a in spec.get("aliases") or []:
_register_alias(a, key)
return urls, name_prefix, primary_cn, alias_to_key
PLATFORM_URLS, _PLATFORM_NAME_CN, _PLATFORM_PRIMARY_CN, _PLATFORM_ALIAS_TO_KEY = _build_platform_derived()
def resolve_platform_key(name: str):
"""将用户输入解析为内部 platform 键;无法识别返回 None。"""
if name is None:
return None
s = str(name).strip()
if not s:
return None
sl = s.lower()
if sl in PLATFORM_URLS:
return sl
return _PLATFORM_ALIAS_TO_KEY.get(s)
def _normalize_phone_digits(phone: str) -> str:
"""从输入中提取数字串,供号段规则校验与入库去重(不对外承诺可随意夹杂符号)。"""
d = re.sub(r"\D", "", (phone or "").strip())
if d.startswith("86") and len(d) >= 13:
d = d[2:]
return d
def _is_valid_cn_mobile11(digits: str) -> bool:
"""中国大陆 11 位手机号1 开头,第二位 39共 11 位数字。"""
return bool(re.fullmatch(r"1[3-9]\d{9}", digits or ""))
def _platform_list_cn_for_help() -> str:
"""帮助文案:一行中文展示名(不重复内部键)。"""
parts = []
for k in sorted(PLATFORM_URLS.keys()):
parts.append(_PLATFORM_PRIMARY_CN.get(k, k))
return "".join(parts)

View File

@@ -0,0 +1,40 @@
"""
淡化 Playwright 启动时的自动化指纹,减轻 Google 账号页「此浏览器或应用可能不安全」类拦截。
关闭:环境变量 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) {}
})();
"""
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_persistent_context。
"""
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"]

View File

@@ -0,0 +1,80 @@
"""数据目录、profile 路径、CLI 路径调试输出。"""
import os
import re
import sys
from typing import Optional
from jiangchang_skill_core.runtime_env import get_data_root, get_user_id
from util.constants import SKILL_SLUG
from util.platforms import _PLATFORM_PRIMARY_CN
def get_skill_data_dir():
path = os.path.join(get_data_root(), get_user_id(), SKILL_SLUG)
os.makedirs(path, exist_ok=True)
return path
def get_db_path():
return os.path.join(get_skill_data_dir(), "account-manager.db")
def _runtime_paths_debug_text():
"""供排查「为何 list 没数据」:终端未注入环境变量时会落到 _anon 等默认目录,与网关里用的库不是同一个文件。"""
env_root = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip() or "(未设置)"
env_uid = (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "(未设置→使用 _anon)"
return (
"[account-manager] "
f"JIANGCHANG_DATA_ROOT={env_root} | "
f"JIANGCHANG_USER_ID={env_uid} | "
f"实际数据根={get_data_root()} | "
f"实际用户目录={get_user_id()} | "
f"数据库文件={get_db_path()}"
)
def _maybe_print_paths_debug_cli():
"""设置 JIANGCHANG_ACCOUNT_DEBUG_PATHS=1 时,每次执行子命令前在 stderr 打印路径。"""
v = (os.getenv("JIANGCHANG_ACCOUNT_DEBUG_PATHS") or "").strip().lower()
if v in ("1", "true", "yes", "on"):
print(_runtime_paths_debug_text(), file=sys.stderr)
_WIN_FS_FORBIDDEN = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
_WIN_RESERVED_NAMES = frozenset(
["CON", "PRN", "AUX", "NUL"]
+ [f"COM{i}" for i in range(1, 10)]
+ [f"LPT{i}" for i in range(1, 10)]
)
def _fs_safe_segment(name: str, fallback: str) -> str:
"""单级目录名:去掉 Windows 非法字符,避免末尾点/空格;空则用 fallback。"""
t = _WIN_FS_FORBIDDEN.sub("_", (name or "").strip())
t = t.rstrip(" .")
if not t:
t = fallback
if t.upper() in _WIN_RESERVED_NAMES:
t = f"_{t}"
return t
def get_default_profile_dir(
platform_key: str, phone: Optional[str], account_id: Optional[int] = None
) -> str:
"""
默认可读路径profiles/<平台展示名>/<手机号>/(便于在资源管理器中辨认)。
无手机号时退化为 profiles/<平台展示名>/no_phone_<id>/。
"""
label = _PLATFORM_PRIMARY_CN.get(platform_key, platform_key or "account")
label_seg = _fs_safe_segment(label, _fs_safe_segment(platform_key or "", "platform"))
ph = (phone or "").strip()
if ph:
phone_seg = _fs_safe_segment(ph, f"id_{account_id}" if account_id is not None else "phone")
else:
phone_seg = _fs_safe_segment(
"", f"no_phone_{account_id}" if account_id is not None else "no_phone"
)
path = os.path.join(get_skill_data_dir(), "profiles", label_seg, phone_seg)
os.makedirs(path, exist_ok=True)
return path