From 01d9c82f800d7ad0d645ff1069a3e7277d9754d6 Mon Sep 17 00:00:00 2001 From: chendelian <116870791@qq.com> Date: Tue, 31 Mar 2026 11:50:56 +0800 Subject: [PATCH] chore: auto release commit (2026-03-31 11:50:55) --- SKILL.md | 3 - scripts/account.py | 210 ++++++++++++++++++++++++++++++++++++++------- 2 files changed, 180 insertions(+), 33 deletions(-) diff --git a/SKILL.md b/SKILL.md index 8c42496..9e5ad1b 100644 --- a/SKILL.md +++ b/SKILL.md @@ -8,9 +8,6 @@ metadata: slug: account-manager emoji: "👤" category: "通用" - skill_type: 1 - monthly_price: 0 - yearly_price: 0 allowed-tools: - bash --- diff --git a/scripts/account.py b/scripts/account.py index 17a79de..4b68772 100644 --- a/scripts/account.py +++ b/scripts/account.py @@ -1,7 +1,9 @@ import sys import json import os +import sqlite3 import subprocess +from datetime import datetime # Windows GBK 编码兼容修复 if sys.platform == "win32": @@ -9,7 +11,8 @@ if sys.platform == "win32": sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) -ACCOUNTS_FILE = os.path.join(BASE_DIR, "accounts.json") +SKILL_SLUG = "account-manager" +LEGACY_ACCOUNTS_FILE = os.path.join(BASE_DIR, "accounts.json") PLATFORM_URLS = { # 自媒体/图文平台 @@ -26,50 +29,197 @@ PLATFORM_URLS = { "yuanbao": "https://yuanbao.tencent.com" } -def load_accounts(): - with open(ACCOUNTS_FILE, "r", encoding="utf-8") as f: - return json.load(f) + +def get_data_root(): + return (os.getenv("JIANGCHANG_DATA_ROOT") or os.path.join(os.path.expanduser("~"), ".jiangchang-data")).strip() + + +def get_user_id(): + uid = (os.getenv("JIANGCHANG_USER_ID") or os.getenv("OPENCLAW_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", 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( + """ + CREATE TABLE IF NOT EXISTS accounts ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + platform TEXT NOT NULL, + phone TEXT, + profile_dir TEXT, + extra_json TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """ + ) + conn.commit() + finally: + conn.close() + + +def migrate_legacy_json_if_needed(): + init_db() + conn = get_conn() + try: + cur = conn.cursor() + cur.execute("SELECT COUNT(1) FROM accounts") + count = cur.fetchone()[0] + if count > 0: + return + finally: + conn.close() + + if not os.path.exists(LEGACY_ACCOUNTS_FILE): + return + + try: + with open(LEGACY_ACCOUNTS_FILE, "r", encoding="utf-8") as f: + raw = json.load(f) + except Exception: + return + + now = datetime.now().isoformat(timespec="seconds") + rows = [] + for platform, accounts in (raw or {}).items(): + for acc in accounts: + acc_id = str(acc.get("id") or "").strip() + if not acc_id: + continue + profile_dir = (acc.get("profile_dir") or "").strip() or get_default_profile_dir(acc_id) + base = { + "id": acc_id, + "name": str(acc.get("name") or acc_id), + "platform": str(acc.get("platform") or platform), + "phone": str(acc.get("phone") or ""), + "profile_dir": profile_dir, + } + extra = {k: v for k, v in acc.items() if k not in base} + rows.append( + ( + base["id"], + base["name"], + base["platform"], + base["phone"], + base["profile_dir"], + json.dumps(extra, ensure_ascii=False) if extra else None, + now, + now, + ) + ) + + if not rows: + return + + conn = get_conn() + try: + cur = conn.cursor() + cur.executemany( + """ + INSERT OR REPLACE INTO accounts + (id, name, platform, phone, profile_dir, extra_json, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + rows, + ) + conn.commit() + finally: + conn.close() + +def get_account_by_id(account_id): + migrate_legacy_json_if_needed() + conn = get_conn() + try: + cur = conn.cursor() + cur.execute( + "SELECT id, name, platform, phone, profile_dir, extra_json FROM accounts WHERE id = ?", + (account_id,), + ) + 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]), + } + if row[5]: + try: + extra = json.loads(row[5]) + if isinstance(extra, dict): + acc.update(extra) + except Exception: + pass + return acc + finally: + conn.close() def cmd_list(platform="all"): """ 修改后的 list 功能，能直观打出带着【手机号】的详细台账供AI查询匹配！ 支持查全部(all)，或指定(sohu) """ - accounts = load_accounts() - if platform == "all": - platforms_to_check = accounts.keys() - else: - platforms_to_check = [platform] - + migrate_legacy_json_if_needed() + conn = get_conn() found = False - for plat in platforms_to_check: - platform_accounts = accounts.get(plat, []) - for acc in platform_accounts: - phone = acc.get('phone', '未绑定手机') - print(f"账号ID:{acc['id']} | 名称:{acc['name']} | 手机号:{phone} | 平台:{acc['platform']}") - found = True - + try: + cur = conn.cursor() + if platform == "all": + cur.execute("SELECT id, name, phone, platform FROM accounts ORDER BY platform, id") + else: + cur.execute( + "SELECT id, name, phone, platform FROM accounts WHERE platform = ? ORDER BY id", + (platform,), + ) + rows = cur.fetchall() + finally: + conn.close() + + found = False + for row in rows: + phone = row[2] or "未绑定手机" + print(f"账号ID:{row[0]} | 名称:{row[1]} | 手机号:{phone} | 平台:{row[3]}") + found = True + if not found: print(f"ERROR:NO_ACCOUNTS_FOUND") def cmd_get(account_id): - accounts = load_accounts() - for platform_accounts in accounts.values(): - for acc in platform_accounts: - if acc["id"] == account_id: - print(json.dumps(acc, ensure_ascii=False)) - return + acc = get_account_by_id(account_id) + if acc: + print(json.dumps(acc, ensure_ascii=False)) + return print("ERROR:ACCOUNT_NOT_FOUND") def cmd_login(account_id): """首次登录：打开浏览器，用户手动登录，登录态自动保存到Profile目录""" - accounts = load_accounts() - target = None - for platform_accounts in accounts.values(): - for acc in platform_accounts: - if acc["id"] == account_id: - target = acc - break + target = get_account_by_id(account_id) if not target: print("ERROR:ACCOUNT_NOT_FOUND")