Add JoinQuant env browser login helper

plugins/joinquant/browser.py

@@ -188,6 +188,23 @@ def load_json(path: str | Path) -> dict[str, Any]:
     return data
 
 
+def load_env_file(path: str | Path) -> dict[str, str]:
+    """Load simple KEY=VALUE env files without requiring shell-safe quoting."""
+    values: dict[str, str] = {}
+    for raw in Path(path).expanduser().read_text(encoding="utf-8").splitlines():
+        line = raw.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        clean = value.strip()
+        if len(clean) >= 2 and clean[0] == clean[-1] and clean[0] in {"'", '"'}:
+            clean = clean[1:-1]
+        else:
+            clean = clean.strip("'\"")
+        values[key.strip()] = clean
+    return values
+
+
 def _get_dotted(data: dict[str, Any], dotted: str) -> Any:
     current: Any = data
     for part in dotted.split("."):
@@ -271,6 +288,80 @@ def save_login_state(
     return state_path
 
 
+def save_login_state_from_env(
+    *,
+    env_path: str | Path,
+    storage_state: str | Path,
+    login_url: str = DEFAULT_LOGIN_URL,
+    headless: bool = True,
+    out_dir: str | Path | None = None,
+    timeout_ms: int = 120_000,
+) -> dict[str, Any]:
+    """Try to log in with env-file credentials and save browser state.
+
+    This handles the normal password-login form. If JoinQuant presents CAPTCHA,
+    slide verification, SMS verification, or 2FA, the report will mark the run
+    as not logged in and save a screenshot for manual diagnosis.
+    """
+    env = load_env_file(env_path)
+    username = env.get("JOINQUANT_USERNAME")
+    password = env.get("JOINQUANT_PASSWORD")
+    if not username or not password:
+        raise ValueError("JOINQUANT_USERNAME and JOINQUANT_PASSWORD are required")
+
+    sync_playwright, _ = _require_playwright()
+    state_path = Path(storage_state).expanduser()
+    state_path.parent.mkdir(parents=True, exist_ok=True)
+    artifact_dir = Path(out_dir or state_path.parent / "joinquant_login_artifacts")
+    artifact_dir.mkdir(parents=True, exist_ok=True)
+    screenshot_path = artifact_dir / "login_after_submit.png"
+    html_path = artifact_dir / "login_after_submit.html"
+
+    with sync_playwright() as pw:
+        browser = pw.chromium.launch(headless=headless)
+        context = browser.new_context()
+        page = context.new_page()
+        page.set_default_timeout(timeout_ms)
+        page.goto(login_url, wait_until="networkidle", timeout=timeout_ms)
+        page.locator('input[name="username"], input.pwd-phone').first.fill(username)
+        page.locator('input[name="pwd"], input.jq-login__password').first.fill(password)
+        checkbox = page.locator("#agreementBox, input.agreement-box").first
+        if checkbox.count():
+            checkbox.check(force=True)
+        page.locator("button.btnPwdSubmit, button.login-submit").first.click()
+        page.wait_for_timeout(5000)
+
+        html = page.content()
+        html_path.write_text(html, encoding="utf-8")
+        page.screenshot(path=str(screenshot_path), full_page=True)
+        login_inputs = page.locator('input[name="username"], input[name="pwd"]').count()
+        current_url = page.url
+        logged_in = login_inputs == 0 and "/user/login" not in current_url
+        if logged_in:
+            context.storage_state(path=str(state_path))
+            state_path.chmod(0o600)
+        browser.close()
+
+    report = {
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "logged_in": bool(logged_in),
+        "storage_state": str(state_path) if logged_in else "",
+        "artifact_dir": str(artifact_dir),
+        "screenshot": str(screenshot_path),
+        "html": str(html_path),
+        "current_url": current_url,
+        "notes": (
+            "Logged in and saved browser state."
+            if logged_in
+            else "Login did not complete. CAPTCHA/SMS/2FA or invalid credentials may be required."
+        ),
+    }
+    report_path = artifact_dir / "login_report.json"
+    report["report_path"] = str(report_path)
+    report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    return report
+
+
 def browser_snapshot(
     *,
     url: str,
@@ -299,7 +390,7 @@ def browser_snapshot(
 
 
 def _locator(page: Any, selector: str):
-    return page.locator(selector).first()
+    return page.locator(selector).first
 
 
 def _action_fail(action: dict[str, Any], exc: Exception) -> None: