Compare commits

...

12 Commits

Author SHA1 Message Date
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
33 changed files with 2212 additions and 551 deletions

View File

@@ -6,6 +6,8 @@ on:
jobs:
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,7 +1,7 @@
---
name: 账号管理
description: 多平台多账号管理。管理各平台账号与Chrome Profile的对应关系供publisher类Skill调用获取账号信息。
version: 1.0.2
version: 1.0.4
author: 深圳匠厂科技有限公司
metadata:
openclaw:
@@ -23,12 +23,46 @@ allowed-tools:
## 执行步骤
### 添加账号(仅需平台,可选手机号ID 数据库自增,名称自动生成如「搜狐1号」
### 添加账号(平台用中文名;须填合法中国大陆 11 位手机号;同平台下同手机号不可重复ID 自增名称如「搜狐1号」
```bash
python3 {baseDir}/scripts/account.py add sohu
python3 {baseDir}/scripts/account.py add sohu 189xxxxxxxx
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>
```
### 列出某平台所有账号
### 登录并自动检测、写入数据库
```bash
python3 {baseDir}/scripts/account.py list <platform>
python3 {baseDir}/scripts/main.py login <id>
```
### 删除账号(同时删库里的记录与 profile 用户数据目录)
```bash
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 搜狐号
```
### 跨技能:单行 JSONget / list-json / pick-web / pick-logged-in / set-login-status
```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 搜狐号
python3 {baseDir}/scripts/main.py set-login-status <id> 1
```
约定与字段说明见 `references/INTEGRATION.md`;错误前缀见 `references/ERRORS.md`
## 扩展文档(便于 AI 精确调用)
- **references/**:完整 CLI、平台表、运行时环境、错误码、集成协议 — 见 [references/README.md](references/README.md)。
- **assets/**:示例 JSON 与单条账号 Schema — 见 [assets/README.md](assets/README.md)。

6
assets/README.md Normal file
View File

@@ -0,0 +1,6 @@
# assets
- `examples/`CLI 成功输出形状示例(虚构路径与数据)。
- `schemas/``get` / `list-json` / `pick-*` 单条对象的轻量 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,12 @@
{
"id": 1,
"name": "搜狐1号",
"platform": "sohu",
"phone": "13800138000",
"profile_dir": "D:/jiangchang-data/_anon/account-manager/profiles/搜狐号/13800138000",
"url": "https://mp.sohu.com",
"login_status": 1,
"last_login_at": "2026-04-01T12:00:00",
"created_at": "2026-03-01T10:00:00",
"updated_at": "2026-04-01T12:00:00"
}

View File

@@ -0,0 +1,26 @@
[
{
"id": 2,
"name": "知乎1号",
"platform": "zhihu",
"phone": "13900001111",
"profile_dir": "D:/jiangchang-data/_anon/account-manager/profiles/知乎/13900001111",
"url": "https://www.zhihu.com",
"login_status": 0,
"last_login_at": null,
"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",
"login_status": 1,
"last_login_at": "2026-04-01T12:00:00",
"created_at": "2026-03-01T10:00:00",
"updated_at": "2026-04-01T12:00:00"
}
]

View File

@@ -0,0 +1,24 @@
{
"$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-* 返回的单条账号对象核心字段extra_json 合并键不枚举)",
"type": "object",
"required": ["id", "name", "platform", "phone", "profile_dir", "url", "login_status", "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" },
"login_status": { "type": "integer", "enum": [0, 1] },
"last_login_at": {
"type": ["string", "null"],
"description": "本地时区 ISO8601 字符串;未登录过为 null"
},
"created_at": { "type": ["string", "null"] },
"updated_at": { "type": ["string", "null"] }
},
"additionalProperties": true
}

116
references/CLI.md Normal file
View File

@@ -0,0 +1,116 @@
# 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-logged-in``pick-web``set-login-status` 的成功/失败约定见 [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-logged-in` — 该平台**已登录**的一条账号(单行 JSON
```bash
python main.py pick-logged-in <平台>
```
- 筛选:`login_status = 1`,按 `last_login_at` 优先。
- 失败:首行以 `ERROR:` 开头(**勿当 JSON 解析**)。
### `pick-web` — 网页自动化候选账号(单行 JSON
```bash
python main.py pick-web <平台>
```
- 优先同 `pick-logged-in`;若无已登录,则取该平台 `updated_at` 最新一条。
- 失败:首行 `ERROR:`
### `set-login-status` — 回写登录态(跨技能)
```bash
python main.py set-login-status <id> <0|1>
```
- 成功:`stdout` 首行 `OK:SET_LOGIN_STATUS``OK:CLEARED_LOGIN_STATUS`,退出码 `0`
- 失败:`stdout` 首行 `ERROR:...`,退出码 `1`
### `open` — 仅打开浏览器(**不写库**
```bash
python main.py open <id>
```
- 需本机 Chrome/Edge + Playwright详见运行环境说明。
### `login` — 打开浏览器并**自动检测登录**,写回 `login_status`
```bash
python main.py login <id>
```
- 依赖 Playwright超时与轮询间隔等见 [RUNTIME.md](RUNTIME.md)。
### `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`

56
references/ERRORS.md Normal file
View File

@@ -0,0 +1,56 @@
# 终端前缀约定account-manager
约定:**机器解析只依赖 `stdout` 首行**时,应先判断首行是否以 `ERROR:``OK:` 开头;多行说明可忽略或仅作展示。
## `OK:`(成功标记)
| 首行 | 场景 |
|------|------|
| `OK:SET_LOGIN_STATUS` | `set-login-status <id> 1` 成功 |
| `OK:CLEARED_LOGIN_STATUS` | `set-login-status <id> 0` 成功 |
## `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_SET_LOGIN_STATUS_MISSING_ARGS` | `set-login-status` 参数不足 |
| `ERROR:CLI_LIST_JSON_MISSING_PLATFORM` | `list-json` 缺少平台参数 |
| `ERROR:CLI_PICK_LOGGED_IN_MISSING_ARGS` | `pick-logged-in` 缺少平台 |
| `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_LOGGED_IN_ACCOUNT` | `pick-logged-in``login_status=1` 的账号 |
| `ERROR:NO_ACCOUNT` | `pick-web` 该平台无任何记录 |
| `ERROR:INVALID_ACCOUNT_ID` | `set-login-status` 的 id 非数字 |
| `ERROR:INVALID_STATUS` | `set-login-status` 状态不是 0/1 |
## `ERROR:` — 浏览器 / 环境
| 前缀 | 含义 |
|------|------|
| `ERROR:需要 playwrightpip install playwright && playwright install chromium` | 未安装 Playwright 或浏览器未安装 |
| `ERROR:PROFILE_DIR_MISSING` | 账号记录中 `profile_dir` 为空 |
## 非前缀输出
- **成功添加账号**、**删除成功**等可能包含 `✅` 等人类可读行,**不宜**作为稳定协议解析。
- **`get` / `list-json` / `pick-*` 成功**`stdout` 应为**单行 JSON**(失败时首行 `ERROR:`,勿当 JSON

39
references/INTEGRATION.md Normal file
View File

@@ -0,0 +1,39 @@
# 与其它 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-logged-in <平台>` | 单行 JSON 对象 |
| 自动化候选(优先已登录) | `pick-web <平台>` | 单行 JSON 对象 |
| 回写登录态 | `set-login-status <id> <0或1>` | 首行 `OK:...` |
## 解析协议(强建议)
1. 读取子进程 **stdout** 全文或首行。
2. 若首行以 **`ERROR:`** 开头 → 失败,不要 `json.loads` 整段 stdout。
3. 若首行以 **`OK:`** 开头 → `set-login-status` 成功(可再检查退出码 `0`)。
4. **`get` / `list-json` / `pick-logged-in` / `pick-web`**:成功时应对**第一行**做 `json.loads`;后续行应不存在,若有则视为非协议内容。
## JSON 对象字段(与 `get` 一致)
典型字段见仓库 `assets/examples/get-response.json``assets/schemas/account-record.schema.json`
常见键名包括:`id`, `name`, `platform`, `phone`, `profile_dir`, `url`, `login_status`, `last_login_at`, `created_at`, `updated_at``extra_json` 列中的扁平扩展可能并入同一对象。
## Publisher / 自动化流水线示例
1. `pick-web 搜狐号`(或目标平台)→ 解析 JSON → 取 `id``profile_dir``url`
2. 若需确认已登录:检查 `login_status == 1`,否则可引导用户执行 `login <id>` 或先 `open <id>` 人工登录。
3. 外部脚本在确认登录后,可调用 `set-login-status <id> 1` 保持与库一致(若你们自有检测逻辑)。
## 退出码
- **`set-login-status`**:失败时进程以 **`1`** 退出(见实现)。
- 其它子命令多数为 **`0`**;若需严格编排,请以 **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:` / `OK:` 前缀一览 |
| [INTEGRATION.md](INTEGRATION.md) | 跨 Skill 调用与 JSON 解析约定 |
示例 JSON 与 Schema 见仓库 `assets/`

54
references/RUNTIME.md Normal file
View File

@@ -0,0 +1,54 @@
# 运行环境与数据路径account-manager
## 数据落盘位置(当前实现)
代码见 `scripts/util/runtime_paths.py``scripts/util/constants.py`
- **数据根目录** `get_data_root()`:优先环境变量 **`JIANGCHANG_DATA_ROOT`**;未设置时 Windows 默认 `D:\jiangchang-data`,其它系统默认 `~/.jiangchang-data`
- **用户/工作空间 id** `get_user_id()`:优先 **`JIANGCHANG_USER_ID`**;未设置时为 **`_anon`**。
- **技能数据目录**`{数据根}/{用户id}/account-manager/`
- **SQLite**`account-manager.db`
- **Profile**`profiles/<平台展示名>/<手机号>/`(由 `get_default_profile_dir` 生成)
- **日志目录**`logs/`(见下)
> OpenClaw 工作区文档中的 **`CLAW_*`** 变量:本技能**当前仍以 `JIANGCHANG_*` 为准**读取数据根与用户 id若宿主已统一注入 `CLAW_*`,需在代码层做别名兼容(未做前请只依赖本文列出的变量)。
## 调试为什么「list 没数据」
- 本地终端未设置 `JIANGCHANG_DATA_ROOT` / `JIANGCHANG_USER_ID` 时,会落到 **默认盘路径****`_anon`**,与网关中运行的实例**不是同一个库文件**。
- 设置 **`JIANGCHANG_ACCOUNT_DEBUG_PATHS=1`**`1`/`true`/`yes`/`on`)后,每次 CLI 子命令前会在 **stderr** 打印实际使用的根目录、用户 id 与数据库路径(由 `_maybe_print_paths_debug_cli` 输出)。
## 本地 CLI 开发注入(慎用)
`util/constants.py`**`_apply_cli_local_dev_env()`** 仅在 CLI 入口调用:
- 若开启本地开发开关(代码内 `_ACCOUNT_MANAGER_CLI_LOCAL_DEV` 或环境变量 **`JIANGCHANG_ACCOUNT_CLI_LOCAL_DEV=1`**),且当前未设置 `JIANGCHANG_DATA_ROOT` / `JIANGCHANG_USER_ID`,会注入代码内配置的默认路径与用户 id。
- **发版前务必关闭**,避免个人路径进入制品。详见该文件内注释。
## 日志
`scripts/util/logging_config.py`
| 变量 | 作用 |
|------|------|
| `JIANGCHANG_LOG_LEVEL` | 日志级别,默认 `INFO` |
| `JIANGCHANG_LOG_TO_STDERR` | 设为 `1`/`true` 等时WARNING 及以上同步 stderr |
| `JIANGCHANG_ACCOUNT_MANAGER_LOG_FILE` | 覆盖日志文件绝对路径(默认 `{技能数据目录}/logs/account-manager.log` |
## 登录 / 浏览器子进程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`** 且已执行 `playwright install chromium`(错误信息见 CLI 输出)。
## 常见问题(简要)
- **`ERROR:需要 playwright`**:安装 `playwright` 并安装浏览器内核。
- **`ERROR:PROFILE_DIR_MISSING`**:库中该账号 `profile_dir` 为空,需检查数据或重新添加账号流程。
- **登录检测一直失败**:关闭占用同一 profile 目录的其它浏览器;可调高超时与环境变量;查看技能日志文件。

View File

@@ -1,545 +0,0 @@
import sys
import json
import os
import sqlite3
import subprocess
import tempfile
import time
from datetime import datetime
# 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__)))
SKILL_SLUG = "account-manager"
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",
}
# 新账号默认名称搜狐1号 / sohu_2 等同平台序号
_PLATFORM_NAME_CN = {
"sohu": "搜狐",
"zhihu": "知乎",
"wechat": "微信",
"kimi": "Kimi",
"deepseek": "DeepSeek",
"doubao": "豆包",
"qianwen": "通义",
"yiyan": "文心",
"yuanbao": "元宝",
}
def get_data_root():
env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip()
if env:
return env
if sys.platform == "win32":
return r"D:\jiangchang-data"
return os.path.join(os.path.expanduser("~"), ".jiangchang-data")
def get_user_id():
uid = (os.getenv("JIANGCHANG_USER_ID") or "").strip()
return uid or "_anon"
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 get_default_profile_dir(account_id):
path = os.path.join(get_skill_data_dir(), "profiles", str(account_id))
os.makedirs(path, exist_ok=True)
return path
def get_conn():
return sqlite3.connect(get_db_path())
def _ensure_column(cur, table_name, col_name, ddl):
cur.execute(f"PRAGMA table_info({table_name})")
cols = [row[1] for row in cur.fetchall()]
if col_name not in cols:
cur.execute(f"ALTER TABLE {table_name} ADD COLUMN {col_name} {ddl}")
def _has_text_primary_id(cur):
cur.execute("PRAGMA table_info(accounts)")
for row in cur.fetchall():
if row[1] == "id":
t = (row[2] or "").upper()
return "TEXT" in t or "CHAR" in t
return False
def _migrate_text_id_to_integer(conn):
cur = conn.cursor()
cur.execute(
"""
CREATE TABLE accounts_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
platform TEXT NOT NULL,
phone TEXT,
profile_dir TEXT,
url TEXT,
login_status INTEGER DEFAULT 0,
last_login_at TEXT,
extra_json TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
cur.execute(
"SELECT name, platform, phone, profile_dir, login_status, last_login_at, extra_json, created_at, updated_at FROM accounts ORDER BY rowid"
)
rows = cur.fetchall()
for row in rows:
name, platform, phone, profile_dir, login_status, last_login_at, extra_json, created_at, updated_at = row
plat = (platform or "").strip().lower()
url = PLATFORM_URLS.get(plat, "")
cur.execute(
"""
INSERT INTO accounts_new
(name, platform, phone, profile_dir, url, login_status, last_login_at, extra_json, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
name,
plat,
phone or "",
profile_dir or "",
url,
int(login_status or 0),
last_login_at,
extra_json,
created_at,
updated_at,
),
)
cur.execute("DROP TABLE accounts")
cur.execute("ALTER TABLE accounts_new RENAME TO accounts")
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.execute(
"""
CREATE TABLE accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
platform TEXT NOT NULL,
phone TEXT,
profile_dir TEXT,
url TEXT,
login_status INTEGER DEFAULT 0,
last_login_at TEXT,
extra_json TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
else:
if _has_text_primary_id(cur):
_migrate_text_id_to_integer(conn)
cur = conn.cursor()
_ensure_column(cur, "accounts", "url", "TEXT")
cur.execute(
"SELECT id, platform FROM accounts WHERE url IS NULL OR TRIM(COALESCE(url,'')) = ''"
)
for rid, plat in cur.fetchall():
pu = PLATFORM_URLS.get((plat or "").strip().lower(), "")
if pu:
cur.execute("UPDATE accounts SET url = ? WHERE id = ?", (pu, rid))
conn.commit()
finally:
conn.close()
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 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, login_status, last_login_at, extra_json FROM accounts WHERE id = ?",
(aid,),
)
row = cur.fetchone()
if not row:
return None
acc = {
"id": row[0],
"name": row[1],
"platform": row[2],
"phone": row[3] or "",
"profile_dir": row[4] or get_default_profile_dir(row[0]),
"url": row[5] or PLATFORM_URLS.get(row[2], ""),
"login_status": int(row[6] or 0),
"last_login_at": row[7],
}
if row[8]:
try:
extra = json.loads(row[8])
if isinstance(extra, dict):
acc.update(extra)
except Exception:
pass
return acc
finally:
conn.close()
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 cmd_list(platform="all"):
init_db()
conn = get_conn()
try:
cur = conn.cursor()
if platform == "all":
cur.execute(
"SELECT id, name, phone, platform, login_status FROM accounts ORDER BY platform, id"
)
else:
cur.execute(
"SELECT id, name, phone, platform, login_status FROM accounts WHERE platform = ? ORDER BY id",
(platform,),
)
rows = cur.fetchall()
finally:
conn.close()
found = False
for row in rows:
phone = row[2] or "未绑定手机"
login_text = "已登录" if int(row[4] or 0) == 1 else "未登录"
print(f"账号ID:{row[0]} | 名称:{row[1]} | 手机号:{phone} | 平台:{row[3]} | 状态:{login_text}")
found = True
if not found:
print("ERROR:NO_ACCOUNTS_FOUND")
def cmd_get(account_id):
acc = get_account_by_id(account_id)
if acc:
print(json.dumps(acc, ensure_ascii=False))
return
print("ERROR:ACCOUNT_NOT_FOUND")
def cmd_add(platform: str, phone: str = ""):
"""添加账号:仅平台 + 可选手机号ID 自增,名称按平台序号自动生成。"""
init_db()
platform = (platform or "").strip().lower()
if platform not in PLATFORM_URLS:
print(f"ERROR:INVALID_PLATFORM (支持: {', '.join(sorted(PLATFORM_URLS.keys()))})")
return
phone = (phone or "").strip()
url = PLATFORM_URLS[platform]
now = datetime.now().isoformat(timespec="seconds")
conn = get_conn()
try:
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM accounts WHERE platform = ?", (platform,))
next_idx = cur.fetchone()[0] + 1
name = _default_name_for_platform(platform, next_idx)
cur.execute(
"""
INSERT INTO accounts (name, platform, phone, profile_dir, url, login_status, last_login_at, extra_json, created_at, updated_at)
VALUES (?, ?, ?, '', ?, 0, NULL, NULL, ?, ?)
""",
(name, platform, phone, url, now, now),
)
new_id = cur.lastrowid
profile_dir = get_default_profile_dir(new_id)
cur.execute(
"UPDATE accounts SET profile_dir = ?, updated_at = ? WHERE id = ?",
(profile_dir, now, new_id),
)
conn.commit()
finally:
conn.close()
print(f"✅ 已保存账号ID {new_id} | {name} | {platform}")
print(f" 登录请执行python account.py login {new_id}")
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 _verify_login(profile_dir: str, platform: str, start_url: str, channel: str) -> bool:
try:
from playwright.sync_api import sync_playwright
except ImportError:
print("⚠️ 未安装 playwright跳过登录校验请: pip install playwright && playwright install")
return False
try:
with sync_playwright() as p:
ctx = p.chromium.launch_persistent_context(
user_data_dir=profile_dir,
headless=True,
channel=channel,
args=["--disable-blink-features=AutomationControlled"],
)
try:
page = ctx.pages[0] if ctx.pages else ctx.new_page()
page.goto(start_url, wait_until="domcontentloaded", timeout=45000)
time.sleep(2)
u = (page.url or "").lower()
if platform == "sohu":
if "passport" in u or "/login" in u:
return False
return True
if "login" in u or "signin" in u or "passport" in u:
return False
return True
finally:
ctx.close()
except Exception:
return False
def cmd_login(account_id):
target = get_account_by_id(account_id)
if not target:
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:需要 playwrightpip install playwright && playwright install chromium")
return
profile_dir = target["profile_dir"]
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()
timeout_sec = _login_timeout_seconds()
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
print(f"正在为账号 [{target['name']}] 打开 {browser_name}")
print(f"访问地址:{url}")
print("请在窗口内完成登录(扫码/输入密码)。到达等待时间后将自动关闭窗口,您也可提前手动关闭。")
print(f"最长等待:{timeout_sec}")
# Playwright 同步 API 宜在独立子进程中跑浏览器,避免线程与事件循环问题
runner = r"""import json, sys
from playwright.sync_api import sync_playwright
with open(sys.argv[1], encoding="utf-8") as f:
c = json.load(f)
with sync_playwright() as p:
ctx = p.chromium.launch_persistent_context(
user_data_dir=c["profile_dir"],
headless=False,
channel=c["channel"],
no_viewport=True,
args=["--start-maximized"],
)
try:
page = ctx.pages[0] if ctx.pages else ctx.new_page()
page.goto(c["url"], wait_until="domcontentloaded", timeout=60000)
try:
ctx.wait_for_event("close", timeout=c["timeout_sec"] * 1000)
except Exception:
pass
try:
ctx.close()
except Exception:
pass
except Exception as e:
print(e, file=sys.stderr)
raise
"""
cfg = {
"channel": channel,
"profile_dir": profile_dir,
"url": url,
"timeout_sec": timeout_sec,
}
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
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", delete=False, encoding="utf-8"
) as pf:
pf.write(runner)
py_path = pf.name
try:
r = subprocess.run(
[sys.executable, py_path, cfg_path],
timeout=timeout_sec + 180,
)
if r.returncode != 0:
print("⚠️ 浏览器进程异常退出,仍将尝试检测登录状态")
except subprocess.TimeoutExpired:
print("⚠️ 等待浏览器超时,仍将尝试检测登录状态")
finally:
try:
os.unlink(cfg_path)
except OSError:
pass
try:
os.unlink(py_path)
except OSError:
pass
time.sleep(1.5)
print("正在检测登录状态…")
ok = _verify_login(profile_dir, platform, url, channel)
_mark_login_status(target["id"], ok)
if ok:
print("✅ 检测到已登录,状态已写入数据库")
else:
print("⚠️ 未检测到有效登录,状态为未登录。请关闭其他占用该用户目录的浏览器后重试,或延长 JIANGCHANG_LOGIN_TIMEOUT_SECONDS 后再登录。")
def _mark_login_status(account_id, success: bool):
now = datetime.now().isoformat(timespec="seconds")
conn = get_conn()
try:
cur = conn.cursor()
if success:
cur.execute(
"UPDATE accounts SET login_status = 1, last_login_at = ?, updated_at = ? WHERE id = ?",
(now, now, account_id),
)
else:
cur.execute(
"UPDATE accounts SET login_status = 0, updated_at = ? WHERE id = ?",
(now, account_id),
)
conn.commit()
finally:
conn.close()
if __name__ == "__main__":
if len(sys.argv) < 2:
print(
"用法python account.py add <platform> [phone] | list [platform] | get <id> | login <id>"
)
sys.exit(1)
cmd = sys.argv[1]
if len(sys.argv) >= 3 and sys.argv[2] == "list" and cmd in PLATFORM_URLS:
cmd_list(cmd)
elif cmd in PLATFORM_URLS and len(sys.argv) == 2:
cmd_list(cmd)
elif cmd == "add" and len(sys.argv) >= 3:
plat = sys.argv[2].lower()
phone_arg = sys.argv[3] if len(sys.argv) >= 4 else ""
cmd_add(plat, phone_arg)
elif cmd == "list":
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."""

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

@@ -0,0 +1,185 @@
"""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_logged_in,
cmd_pick_web,
cmd_set_login_status,
)
from service.browser_service import cmd_login, cmd_open
from util.constants import _apply_cli_local_dev_env
from util.logging_config import get_skill_logger
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-logged-in <平台> # 跨技能用:仅已登录账号,一行 JSON 或 ERROR")
print(" python main.py pick-web <平台> # 跨技能用:优先已登录,否则最新一条账号")
print(" python main.py list-json <平台|all> [--limit N] # 跨技能用:一行 JSON 数组,元素同 get")
print(" python main.py get <id>")
print(" python main.py set-login-status <id> <0|1> # 跨技能回写登录态")
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)
_apply_cli_local_dev_env()
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 == "set-login-status":
if len(sys.argv) < 4:
print("ERROR:CLI_SET_LOGIN_STATUS_MISSING_ARGS")
print("用法python main.py set-login-status <account_id> <0|1>")
print("说明:供 llm-manager 等脚本回写登录态;成功时 stdout 首行以 OK: 开头。")
sys.exit(1)
cmd_set_login_status(sys.argv[2], sys.argv[3])
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-logged-in":
if len(sys.argv) < 3:
print("ERROR:CLI_PICK_LOGGED_IN_MISSING_ARGS")
print("用法python main.py pick-logged-in <平台中文名或英文键>")
print("说明:供 llm-manager 等技能查询该平台已登录账号;成功时 stdout 仅一行 JSON。")
sys.exit(1)
cmd_pick_logged_in(sys.argv[2])
elif cmd == "pick-web":
if len(sys.argv) < 3:
print("ERROR:CLI_PICK_WEB_MISSING_ARGS")
print("用法python main.py pick-web <平台中文名或英文键>")
print("说明:供 llm-manager 等;优先已登录账号,否则返回该平台最新一条账号(未登录也可用于打开浏览器登录)。")
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"]

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

@@ -0,0 +1,269 @@
"""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, login_status, last_login_at, 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], ""),
"login_status": int(row[6] or 0),
"last_login_at": _unix_to_iso_local(row[7]),
"created_at": _unix_to_iso_local(row[9]),
"updated_at": _unix_to_iso_local(row[10]),
}
if row[8]:
try:
extra = json.loads(row[8])
if isinstance(extra, dict):
acc.update(extra)
except Exception:
pass
return acc
finally:
conn.close()
def mark_login_status(account_id, success: bool):
now = _now_unix()
conn = get_conn()
try:
cur = conn.cursor()
if success:
cur.execute(
"UPDATE accounts SET login_status = 1, last_login_at = ?, updated_at = ? WHERE id = ?",
(now, now, account_id),
)
else:
cur.execute(
"UPDATE accounts SET login_status = 0, 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, login_status, "
"last_login_at, 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_logged_in_id(platform_key: str) -> Optional[int]:
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ? AND login_status = 1
ORDER BY (last_login_at IS NULL), last_login_at DESC
LIMIT 1
""",
(platform_key,),
)
row = cur.fetchone()
return int(row[0]) if row else None
finally:
conn.close()
def fetch_pick_web_candidate_id(platform_key: str) -> Optional[int]:
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ? AND login_status = 1
ORDER BY (last_login_at IS NULL), last_login_at DESC
LIMIT 1
""",
(platform_key,),
)
row = cur.fetchone()
if not row:
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ?
ORDER BY updated_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, login_status, last_login_at, extra_json, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, 0, NULL, 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()

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

@@ -0,0 +1,18 @@
# 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
login_status INTEGER NOT NULL DEFAULT 0, -- 是否已登录: 0 否 1 是 (由脚本校验后写入)
last_login_at INTEGER, -- 最近一次登录成功时间, Unix 秒 UTC; 未登录过为 NULL
extra_json TEXT, -- 扩展字段 JSON
created_at INTEGER NOT NULL, -- 记录创建时间, Unix 秒 UTC
updated_at INTEGER NOT NULL -- 记录最后更新时间, Unix 秒 UTC
);
"""

18
scripts/main.py Normal file
View File

@@ -0,0 +1,18 @@
"""
account-manager CLI 入口。
分层cliargv→ service用例→ dbSQLite共用工具在 util/。
"""
import sys
# 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,380 @@
"""账号用例list / get / add / delete / pick / set-login-status含终端输出供 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_logged_in_id,
fetch_pick_web_candidate_id,
fetch_row_platform_phone_conn,
get_account_by_id,
has_duplicate_phone_conn,
insert_account_row,
mark_login_status,
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,
login_status,
last_login_at,
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"login_status{int(login_status) if login_status is not None else ''}")
print(f"last_login_at{int(last_login_at) if last_login_at is not None else ''}")
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_logged_in(platform_input: str):
"""
机器可读跨技能接口查询指定平台下「已登录」的一条账号login_status=1按 last_login_at 优先)。
成功stdout 仅输出一行 JSON结构与 get 子命令一致。
失败stdout 首行以 ERROR: 开头(由调用方判断,勿解析为 JSON
"""
get_skill_logger().info("pick_logged_in 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_logged_in_id(key)
if picked is None:
print("ERROR:NO_LOGGED_IN_ACCOUNT 该平台暂无已登录账号login_status=1")
print("请先 list 查看账号 id再执行python main.py login <id>")
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_pick_web(platform_input: str):
"""
供 llm-manager 等:取该平台用于网页自动化的账号候选。
优先 login_status=1与 pick-logged-in 一致);若无,则取该平台 updated_at 最新的一条,
便于「已 add 未标登录」时仍打开 profile在浏览器内登录后继续任务。
成功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)
def cmd_set_login_status(account_id_str: str, status_str: str) -> None:
"""
跨技能机器接口:回写 accounts.login_status。
成功stdout 首行 OK:SET_LOGIN_STATUS 或 OK:CLEARED_LOGIN_STATUS进程退出码 0。
失败stdout 首行 ERROR:...,退出码 1。
"""
aid_s = (account_id_str or "").strip()
st_s = (status_str or "").strip()
if not aid_s.isdigit():
print("ERROR:INVALID_ACCOUNT_ID")
sys.exit(1)
if st_s not in ("0", "1"):
print("ERROR:INVALID_STATUS 须为 0 或 1")
sys.exit(1)
aid = int(aid_s)
init_db()
if get_account_by_id(aid) is None:
get_skill_logger().warning("set_login_status_not_found account_id=%r", aid)
print("ERROR:ACCOUNT_NOT_FOUND")
sys.exit(1)
ok = st_s == "1"
mark_login_status(aid, ok)
get_skill_logger().info("set_login_status account_id=%s success=%s", aid, ok)
print("OK:SET_LOGIN_STATUS" if ok else "OK:CLEARED_LOGIN_STATUS")

View File

@@ -0,0 +1,295 @@
"""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, mark_login_status
from util.logging_config import get_skill_logger, get_skill_log_file_path
from util.platforms import (
PLATFORMS,
PLATFORM_URLS,
_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS,
)
# 子进程入口脚本与本模块同目录(独立解释器执行,非 import
_SERVICE_DIR = os.path.dirname(os.path.abspath(__file__))
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 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
channel = resolve_chromium_channel()
if not channel:
log.warning("login_aborted no_chromium_channel")
_print_browser_install_hint()
return
try:
import playwright # noqa: F401
except ImportError:
log.error("login_aborted playwright_missing")
print("ERROR:需要 playwrightpip install playwright && playwright install chromium")
return
profile_dir = (target.get("profile_dir") or "").strip()
if not profile_dir:
log.warning("login_aborted profile_dir_empty account_id=%s", target.get("id"))
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")
platform = (target.get("platform") or "").strip().lower()
timeout_sec = _login_timeout_seconds()
poll_sec = _login_poll_interval_seconds()
dom_grace_sec = _login_dom_grace_seconds()
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
print(f"正在为账号 [{target['name']}] 打开 {browser_name}")
print(f"访问地址:{url}")
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",
target.get("id"),
platform,
channel,
timeout_sec,
poll_sec,
dom_grace_sec,
profile_dir,
url,
log_file,
)
login_runner_path = os.path.join(_SERVICE_DIR, "login_child_runner.py")
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
try:
r = subprocess.run(
[sys.executable, login_runner_path, cfg_path],
timeout=timeout_sec + 180,
)
proc_rc = r.returncode
if proc_rc != 0:
print("⚠️ 浏览器进程异常退出,将仅根据已写入的检测结果更新状态")
log.warning("login_subprocess_nonzero_return code=%s", proc_rc)
except subprocess.TimeoutExpired as ex:
print("⚠️ 等待浏览器超时,将仅根据已写入的检测结果更新状态")
log.warning("login_subprocess_timeout err=%s", ex)
finally:
try:
os.unlink(cfg_path)
except OSError:
pass
interactive_ok = False
try:
with open(result_path, encoding="utf-8") as rf:
interactive_ok = bool(json.load(rf).get("interactive_ok"))
except Exception:
pass
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 subprocess_rc=%s marking_db=%s",
target.get("id"),
interactive_ok,
proc_rc,
ok,
)
mark_login_status(target["id"], ok)
if ok:
print("✅ 已判定登录成功,状态已写入数据库")
else:
log.warning(
"login_not_detected account_id=%s platform=%s see_log=%s",
target.get("id"),
platform,
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:需要 playwrightpip install playwright && 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>")
open_runner_path = os.path.join(_SERVICE_DIR, "open_child_runner.py")
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, open_runner_path, cfg_path])
finally:
try:
os.unlink(cfg_path)
except OSError:
pass

View File

@@ -0,0 +1,223 @@
"""Playwright 登录检测子进程:由 browser_service.cmd_login 以独立解释器启动argv[1] 为 JSON 配置路径。"""
import json
import logging
import sys
import time
from urllib.parse import urlparse
from playwright.sync_api import sync_playwright
def page_location_href(p):
# Prefer location.href over page.url for SPA (e.g. Sohu shell path lag).
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 main():
with open(sys.argv[1], encoding="utf-8") as f:
c = json.load(f)
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"])
interactive_ok = False
result_path = c.get("result_path") or ""
log_path = (c.get("log_file") or "").strip()
lg = logging.getLogger("openclaw.skill.account_manager.login_child")
if log_path:
lg.handlers.clear()
lg.setLevel(logging.DEBUG)
_fh = logging.FileHandler(log_path, encoding="utf-8")
_fh.setFormatter(
logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
)
lg.addHandler(_fh)
lg.propagate = False
last_eval_detail = ""
with sync_playwright() as p:
ctx = p.chromium.launch_persistent_context(
user_data_dir=c["profile_dir"],
headless=False,
channel=c["channel"],
no_viewport=True,
args=["--start-maximized"],
)
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.handlers:
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, detail = evaluate_logged_out_dom(ctx, page, bundle)
ok = not still_logged_out
last_eval_detail = detail
if lg.handlers:
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 iter_start_elapsed=%.1fs deadline_in=%.1fs tabs=%s poll_interval_sec=%s logged_out=%s ok=%s detail=%s | %s",
iter_start - t0,
deadline - iter_start,
len(ctx.pages or []),
poll,
still_logged_out,
ok,
detail,
" ; ".join(parts) if parts else "no_tabs",
)
if ok:
interactive_ok = True
if lg.handlers:
lg.info("login_detected_ok %s", detail)
break
except Exception as ex:
if lg.handlers:
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:
if lg.handlers:
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
else:
if lg.handlers:
lg.info("login_success_closing_browser_immediately")
except Exception as e:
if lg.handlers:
lg.exception("login_runner_fatal err=%s", e)
print(e, file=sys.stderr)
raise
finally:
if result_path:
try:
with open(result_path, "w", encoding="utf-8") as rf:
json.dump({"interactive_ok": interactive_ok}, rf, ensure_ascii=False)
except Exception:
pass
try:
ctx.close()
except Exception:
pass
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,34 @@
"""Playwright 仅打开浏览器子进程:由 browser_service.cmd_open 启动argv[1] 为 JSON 配置路径。"""
import json
import sys
from playwright.sync_api import sync_playwright
def main():
with open(sys.argv[1], encoding="utf-8") as f:
c = json.load(f)
with sync_playwright() as p:
ctx = p.chromium.launch_persistent_context(
user_data_dir=c["profile_dir"],
headless=False,
channel=c["channel"],
no_viewport=True,
args=["--start-maximized"],
)
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)."""

35
scripts/util/constants.py Normal file
View File

@@ -0,0 +1,35 @@
"""技能级常量与本地 CLI 调试环境注入(仅 CLI 入口显式调用 _apply_cli_local_dev_env 时生效)。"""
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"
# ---------------------------------------------------------------------------
# 本地 CLI 调试:直接运行 python main.py 时,若未设置 JIANGCHANG_*,可自动注入下面默认值。
# - 仅当 CLI 里调用了 _apply_cli_local_dev_env() 时生效;被其它脚本 import 不会改 os.environ。
# - 不会覆盖宿主/终端已设置的 JIANGCHANG_DATA_ROOT、JIANGCHANG_USER_ID。
# - 开启方式(二选一):
# 1) 将 _ACCOUNT_MANAGER_CLI_LOCAL_DEV 改为 True
# 2) 或设置环境变量 JIANGCHANG_ACCOUNT_CLI_LOCAL_DEV=1无需改代码
# 合并/发版前请关闭开关,避免把个人机器路径带进生产制品。
# ---------------------------------------------------------------------------
_ACCOUNT_MANAGER_CLI_LOCAL_DEV = True
_ACCOUNT_MANAGER_CLI_LOCAL_DATA_ROOT = r"D:\jiangchang-data"
_ACCOUNT_MANAGER_CLI_LOCAL_USER_ID = "10032"
def _apply_cli_local_dev_env() -> None:
enabled = _ACCOUNT_MANAGER_CLI_LOCAL_DEV
if not enabled:
v = (os.getenv("JIANGCHANG_ACCOUNT_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"] = _ACCOUNT_MANAGER_CLI_LOCAL_DATA_ROOT.strip()
if not (os.getenv("JIANGCHANG_USER_ID") or "").strip():
os.environ["JIANGCHANG_USER_ID"] = _ACCOUNT_MANAGER_CLI_LOCAL_USER_ID.strip()

View File

@@ -0,0 +1,69 @@
"""技能级文件日志。"""
import logging
import os
import sys
from logging.handlers import TimedRotatingFileHandler
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
from util.runtime_paths import get_skill_logs_dir
def _skill_log_file_path() -> str:
"""主日志文件路径;可由 JIANGCHANG_ACCOUNT_MANAGER_LOG_FILE 覆盖为绝对路径。"""
override = (os.getenv("JIANGCHANG_ACCOUNT_MANAGER_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_skill_logs_dir(), f"{SKILL_SLUG}.log")
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 get_skill_log_file_path() -> str:
"""主日志文件绝对路径(与 get_skill_logger 写入文件一致)。"""
return _skill_log_file_path()
def get_skill_logger() -> logging.Logger:
"""
技能级日志单文件按日切分TimedRotatingFileHandler midnightUTF-8。
环境变量JIANGCHANG_LOG_LEVEL默认 INFO、JIANGCHANG_LOG_TO_STDERR=1 时 WARNING+ 同步 stderr、
JIANGCHANG_ACCOUNT_MANAGER_LOG_FILE 覆盖日志文件路径。
"""
log = logging.getLogger(LOG_LOGGER_NAME)
if log.handlers:
return log
log.setLevel(_log_level_from_env())
path = _skill_log_file_path()
fh = TimedRotatingFileHandler(
path,
when="midnight",
interval=1,
backupCount=30,
encoding="utf-8",
delay=True,
)
fh.setFormatter(
logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
)
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(fh.formatter)
log.addHandler(sh)
log.propagate = False
return log

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

@@ -0,0 +1,139 @@
"""平台配置、别名解析、手机号校验。"""
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": ["元宝"],
},
}
# 未登录页共性主行动点含「登录」——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,100 @@
"""数据目录、profile 路径、CLI 路径调试输出。"""
import os
import re
import sys
from typing import Optional
from util.constants import SKILL_SLUG
from util.platforms import _PLATFORM_PRIMARY_CN
def get_data_root():
env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip()
if env:
return env
if sys.platform == "win32":
return r"D:\jiangchang-data"
return os.path.join(os.path.expanduser("~"), ".jiangchang-data")
def get_user_id():
uid = (os.getenv("JIANGCHANG_USER_ID") or "").strip()
return uid or "_anon"
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_skill_logs_dir() -> str:
"""{DATA_ROOT}/{USER_ID}/account-manager/logs/"""
path = os.path.join(get_skill_data_dir(), "logs")
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