diff --git a/SKILL.md b/SKILL.md index a3225d7..c821d85 100644 --- a/SKILL.md +++ b/SKILL.md @@ -1,7 +1,7 @@ --- name: 账号管理 description: 多平台多账号管理。管理各平台账号与Chrome Profile的对应关系，供publisher类Skill调用获取账号信息。 -version: 1.0.2 +version: 1.0.3 author: 深圳匠厂科技有限公司 metadata: openclaw: @@ -29,6 +29,16 @@ python3 {baseDir}/scripts/account.py add sohu python3 {baseDir}/scripts/account.py add sohu 189xxxxxxxx ``` +### 仅打开浏览器核对是否已登录（不写数据库） +```bash +python3 {baseDir}/scripts/account.py open +``` + +### 登录并自动检测、写入数据库 +```bash +python3 {baseDir}/scripts/account.py login +``` + ### 列出某平台所有账号 ```bash python3 {baseDir}/scripts/account.py list \ No newline at end of file diff --git a/scripts/__pycache__/account.cpython-312.pyc b/scripts/__pycache__/account.cpython-312.pyc index 4aef0a7..2acca90 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 a979e45..d4d9282 100644 --- a/scripts/account.py +++ b/scripts/account.py @@ -1,7 +1,9 @@ import sys import json import os +import re import sqlite3 +from urllib.parse import urlparse import subprocess import tempfile import time @@ -319,23 +321,95 @@ def _login_poll_interval_seconds(): 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: - """根据当前 URL 判断是否已进入后台（与交互窗口、无头校验共用逻辑）。""" - u = (page_url or "").lower() - if not u: + """是否判定为已登录：搜狐用白名单+负例；其它平台仅用「非明确登录页」宽松策略。""" + if not page_url or not page_url.strip(): 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 _sohu_url_logged_in(page_url) + if _definitely_logged_out_url(page_url): return False - if "login" in u or "signin" in u or "passport" in u: + 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 @@ -354,12 +428,11 @@ 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) - # 无头启动后可能仍在跳转，短轮询几次再判定 for _ in range(20): - if _url_looks_logged_in(platform, page.url): + if _verify_logged_in_page(platform, page): return True time.sleep(1.0) - return _url_looks_logged_in(platform, page.url) + return _verify_logged_in_page(platform, page) finally: ctx.close() except Exception: @@ -398,20 +471,43 @@ def cmd_login(account_id): ) print(f"轮询间隔约 {poll_sec} 秒；全程最长 {timeout_sec} 秒。") - # 子进程：轮询 URL，已登录则立即关窗；勿仅用 wait_for_event(close)，否则用户不关窗口会卡满整段超时。 + # 子进程：搜狐轮询 URL+页面；勿用「含 mpfe 即登录」，/mpfe/v4/login 会误判。 runner = r"""import json, sys, time +from urllib.parse import urlparse from playwright.sync_api import sync_playwright -def looks_sohu(u): - u = (u or "").lower() +def sohu_logged_in(url): + u = (url or "").strip() if not u: return False - if "passport" in u or "/login" in u: + try: + p = urlparse(u) + except Exception: return False - if "mpfe" in u or "contentmanagement" in u: + 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() @@ -431,7 +527,7 @@ with sync_playwright() as p: if platform == "sohu": while time.time() < deadline: try: - if looks_sohu(page.url): + if sohu_logged_in(page.url) and not sohu_not_login_page(page): break except Exception: break @@ -498,6 +594,84 @@ with sync_playwright() as p: 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 ") + + 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() @@ -521,7 +695,7 @@ def _mark_login_status(account_id, success: bool): if __name__ == "__main__": if len(sys.argv) < 2: print( - "用法：python account.py add [phone] | list [platform] | get | login " + "用法：python account.py add [phone] | list [platform] | get | open | login " ) sys.exit(1) @@ -538,6 +712,8 @@ if __name__ == "__main__": 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: