client-commons/account-manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: auto release commit (2026-04-06 12:10:31)
All checks were successful
技能自动化发布 / release (push) Successful in 13s
Details

Browse Source
This commit is contained in:
chendelian
2026-04-06 12:10:31 +08:00
parent c428bedfd5
commit 39daeaca59
21 changed files with 728 additions and 1420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

721
scripts/account.py
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:需要 playwright：pip 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:需要 playwright：pip 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)

285
scripts/am_commands_read.py
View File

@@ -1,285 +0,0 @@
"""只读类 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 <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))

246
scripts/am_commands_write.py
View File

@@ -1,246 +0,0 @@
"""写入类 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")

136
scripts/am_db.py
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()

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

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

22
scripts/am_cli.py → scripts/cli/app.py
View File

@@ -1,25 +1,23 @@
"""CLI 入口：用法说明与 argv 分发。"""
import sys
from am_browser import cmd_login, cmd_open
from am_commands_read import (
from service.account_service import (
cmd_add,
cmd_delete_by_id,
cmd_delete_by_platform,
cmd_delete_by_platform_phone,
cmd_get,
cmd_list,
cmd_list_json,
cmd_pick_logged_in,
cmd_pick_web,
)
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
from service.browser_service import cmd_login, cmd_open
from util.constants import _apply_cli_local_dev_env
from util.logging_config import get_skill_logger
from util.platforms import _platform_list_cn_for_help, resolve_platform_key
from util.runtime_paths import _maybe_print_paths_debug_cli
def _cli_print_full_usage() -> None:

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

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

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

@@ -0,0 +1,269 @@
"""Accounts table: CRUD and queries (no stdout)."""
import json
import time
from datetime import datetime
from typing import Any, Optional
from db.connection import get_conn, init_db
from util.platforms import PLATFORM_URLS, _PLATFORM_NAME_CN
def _now_unix() -> int:
return int(time.time())
def _unix_to_iso_local(ts: Optional[int]) -> Optional[str]:
"""对外 JSON：本地时区 ISO8601 字符串，便于人读。"""
if ts is None:
return None
try:
return datetime.fromtimestamp(int(ts)).isoformat(timespec="seconds")
except (ValueError, OSError, OverflowError):
return None
def normalize_account_id(account_id):
if account_id is None:
return None
s = str(account_id).strip()
if s.isdigit():
return int(s)
return s
def default_name_for_platform(platform: str, index: int) -> str:
cn = _PLATFORM_NAME_CN.get(platform)
if cn:
return f"{cn}{index}号"
return f"{platform}_{index}"
def get_account_by_id(account_id):
init_db()
aid = normalize_account_id(account_id)
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT id, name, platform, phone, profile_dir, url, login_status, last_login_at, extra_json,
created_at, updated_at
FROM accounts WHERE id = ?
""",
(aid,),
)
row = cur.fetchone()
if not row:
return None
acc: dict[str, Any] = {
"id": row[0],
"name": row[1],
"platform": row[2],
"phone": row[3] or "",
"profile_dir": (row[4] or "").strip(),
"url": row[5] or PLATFORM_URLS.get(row[2], ""),
"login_status": int(row[6] or 0),
"last_login_at": _unix_to_iso_local(row[7]),
"created_at": _unix_to_iso_local(row[9]),
"updated_at": _unix_to_iso_local(row[10]),
}
if row[8]:
try:
extra = json.loads(row[8])
if isinstance(extra, dict):
acc.update(extra)
except Exception:
pass
return acc
finally:
conn.close()
def mark_login_status(account_id, success: bool):
now = _now_unix()
conn = get_conn()
try:
cur = conn.cursor()
if success:
cur.execute(
"UPDATE accounts SET login_status = 1, last_login_at = ?, updated_at = ? WHERE id = ?",
(now, now, account_id),
)
else:
cur.execute(
"UPDATE accounts SET login_status = 0, updated_at = ? WHERE id = ?",
(now, account_id),
)
conn.commit()
finally:
conn.close()
def fetch_list_rows(platform_key: str, limit: int):
"""platform_key 'all' or platform slug; returns list of row tuples (full row)."""
conn = get_conn()
try:
cur = conn.cursor()
sql = (
"SELECT id, name, platform, phone, profile_dir, url, login_status, "
"last_login_at, extra_json, created_at, updated_at FROM accounts "
)
if platform_key == "all":
cur.execute(sql + "ORDER BY created_at DESC, id DESC LIMIT ?", (int(limit),))
else:
cur.execute(
sql + "WHERE platform = ? ORDER BY created_at DESC, id DESC LIMIT ?",
(platform_key, int(limit)),
)
return cur.fetchall()
finally:
conn.close()
def fetch_ids_for_list_json(platform_key: str, lim: int):
conn = get_conn()
try:
cur = conn.cursor()
if platform_key == "all":
cur.execute(
"SELECT id FROM accounts ORDER BY updated_at DESC, id DESC LIMIT ?",
(lim,),
)
else:
cur.execute(
"SELECT id FROM accounts WHERE platform = ? "
"ORDER BY updated_at DESC, id DESC LIMIT ?",
(platform_key, lim),
)
return [r[0] for r in cur.fetchall()]
finally:
conn.close()
def fetch_pick_logged_in_id(platform_key: str) -> Optional[int]:
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ? AND login_status = 1
ORDER BY (last_login_at IS NULL), last_login_at DESC
LIMIT 1
""",
(platform_key,),
)
row = cur.fetchone()
return int(row[0]) if row else None
finally:
conn.close()
def fetch_pick_web_candidate_id(platform_key: str) -> Optional[int]:
conn = get_conn()
try:
cur = conn.cursor()
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ? AND login_status = 1
ORDER BY (last_login_at IS NULL), last_login_at DESC
LIMIT 1
""",
(platform_key,),
)
row = cur.fetchone()
if not row:
cur.execute(
"""
SELECT id FROM accounts
WHERE platform = ?
ORDER BY updated_at DESC, id DESC
LIMIT 1
""",
(platform_key,),
)
row = cur.fetchone()
return int(row[0]) if row else None
finally:
conn.close()
def has_duplicate_phone_conn(conn, platform_key: str, phone_store: str) -> bool:
cur = conn.cursor()
cur.execute(
"SELECT id FROM accounts WHERE platform = ? AND phone = ?",
(platform_key, phone_store),
)
return cur.fetchone() is not None
def count_platform_accounts_conn(conn, platform_key: str) -> int:
cur = conn.cursor()
cur.execute("SELECT COUNT(*) FROM accounts WHERE platform = ?", (platform_key,))
return int(cur.fetchone()[0])
def insert_account_row(
conn,
name: str,
platform_key: str,
phone_store: str,
profile_dir: str,
url: str,
now: int,
) -> int:
cur = conn.cursor()
cur.execute(
"""
INSERT INTO accounts (name, platform, phone, profile_dir, url, login_status, last_login_at, extra_json, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, 0, NULL, NULL, ?, ?)
""",
(name, platform_key, phone_store, profile_dir, url, now, now),
)
return int(cur.lastrowid)
def fetch_delete_row_by_id_conn(conn, aid: int):
cur = conn.cursor()
cur.execute(
"SELECT id, name, platform, phone, profile_dir FROM accounts WHERE id = ?",
(aid,),
)
return cur.fetchone()
def delete_account_by_id_conn(conn, rid: int) -> None:
cur = conn.cursor()
cur.execute("DELETE FROM accounts WHERE id = ?", (rid,))
def fetch_ids_profile_for_platform_conn(conn, platform_key: str):
cur = conn.cursor()
cur.execute(
"SELECT id, profile_dir FROM accounts WHERE platform = ?",
(platform_key,),
)
return cur.fetchall()
def delete_accounts_by_platform_conn(conn, platform_key: str) -> None:
cur = conn.cursor()
cur.execute("DELETE FROM accounts WHERE platform = ?", (platform_key,))
def fetch_row_platform_phone_conn(conn, platform_key: str, phone_norm: str):
cur = conn.cursor()
cur.execute(
"SELECT id, name, profile_dir FROM accounts WHERE platform = ? AND phone = ?",
(platform_key, phone_norm),
)
return cur.fetchone()
def delete_account_platform_phone_conn(conn, platform_key: str, phone_norm: str) -> None:
cur = conn.cursor()
cur.execute(
"DELETE FROM accounts WHERE platform = ? AND phone = ?",
(platform_key, phone_norm),
)

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

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

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

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

4
scripts/main.py
View File

@@ -1,7 +1,7 @@
"""
account-manager CLI 入口。
业务逻辑按模块拆分（am_*.py），便于维护与满足 PyArmor trial 对单文件体积的限制。
分层：cli（argv）→ service（用例）→ db（SQLite）；共用工具在 util/。
"""
import sys
@@ -12,7 +12,7 @@ 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")
from am_cli import main
from cli.app import main
if __name__ == "__main__":
main()

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

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

380
scripts/service/account_service.py Normal file
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 开头、第二位为 3～9（示例 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 开头、第二位为 3～9。"
)
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")

18
scripts/am_browser.py → scripts/service/browser_service.py
View File

@@ -7,14 +7,17 @@ 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 (
from db.accounts_repo import get_account_by_id, mark_login_status
from util.logging_config import get_skill_logger, get_skill_log_file_path
from util.platforms import (
PLATFORMS,
PLATFORM_URLS,
_LOGIN_LOGGED_OUT_DOM_GENERIC_SELECTORS,
)
# 子进程入口脚本与本模块同目录（独立解释器执行，非 import）
_SERVICE_DIR = os.path.dirname(os.path.abspath(__file__))
def _win_find_exe(candidates):
for p in candidates:
@@ -154,7 +157,6 @@ def cmd_login(account_id):
)
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 "
@@ -169,8 +171,7 @@ def cmd_login(account_id):
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")
login_runner_path = os.path.join(_SERVICE_DIR, "login_child_runner.py")
cfg = {
"channel": channel,
"profile_dir": profile_dir,
@@ -230,7 +231,7 @@ def cmd_login(account_id):
proc_rc,
ok,
)
_mark_login_status(target["id"], ok)
mark_login_status(target["id"], ok)
if ok:
print("✅ 已判定登录成功，状态已写入数据库")
else:
@@ -278,8 +279,7 @@ def cmd_open(account_id):
print("请在本窗口中自行确认登录态。关闭浏览器后命令结束。")
print("需要自动检测并写入数据库时，请执行：python main.py login <id>")
_scripts_dir = os.path.dirname(os.path.abspath(__file__))
open_runner_path = os.path.join(_scripts_dir, "am_open_child_runner.py")
open_runner_path = os.path.join(_SERVICE_DIR, "open_child_runner.py")
cfg = {"channel": channel, "profile_dir": profile_dir, "url": url}
with tempfile.NamedTemporaryFile(
mode="w", suffix=".json", delete=False, encoding="utf-8"

2
scripts/am_login_child_runner.py → scripts/service/login_child_runner.py
View File

@@ -1,4 +1,4 @@
"""Playwright 登录检测子进程：由 main.cmd_login 以独立解释器启动，argv[1] 为 JSON 配置路径。"""
"""Playwright 登录检测子进程：由 browser_service.cmd_login 以独立解释器启动，argv[1] 为 JSON 配置路径。"""
import json
import logging
import sys

2
scripts/am_open_child_runner.py → scripts/service/open_child_runner.py
View File

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

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

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

8
scripts/am_constants.py → scripts/util/constants.py
View File

@@ -1,14 +1,16 @@
"""技能级常量与本地 CLI 调试环境注入（仅 main 入口显式调用 _apply_cli_local_dev_env 时生效）。"""
"""技能级常量与本地 CLI 调试环境注入（仅 CLI 入口显式调用 _apply_cli_local_dev_env 时生效）。"""
import os
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# scripts/util/constants.py -> skill root = parents[2]
_BASE = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
BASE_DIR = _BASE
SKILL_SLUG = "account-manager"
# 与其它 OpenClaw 技能对齐：openclaw.skill.<slug_下划线>
LOG_LOGGER_NAME = "openclaw.skill.account_manager"
# ---------------------------------------------------------------------------
# 本地 CLI 调试：直接运行 python main.py 时，若未设置 JIANGCHANG_*，可自动注入下面默认值。
# - 仅当 __main__ 里调用了 _apply_cli_local_dev_env() 时生效；被其它脚本 import 不会改 os.environ。
# - 仅当 CLI 里调用了 _apply_cli_local_dev_env() 时生效；被其它脚本 import 不会改 os.environ。
# - 不会覆盖宿主/终端已设置的 JIANGCHANG_DATA_ROOT、JIANGCHANG_USER_ID。
# - 开启方式（二选一）：
# 1) 将 _ACCOUNT_MANAGER_CLI_LOCAL_DEV 改为 True；

4
scripts/am_logging.py → scripts/util/logging_config.py
View File

@@ -4,8 +4,8 @@ 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
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
from util.runtime_paths import get_skill_logs_dir
def _skill_log_file_path() -> str:

0
scripts/am_platforms.py → scripts/util/platforms.py
View File

4
scripts/am_runtime_paths.py → scripts/util/runtime_paths.py
View File

@@ -4,8 +4,8 @@ import re
import sys
from typing import Optional
from am_constants import SKILL_SLUG
from am_platforms import _PLATFORM_PRIMARY_CN
from util.constants import SKILL_SLUG
from util.platforms import _PLATFORM_PRIMARY_CN
def get_data_root():