diff --git a/scripts/am_browser.py b/scripts/am_browser.py new file mode 100644 index 0000000..2029508 --- /dev/null +++ b/scripts/am_browser.py @@ -0,0 +1,295 @@ +"""Chromium/Edge 检测与 Playwright 登录、仅打开浏览器。""" +import json +import os +import subprocess +import sys +import tempfile +import time +from urllib.parse import urlparse + +from am_db import _mark_login_status, get_account_by_id +from am_logging import get_skill_logger, get_skill_log_file_path +from am_platforms import ( + PLATFORMS, + PLATFORM_URLS, + _LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS, +) + + +def _win_find_exe(candidates): + for p in candidates: + if p and os.path.isfile(p): + return p + return None + + +def resolve_chromium_channel(): + """ + Playwright channel: 'chrome' | 'msedge' | None + Windows 优先 Chrome，其次 Edge；其它系统默认 chrome（由 Playwright 解析 PATH）。 + """ + if sys.platform != "win32": + return "chrome" + + pf = os.environ.get("ProgramFiles", r"C:\Program Files") + pfx86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)") + local = os.environ.get("LocalAppData", "") + + chrome = _win_find_exe( + [ + os.path.join(pf, "Google", "Chrome", "Application", "chrome.exe"), + os.path.join(pfx86, "Google", "Chrome", "Application", "chrome.exe"), + os.path.join(local, "Google", "Chrome", "Application", "chrome.exe") if local else "", + ] + ) + if chrome: + return "chrome" + + edge = _win_find_exe( + [ + os.path.join(pfx86, "Microsoft", "Edge", "Application", "msedge.exe"), + os.path.join(pf, "Microsoft", "Edge", "Application", "msedge.exe"), + os.path.join(local, "Microsoft", "Edge", "Application", "msedge.exe") if local else "", + ] + ) + if edge: + return "msedge" + + return None + + +def _print_browser_install_hint(): + print("❌ 未检测到 Google Chrome 或 Microsoft Edge，请先安装后再登录：") + print(" • Chrome: https://www.google.com/chrome/") + print(" • Edge: https://www.microsoft.com/zh-cn/edge") + + +def _login_timeout_seconds(): + try: + return max(60, int((os.getenv("JIANGCHANG_LOGIN_TIMEOUT_SECONDS") or "300").strip())) + except ValueError: + return 300 + + +def _login_poll_interval_seconds(): + """轮询间隔，默认 1.5s。""" + try: + v = float((os.getenv("JIANGCHANG_LOGIN_POLL_INTERVAL_SECONDS") or "1.5").strip()) + return max(0.5, min(v, 10.0)) + except ValueError: + return 1.5 + + +def _login_dom_grace_seconds() -> float: + """goto 后等待再开始 DOM 判定，减轻跳转中途误判。默认 4s，JIANGCHANG_LOGIN_DOM_GRACE_SECONDS。""" + try: + v = float((os.getenv("JIANGCHANG_LOGIN_DOM_GRACE_SECONDS") or "4").strip()) + return max(0.0, min(v, 60.0)) + except ValueError: + return 4.0 + + +def _login_out_bundle_for(platform_key: str) -> dict: + """传给子进程：anchor + 未登录可见的选择器列表。""" + spec = PLATFORMS.get(platform_key, {}) if platform_key else {} + anchor = (urlparse(PLATFORM_URLS.get(platform_key, "") or "").netloc or "").lower() + spec_lo_dom = [ + str(s).strip() + for s in (spec.get("login_logged_out_selectors") or []) + if s is not None and str(s).strip() + ] + if bool(spec.get("login_skip_generic_logged_out_dom")): + logged_out_dom_selectors = list(spec_lo_dom) + else: + logged_out_dom_selectors = list(_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS) + spec_lo_dom + return { + "logged_out_dom_selectors": logged_out_dom_selectors, + "anchor_host": anchor, + } + + +def cmd_login(account_id): + log = get_skill_logger() + log.info("login_command account_id=%r", account_id) + target = get_account_by_id(account_id) + if not target: + log.warning("login_aborted account_not_found account_id=%r", account_id) + print("ERROR:ACCOUNT_NOT_FOUND") + return + + channel = resolve_chromium_channel() + if not channel: + log.warning("login_aborted no_chromium_channel") + _print_browser_install_hint() + return + + try: + import playwright # noqa: F401 + except ImportError: + log.error("login_aborted playwright_missing") + print("ERROR:需要 playwright：pip install playwright && playwright install chromium") + return + + profile_dir = (target.get("profile_dir") or "").strip() + if not profile_dir: + log.warning("login_aborted profile_dir_empty account_id=%s", target.get("id")) + print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。") + return + os.makedirs(profile_dir, exist_ok=True) + url = (target.get("url") or "").strip() or PLATFORM_URLS.get(target["platform"], "https://www.google.com") + platform = (target.get("platform") or "").strip().lower() + timeout_sec = _login_timeout_seconds() + poll_sec = _login_poll_interval_seconds() + dom_grace_sec = _login_dom_grace_seconds() + + browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge" + print(f"正在为账号 [{target['name']}] 打开 {browser_name} …") + print(f"访问地址：{url}") + print( + "请在窗口内完成登录。系统根据页面是否仍出现「登录」等未登录控件判断；" + "成功后自动关闭窗口并写入状态（无需手动关浏览器）。" + ) + print( + f"页面加载后先等待约 {dom_grace_sec:g} 秒再开始检测（减轻跳转误判，可用 JIANGCHANG_LOGIN_DOM_GRACE_SECONDS 调整）。" + ) + print(f"轮询间隔约 {poll_sec} 秒；全程最长 {timeout_sec} 秒。") + + # 子进程：login_detect_bundle + 多 tab DOM；日志与主进程同一文件（追加）。 + log_file = get_skill_log_file_path() + log.info( + "login_browser_start account_id=%s platform=%s channel=%s timeout_sec=%s poll_sec=%s " + "dom_grace_sec=%s profile_dir=%s start_url=%s log_file=%s", + target.get("id"), + platform, + channel, + timeout_sec, + poll_sec, + dom_grace_sec, + profile_dir, + url, + log_file, + ) + _scripts_dir = os.path.dirname(os.path.abspath(__file__)) + login_runner_path = os.path.join(_scripts_dir, "am_login_child_runner.py") + cfg = { + "channel": channel, + "profile_dir": profile_dir, + "url": url, + "timeout_sec": timeout_sec, + "poll_interval": poll_sec, + "dom_grace_sec": dom_grace_sec, + "login_detect_bundle": _login_out_bundle_for(platform), + "log_file": log_file, + } + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False, encoding="utf-8" + ) as rf: + result_path = rf.name + cfg["result_path"] = result_path + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False, encoding="utf-8" + ) as jf: + json.dump(cfg, jf, ensure_ascii=False) + cfg_path = jf.name + proc_rc = None + try: + r = subprocess.run( + [sys.executable, login_runner_path, cfg_path], + timeout=timeout_sec + 180, + ) + proc_rc = r.returncode + if proc_rc != 0: + print("⚠️ 浏览器进程异常退出，将仅根据已写入的检测结果更新状态") + log.warning("login_subprocess_nonzero_return code=%s", proc_rc) + except subprocess.TimeoutExpired as ex: + print("⚠️ 等待浏览器超时，将仅根据已写入的检测结果更新状态") + log.warning("login_subprocess_timeout err=%s", ex) + finally: + try: + os.unlink(cfg_path) + except OSError: + pass + + interactive_ok = False + try: + with open(result_path, encoding="utf-8") as rf: + interactive_ok = bool(json.load(rf).get("interactive_ok")) + except Exception: + pass + try: + os.unlink(result_path) + except OSError: + pass + + time.sleep(0.5) + ok = interactive_ok + log.info( + "login_finished account_id=%s interactive_ok=%s subprocess_rc=%s marking_db=%s", + target.get("id"), + interactive_ok, + proc_rc, + ok, + ) + _mark_login_status(target["id"], ok) + if ok: + print("✅ 已判定登录成功，状态已写入数据库") + else: + log.warning( + "login_not_detected account_id=%s platform=%s see_log=%s", + target.get("id"), + platform, + log_file, + ) + print("⚠️ 未检测到有效登录，状态为未登录。请关闭其他占用该用户目录的浏览器后重试，或延长 JIANGCHANG_LOGIN_TIMEOUT_SECONDS 后再登录。") + print(f"ℹ️ 详细轮询日志见：{log_file}（可将 JIANGCHANG_LOG_LEVEL=DEBUG 打开更细粒度）") + + +def cmd_open(account_id): + """打开该账号的持久化浏览器，仅用于肉眼确认是否已登录；不写数据库。""" + get_skill_logger().info("open account_id=%r", account_id) + target = get_account_by_id(account_id) + if not target: + get_skill_logger().warning("open_not_found account_id=%r", account_id) + print("ERROR:ACCOUNT_NOT_FOUND") + return + + channel = resolve_chromium_channel() + if not channel: + _print_browser_install_hint() + return + + try: + import playwright # noqa: F401 + except ImportError: + print("ERROR:需要 playwright：pip install playwright && playwright install chromium") + return + + profile_dir = (target.get("profile_dir") or "").strip() + if not profile_dir: + print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。") + return + os.makedirs(profile_dir, exist_ok=True) + url = (target.get("url") or "").strip() or PLATFORM_URLS.get( + target["platform"], "https://www.google.com" + ) + browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge" + print(f"正在打开 [{target['name']}] 的 {browser_name}（仅查看，不写入数据库）") + print(f"地址：{url}") + print("请在本窗口中自行确认登录态。关闭浏览器后命令结束。") + print("需要自动检测并写入数据库时，请执行：python main.py login ") + + _scripts_dir = os.path.dirname(os.path.abspath(__file__)) + open_runner_path = os.path.join(_scripts_dir, "am_open_child_runner.py") + cfg = {"channel": channel, "profile_dir": profile_dir, "url": url} + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False, encoding="utf-8" + ) as jf: + json.dump(cfg, jf, ensure_ascii=False) + cfg_path = jf.name + try: + subprocess.run([sys.executable, open_runner_path, cfg_path]) + finally: + try: + os.unlink(cfg_path) + except OSError: + pass diff --git a/scripts/am_cli.py b/scripts/am_cli.py new file mode 100644 index 0000000..0c0c410 --- /dev/null +++ b/scripts/am_cli.py @@ -0,0 +1,187 @@ +"""CLI 入口：用法说明与 argv 分发。""" +import sys + +from am_browser import cmd_login, cmd_open +from am_commands_read import ( + cmd_get, + cmd_list, + cmd_list_json, + cmd_pick_logged_in, + 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, +) +from am_constants import _apply_cli_local_dev_env +from am_logging import get_skill_logger +from am_platforms import _platform_list_cn_for_help, resolve_platform_key +from am_runtime_paths import _maybe_print_paths_debug_cli + + +def _cli_print_full_usage() -> None: + """总览（未带子命令或需要完整说明时）。""" + print("用法概览（将 main.py 换为你的路径）：") + print(" python main.py add <平台中文名> <手机号>") + print(" python main.py list [平台|all|全部]") + print(" python main.py pick-logged-in <平台> # 跨技能用：仅已登录账号，一行 JSON 或 ERROR") + print(" python main.py pick-web <平台> # 跨技能用：优先已登录，否则最新一条账号") + print(" python main.py list-json <平台|all> [--limit N] # 跨技能用：一行 JSON 数组，元素同 get") + print(" python main.py get ") + print(" python main.py set-login-status <0|1> # 跨技能回写登录态") + print(" python main.py open ") + print(" python main.py login ") + print(" python main.py delete id ") + print(" python main.py delete platform <平台>") + print(" python main.py delete platform <平台> <手机号>") + print(" python main.py <平台> # 等价于只列该平台") + print(" python main.py <平台> list # 同上") + print() + print("支持的平台（可用下列中文称呼，亦支持英文键如 sohu）：") + print(" " + _platform_list_cn_for_help()) + + +def _cli_fail_add() -> None: + print("ERROR:CLI_ADD_MISSING_ARGS") + print() + print("【add 添加账号】参数不足。") + print(" 必填：平台名称（中文即可，如 搜狐号、知乎、微信公众号）") + print(" 必填：中国大陆 11 位手机号（1 开头，第二位 3～9），同平台下不可重复") + print() + print("示例：") + print(" python main.py add 搜狐号 13800138000") + print(" python main.py add 知乎 13900001111") + print() + print("支持的平台：" + _platform_list_cn_for_help()) + + +def _cli_fail_need_account_id(subcmd: str) -> None: + print(f"ERROR:CLI_{subcmd.upper()}_MISSING_ID") + print() + print(f"【{subcmd}】缺少必填参数：账号 id（数据库自增整数，可先 list 查看）。") + print("示例：") + print(f" python main.py {subcmd} 1") + + +def _cli_fail_delete() -> None: + print("ERROR:CLI_DELETE_MISSING_ARGS") + print() + print("【delete 删除账号】参数不足。支持三种方式：") + print(" 按 id：python main.py delete id ") + print(" 按平台（删该平台下全部）：python main.py delete platform <平台中文名或英文键>") + print(" 按平台+手机号：python main.py delete platform <平台> <手机号>") + print() + print("示例：") + print(" python main.py delete id 3") + print(" python main.py delete platform 搜狐号") + print(" python main.py delete platform 头条号 18925203701") + print() + print("说明：会同时删除数据库记录与 profile 用户数据目录（若存在）。") + + +def main() -> None: + if len(sys.argv) < 2: + _cli_print_full_usage() + sys.exit(1) + + _apply_cli_local_dev_env() + get_skill_logger().info("cli_start argv=%s", sys.argv) + cmd = sys.argv[1] + _maybe_print_paths_debug_cli() + + plat_from_first = resolve_platform_key(cmd) + + if len(sys.argv) >= 3 and sys.argv[2] == "list" and plat_from_first: + cmd_list(plat_from_first) + elif plat_from_first and len(sys.argv) == 2: + cmd_list(plat_from_first) + elif cmd == "add": + if len(sys.argv) < 4: + _cli_fail_add() + sys.exit(1) + cmd_add(sys.argv[2], sys.argv[3]) + elif cmd == "list": + cmd_list(sys.argv[2] if len(sys.argv) >= 3 else "all") + elif cmd == "get": + if len(sys.argv) < 3: + _cli_fail_need_account_id("get") + sys.exit(1) + cmd_get(sys.argv[2]) + elif cmd == "set-login-status": + if len(sys.argv) < 4: + print("ERROR:CLI_SET_LOGIN_STATUS_MISSING_ARGS") + print("用法：python main.py set-login-status <0|1>") + print("说明：供 llm-manager 等脚本回写登录态；成功时 stdout 首行以 OK: 开头。") + sys.exit(1) + cmd_set_login_status(sys.argv[2], sys.argv[3]) + elif cmd == "list-json": + if len(sys.argv) < 3: + print("ERROR:CLI_LIST_JSON_MISSING_PLATFORM") + print("用法：python main.py list-json <平台|all|全部> [--limit N]") + print("说明：跨技能编排；成功时 stdout 仅一行 JSON 数组（元素与 get 一致）。") + sys.exit(1) + av = sys.argv[2:] + platform_arg = av[0] + lim = 200 + if "--limit" in av: + i = av.index("--limit") + if i + 1 < len(av): + try: + lim = int(av[i + 1]) + except ValueError: + pass + cmd_list_json(platform_arg, limit=lim) + elif cmd == "pick-logged-in": + if len(sys.argv) < 3: + print("ERROR:CLI_PICK_LOGGED_IN_MISSING_ARGS") + print("用法：python main.py pick-logged-in <平台中文名或英文键>") + print("说明：供 llm-manager 等技能查询该平台已登录账号；成功时 stdout 仅一行 JSON。") + sys.exit(1) + cmd_pick_logged_in(sys.argv[2]) + elif cmd == "pick-web": + if len(sys.argv) < 3: + print("ERROR:CLI_PICK_WEB_MISSING_ARGS") + print("用法：python main.py pick-web <平台中文名或英文键>") + print("说明：供 llm-manager 等；优先已登录账号，否则返回该平台最新一条账号（未登录也可用于打开浏览器登录）。") + sys.exit(1) + cmd_pick_web(sys.argv[2]) + elif cmd == "open": + if len(sys.argv) < 3: + _cli_fail_need_account_id("open") + sys.exit(1) + cmd_open(sys.argv[2]) + elif cmd == "login": + if len(sys.argv) < 3: + _cli_fail_need_account_id("login") + sys.exit(1) + cmd_login(sys.argv[2]) + elif cmd == "delete": + if len(sys.argv) < 4: + _cli_fail_delete() + sys.exit(1) + mode = (sys.argv[2] or "").strip().lower() + if mode == "id": + cmd_delete_by_id(sys.argv[3]) + elif mode == "platform": + if len(sys.argv) >= 5: + cmd_delete_by_platform_phone(sys.argv[3], sys.argv[4]) + else: + cmd_delete_by_platform(sys.argv[3]) + else: + print(f"ERROR:CLI_DELETE_BAD_MODE 不支持的删除方式「{sys.argv[2]}」，请使用 id 或 platform。") + _cli_fail_delete() + sys.exit(1) + else: + print(f"ERROR:CLI_UNKNOWN_OR_BAD_ARGS 子命令「{cmd}」无法识别，或参数组合不合法。") + print() + _cli_print_full_usage() + print() + print("说明：list 的筛选可为 all/全部，或上面列出的任一平台中文名。") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/scripts/am_commands_read.py b/scripts/am_commands_read.py new file mode 100644 index 0000000..27374f7 --- /dev/null +++ b/scripts/am_commands_read.py @@ -0,0 +1,285 @@ +"""只读类 CLI：list / 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 ") + 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)) diff --git a/scripts/am_commands_write.py b/scripts/am_commands_write.py new file mode 100644 index 0000000..aac69e5 --- /dev/null +++ b/scripts/am_commands_write.py @@ -0,0 +1,246 @@ +"""写入类 CLI：add / 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 开头、第二位为 3～9（示例 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 开头、第二位为 3～9。" + ) + 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") diff --git a/scripts/am_constants.py b/scripts/am_constants.py new file mode 100644 index 0000000..288c3a0 --- /dev/null +++ b/scripts/am_constants.py @@ -0,0 +1,33 @@ +"""技能级常量与本地 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. +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() diff --git a/scripts/am_db.py b/scripts/am_db.py new file mode 100644 index 0000000..4fca795 --- /dev/null +++ b/scripts/am_db.py @@ -0,0 +1,136 @@ +"""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() diff --git a/scripts/am_logging.py b/scripts/am_logging.py new file mode 100644 index 0000000..64e341c --- /dev/null +++ b/scripts/am_logging.py @@ -0,0 +1,69 @@ +"""技能级文件日志。""" +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 midnight），UTF-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 diff --git a/scripts/am_platforms.py b/scripts/am_platforms.py new file mode 100644 index 0000000..17d0846 --- /dev/null +++ b/scripts/am_platforms.py @@ -0,0 +1,139 @@ +"""平台配置、别名解析、手机号校验。""" +import re + +# 平台唯一配置表：只改这里即可。键 = 入库的 platform；label = 帮助/提示里的展示名； +# prefix = 默认账号名「{prefix}{序号}号」，省略时等于 label； +# aliases = 额外 CLI 称呼（英文键与 label 已自动参与解析，不必重复写）。 +# 登录检测：仅看页面 DOM。匹配 anchor 的标签页上出现「未登录」选择器则未登录；否则视为已登录。 +# - _LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS：通用「登录」按钮/链接等（可被 login_skip_generic_logged_out_dom 关闭）。 +# - login_logged_out_selectors：按平台追加；误判时也可用 login_skip_generic_logged_out_dom + 仅平台自选。 +PLATFORMS = { + "sohu": { + "url": "https://mp.sohu.com", + "label": "搜狐号", + "prefix": "搜狐", + "aliases": ["搜狐"], + }, + "toutiao": { + "url": "https://mp.toutiao.com/", + "label": "头条号", + "prefix": "头条", + "aliases": ["头条"], + }, + "zhihu": { + "url": "https://www.zhihu.com", + "label": "知乎", + "aliases": ["知乎号"], + }, + "wechat": { + "url": "https://mp.weixin.qq.com", + "label": "微信公众号", + "prefix": "微信", + "aliases": ["公众号", "微信"], + }, + "kimi": { + "url": "https://kimi.moonshot.cn", + "label": "Kimi", + "aliases": ["月之暗面"], + }, + "deepseek": { + "url": "https://chat.deepseek.com", + "label": "DeepSeek", + }, + "doubao": { + "url": "https://www.doubao.com", + "label": "豆包", + }, + "qianwen": { + "url": "https://tongyi.aliyun.com", + "label": "通义千问", + "prefix": "通义", + "aliases": ["通义", "千问"], + }, + "yiyan": { + "url": "https://yiyan.baidu.com", + "label": "文心一言", + "prefix": "文心", + "aliases": ["文心", "一言"], + }, + "yuanbao": { + "url": "https://yuanbao.tencent.com", + "label": "腾讯元宝", + "prefix": "元宝", + "aliases": ["元宝"], + }, +} + +# 未登录页共性：主行动点含「登录」——submit 按钮、Element 主按钮、Semi 文案、标题、通栏主按钮等，Playwright :has-text 可覆盖多数站点。 +_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS = ( + 'button:has-text("登录")', + 'role=button[name="登录"]', + 'a:has-text("登录")', + 'h4:has-text("登录")', + 'span.semi-button-content:has-text("登录")', +) + + +def _build_platform_derived(): + urls = {} + name_prefix = {} + primary_cn = {} + alias_to_key = {} + + def _register_alias(alias: str, key: str) -> None: + if not alias or not str(alias).strip(): + return + a = str(alias).strip() + if a not in alias_to_key: + alias_to_key[a] = key + lo = a.lower() + if lo not in alias_to_key: + alias_to_key[lo] = key + + for key, spec in PLATFORMS.items(): + urls[key] = spec["url"] + primary_cn[key] = spec["label"] + name_prefix[key] = (spec.get("prefix") or spec["label"]).strip() or key + _register_alias(key, key) + _register_alias(spec["label"], key) + for a in spec.get("aliases") or []: + _register_alias(a, key) + + return urls, name_prefix, primary_cn, alias_to_key + + +PLATFORM_URLS, _PLATFORM_NAME_CN, _PLATFORM_PRIMARY_CN, _PLATFORM_ALIAS_TO_KEY = _build_platform_derived() + + +def resolve_platform_key(name: str): + """将用户输入解析为内部 platform 键；无法识别返回 None。""" + if name is None: + return None + s = str(name).strip() + if not s: + return None + sl = s.lower() + if sl in PLATFORM_URLS: + return sl + return _PLATFORM_ALIAS_TO_KEY.get(s) + + +def _normalize_phone_digits(phone: str) -> str: + """从输入中提取数字串，供号段规则校验与入库去重（不对外承诺可随意夹杂符号）。""" + d = re.sub(r"\D", "", (phone or "").strip()) + if d.startswith("86") and len(d) >= 13: + d = d[2:] + return d + + +def _is_valid_cn_mobile11(digits: str) -> bool: + """中国大陆 11 位手机号：1 开头，第二位 3–9，共 11 位数字。""" + return bool(re.fullmatch(r"1[3-9]\d{9}", digits or "")) + + +def _platform_list_cn_for_help() -> str: + """帮助文案：一行中文展示名（不重复内部键）。""" + parts = [] + for k in sorted(PLATFORM_URLS.keys()): + parts.append(_PLATFORM_PRIMARY_CN.get(k, k)) + return "、".join(parts) diff --git a/scripts/am_runtime_paths.py b/scripts/am_runtime_paths.py new file mode 100644 index 0000000..f014ff3 --- /dev/null +++ b/scripts/am_runtime_paths.py @@ -0,0 +1,100 @@ +"""数据目录、profile 路径、CLI 路径调试输出。""" +import os +import re +import sys +from typing import Optional + +from am_constants import SKILL_SLUG +from am_platforms import _PLATFORM_PRIMARY_CN + + +def get_data_root(): + env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip() + if env: + return env + if sys.platform == "win32": + return r"D:\jiangchang-data" + return os.path.join(os.path.expanduser("~"), ".jiangchang-data") + + +def get_user_id(): + uid = (os.getenv("JIANGCHANG_USER_ID") or "").strip() + return uid or "_anon" + + +def get_skill_data_dir(): + path = os.path.join(get_data_root(), get_user_id(), SKILL_SLUG) + os.makedirs(path, exist_ok=True) + return path + + +def get_skill_logs_dir() -> str: + """{DATA_ROOT}/{USER_ID}/account-manager/logs/""" + path = os.path.join(get_skill_data_dir(), "logs") + os.makedirs(path, exist_ok=True) + return path + + +def get_db_path(): + return os.path.join(get_skill_data_dir(), "account-manager.db") + + +def _runtime_paths_debug_text(): + """供排查「为何 list 没数据」：终端未注入环境变量时会落到 _anon 等默认目录，与网关里用的库不是同一个文件。""" + env_root = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip() or "(未设置)" + env_uid = (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "(未设置→使用 _anon)" + return ( + "[account-manager] " + f"JIANGCHANG_DATA_ROOT={env_root} | " + f"JIANGCHANG_USER_ID={env_uid} | " + f"实际数据根={get_data_root()} | " + f"实际用户目录={get_user_id()} | " + f"数据库文件={get_db_path()}" + ) + + +def _maybe_print_paths_debug_cli(): + """设置 JIANGCHANG_ACCOUNT_DEBUG_PATHS=1 时，每次执行子命令前在 stderr 打印路径。""" + v = (os.getenv("JIANGCHANG_ACCOUNT_DEBUG_PATHS") or "").strip().lower() + if v in ("1", "true", "yes", "on"): + print(_runtime_paths_debug_text(), file=sys.stderr) + + +_WIN_FS_FORBIDDEN = re.compile(r'[<>:"/\\|?*\x00-\x1f]') +_WIN_RESERVED_NAMES = frozenset( + ["CON", "PRN", "AUX", "NUL"] + + [f"COM{i}" for i in range(1, 10)] + + [f"LPT{i}" for i in range(1, 10)] +) + + +def _fs_safe_segment(name: str, fallback: str) -> str: + """单级目录名：去掉 Windows 非法字符，避免末尾点/空格；空则用 fallback。""" + t = _WIN_FS_FORBIDDEN.sub("_", (name or "").strip()) + t = t.rstrip(" .") + if not t: + t = fallback + if t.upper() in _WIN_RESERVED_NAMES: + t = f"_{t}" + return t + + +def get_default_profile_dir( + platform_key: str, phone: Optional[str], account_id: Optional[int] = None +) -> str: + """ + 默认可读路径：profiles/<平台展示名>/<手机号>/（便于在资源管理器中辨认）。 + 无手机号时退化为 profiles/<平台展示名>/no_phone_/。 + """ + label = _PLATFORM_PRIMARY_CN.get(platform_key, platform_key or "account") + label_seg = _fs_safe_segment(label, _fs_safe_segment(platform_key or "", "platform")) + ph = (phone or "").strip() + if ph: + phone_seg = _fs_safe_segment(ph, f"id_{account_id}" if account_id is not None else "phone") + else: + phone_seg = _fs_safe_segment( + "", f"no_phone_{account_id}" if account_id is not None else "no_phone" + ) + path = os.path.join(get_skill_data_dir(), "profiles", label_seg, phone_seg) + os.makedirs(path, exist_ok=True) + return path diff --git a/scripts/main.py b/scripts/main.py index 74e1b04..c4813f5 100644 --- a/scripts/main.py +++ b/scripts/main.py @@ -1,1407 +1,18 @@ -import sys -import json -import logging -import os -import re -import sqlite3 -from logging.handlers import TimedRotatingFileHandler -from urllib.parse import urlparse -import shutil -import subprocess -import tempfile -import time -from datetime import datetime -from typing import Optional +""" +account-manager CLI 入口。 -# Windows GBK 编码兼容修复 +业务逻辑按模块拆分（am_*.py），便于维护与满足 PyArmor trial 对单文件体积的限制。 +""" +import sys + +# Windows GBK 编码兼容修复（须在任何业务 import 与 print 之前） if sys.platform == "win32": import io + sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace") -BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -SKILL_SLUG = "account-manager" -# 与其它 OpenClaw 技能对齐：openclaw.skill. -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() - - -# 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 - - -# 平台唯一配置表：只改这里即可。键 = 入库的 platform；label = 帮助/提示里的展示名； -# prefix = 默认账号名「{prefix}{序号}号」，省略时等于 label； -# aliases = 额外 CLI 称呼（英文键与 label 已自动参与解析，不必重复写）。 -# 登录检测：仅看页面 DOM。匹配 anchor 的标签页上出现「未登录」选择器则未登录；否则视为已登录。 -# - _LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS：通用「登录」按钮/链接等（可被 login_skip_generic_logged_out_dom 关闭）。 -# - login_logged_out_selectors：按平台追加；误判时也可用 login_skip_generic_logged_out_dom + 仅平台自选。 -PLATFORMS = { - "sohu": { - "url": "https://mp.sohu.com", - "label": "搜狐号", - "prefix": "搜狐", - "aliases": ["搜狐"], - }, - "toutiao": { - "url": "https://mp.toutiao.com/", - "label": "头条号", - "prefix": "头条", - "aliases": ["头条"], - }, - "zhihu": { - "url": "https://www.zhihu.com", - "label": "知乎", - "aliases": ["知乎号"], - }, - "wechat": { - "url": "https://mp.weixin.qq.com", - "label": "微信公众号", - "prefix": "微信", - "aliases": ["公众号", "微信"], - }, - "kimi": { - "url": "https://kimi.moonshot.cn", - "label": "Kimi", - "aliases": ["月之暗面"], - }, - "deepseek": { - "url": "https://chat.deepseek.com", - "label": "DeepSeek", - }, - "doubao": { - "url": "https://www.doubao.com", - "label": "豆包", - }, - "qianwen": { - "url": "https://tongyi.aliyun.com", - "label": "通义千问", - "prefix": "通义", - "aliases": ["通义", "千问"], - }, - "yiyan": { - "url": "https://yiyan.baidu.com", - "label": "文心一言", - "prefix": "文心", - "aliases": ["文心", "一言"], - }, - "yuanbao": { - "url": "https://yuanbao.tencent.com", - "label": "腾讯元宝", - "prefix": "元宝", - "aliases": ["元宝"], - }, -} - -# 未登录页共性：主行动点含「登录」——submit 按钮、Element 主按钮、Semi 文案、标题、通栏主按钮等，Playwright :has-text 可覆盖多数站点。 -_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS = ( - 'button:has-text("登录")', - 'role=button[name="登录"]', - 'a:has-text("登录")', - 'h4:has-text("登录")', - 'span.semi-button-content:has-text("登录")', -) - - -def _build_platform_derived(): - urls = {} - name_prefix = {} - primary_cn = {} - alias_to_key = {} - - def _register_alias(alias: str, key: str) -> None: - if not alias or not str(alias).strip(): - return - a = str(alias).strip() - if a not in alias_to_key: - alias_to_key[a] = key - lo = a.lower() - if lo not in alias_to_key: - alias_to_key[lo] = key - - for key, spec in PLATFORMS.items(): - urls[key] = spec["url"] - primary_cn[key] = spec["label"] - name_prefix[key] = (spec.get("prefix") or spec["label"]).strip() or key - _register_alias(key, key) - _register_alias(spec["label"], key) - for a in spec.get("aliases") or []: - _register_alias(a, key) - - return urls, name_prefix, primary_cn, alias_to_key - - -PLATFORM_URLS, _PLATFORM_NAME_CN, _PLATFORM_PRIMARY_CN, _PLATFORM_ALIAS_TO_KEY = _build_platform_derived() - - -def resolve_platform_key(name: str): - """将用户输入解析为内部 platform 键；无法识别返回 None。""" - if name is None: - return None - s = str(name).strip() - if not s: - return None - sl = s.lower() - if sl in PLATFORM_URLS: - return sl - return _PLATFORM_ALIAS_TO_KEY.get(s) - - -def _normalize_phone_digits(phone: str) -> str: - """从输入中提取数字串，供号段规则校验与入库去重（不对外承诺可随意夹杂符号）。""" - d = re.sub(r"\D", "", (phone or "").strip()) - if d.startswith("86") and len(d) >= 13: - d = d[2:] - return d - - -def _is_valid_cn_mobile11(digits: str) -> bool: - """中国大陆 11 位手机号：1 开头，第二位 3–9，共 11 位数字。""" - return bool(re.fullmatch(r"1[3-9]\d{9}", digits or "")) - - -def _platform_list_cn_for_help() -> str: - """帮助文案：一行中文展示名（不重复内部键）。""" - parts = [] - for k in sorted(PLATFORM_URLS.keys()): - parts.append(_PLATFORM_PRIMARY_CN.get(k, k)) - return "、".join(parts) - - -def get_data_root(): - env = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip() - if env: - return env - if sys.platform == "win32": - return r"D:\jiangchang-data" - return os.path.join(os.path.expanduser("~"), ".jiangchang-data") - - -def get_user_id(): - uid = (os.getenv("JIANGCHANG_USER_ID") or "").strip() - return uid or "_anon" - - -def get_skill_data_dir(): - path = os.path.join(get_data_root(), get_user_id(), SKILL_SLUG) - os.makedirs(path, exist_ok=True) - return path - - -def get_skill_logs_dir() -> str: - """{DATA_ROOT}/{USER_ID}/account-manager/logs/""" - path = os.path.join(get_skill_data_dir(), "logs") - os.makedirs(path, exist_ok=True) - return path - - -def get_skill_log_file_path() -> str: - """主日志文件绝对路径（与 get_skill_logger 写入文件一致）。""" - return _skill_log_file_path() - - -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_logger() -> logging.Logger: - """ - 技能级日志：单文件按日切分（TimedRotatingFileHandler midnight），UTF-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 - - -def get_db_path(): - return os.path.join(get_skill_data_dir(), "account-manager.db") - - -def _runtime_paths_debug_text(): - """供排查「为何 list 没数据」：终端未注入环境变量时会落到 _anon 等默认目录，与网关里用的库不是同一个文件。""" - env_root = (os.getenv("JIANGCHANG_DATA_ROOT") or "").strip() or "(未设置)" - env_uid = (os.getenv("JIANGCHANG_USER_ID") or "").strip() or "(未设置→使用 _anon)" - return ( - "[account-manager] " - f"JIANGCHANG_DATA_ROOT={env_root} | " - f"JIANGCHANG_USER_ID={env_uid} | " - f"实际数据根={get_data_root()} | " - f"实际用户目录={get_user_id()} | " - f"数据库文件={get_db_path()}" - ) - - -def _maybe_print_paths_debug_cli(): - """设置 JIANGCHANG_ACCOUNT_DEBUG_PATHS=1 时，每次执行子命令前在 stderr 打印路径。""" - v = (os.getenv("JIANGCHANG_ACCOUNT_DEBUG_PATHS") or "").strip().lower() - if v in ("1", "true", "yes", "on"): - print(_runtime_paths_debug_text(), file=sys.stderr) - - -_WIN_FS_FORBIDDEN = re.compile(r'[<>:"/\\|?*\x00-\x1f]') -_WIN_RESERVED_NAMES = frozenset( - ["CON", "PRN", "AUX", "NUL"] - + [f"COM{i}" for i in range(1, 10)] - + [f"LPT{i}" for i in range(1, 10)] -) - - -def _fs_safe_segment(name: str, fallback: str) -> str: - """单级目录名：去掉 Windows 非法字符，避免末尾点/空格；空则用 fallback。""" - t = _WIN_FS_FORBIDDEN.sub("_", (name or "").strip()) - t = t.rstrip(" .") - if not t: - t = fallback - if t.upper() in _WIN_RESERVED_NAMES: - t = f"_{t}" - return t - - -def get_default_profile_dir( - platform_key: str, phone: Optional[str], account_id: Optional[int] = None -) -> str: - """ - 默认可读路径：profiles/<平台展示名>/<手机号>/（便于在资源管理器中辨认）。 - 无手机号时退化为 profiles/<平台展示名>/no_phone_/。 - """ - label = _PLATFORM_PRIMARY_CN.get(platform_key, platform_key or "account") - label_seg = _fs_safe_segment(label, _fs_safe_segment(platform_key or "", "platform")) - ph = (phone or "").strip() - if ph: - phone_seg = _fs_safe_segment(ph, f"id_{account_id}" if account_id is not None else "phone") - else: - phone_seg = _fs_safe_segment( - "", f"no_phone_{account_id}" if account_id is not None else "no_phone" - ) - path = os.path.join(get_skill_data_dir(), "profiles", label_seg, phone_seg) - os.makedirs(path, exist_ok=True) - return path - - -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}" - - -# 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 ") - 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)) - - -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 开头、第二位为 3～9（示例 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 _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_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 开头、第二位为 3～9。" - ) - 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 _win_find_exe(candidates): - for p in candidates: - if p and os.path.isfile(p): - return p - return None - - -def resolve_chromium_channel(): - """ - Playwright channel: 'chrome' | 'msedge' | None - Windows 优先 Chrome，其次 Edge；其它系统默认 chrome（由 Playwright 解析 PATH）。 - """ - if sys.platform != "win32": - return "chrome" - - pf = os.environ.get("ProgramFiles", r"C:\Program Files") - pfx86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)") - local = os.environ.get("LocalAppData", "") - - chrome = _win_find_exe( - [ - os.path.join(pf, "Google", "Chrome", "Application", "chrome.exe"), - os.path.join(pfx86, "Google", "Chrome", "Application", "chrome.exe"), - os.path.join(local, "Google", "Chrome", "Application", "chrome.exe") if local else "", - ] - ) - if chrome: - return "chrome" - - edge = _win_find_exe( - [ - os.path.join(pfx86, "Microsoft", "Edge", "Application", "msedge.exe"), - os.path.join(pf, "Microsoft", "Edge", "Application", "msedge.exe"), - os.path.join(local, "Microsoft", "Edge", "Application", "msedge.exe") if local else "", - ] - ) - if edge: - return "msedge" - - return None - - -def _print_browser_install_hint(): - print("❌ 未检测到 Google Chrome 或 Microsoft Edge，请先安装后再登录：") - print(" • Chrome: https://www.google.com/chrome/") - print(" • Edge: https://www.microsoft.com/zh-cn/edge") - - -def _login_timeout_seconds(): - try: - return max(60, int((os.getenv("JIANGCHANG_LOGIN_TIMEOUT_SECONDS") or "300").strip())) - except ValueError: - return 300 - - -def _login_poll_interval_seconds(): - """轮询间隔，默认 1.5s。""" - try: - v = float((os.getenv("JIANGCHANG_LOGIN_POLL_INTERVAL_SECONDS") or "1.5").strip()) - return max(0.5, min(v, 10.0)) - except ValueError: - return 1.5 - - -def _login_dom_grace_seconds() -> float: - """goto 后等待再开始 DOM 判定，减轻跳转中途误判。默认 4s，JIANGCHANG_LOGIN_DOM_GRACE_SECONDS。""" - try: - v = float((os.getenv("JIANGCHANG_LOGIN_DOM_GRACE_SECONDS") or "4").strip()) - return max(0.0, min(v, 60.0)) - except ValueError: - return 4.0 - - -def _login_out_bundle_for(platform_key: str) -> dict: - """传给子进程：anchor + 未登录可见的选择器列表。""" - spec = PLATFORMS.get(platform_key, {}) if platform_key else {} - anchor = (urlparse(PLATFORM_URLS.get(platform_key, "") or "").netloc or "").lower() - spec_lo_dom = [ - str(s).strip() - for s in (spec.get("login_logged_out_selectors") or []) - if s is not None and str(s).strip() - ] - if bool(spec.get("login_skip_generic_logged_out_dom")): - logged_out_dom_selectors = list(spec_lo_dom) - else: - logged_out_dom_selectors = list(_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS) + spec_lo_dom - return { - "logged_out_dom_selectors": logged_out_dom_selectors, - "anchor_host": anchor, - } - - -def cmd_login(account_id): - log = get_skill_logger() - log.info("login_command account_id=%r", account_id) - target = get_account_by_id(account_id) - if not target: - log.warning("login_aborted account_not_found account_id=%r", account_id) - print("ERROR:ACCOUNT_NOT_FOUND") - return - - channel = resolve_chromium_channel() - if not channel: - log.warning("login_aborted no_chromium_channel") - _print_browser_install_hint() - return - - try: - import playwright # noqa: F401 - except ImportError: - log.error("login_aborted playwright_missing") - print("ERROR:需要 playwright：pip install playwright && playwright install chromium") - return - - profile_dir = (target.get("profile_dir") or "").strip() - if not profile_dir: - log.warning("login_aborted profile_dir_empty account_id=%s", target.get("id")) - print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。") - return - os.makedirs(profile_dir, exist_ok=True) - url = (target.get("url") or "").strip() or PLATFORM_URLS.get(target["platform"], "https://www.google.com") - platform = (target.get("platform") or "").strip().lower() - timeout_sec = _login_timeout_seconds() - poll_sec = _login_poll_interval_seconds() - dom_grace_sec = _login_dom_grace_seconds() - - browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge" - print(f"正在为账号 [{target['name']}] 打开 {browser_name} …") - print(f"访问地址：{url}") - print( - "请在窗口内完成登录。系统根据页面是否仍出现「登录」等未登录控件判断；" - "成功后自动关闭窗口并写入状态（无需手动关浏览器）。" - ) - print( - f"页面加载后先等待约 {dom_grace_sec:g} 秒再开始检测（减轻跳转误判，可用 JIANGCHANG_LOGIN_DOM_GRACE_SECONDS 调整）。" - ) - print(f"轮询间隔约 {poll_sec} 秒；全程最长 {timeout_sec} 秒。") - - # 子进程：login_detect_bundle + 多 tab DOM；日志与主进程同一文件（追加）。 - log_file = _skill_log_file_path() - log.info( - "login_browser_start account_id=%s platform=%s channel=%s timeout_sec=%s poll_sec=%s " - "dom_grace_sec=%s profile_dir=%s start_url=%s log_file=%s", - target.get("id"), - platform, - channel, - timeout_sec, - poll_sec, - dom_grace_sec, - profile_dir, - url, - log_file, - ) - _scripts_dir = os.path.dirname(os.path.abspath(__file__)) - login_runner_path = os.path.join(_scripts_dir, "am_login_child_runner.py") - cfg = { - "channel": channel, - "profile_dir": profile_dir, - "url": url, - "timeout_sec": timeout_sec, - "poll_interval": poll_sec, - "dom_grace_sec": dom_grace_sec, - "login_detect_bundle": _login_out_bundle_for(platform), - "log_file": log_file, - } - with tempfile.NamedTemporaryFile( - mode="w", suffix=".json", delete=False, encoding="utf-8" - ) as rf: - result_path = rf.name - cfg["result_path"] = result_path - with tempfile.NamedTemporaryFile( - mode="w", suffix=".json", delete=False, encoding="utf-8" - ) as jf: - json.dump(cfg, jf, ensure_ascii=False) - cfg_path = jf.name - proc_rc = None - try: - r = subprocess.run( - [sys.executable, login_runner_path, cfg_path], - timeout=timeout_sec + 180, - ) - proc_rc = r.returncode - if proc_rc != 0: - print("⚠️ 浏览器进程异常退出，将仅根据已写入的检测结果更新状态") - log.warning("login_subprocess_nonzero_return code=%s", proc_rc) - except subprocess.TimeoutExpired as ex: - print("⚠️ 等待浏览器超时，将仅根据已写入的检测结果更新状态") - log.warning("login_subprocess_timeout err=%s", ex) - finally: - try: - os.unlink(cfg_path) - except OSError: - pass - - interactive_ok = False - try: - with open(result_path, encoding="utf-8") as rf: - interactive_ok = bool(json.load(rf).get("interactive_ok")) - except Exception: - pass - try: - os.unlink(result_path) - except OSError: - pass - - time.sleep(0.5) - ok = interactive_ok - log.info( - "login_finished account_id=%s interactive_ok=%s subprocess_rc=%s marking_db=%s", - target.get("id"), - interactive_ok, - proc_rc, - ok, - ) - _mark_login_status(target["id"], ok) - if ok: - print("✅ 已判定登录成功，状态已写入数据库") - else: - log.warning( - "login_not_detected account_id=%s platform=%s see_log=%s", - target.get("id"), - platform, - log_file, - ) - print("⚠️ 未检测到有效登录，状态为未登录。请关闭其他占用该用户目录的浏览器后重试，或延长 JIANGCHANG_LOGIN_TIMEOUT_SECONDS 后再登录。") - print(f"ℹ️ 详细轮询日志见：{log_file}（可将 JIANGCHANG_LOG_LEVEL=DEBUG 打开更细粒度）") - - -def cmd_open(account_id): - """打开该账号的持久化浏览器，仅用于肉眼确认是否已登录；不写数据库。""" - get_skill_logger().info("open account_id=%r", account_id) - target = get_account_by_id(account_id) - if not target: - get_skill_logger().warning("open_not_found account_id=%r", account_id) - print("ERROR:ACCOUNT_NOT_FOUND") - return - - channel = resolve_chromium_channel() - if not channel: - _print_browser_install_hint() - return - - try: - import playwright # noqa: F401 - except ImportError: - print("ERROR:需要 playwright：pip install playwright && playwright install chromium") - return - - profile_dir = (target.get("profile_dir") or "").strip() - if not profile_dir: - print("ERROR:PROFILE_DIR_MISSING 库中 profile_dir 为空。") - return - os.makedirs(profile_dir, exist_ok=True) - url = (target.get("url") or "").strip() or PLATFORM_URLS.get( - target["platform"], "https://www.google.com" - ) - browser_name = "Google Chrome" if channel == "chrome" else "Microsoft Edge" - print(f"正在打开 [{target['name']}] 的 {browser_name}（仅查看，不写入数据库）") - print(f"地址：{url}") - print("请在本窗口中自行确认登录态。关闭浏览器后命令结束。") - print("需要自动检测并写入数据库时，请执行：python main.py login ") - - _scripts_dir = os.path.dirname(os.path.abspath(__file__)) - open_runner_path = os.path.join(_scripts_dir, "am_open_child_runner.py") - cfg = {"channel": channel, "profile_dir": profile_dir, "url": url} - with tempfile.NamedTemporaryFile( - mode="w", suffix=".json", delete=False, encoding="utf-8" - ) as jf: - json.dump(cfg, jf, ensure_ascii=False) - cfg_path = jf.name - try: - subprocess.run([sys.executable, open_runner_path, cfg_path]) - finally: - try: - os.unlink(cfg_path) - except OSError: - pass - - -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 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") - - -def _cli_print_full_usage() -> None: - """总览（未带子命令或需要完整说明时）。""" - print("用法概览（将 main.py 换为你的路径）：") - print(" python main.py add <平台中文名> <手机号>") - print(" python main.py list [平台|all|全部]") - print(" python main.py pick-logged-in <平台> # 跨技能用：仅已登录账号，一行 JSON 或 ERROR") - print(" python main.py pick-web <平台> # 跨技能用：优先已登录，否则最新一条账号") - print(" python main.py list-json <平台|all> [--limit N] # 跨技能用：一行 JSON 数组，元素同 get") - print(" python main.py get ") - print(" python main.py set-login-status <0|1> # 跨技能回写登录态") - print(" python main.py open ") - print(" python main.py login ") - print(" python main.py delete id ") - print(" python main.py delete platform <平台>") - print(" python main.py delete platform <平台> <手机号>") - print(" python main.py <平台> # 等价于只列该平台") - print(" python main.py <平台> list # 同上") - print() - print("支持的平台（可用下列中文称呼，亦支持英文键如 sohu）：") - print(" " + _platform_list_cn_for_help()) - - -def _cli_fail_add() -> None: - print("ERROR:CLI_ADD_MISSING_ARGS") - print() - print("【add 添加账号】参数不足。") - print(" 必填：平台名称（中文即可，如 搜狐号、知乎、微信公众号）") - print(" 必填：中国大陆 11 位手机号（1 开头，第二位 3～9），同平台下不可重复") - print() - print("示例：") - print(" python main.py add 搜狐号 13800138000") - print(" python main.py add 知乎 13900001111") - print() - print("支持的平台：" + _platform_list_cn_for_help()) - - -def _cli_fail_need_account_id(subcmd: str) -> None: - print(f"ERROR:CLI_{subcmd.upper()}_MISSING_ID") - print() - print(f"【{subcmd}】缺少必填参数：账号 id（数据库自增整数，可先 list 查看）。") - print("示例：") - print(f" python main.py {subcmd} 1") - - -def _cli_fail_delete() -> None: - print("ERROR:CLI_DELETE_MISSING_ARGS") - print() - print("【delete 删除账号】参数不足。支持三种方式：") - print(" 按 id：python main.py delete id ") - print(" 按平台（删该平台下全部）：python main.py delete platform <平台中文名或英文键>") - print(" 按平台+手机号：python main.py delete platform <平台> <手机号>") - print() - print("示例：") - print(" python main.py delete id 3") - print(" python main.py delete platform 搜狐号") - print(" python main.py delete platform 头条号 18925203701") - print() - print("说明：会同时删除数据库记录与 profile 用户数据目录（若存在）。") +from am_cli import main if __name__ == "__main__": - if len(sys.argv) < 2: - _cli_print_full_usage() - sys.exit(1) - - _apply_cli_local_dev_env() - get_skill_logger().info("cli_start argv=%s", sys.argv) - cmd = sys.argv[1] - _maybe_print_paths_debug_cli() - - plat_from_first = resolve_platform_key(cmd) - - if len(sys.argv) >= 3 and sys.argv[2] == "list" and plat_from_first: - cmd_list(plat_from_first) - elif plat_from_first and len(sys.argv) == 2: - cmd_list(plat_from_first) - elif cmd == "add": - if len(sys.argv) < 4: - _cli_fail_add() - sys.exit(1) - cmd_add(sys.argv[2], sys.argv[3]) - elif cmd == "list": - cmd_list(sys.argv[2] if len(sys.argv) >= 3 else "all") - elif cmd == "get": - if len(sys.argv) < 3: - _cli_fail_need_account_id("get") - sys.exit(1) - cmd_get(sys.argv[2]) - elif cmd == "set-login-status": - if len(sys.argv) < 4: - print("ERROR:CLI_SET_LOGIN_STATUS_MISSING_ARGS") - print("用法：python main.py set-login-status <0|1>") - print("说明：供 llm-manager 等脚本回写登录态；成功时 stdout 首行以 OK: 开头。") - sys.exit(1) - cmd_set_login_status(sys.argv[2], sys.argv[3]) - elif cmd == "list-json": - if len(sys.argv) < 3: - print("ERROR:CLI_LIST_JSON_MISSING_PLATFORM") - print("用法：python main.py list-json <平台|all|全部> [--limit N]") - print("说明：跨技能编排；成功时 stdout 仅一行 JSON 数组（元素与 get 一致）。") - sys.exit(1) - av = sys.argv[2:] - platform_arg = av[0] - lim = 200 - if "--limit" in av: - i = av.index("--limit") - if i + 1 < len(av): - try: - lim = int(av[i + 1]) - except ValueError: - pass - cmd_list_json(platform_arg, limit=lim) - elif cmd == "pick-logged-in": - if len(sys.argv) < 3: - print("ERROR:CLI_PICK_LOGGED_IN_MISSING_ARGS") - print("用法：python main.py pick-logged-in <平台中文名或英文键>") - print("说明：供 llm-manager 等技能查询该平台已登录账号；成功时 stdout 仅一行 JSON。") - sys.exit(1) - cmd_pick_logged_in(sys.argv[2]) - elif cmd == "pick-web": - if len(sys.argv) < 3: - print("ERROR:CLI_PICK_WEB_MISSING_ARGS") - print("用法：python main.py pick-web <平台中文名或英文键>") - print("说明：供 llm-manager 等；优先已登录账号，否则返回该平台最新一条账号（未登录也可用于打开浏览器登录）。") - sys.exit(1) - cmd_pick_web(sys.argv[2]) - elif cmd == "open": - if len(sys.argv) < 3: - _cli_fail_need_account_id("open") - sys.exit(1) - cmd_open(sys.argv[2]) - elif cmd == "login": - if len(sys.argv) < 3: - _cli_fail_need_account_id("login") - sys.exit(1) - cmd_login(sys.argv[2]) - elif cmd == "delete": - if len(sys.argv) < 4: - _cli_fail_delete() - sys.exit(1) - mode = (sys.argv[2] or "").strip().lower() - if mode == "id": - cmd_delete_by_id(sys.argv[3]) - elif mode == "platform": - if len(sys.argv) >= 5: - cmd_delete_by_platform_phone(sys.argv[3], sys.argv[4]) - else: - cmd_delete_by_platform(sys.argv[3]) - else: - print(f"ERROR:CLI_DELETE_BAD_MODE 不支持的删除方式「{sys.argv[2]}」，请使用 id 或 platform。") - _cli_fail_delete() - sys.exit(1) - else: - print(f"ERROR:CLI_UNKNOWN_OR_BAD_ARGS 子命令「{cmd}」无法识别，或参数组合不合法。") - print() - _cli_print_full_usage() - print() - print("说明：list 的筛选可为 all/全部，或上面列出的任一平台中文名。") - sys.exit(1) + main()