From dbf5604090e22df754684c29d3369d332ae0d9d4 Mon Sep 17 00:00:00 2001
From: chendelian <116870791@qq.com>
Date: Mon, 11 May 2026 16:44:46 +0800
Subject: [PATCH] fix(rpa_helpers): require positive evidence in
 _is_already_logged_in to avoid false success

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 scripts/service/rpa_helpers/auth_flows.py | 42 +++++++++---------
 tests/test_rpa_helpers_auth_flows.py      | 53 +++++++++++++++++++++++
 2 files changed, 75 insertions(+), 20 deletions(-)

diff --git a/scripts/service/rpa_helpers/auth_flows.py b/scripts/service/rpa_helpers/auth_flows.py
index c2bc3c3..7192f40 100644
--- a/scripts/service/rpa_helpers/auth_flows.py
+++ b/scripts/service/rpa_helpers/auth_flows.py
@@ -68,35 +68,37 @@ def _account_extra(account: dict) -> dict:
 
 
 def _is_already_logged_in(page, account: dict) -> bool:
-    """粗略判断是否已登录（可被 extra_json.logged_in_url_pattern 覆盖）。"""
+    """是否已登录：仅当 extra_json 提供正向证据（URL 正则或可见 selector）时为 True。"""
     try:
         cur = page.url or ""
     except Exception:
         return False
+    cur_stripped = cur.strip()
+    if not cur_stripped:
+        return False
+    ul = cur_stripped.lower()
+    if ul == "about:blank" or ul.startswith("about:blank"):
+        return False
+
     extra = _account_extra(account)
+
     pattern = extra.get("logged_in_url_pattern")
-    if isinstance(pattern, str) and pattern:
+    if isinstance(pattern, str) and pattern.strip():
         try:
-            return bool(re.search(pattern, cur))
+            if re.search(pattern, cur_stripped):
+                return True
         except re.error:
             pass
-    lc = cur.lower()
-    if any(k in lc for k in ("login", "signin", "sign-in", "auth", "/sso/")):
-        return False
-    try:
-        login_btn = page.locator(
-            "button:has-text('登录'), button:has-text('Sign in'), "
-            "a:has-text('登录'), a:has-text('Sign in')"
-        )
-        if login_btn.count() > 0:
-            try:
-                if login_btn.first.is_visible(timeout=500):
-                    return False
-            except Exception:
-                pass
-    except Exception:
-        pass
-    return True
+
+    sel = extra.get("logged_in_selector")
+    if isinstance(sel, str) and sel.strip():
+        try:
+            if page.locator(sel.strip()).first.is_visible(timeout=500):
+                return True
+        except Exception:
+            pass
+
+    return False
 
 
 def _current_machine_fingerprint() -> str:
diff --git a/tests/test_rpa_helpers_auth_flows.py b/tests/test_rpa_helpers_auth_flows.py
index 9735722..a618e60 100644
--- a/tests/test_rpa_helpers_auth_flows.py
+++ b/tests/test_rpa_helpers_auth_flows.py
@@ -102,3 +102,56 @@ def test_device_bound_fingerprint_mismatch(monkeypatch):
         ensure_logged_in(page, account, credentials={"plaintext": "pw"})
     assert ei.value.code == "ERR_DEVICE_FINGERPRINT_MISMATCH"
 
+
+def test_is_already_logged_in_default_false():
+    from service.rpa_helpers.auth_flows import _is_already_logged_in
+
+    p = MagicMock()
+    p.url = "about:blank"
+    assert _is_already_logged_in(p, {}) is False
+
+    p.url = "https://www.douyin.com"
+    assert _is_already_logged_in(p, {}) is False
+
+    p.url = "https://www.douyin.com/user/abc"
+    assert _is_already_logged_in(
+        p,
+        {"extra_json": {"logged_in_url_pattern": r"/user/"}},
+    ) is True
+
+    loc_first = MagicMock()
+    loc_first.is_visible.return_value = True
+    loc = MagicMock()
+    loc.first = loc_first
+    p2 = MagicMock()
+    p2.url = "https://example.com/home"
+    p2.locator.return_value = loc
+    assert _is_already_logged_in(
+        p2,
+        {"extra_json": {"logged_in_selector": "#avatar"}},
+    ) is True
+    p2.locator.assert_called_with("#avatar")
+
+
+def test_qr_code_manual_default_no_hooks():
+    from service.rpa_helpers.auth_flows import ensure_logged_in
+
+    qr_first = MagicMock()
+    qr_first.wait_for.side_effect = TimeoutError("mock qr timeout")
+    qr_loc = MagicMock()
+    qr_loc.first = qr_first
+    page = MagicMock()
+    page.url = "about:blank"
+    page.locator.return_value = qr_loc
+
+    account = {"auth_strategy": "qr_code_manual", "url": "http://login.example"}
+    r = ensure_logged_in(page, account)
+    page.goto.assert_called_once_with(
+        "http://login.example",
+        wait_until="domcontentloaded",
+        timeout=60000,
+    )
+    qr_first.wait_for.assert_called_once()
+    assert r.ok is False
+    assert r.message == "qr code not found"
+