Compare commits

...

6 Commits

Author SHA1 Message Date
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
38 changed files with 1665 additions and 1601 deletions

View File

@@ -50,4 +50,19 @@ python3 {baseDir}/scripts/main.py delete platform <平台> <手机号>
### 列出某平台所有账号platform 可为中文名、all 或 全部) ### 列出某平台所有账号platform 可为中文名、all 或 全部)
```bash ```bash
python3 {baseDir}/scripts/main.py list all python3 {baseDir}/scripts/main.py list all
python3 {baseDir}/scripts/main.py list 搜狐号 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/`

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`** 且已执行 `playwright install chromium`(错误信息见 CLI 输出)。
## 常见问题(简要)
- **`ERROR:需要 playwright`**:安装 `playwright` 并安装浏览器内核。
- **`ERROR:PROFILE_DIR_MISSING`**:库中该账号 `profile_dir` 为空,需检查数据或重新添加账号流程。
- **登录检测一直失败**:关闭占用同一 profile 目录的其它浏览器;可调高超时与环境变量;查看统一日志文件。

View File

@@ -1,721 +0,0 @@
import sys
import json
import os
import re
import sqlite3
from urllib.parse import urlparse
import subprocess
import tempfile
import time
from datetime import datetime
from typing import Optional
# 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"
# 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, -- 平台标识,与内置 PLATFORM_URLS 键一致,如 sohu
phone TEXT, -- 可选绑定手机号
profile_dir TEXT, -- Playwright 持久化用户数据目录(绝对路径)
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
);
"""
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
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 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()
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,
created_at, updated_at
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": _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 _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 = _now_unix()
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_unix(), 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 _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 _definitely_logged_out_url(page_url: str) -> bool:
"""
通用:是否「明显仍在登录/认证页」。
用路径段判断,避免 mpfe/v4/login 因含 mpfe 被误判为已登录。
"""
if not page_url or not page_url.strip():
return True
try:
p = urlparse(page_url.strip())
except Exception:
return True
host = (p.netloc or "").lower()
path = (p.path or "").lower()
query = (p.query or "").lower()
if "passport" in host or "login." in host or "signin." in host:
return True
seg = [s for s in path.split("/") if s]
login_segments = (
"login",
"signin",
"signup",
"register",
"authorize",
"authentication",
)
if any(s in login_segments for s in seg):
return True
if re.search(r"/(login|signin)(/|$|\?)", path):
return True
if "redirect_uri" in query and ("login" in query or "passport" in query):
return True
return False
def _sohu_url_logged_in(page_url: str) -> bool:
"""
搜狐:仅白名单路径视为已进入后台(不用裸 mpfe避免 /mpfe/v4/login 误判)。
"""
if _definitely_logged_out_url(page_url):
return False
try:
path = (urlparse(page_url).path or "").lower()
except Exception:
return False
if "contentmanagement" in path:
return True
if "/mpfe/v4/main" in path or path.rstrip("/").endswith("/main"):
return True
if "/mpfe/v4/publish" in path or "/mpfe/v4/data" in path:
return True
return False
def _sohu_page_shows_login_form(page) -> bool:
"""搜狐登录页常见密码框;可见则认为未登录(与 URL 交叉验证)。"""
try:
loc = page.locator('input[type="password"]')
n = loc.count()
if n == 0:
return False
return loc.first.is_visible(timeout=800)
except Exception:
return False
def _url_looks_logged_in(platform: str, page_url: str) -> bool:
"""是否判定为已登录:搜狐用白名单+负例;其它平台仅用「非明确登录页」宽松策略。"""
if not page_url or not page_url.strip():
return False
if platform == "sohu":
return _sohu_url_logged_in(page_url)
if _definitely_logged_out_url(page_url):
return False
u = page_url.lower()
if "passport" in u or "signin" in u:
return False
return True
def _verify_logged_in_page(platform: str, page) -> bool:
"""结合 URL 与页面(搜狐加验登录表单)。"""
url = page.url or ""
if platform == "sohu":
if _sohu_page_shows_login_form(page):
return False
return _sohu_url_logged_in(url)
return _url_looks_logged_in(platform, url)
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)
for _ in range(20):
if _verify_logged_in_page(platform, page):
return True
time.sleep(1.0)
return _verify_logged_in_page(platform, page)
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()
poll_sec = _login_poll_interval_seconds()
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
print(f"正在为账号 [{target['name']}] 打开 {browser_name}")
print(f"访问地址:{url}")
print(
"请在窗口内完成登录。系统会轮询地址栏,进入后台后将自动关闭窗口并写入状态(无需手动关浏览器)。"
)
print(f"轮询间隔约 {poll_sec} 秒;全程最长 {timeout_sec} 秒。")
# 子进程:搜狐轮询 URL+页面;勿用「含 mpfe 即登录」,/mpfe/v4/login 会误判。
runner = r"""import json, sys, time
from urllib.parse import urlparse
from playwright.sync_api import sync_playwright
def sohu_logged_in(url):
u = (url or "").strip()
if not u:
return False
try:
p = urlparse(u)
except Exception:
return False
path = (p.path or "").lower()
host = (p.netloc or "").lower()
if "passport" in host:
return False
segs = [s for s in path.split("/") if s]
if any(s in ("login", "signin", "signup", "register", "authorize") for s in segs):
return False
if "contentmanagement" in path:
return True
if "/mpfe/v4/main" in path or path.rstrip("/").endswith("/main"):
return True
if "/mpfe/v4/publish" in path or "/mpfe/v4/data" in path:
return True
return False
def sohu_not_login_page(page):
try:
loc = page.locator('input[type="password"]')
if loc.count() == 0:
return False
return loc.first.is_visible(timeout=600)
except Exception:
return False
with open(sys.argv[1], encoding="utf-8") as f:
c = json.load(f)
platform = (c.get("platform") or "").lower()
poll = float(c.get("poll_interval", 1.5))
deadline = time.time() + float(c["timeout_sec"])
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 platform == "sohu":
while time.time() < deadline:
try:
if sohu_logged_in(page.url) and not sohu_not_login_page(page):
break
except Exception:
break
time.sleep(poll)
else:
rem = max(0.0, deadline - time.time())
try:
ctx.wait_for_event("close", timeout=rem * 1000)
except Exception:
pass
except Exception as e:
print(e, file=sys.stderr)
raise
finally:
try:
ctx.close()
except Exception:
pass
"""
cfg = {
"channel": channel,
"profile_dir": profile_dir,
"url": url,
"timeout_sec": timeout_sec,
"platform": platform,
"poll_interval": poll_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 cmd_open(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"
)
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
print(f"正在打开 [{target['name']}] 的 {browser_name}(仅查看,不写入数据库)")
print(f"地址:{url}")
print("请在本窗口中自行确认登录态。关闭浏览器后命令结束。")
print("需要自动检测并写入数据库时请执行python account.py login <id>")
runner_open = 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)
ctx.wait_for_event("close", timeout=86400000)
except Exception as e:
print(e, file=sys.stderr)
raise
finally:
try:
ctx.close()
except Exception:
pass
"""
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
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", delete=False, encoding="utf-8"
) as pf:
pf.write(runner_open)
py_path = pf.name
try:
subprocess.run([sys.executable, py_path, cfg_path])
finally:
try:
os.unlink(cfg_path)
except OSError:
pass
try:
os.unlink(py_path)
except OSError:
pass
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()
if __name__ == "__main__":
if len(sys.argv) < 2:
print(
"用法python account.py add <platform> [phone] | list [platform] | get <id> | open <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 == "open" and len(sys.argv) >= 3:
cmd_open(sys.argv[2])
elif cmd == "login" and len(sys.argv) >= 3:
cmd_login(sys.argv[2])
else:
print("参数错误")
sys.exit(1)

View File

@@ -1,285 +0,0 @@
"""只读类 CLIlist / get / list-json / pick-*。"""
import json
import sys
from am_db import get_account_by_id, get_conn, init_db
from am_logging import get_skill_logger
from am_platforms import _platform_list_cn_for_help, resolve_platform_key
from am_runtime_paths import _runtime_paths_debug_text
# list 表头:与 accounts 表列顺序、列名一致(便于对照库结构)
_LIST_TABLE_COLUMNS = (
"id",
"name",
"platform",
"phone",
"profile_dir",
"url",
"login_status",
"last_login_at",
"extra_json",
"created_at",
"updated_at",
)
def _list_row_to_cells(row: tuple) -> list:
"""将 SELECT 整行转为展示用字符串列表(与 _LIST_TABLE_COLUMNS 顺序一致)。"""
rid, name, plat, phone, pdir, url, lstat, lla, exj, cat, uat = row
return [
str(rid) if rid is not None else "",
name if name is not None else "",
plat if plat is not None else "",
phone if phone is not None else "",
pdir if pdir is not None else "",
url if url is not None else "",
str(int(lstat)) if lstat is not None else "",
str(int(lla)) if lla is not None else "",
(exj or "").replace("\n", "\\n").replace("\r", ""),
str(int(cat)) if cat is not None else "",
str(int(uat)) if uat is not None else "",
]
def _print_accounts_table(rows: list) -> None:
"""表头 + 分隔线 + 每账号一行;列宽按内容对齐(终端等宽字体)。"""
headers = list(_LIST_TABLE_COLUMNS)
body = [_list_row_to_cells(r) for r in rows]
n = len(headers)
widths = [len(headers[j]) for j in range(n)]
for cells in body:
for j in range(n):
widths[j] = max(widths[j], len(cells[j]))
sep = " | "
def fmt(cells):
return sep.join(cells[j].ljust(widths[j]) for j in range(n))
print(fmt(headers))
print(sep.join("-" * widths[j] for j in range(n)))
for cells in body:
print(fmt(cells))
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
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 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 ?",
(key, int(limit)),
)
rows = cur.fetchall()
finally:
conn.close()
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))
conn = get_conn()
try:
cur = conn.cursor()
if 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 ?",
(key, lim),
)
ids = [r[0] for r in cur.fetchall()]
finally:
conn.close()
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()
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
""",
(key,),
)
row = cur.fetchone()
finally:
conn.close()
if not row:
print("ERROR:NO_LOGGED_IN_ACCOUNT 该平台暂无已登录账号login_status=1")
print("请先 list 查看账号 id再执行python main.py login <id>")
return
acc = get_account_by_id(row[0])
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()
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
""",
(key,),
)
row = cur.fetchone()
if not row:
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ?
ORDER BY updated_at DESC, id DESC
LIMIT 1
""",
(key,),
)
row = cur.fetchone()
finally:
conn.close()
if not row:
print("ERROR:NO_ACCOUNT 该平台在账号库中没有任何记录,请先执行 add。")
return
acc = get_account_by_id(row[0])
if not acc:
print("ERROR:ACCOUNT_NOT_FOUND")
print(_runtime_paths_debug_text(), file=sys.stderr)
return
print(json.dumps(acc, ensure_ascii=False))

View File

@@ -1,246 +0,0 @@
"""写入类 CLIadd / delete / set-login-status。"""
import os
import shutil
import sys
from am_db import (
_default_name_for_platform,
_mark_login_status,
_normalize_account_id,
_now_unix,
get_account_by_id,
get_conn,
init_db,
)
from am_logging import get_skill_logger
from am_platforms import (
PLATFORM_URLS,
_PLATFORM_PRIMARY_CN,
_is_valid_cn_mobile11,
_normalize_phone_digits,
_platform_list_cn_for_help,
resolve_platform_key,
)
from am_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_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 = _now_unix()
phone_store = phone_norm
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"SELECT id FROM accounts WHERE platform = ? AND phone = ?",
(key, phone_store),
)
if cur.fetchone():
log.warning("add_duplicate platform=%s phone_suffix=%s", key, phone_store[-4:])
print(
f"ERROR:DUPLICATE_PHONE_PLATFORM 该平台下已存在手机号 {phone_store},请勿重复添加。"
)
return
cur.execute("SELECT COUNT(*) FROM accounts WHERE platform = ?", (key,))
next_idx = cur.fetchone()[0] + 1
name = _default_name_for_platform(key, next_idx)
profile_dir = get_default_profile_dir(key, phone_store, None)
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, key, phone_store, profile_dir, url, now, now),
)
new_id = cur.lastrowid
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:
cur = conn.cursor()
cur.execute(
"SELECT id, name, platform, phone, profile_dir FROM accounts WHERE id = ?",
(int(aid),),
)
row = cur.fetchone()
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
cur.execute("DELETE FROM accounts WHERE id = ?", (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:
cur = conn.cursor()
cur.execute(
"SELECT id, profile_dir FROM accounts WHERE platform = ?",
(key,),
)
rows = cur.fetchall()
if not rows:
log.warning("delete_by_platform_empty platform=%s", key)
print("ERROR:NO_ACCOUNTS_FOUND 该平台下没有账号记录。")
return
cur.execute("DELETE FROM accounts WHERE platform = ?", (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:
cur = conn.cursor()
cur.execute(
"SELECT id, name, profile_dir FROM accounts WHERE platform = ? AND phone = ?",
(key, phone_norm),
)
row = cur.fetchone()
if not row:
print("ERROR:ACCOUNT_NOT_FOUND 该平台下无此手机号。")
return
rid, name, profile_dir = row
cur.execute(
"DELETE FROM accounts WHERE platform = ? AND phone = ?",
(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

@@ -1,33 +0,0 @@
"""技能级常量与本地 CLI 调试环境注入(仅 main 入口显式调用 _apply_cli_local_dev_env 时生效)。"""
import os
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SKILL_SLUG = "account-manager"
# 与其它 OpenClaw 技能对齐openclaw.skill.<slug_下划线>
LOG_LOGGER_NAME = "openclaw.skill.account_manager"
# ---------------------------------------------------------------------------
# 本地 CLI 调试:直接运行 python main.py 时,若未设置 JIANGCHANG_*,可自动注入下面默认值。
# - 仅当 __main__ 里调用了 _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

@@ -1,136 +0,0 @@
"""SQLite 账号库:建表、读写、登录态字段。"""
import json
import sqlite3
import time
from datetime import datetime
from typing import Optional
from am_platforms import PLATFORM_URLS, _PLATFORM_NAME_CN
from am_runtime_paths import get_db_path
# 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
);
"""
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 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()
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,
created_at, updated_at
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 "").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 _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 _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()

