chore: auto release commit (2026-04-06 10:37:07)

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 <id>")
+
+    _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