"""Playwright 登录检测子进程：由 browser_service.cmd_login 以独立解释器启动，argv[1] 为 JSON 配置路径。""" import json import os import sys import time 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 jiangchang_skill_core.unified_logging import attach_unified_file_handler from playwright.sync_api import sync_playwright try: from playwright.sync_api import Error as PlaywrightError except ImportError: PlaywrightError = Exception # type: ignore[misc,assignment] 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 isinstance(ex, PlaywrightError) and ( "closed" in s or "destroyed" in s ): return True return False def page_location_href(p): # Prefer location.href over page.url for SPA (e.g. Sohu shell path lag). 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 main(): with open(sys.argv[1], encoding="utf-8") as f: c = json.load(f) 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"]) interactive_ok = False end_reason = "timeout" result_path = c.get("result_path") or "" log_path = (c.get("log_file") or "").strip() lg = None if log_path: lg = attach_unified_file_handler( log_path, skill_slug="account-manager", logger_name="openclaw.skill.account_manager.login_child", ) last_eval_detail = "" 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 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, detail = evaluate_logged_out_dom(ctx, page, bundle) ok = not still_logged_out last_eval_detail = 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 iter_start_elapsed=%.1fs deadline_in=%.1fs tabs=%s poll_interval_sec=%s logged_out=%s ok=%s detail=%s | %s", iter_start - t0, deadline - iter_start, len(ctx.pages or []), poll, still_logged_out, ok, 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", 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 else: if lg is not None: lg.info("login_dom_ok_preparing_to_release_profile_dir") if interactive_ok: # 给 Cookie/会话落盘一点时间；随后必须关闭上下文，否则 llm-manager 无法再对同一 profile 启动持久化浏览器 print( "INFO:LOGIN_DOM_OK 页面检测已判定登录成功；约 2 秒后关闭本窗口以释放用户目录，供后续大模型网页会话使用。", flush=True, ) time.sleep(2.0) except Exception as e: if _is_context_closed_error(e): end_reason = "user_closed" interactive_ok = False if lg is not None: lg.warning("login_runner_browser_closed err=%s", e) else: if lg is not None: lg.exception("login_runner_fatal err=%s", e) print(e, file=sys.stderr) raise finally: if result_path: try: with open(result_path, "w", encoding="utf-8") as rf: json.dump( { "interactive_ok": interactive_ok, "end_reason": end_reason, }, rf, ensure_ascii=False, ) except Exception: pass try: ctx.close() except Exception: pass if __name__ == "__main__": main()