View File

@@ -1,69 +0,0 @@
"""技能级文件日志。"""
import logging
import os
import sys
from logging.handlers import TimedRotatingFileHandler
from am_constants import LOG_LOGGER_NAME, SKILL_SLUG
from am_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

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

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

View File

@@ -1,25 +1,23 @@
"""CLI 入口:用法说明与 argv 分发。""" """CLI 入口:用法说明与 argv 分发。"""
import sys import sys
from am_browser import cmd_login, cmd_open from service.account_service import (
from am_commands_read import ( cmd_add,
cmd_delete_by_id,
cmd_delete_by_platform,
cmd_delete_by_platform_phone,
cmd_get, cmd_get,
cmd_list, cmd_list,
cmd_list_json, cmd_list_json,
cmd_pick_logged_in, cmd_pick_logged_in,
cmd_pick_web, cmd_pick_web,
)
from am_commands_write import (
cmd_add,
cmd_delete_by_id,
cmd_delete_by_platform,
cmd_delete_by_platform_phone,
cmd_set_login_status, cmd_set_login_status,
) )
from am_constants import _apply_cli_local_dev_env from service.browser_service import cmd_login, cmd_open, cmd_wait_login
from am_logging import get_skill_logger from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
from am_platforms import _platform_list_cn_for_help, resolve_platform_key from util.logging_config import get_skill_logger, setup_skill_logging
from am_runtime_paths import _maybe_print_paths_debug_cli 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: def _cli_print_full_usage() -> None:
@@ -34,6 +32,7 @@ def _cli_print_full_usage() -> None:
print(" python main.py set-login-status <id> <0|1> # 跨技能回写登录态") print(" python main.py set-login-status <id> <0|1> # 跨技能回写登录态")
print(" python main.py open <id>") print(" python main.py open <id>")
print(" python main.py login <id>") print(" python main.py login <id>")
print(" python main.py wait-login <id> [--force] # 跨技能:限时内等待网页登录(默认 180s--force 忽略库内已登录标记")
print(" python main.py delete id <id>") print(" python main.py delete id <id>")
print(" python main.py delete platform <平台>") print(" python main.py delete platform <平台>")
print(" python main.py delete platform <平台> <手机号>") print(" python main.py delete platform <平台> <手机号>")
@@ -87,7 +86,7 @@ def main() -> None:
_cli_print_full_usage() _cli_print_full_usage()
sys.exit(1) sys.exit(1)
_apply_cli_local_dev_env() setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
get_skill_logger().info("cli_start argv=%s", sys.argv) get_skill_logger().info("cli_start argv=%s", sys.argv)
cmd = sys.argv[1] cmd = sys.argv[1]
_maybe_print_paths_debug_cli() _maybe_print_paths_debug_cli()
@@ -153,6 +152,17 @@ def main() -> None:
_cli_fail_need_account_id("open") _cli_fail_need_account_id("open")
sys.exit(1) sys.exit(1)
cmd_open(sys.argv[2]) cmd_open(sys.argv[2])
elif cmd == "wait-login":
av = [a for a in sys.argv[2:] if a]
force = "--force" in av or "-f" in av
av = [a for a in av if a not in ("--force", "-f")]
if len(av) < 1:
print("ERROR:CLI_WAIT_LOGIN_MISSING_ID")
print("用法python main.py wait-login <账号 id> [--force]")
print("说明:供 llm-manager / 搜狐发布等编排;未登录时打开浏览器,限时内等待登录或用户关窗即结束。")
print("--force即使库内已标为登录也重新打开浏览器检测Cookie 失效时用)。")
sys.exit(1)
sys.exit(cmd_wait_login(av[0], force=force))
elif cmd == "login": elif cmd == "login":
if len(sys.argv) < 3: if len(sys.argv) < 3:
_cli_fail_need_account_id("login") _cli_fail_need_account_id("login")

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, -- 展示/排序用尽力同步;编排上请以 Playwright DOM 检测为准,勿单独依赖本字段
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
);
"""

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,127 @@
"""
JIANGCHANG_* 数据根与用户目录:解析规则 + 可选本地 CLI 默认值。
各技能 scripts/jiangchang_skill_core/ 为发布用副本,与 kit 保持同步。
"""
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",
"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 即根

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

View File

@@ -1,10 +1,19 @@
""" """
account-manager CLI 入口。 account-manager CLI 入口。
业务逻辑按模块拆分am_*.py便于维护与满足 PyArmor trial 对单文件体积的限制 分层cliargv→ service用例→ dbSQLite共用工具在 util/
""" """
import os
import sys 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 之前) # Windows GBK 编码兼容修复(须在任何业务 import 与 print 之前)
if sys.platform == "win32": if sys.platform == "win32":
import io import io
@@ -12,7 +21,7 @@ if sys.platform == "win32":
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace") sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
from am_cli import main from cli.app import main
if __name__ == "__main__": if __name__ == "__main__":
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

@@ -7,14 +7,21 @@ import tempfile
import time import time
from urllib.parse import urlparse from urllib.parse import urlparse
from am_db import _mark_login_status, get_account_by_id from db.accounts_repo import get_account_by_id, mark_login_status
from am_logging import get_skill_logger, get_skill_log_file_path from util.logging_config import (
from am_platforms import ( get_skill_logger,
get_skill_log_file_path,
subprocess_env_with_trace,
)
from util.platforms import (
PLATFORMS, PLATFORMS,
PLATFORM_URLS, PLATFORM_URLS,
_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS, _LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS,
) )
# 子进程入口脚本与本模块同目录(独立解释器执行,非 import
_SERVICE_DIR = os.path.dirname(os.path.abspath(__file__))
def _win_find_exe(candidates): def _win_find_exe(candidates):
for p in candidates: for p in candidates:
@@ -89,6 +96,14 @@ def _login_dom_grace_seconds() -> float:
return 4.0 return 4.0
def _wait_login_timeout_seconds() -> int:
"""供 wait-login / llm 编排:默认 180 秒JIANGCHANG_WAIT_LOGIN_TIMEOUT_SECONDS。"""
try:
return max(30, min(600, int((os.getenv("JIANGCHANG_WAIT_LOGIN_TIMEOUT_SECONDS") or "180").strip())))
except ValueError:
return 180
def _login_out_bundle_for(platform_key: str) -> dict: def _login_out_bundle_for(platform_key: str) -> dict:
"""传给子进程anchor + 未登录可见的选择器列表。""" """传给子进程anchor + 未登录可见的选择器列表。"""
spec = PLATFORMS.get(platform_key, {}) if platform_key else {} spec = PLATFORMS.get(platform_key, {}) if platform_key else {}
@@ -108,58 +123,70 @@ def _login_out_bundle_for(platform_key: str) -> dict:
} }
def cmd_login(account_id): def _run_login_browser_session(
log = get_skill_logger() target: dict,
log.info("login_command account_id=%r", account_id) timeout_sec: int,
target = get_account_by_id(account_id) *,
if not target: verbose_ui: bool = True,
log.warning("login_aborted account_not_found account_id=%r", account_id) ) -> tuple[bool, str]:
print("ERROR:ACCOUNT_NOT_FOUND") """
return 启动 Playwright 登录检测子进程**是否已登录以子进程 DOM 判定为准**
成功后尽力 `mark_login_status` list/排序展示**跨技能编排请勿再依赖该字段**可能与真实 Cookie 会话脱节
返回 (是否已登录成功, 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() channel = resolve_chromium_channel()
if not channel: if not channel:
log.warning("login_aborted no_chromium_channel") log.warning("login_aborted no_chromium_channel")
_print_browser_install_hint() if verbose_ui:
return _print_browser_install_hint()
return False, "no_channel"
try: try:
import playwright # noqa: F401 import playwright # noqa: F401
except ImportError: except ImportError:
log.error("login_aborted playwright_missing") log.error("login_aborted playwright_missing")
print("ERROR:需要 playwrightpip install playwright && playwright install chromium") if verbose_ui:
return print("ERROR:需要 playwrightpip install playwright && playwright install chromium")
return False, "playwright_missing"
profile_dir = (target.get("profile_dir") or "").strip() profile_dir = (target.get("profile_dir") or "").strip()
if not profile_dir: if not profile_dir:
log.warning("login_aborted profile_dir_empty account_id=%s", target.get("id")) log.warning("login_aborted profile_dir_empty account_id=%s", account_id)
print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。") if verbose_ui:
return print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。")
return False, "profile_missing"
os.makedirs(profile_dir, exist_ok=True) os.makedirs(profile_dir, exist_ok=True)
url = (target.get("url") or "").strip() or PLATFORM_URLS.get(target["platform"], "https://www.google.com") url = (target.get("url") or "").strip() or PLATFORM_URLS.get(
target["platform"], "https://www.google.com"
)
platform = (target.get("platform") or "").strip().lower() platform = (target.get("platform") or "").strip().lower()
timeout_sec = _login_timeout_seconds()
poll_sec = _login_poll_interval_seconds() poll_sec = _login_poll_interval_seconds()
dom_grace_sec = _login_dom_grace_seconds() dom_grace_sec = _login_dom_grace_seconds()
browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge" browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge"
print(f"正在为账号 [{target['name']}] 打开 {browser_name}") if verbose_ui:
print(f"访问地址:{url}") print(f"正在为账号 [{target['name']}] 打开 {browser_name}")
print( print(f"访问地址:{url}")
"请在窗口内完成登录。系统根据页面是否仍出现「登录」等未登录控件判断;" print(
"成功后自动关闭窗口并写入状态(无需手动关浏览器)。" "请在窗口内完成登录。系统根据页面是否仍出现「登录」等未登录控件判断;"
) "成功后自动关闭窗口并写入状态(无需手动关浏览器)。"
print( )
f"页面加载后先等待约 {dom_grace_sec:g} 秒再开始检测(减轻跳转误判,可用 JIANGCHANG_LOGIN_DOM_GRACE_SECONDS 调整)。" print(
) f"页面加载后先等待约 {dom_grace_sec:g} 秒再开始检测(减轻跳转误判,可用 JIANGCHANG_LOGIN_DOM_GRACE_SECONDS 调整)。"
print(f"轮询间隔约 {poll_sec} 秒;全程最长 {timeout_sec} 秒。") )
print(f"轮询间隔约 {poll_sec} 秒;全程最长 {timeout_sec} 秒。")
# 子进程login_detect_bundle + 多 tab DOM日志与主进程同一文件追加
log_file = get_skill_log_file_path() log_file = get_skill_log_file_path()
log.info( log.info(
"login_browser_start account_id=%s platform=%s channel=%s timeout_sec=%s poll_sec=%s " "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", "dom_grace_sec=%s profile_dir=%s start_url=%s log_file=%s",
target.get("id"), account_id,
platform, platform,
channel, channel,
timeout_sec, timeout_sec,
@@ -169,8 +196,7 @@ def cmd_login(account_id):
url, url,
log_file, log_file,
) )
_scripts_dir = os.path.dirname(os.path.abspath(__file__)) login_runner_path = os.path.join(_SERVICE_DIR, "login_child_runner.py")
login_runner_path = os.path.join(_scripts_dir, "am_login_child_runner.py")
cfg = { cfg = {
"channel": channel, "channel": channel,
"profile_dir": profile_dir, "profile_dir": profile_dir,
@@ -196,13 +222,16 @@ def cmd_login(account_id):
r = subprocess.run( r = subprocess.run(
[sys.executable, login_runner_path, cfg_path], [sys.executable, login_runner_path, cfg_path],
timeout=timeout_sec + 180, timeout=timeout_sec + 180,
env=subprocess_env_with_trace(),
) )
proc_rc = r.returncode proc_rc = r.returncode
if proc_rc != 0: if proc_rc != 0:
print("⚠️ 浏览器进程异常退出,将仅根据已写入的检测结果更新状态") if verbose_ui:
print("⚠️ 浏览器进程异常退出,将仅根据已写入的检测结果更新状态")
log.warning("login_subprocess_nonzero_return code=%s", proc_rc) log.warning("login_subprocess_nonzero_return code=%s", proc_rc)
except subprocess.TimeoutExpired as ex: except subprocess.TimeoutExpired as ex:
print("⚠️ 等待浏览器超时,将仅根据已写入的检测结果更新状态") if verbose_ui:
print("⚠️ 等待浏览器超时,将仅根据已写入的检测结果更新状态")
log.warning("login_subprocess_timeout err=%s", ex) log.warning("login_subprocess_timeout err=%s", ex)
finally: finally:
try: try:
@@ -211,11 +240,16 @@ def cmd_login(account_id):
pass pass
interactive_ok = False interactive_ok = False
end_reason = "subprocess_error"
try: try:
with open(result_path, encoding="utf-8") as rf: with open(result_path, encoding="utf-8") as rf:
interactive_ok = bool(json.load(rf).get("interactive_ok")) data = json.load(rf)
interactive_ok = bool(data.get("interactive_ok"))
end_reason = str(data.get("end_reason") or "").strip() or (
"login_ok" if interactive_ok else "timeout"
)
except Exception: except Exception:
pass end_reason = "subprocess_error"
try: try:
os.unlink(result_path) os.unlink(result_path)
except OSError: except OSError:
@@ -224,26 +258,102 @@ def cmd_login(account_id):
time.sleep(0.5) time.sleep(0.5)
ok = interactive_ok ok = interactive_ok
log.info( log.info(
"login_finished account_id=%s interactive_ok=%s subprocess_rc=%s marking_db=%s", "login_finished account_id=%s interactive_ok=%s end_reason=%s subprocess_rc=%s marking_db=%s",
target.get("id"), account_id,
interactive_ok, interactive_ok,
end_reason,
proc_rc, proc_rc,
ok, ok,
) )
_mark_login_status(target["id"], ok) mark_login_status(account_id, ok)
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: if ok:
print("✅ 已判定登录成功,状态已写入数据库") print("✅ 已判定登录成功,状态已写入数据库")
else: else:
log.warning( log.warning(
"login_not_detected account_id=%s platform=%s see_log=%s", "login_not_detected account_id=%s platform=%s see_log=%s",
target.get("id"), target.get("id"),
platform, (target.get("platform") or "").strip().lower(),
log_file, log_file,
) )
print("⚠️ 未检测到有效登录,状态为未登录。请关闭其他占用该用户目录的浏览器后重试,或延长 JIANGCHANG_LOGIN_TIMEOUT_SECONDS 后再登录。") print("⚠️ 未检测到有效登录,状态为未登录。请关闭其他占用该用户目录的浏览器后重试,或延长 JIANGCHANG_LOGIN_TIMEOUT_SECONDS 后再登录。")
print(f" 详细轮询日志见:{log_file}(可将 JIANGCHANG_LOG_LEVEL=DEBUG 打开更细粒度)") print(f" 详细轮询日志见:{log_file}(可将 JIANGCHANG_LOG_LEVEL=DEBUG 打开更细粒度)")
def cmd_wait_login(account_id, force: bool = False) -> int:
"""
跨技能编排若未登录则打开浏览器并在限时内等待登录
返回 0已登录原本已登录或本次登录成功1失败超时 / 用户关浏览器 / 其它
stdout OK:WAIT_LOGIN ERROR:WAIT_LOGIN_*
force=True即使库内 login_status=1 也仍打开浏览器并限时检测供搜狐发布等页面仍要求登录场景
"""
log = get_skill_logger()
log.info("wait_login_command account_id=%r force=%s", account_id, force)
target = get_account_by_id(account_id)
if not target:
log.warning("wait_login_aborted account_not_found account_id=%r", account_id)
print("ERROR:ACCOUNT_NOT_FOUND")
return 1
# 库内 login_status=1 时跳过开窗(与真实 Cookie 可能不同步时用 force=True
if int(target.get("login_status") or 0) == 1 and not force:
print("OK:WAIT_LOGIN")
return 0
if not resolve_chromium_channel():
_print_browser_install_hint()
return 1
try:
import playwright # noqa: F401
except ImportError:
print("ERROR:需要 playwrightpip install playwright && playwright install chromium")
return 1
if not (target.get("profile_dir") or "").strip():
print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。")
return 1
sec = _wait_login_timeout_seconds()
print(
f"⏳ 账号「{target.get('name', target['id'])}」尚未登录,将打开浏览器;请在 {sec} 秒内完成登录。"
"关闭浏览器将视为放弃。"
)
ok, end_reason = _run_login_browser_session(target, sec, verbose_ui=False)
if ok:
print("OK:WAIT_LOGIN")
return 0
if end_reason == "user_closed":
print("ERROR:WAIT_LOGIN_ABORTED 已关闭浏览器,未完成登录。")
return 1
if end_reason == "timeout":
print("ERROR:WAIT_LOGIN_TIMEOUT 限时内未完成登录。")
return 1
if end_reason == "playwright_missing":
print("ERROR:需要 playwrightpip install playwright && playwright install chromium")
return 1
if end_reason == "profile_missing":
print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。")
return 1
if end_reason == "no_channel":
_print_browser_install_hint()
return 1
print(f"ERROR:WAIT_LOGIN_FAILED end_reason={end_reason}")
return 1
def cmd_open(account_id): def cmd_open(account_id):
"""打开该账号的持久化浏览器,仅用于肉眼确认是否已登录;不写数据库。""" """打开该账号的持久化浏览器,仅用于肉眼确认是否已登录;不写数据库。"""
get_skill_logger().info("open account_id=%r", account_id) get_skill_logger().info("open account_id=%r", account_id)
@@ -278,8 +388,7 @@ def cmd_open(account_id):
print("请在本窗口中自行确认登录态。关闭浏览器后命令结束。") print("请在本窗口中自行确认登录态。关闭浏览器后命令结束。")
print("需要自动检测并写入数据库时请执行python main.py login <id>") print("需要自动检测并写入数据库时请执行python main.py login <id>")
_scripts_dir = os.path.dirname(os.path.abspath(__file__)) open_runner_path = os.path.join(_SERVICE_DIR, "open_child_runner.py")
open_runner_path = os.path.join(_scripts_dir, "am_open_child_runner.py")
cfg = {"channel": channel, "profile_dir": profile_dir, "url": url} cfg = {"channel": channel, "profile_dir": profile_dir, "url": url}
with tempfile.NamedTemporaryFile( with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False, encoding="utf-8" mode="w", suffix=".json", delete=False, encoding="utf-8"
@@ -287,7 +396,10 @@ def cmd_open(account_id):
json.dump(cfg, jf, ensure_ascii=False) json.dump(cfg, jf, ensure_ascii=False)
cfg_path = jf.name cfg_path = jf.name
try: try:
subprocess.run([sys.executable, open_runner_path, cfg_path]) subprocess.run(
[sys.executable, open_runner_path, cfg_path],
env=subprocess_env_with_trace(),
)
finally: finally:
try: try:
os.unlink(cfg_path) os.unlink(cfg_path)

View File

@@ -1,12 +1,33 @@
"""Playwright 登录检测子进程:由 main.cmd_login 以独立解释器启动argv[1] 为 JSON 配置路径。""" """Playwright 登录检测子进程:由 browser_service.cmd_login 以独立解释器启动argv[1] 为 JSON 配置路径。"""
import json import json
import logging import os
import sys import sys
import time import time
from urllib.parse import urlparse 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 jiangchang_skill_core.unified_logging import attach_unified_file_handler
from playwright.sync_api import sync_playwright from playwright.sync_api import sync_playwright
try:
from playwright.sync_api import Error as PlaywrightError
except ImportError:
PlaywrightError = Exception # type: ignore[misc,assignment]
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 isinstance(ex, PlaywrightError) and (
"closed" in s or "destroyed" in s
):
return True
return False
def page_location_href(p): def page_location_href(p):
# Prefer location.href over page.url for SPA (e.g. Sohu shell path lag). # Prefer location.href over page.url for SPA (e.g. Sohu shell path lag).
@@ -117,21 +138,16 @@ def main():
t0 = time.time() t0 = time.time()
deadline = t0 + float(c["timeout_sec"]) deadline = t0 + float(c["timeout_sec"])
interactive_ok = False interactive_ok = False
end_reason = "timeout"
result_path = c.get("result_path") or "" result_path = c.get("result_path") or ""
log_path = (c.get("log_file") or "").strip() log_path = (c.get("log_file") or "").strip()
lg = logging.getLogger("openclaw.skill.account_manager.login_child") lg = None
if log_path: if log_path:
lg.handlers.clear() lg = attach_unified_file_handler(
lg.setLevel(logging.DEBUG) log_path,
_fh = logging.FileHandler(log_path, encoding="utf-8") skill_slug="account-manager",
_fh.setFormatter( logger_name="openclaw.skill.account_manager.login_child",
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 = "" last_eval_detail = ""
with sync_playwright() as p: with sync_playwright() as p:
ctx = p.chromium.launch_persistent_context( ctx = p.chromium.launch_persistent_context(
@@ -146,7 +162,7 @@ def main():
page.goto(c["url"], wait_until="domcontentloaded", timeout=60000) page.goto(c["url"], wait_until="domcontentloaded", timeout=60000)
if dom_grace > 0: if dom_grace > 0:
time.sleep(dom_grace) time.sleep(dom_grace)
if lg.handlers: if lg is not None:
lg.info( lg.info(
"goto_done initial_pages=%s poll_interval_sec=%s dom_grace_sec=%s", "goto_done initial_pages=%s poll_interval_sec=%s dom_grace_sec=%s",
len(ctx.pages or []), len(ctx.pages or []),
@@ -159,7 +175,7 @@ def main():
still_logged_out, detail = evaluate_logged_out_dom(ctx, page, bundle) still_logged_out, detail = evaluate_logged_out_dom(ctx, page, bundle)
ok = not still_logged_out ok = not still_logged_out
last_eval_detail = detail last_eval_detail = detail
if lg.handlers: if lg is not None:
parts = [] parts = []
for tab in list(ctx.pages or []): for tab in list(ctx.pages or []):
try: try:
@@ -181,16 +197,22 @@ def main():
) )
if ok: if ok:
interactive_ok = True interactive_ok = True
if lg.handlers: end_reason = "login_ok"
if lg is not None:
lg.info("login_detected_ok %s", detail) lg.info("login_detected_ok %s", detail)
break break
except Exception as ex: except Exception as ex:
if lg.handlers: 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) lg.warning("poll_exception err=%s", ex, exc_info=True)
spent = time.time() - iter_start spent = time.time() - iter_start
time.sleep(max(0.0, poll - spent)) time.sleep(max(0.0, poll - spent))
if not interactive_ok: if not interactive_ok and end_reason != "user_closed":
if lg.handlers: if lg is not None:
lg.warning("login_poll_exhausted detail=%s", last_eval_detail) lg.warning("login_poll_exhausted detail=%s", last_eval_detail)
rem = max(0.0, deadline - time.time()) rem = max(0.0, deadline - time.time())
if rem > 0: if rem > 0:
@@ -199,18 +221,38 @@ def main():
except Exception: except Exception:
pass pass
else: else:
if lg.handlers: if lg is not None:
lg.info("login_success_closing_browser_immediately") lg.info("login_dom_ok_preparing_to_release_profile_dir")
if interactive_ok:
# 给 Cookie/会话落盘一点时间;随后必须关闭上下文,否则 llm-manager 无法再对同一 profile 启动持久化浏览器
print(
"INFO:LOGIN_DOM_OK 页面检测已判定登录成功;约 2 秒后关闭本窗口以释放用户目录,供后续大模型网页会话使用。",
flush=True,
)
time.sleep(2.0)
except Exception as e: except Exception as e:
if lg.handlers: if _is_context_closed_error(e):
lg.exception("login_runner_fatal err=%s", e) end_reason = "user_closed"
print(e, file=sys.stderr) interactive_ok = False
raise if lg is not None:
lg.warning("login_runner_browser_closed err=%s", e)
else:
if lg is not None:
lg.exception("login_runner_fatal err=%s", e)
print(e, file=sys.stderr)
raise
finally: finally:
if result_path: if result_path:
try: try:
with open(result_path, "w", encoding="utf-8") as rf: with open(result_path, "w", encoding="utf-8") as rf:
json.dump({"interactive_ok": interactive_ok}, rf, ensure_ascii=False) json.dump(
{
"interactive_ok": interactive_ok,
"end_reason": end_reason,
},
rf,
ensure_ascii=False,
)
except Exception: except Exception:
pass pass
try: try:

View File

@@ -1,4 +1,4 @@
"""Playwright 仅打开浏览器子进程:由 main.cmd_open 启动argv[1] 为 JSON 配置路径。""" """Playwright 仅打开浏览器子进程:由 browser_service.cmd_open 启动argv[1] 为 JSON 配置路径。"""
import json import json
import sys import sys

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",
]

View File

@@ -4,22 +4,9 @@ import re
import sys import sys
from typing import Optional from typing import Optional
from am_constants import SKILL_SLUG from jiangchang_skill_core.runtime_env import get_data_root, get_user_id
from am_platforms import _PLATFORM_PRIMARY_CN 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(): def get_skill_data_dir():
@@ -28,13 +15,6 @@ def get_skill_data_dir():
return path 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(): def get_db_path():
return os.path.join(get_skill_data_dir(), "account-manager.db") return os.path.join(get_skill_data_dir(), "account-manager.db")