224 lines
7.4 KiB
Python
224 lines
7.4 KiB
Python
"""Playwright 登录检测子进程：由 browser_service.cmd_login 以独立解释器启动，argv[1] 为 JSON 配置路径。"""
|
||
import json
|
||
import logging
|
||
import sys
|
||
import time
|
||
from urllib.parse import urlparse
|
||
|
||
from playwright.sync_api import sync_playwright
|
||
|
||
|
||
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
|
||
result_path = c.get("result_path") or ""
|
||
log_path = (c.get("log_file") or "").strip()
|
||
lg = logging.getLogger("openclaw.skill.account_manager.login_child")
|
||
if log_path:
|
||
lg.handlers.clear()
|
||
lg.setLevel(logging.DEBUG)
|
||
_fh = logging.FileHandler(log_path, encoding="utf-8")
|
||
_fh.setFormatter(
|
||
logging.Formatter(
|
||
"%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
|
||
datefmt="%Y-%m-%dT%H:%M:%S",
|
||
)
|
||
)
|
||
lg.addHandler(_fh)
|
||
lg.propagate = False
|
||
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.handlers:
|
||
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.handlers:
|
||
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
|
||
if lg.handlers:
|
||
lg.info("login_detected_ok %s", detail)
|
||
break
|
||
except Exception as ex:
|
||
if lg.handlers:
|
||
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:
|
||
if lg.handlers:
|
||
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.handlers:
|
||
lg.info("login_success_closing_browser_immediately")
|
||
except Exception as e:
|
||
if lg.handlers:
|
||
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}, rf, ensure_ascii=False)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
ctx.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|