diff --git a/scripts/__pycache__/account.cpython-312.pyc b/scripts/__pycache__/account.cpython-312.pyc index dbd6524..4aef0a7 100644 Binary files a/scripts/__pycache__/account.cpython-312.pyc and b/scripts/__pycache__/account.cpython-312.pyc differ diff --git a/scripts/account.py b/scripts/account.py index 0e36a16..a979e45 100644 --- a/scripts/account.py +++ b/scripts/account.py @@ -6,6 +6,7 @@ import subprocess import tempfile import time from datetime import datetime +from typing import Optional # Windows GBK 编码兼容修复 if sys.platform == "win32": @@ -15,6 +16,40 @@ if sys.platform == "win32": 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", @@ -75,107 +110,13 @@ def get_conn(): return sqlite3.connect(get_db_path()) -def _ensure_column(cur, table_name, col_name, ddl): - cur.execute(f"PRAGMA table_info({table_name})") - cols = [row[1] for row in cur.fetchall()] - if col_name not in cols: - cur.execute(f"ALTER TABLE {table_name} ADD COLUMN {col_name} {ddl}") - - -def _has_text_primary_id(cur): - cur.execute("PRAGMA table_info(accounts)") - for row in cur.fetchall(): - if row[1] == "id": - t = (row[2] or "").upper() - return "TEXT" in t or "CHAR" in t - return False - - -def _migrate_text_id_to_integer(conn): - cur = conn.cursor() - cur.execute( - """ - CREATE TABLE accounts_new ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL, - platform TEXT NOT NULL, - phone TEXT, - profile_dir TEXT, - url TEXT, - login_status INTEGER DEFAULT 0, - last_login_at TEXT, - extra_json TEXT, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL - ) - """ - ) - cur.execute( - "SELECT name, platform, phone, profile_dir, login_status, last_login_at, extra_json, created_at, updated_at FROM accounts ORDER BY rowid" - ) - rows = cur.fetchall() - for row in rows: - name, platform, phone, profile_dir, login_status, last_login_at, extra_json, created_at, updated_at = row - plat = (platform or "").strip().lower() - url = PLATFORM_URLS.get(plat, "") - cur.execute( - """ - INSERT INTO accounts_new - (name, platform, phone, profile_dir, url, login_status, last_login_at, extra_json, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - name, - plat, - phone or "", - profile_dir or "", - url, - int(login_status or 0), - last_login_at, - extra_json, - created_at, - updated_at, - ), - ) - cur.execute("DROP TABLE accounts") - cur.execute("ALTER TABLE accounts_new RENAME TO accounts") - - def init_db(): conn = get_conn() try: cur = conn.cursor() cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='accounts'") if not cur.fetchone(): - cur.execute( - """ - CREATE TABLE accounts ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL, - platform TEXT NOT NULL, - phone TEXT, - profile_dir TEXT, - url TEXT, - login_status INTEGER DEFAULT 0, - last_login_at TEXT, - extra_json TEXT, - created_at TEXT NOT NULL, - updated_at TEXT NOT NULL - ) - """ - ) - else: - if _has_text_primary_id(cur): - _migrate_text_id_to_integer(conn) - cur = conn.cursor() - _ensure_column(cur, "accounts", "url", "TEXT") - cur.execute( - "SELECT id, platform FROM accounts WHERE url IS NULL OR TRIM(COALESCE(url,'')) = ''" - ) - for rid, plat in cur.fetchall(): - pu = PLATFORM_URLS.get((plat or "").strip().lower(), "") - if pu: - cur.execute("UPDATE accounts SET url = ? WHERE id = ?", (pu, rid)) + cur.executescript(ACCOUNTS_TABLE_SQL) conn.commit() finally: conn.close() @@ -197,7 +138,11 @@ def get_account_by_id(account_id): try: cur = conn.cursor() cur.execute( - "SELECT id, name, platform, phone, profile_dir, url, login_status, last_login_at, extra_json FROM accounts WHERE id = ?", + """ + 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() @@ -211,7 +156,9 @@ def get_account_by_id(account_id): "profile_dir": row[4] or get_default_profile_dir(row[0]), "url": row[5] or PLATFORM_URLS.get(row[2], ""), "login_status": int(row[6] or 0), - "last_login_at": row[7], + "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: @@ -279,7 +226,7 @@ def cmd_add(platform: str, phone: str = ""): phone = (phone or "").strip() url = PLATFORM_URLS[platform] - now = datetime.now().isoformat(timespec="seconds") + now = _now_unix() conn = get_conn() try: @@ -298,7 +245,7 @@ def cmd_add(platform: str, phone: str = ""): profile_dir = get_default_profile_dir(new_id) cur.execute( "UPDATE accounts SET profile_dir = ?, updated_at = ? WHERE id = ?", - (profile_dir, now, new_id), + (profile_dir, _now_unix(), new_id), ) conn.commit() finally: @@ -363,6 +310,32 @@ def _login_timeout_seconds(): 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 _url_looks_logged_in(platform: str, page_url: str) -> bool: + """根据当前 URL 判断是否已进入后台（与交互窗口、无头校验共用逻辑）。""" + u = (page_url or "").lower() + if not u: + return False + if platform == "sohu": + if "passport" in u or "/login" in u: + return False + # 登录成功后常见：…/mpfe/v4/contentManagement/… + if "mpfe" in u or "contentmanagement" in u: + return True + return False + if "login" in u or "signin" in u or "passport" in u: + return False + return True + + def _verify_login(profile_dir: str, platform: str, start_url: str, channel: str) -> bool: try: from playwright.sync_api import sync_playwright @@ -381,15 +354,12 @@ def _verify_login(profile_dir: str, platform: str, start_url: str, channel: str) try: page = ctx.pages[0] if ctx.pages else ctx.new_page() page.goto(start_url, wait_until="domcontentloaded", timeout=45000) - time.sleep(2) - u = (page.url or "").lower() - if platform == "sohu": - if "passport" in u or "/login" in u: - return False - return True - if "login" in u or "signin" in u or "passport" in u: - return False - return True + # 无头启动后可能仍在跳转，短轮询几次再判定 + for _ in range(20): + if _url_looks_logged_in(platform, page.url): + return True + time.sleep(1.0) + return _url_looks_logged_in(platform, page.url) finally: ctx.close() except Exception: @@ -418,18 +388,35 @@ def cmd_login(account_id): 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"最长等待：{timeout_sec} 秒") + print( + "请在窗口内完成登录。系统会轮询地址栏，进入后台后将自动关闭窗口并写入状态（无需手动关浏览器）。" + ) + print(f"轮询间隔约 {poll_sec} 秒；全程最长 {timeout_sec} 秒。") - # Playwright 同步 API 宜在独立子进程中跑浏览器，避免线程与事件循环问题 - runner = r"""import json, sys + # 子进程：轮询 URL，已登录则立即关窗；勿仅用 wait_for_event(close)，否则用户不关窗口会卡满整段超时。 + runner = r"""import json, sys, time from playwright.sync_api import sync_playwright + +def looks_sohu(u): + u = (u or "").lower() + if not u: + return False + if "passport" in u or "/login" in u: + return False + if "mpfe" in u or "contentmanagement" in u: + return True + 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"], @@ -441,23 +428,36 @@ with sync_playwright() as p: try: page = ctx.pages[0] if ctx.pages else ctx.new_page() page.goto(c["url"], wait_until="domcontentloaded", timeout=60000) - try: - ctx.wait_for_event("close", timeout=c["timeout_sec"] * 1000) - except Exception: - pass + if platform == "sohu": + while time.time() < deadline: + try: + if looks_sohu(page.url): + 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 - except Exception as e: - print(e, file=sys.stderr) - raise """ 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" @@ -499,7 +499,7 @@ with sync_playwright() as p: def _mark_login_status(account_id, success: bool): - now = datetime.now().isoformat(timespec="seconds") + now = _now_unix() conn = get_conn() try: cur = conn.cursor()