"""Playwright 登录检测子进程：由 browser_service 以独立解释器启动，argv[1] 为 JSON 配置路径。 注意：不得在模块顶层 import playwright / jiangchang_skill_core —— 若导入失败，进程会在 main() 之前以退出码 1 退出， 且写不了 result_path，父进程只能得到 subprocess_error。所有重依赖均在 _run_login_child 内延迟加载。 """ from __future__ import annotations import json import os import sys import time import traceback from urllib.parse import urlparse _scripts_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _scripts_dir not in sys.path: sys.path.insert(0, _scripts_dir) from util.playwright_stealth import ( STEALTH_INIT_SCRIPT, persistent_context_launch_parts, stealth_enabled, ) def _is_context_closed_error(ex: BaseException) -> bool: s = str(ex).lower() if "target closed" in s or "browser has been closed" in s or "context was destroyed" in s: return True if "closed" in s or "destroyed" in s: return True return False def _write_result_file( result_path: str, interactive_ok: bool, end_reason: str, detail: str = "", ) -> None: if not (result_path or "").strip(): return payload: dict = {"interactive_ok": interactive_ok, "end_reason": end_reason} d = (detail or "").strip() if d: payload["detail"] = d[:8000] try: with open(result_path, "w", encoding="utf-8") as rf: json.dump(payload, rf, ensure_ascii=False) except Exception as write_ex: print(f"login_child: failed to write result file: {write_ex}", file=sys.stderr) def page_location_href(p): try: href = p.evaluate("() => location.href") if href and str(href).strip(): return str(href).strip() except Exception: pass try: return (p.url or "").strip() except Exception: return "" def host_matches_anchor(netloc, anchor): h = (netloc or "").lower().strip() a = (anchor or "").lower().strip() if not a: return True if not h: return False return h == a or h.endswith("." + a) or a.endswith("." + h) def loc_first_visible(pg, selectors): for sel in selectors: s = str(sel).strip() if not s: continue try: loc = pg.locator(s).first if loc.is_visible(timeout=500): return True, s except Exception: pass return False, None def list_pages_on_anchor(ctx, main_page, bundle): anchor = (bundle.get("anchor_host") or "").strip().lower() seen_ids = set() pages = [] for pg in list(ctx.pages or []): try: i = id(pg) if i not in seen_ids: seen_ids.add(i) pages.append(pg) except Exception: pages.append(pg) if main_page is not None: try: i = id(main_page) if i not in seen_ids: seen_ids.add(i) pages.append(main_page) except Exception: pages.append(main_page) out = [] for pg in pages: try: href = page_location_href(pg) except Exception: continue if not href or str(href).lower().startswith("about:"): continue try: host = urlparse(href).netloc.lower() except Exception: continue if anchor and not host_matches_anchor(host, anchor): continue out.append(pg) return out def evaluate_logged_out_dom(ctx, main_page, bundle): selectors = [ str(x).strip() for x in (bundle.get("logged_out_dom_selectors") or []) if x is not None and str(x).strip() ] if not selectors: return True, "no_logged_out_dom_selectors_configured" for pg in list_pages_on_anchor(ctx, main_page, bundle): try: href = page_location_href(pg) except Exception: href = "" hit, which = loc_first_visible(pg, selectors) if hit: return True, "logged_out_dom=%r url=%r" % (which, href) return False, "" def _run_login_child(c: dict) -> None: """单次登录检测；结果一律写入 c['result_path']（由 finally 保证）。""" result_path = (c.get("result_path") or "").strip() try: from playwright.sync_api import sync_playwright except ImportError: _write_result_file( result_path, False, "playwright_missing", traceback.format_exc(), ) return interactive_ok = False end_reason = "timeout" ctx = None lg = None last_eval_detail = "" detail_msg = "" try: try: from jiangchang_skill_core.unified_logging import attach_unified_file_handler except Exception as e: end_reason = "logging_import_failed" detail_msg = traceback.format_exc() print(detail_msg, file=sys.stderr) return try: profile_dir = (c.get("profile_dir") or "").strip() if not profile_dir: end_reason = "profile_missing" detail_msg = "profile_dir empty in config" return os.makedirs(profile_dir, exist_ok=True) try: n_entries = len(os.listdir(profile_dir)) except OSError as e: end_reason = "profile_dir_unusable" detail_msg = f"{type(e).__name__}: {e}" return except Exception as e: end_reason = "profile_dir_unusable" detail_msg = traceback.format_exc() print(detail_msg, file=sys.stderr) return log_path = (c.get("log_file") or "").strip() if log_path: try: lg = attach_unified_file_handler( log_path, skill_slug="account-manager", logger_name="openclaw.skill.account_manager.login_child", ) except Exception as e: lg = None print(f"login_child: attach_unified_file_handler failed: {e}", file=sys.stderr) if lg is not None: lg.info( "login_child_start profile_dir=%r profile_entry_count=%s channel=%r url=%r timeout_sec=%r", profile_dir, n_entries, c.get("channel"), c.get("url"), c.get("timeout_sec"), ) bundle = c.get("login_detect_bundle") or {} poll = float(c.get("poll_interval", 1.5)) try: dom_grace = float(c.get("dom_grace_sec", 4.0)) except (TypeError, ValueError): dom_grace = 4.0 if dom_grace < 0: dom_grace = 0.0 t0 = time.time() deadline = t0 + float(c["timeout_sec"]) with sync_playwright() as p: try: args, ignore_automation = persistent_context_launch_parts() lc_kw = dict( user_data_dir=profile_dir, headless=False, channel=c["channel"], no_viewport=True, args=args, ) if ignore_automation is not None: lc_kw["ignore_default_args"] = ignore_automation ctx = p.chromium.launch_persistent_context(**lc_kw) if stealth_enabled(): ctx.add_init_script(STEALTH_INIT_SCRIPT) except Exception as e: if _is_context_closed_error(e): end_reason = "user_closed" else: end_reason = "launch_failed" detail_msg = f"{type(e).__name__}: {e}" if lg is not None: lg.exception("launch_persistent_context failed: %s", e) print(detail_msg, file=sys.stderr) return try: page = ctx.pages[0] if ctx.pages else ctx.new_page() page.goto(c["url"], wait_until="domcontentloaded", timeout=60000) if dom_grace > 0: time.sleep(dom_grace) if lg is not None: lg.info( "goto_done initial_pages=%s poll_interval_sec=%s dom_grace_sec=%s", len(ctx.pages or []), poll, dom_grace, ) while time.time() < deadline: iter_start = time.time() try: still_logged_out, dom_detail = evaluate_logged_out_dom(ctx, page, bundle) ok = not still_logged_out last_eval_detail = dom_detail if lg is not None: parts = [] for tab in list(ctx.pages or []): try: href = page_location_href(tab) pu = (tab.url or "").strip() parts.append("href=%r playwright_url=%r" % (href, pu)) except Exception: parts.append("(tab_read_error)") lg.debug( "poll elapsed=%.1fs left=%.1fs tabs=%s logged_out=%s ok=%s detail=%s | %s", iter_start - t0, deadline - iter_start, len(ctx.pages or []), still_logged_out, ok, dom_detail, " ; ".join(parts) if parts else "no_tabs", ) if ok: interactive_ok = True end_reason = "login_ok" if lg is not None: lg.info("login_detected_ok %s", dom_detail) break except Exception as ex: if _is_context_closed_error(ex): end_reason = "user_closed" if lg is not None: lg.warning("login_poll_browser_closed err=%s", ex) break if lg is not None: lg.warning("poll_exception err=%s", ex, exc_info=True) spent = time.time() - iter_start time.sleep(max(0.0, poll - spent)) if not interactive_ok and end_reason != "user_closed": if lg is not None: lg.warning("login_poll_exhausted detail=%s", last_eval_detail) rem = max(0.0, deadline - time.time()) if rem > 0: try: ctx.wait_for_event("close", timeout=rem * 1000) except Exception: pass elif interactive_ok: if lg is not None: lg.info("login_dom_ok_preparing_to_release_profile_dir") print( "INFO:LOGIN_DOM_OK 页面检测已判定登录成功；约 2 秒后关闭本窗口以释放用户目录，供后续大模型网页会话使用。", flush=True, ) time.sleep(2.0) except Exception as e: interactive_ok = False if _is_context_closed_error(e): end_reason = "user_closed" if lg is not None: lg.warning("login_runner_browser_closed err=%s", e) else: end_reason = "page_error" detail_msg = traceback.format_exc() if lg is not None: lg.exception("login_runner_page_err %s", e) print(detail_msg, file=sys.stderr) finally: try: ctx.close() except Exception: pass ctx = None except Exception as e: interactive_ok = False end_reason = "child_fatal" detail_msg = traceback.format_exc() if lg is not None: lg.exception("login_child_unhandled %s", e) print(detail_msg, file=sys.stderr) finally: if ctx is not None: try: ctx.close() except Exception: pass _write_result_file(result_path, interactive_ok, end_reason, detail_msg) def _entrypoint() -> None: """读取配置 → 运行检测；任意异常都写入 result JSON，进程以 0 退出（由父进程读 JSON 判定成败）。""" result_path = "" if len(sys.argv) < 2: print("login_child: missing cfg path argv", file=sys.stderr) return cfg_path = sys.argv[1] try: with open(cfg_path, encoding="utf-8") as f: c = json.load(f) except Exception: tb = traceback.format_exc() print(tb, file=sys.stderr) return result_path = (c.get("result_path") or "").strip() try: _run_login_child(c) except Exception: tb = traceback.format_exc() print(tb, file=sys.stderr) _write_result_file(result_path, False, "child_fatal", tb) if __name__ == "__main__": _entrypoint()