Fail fast on CPA management login errors

AI-Co-Authored-By: Codex

cpa_oauth.py

@@ -201,6 +201,7 @@ def _ensure_cpa_oauth_panel_ready(
 ) -> str:
     """确保 CPA 面板已越过管理登录页，返回已存在的 OAuth URL（如有）。"""
     deadline = time.time() + timeout_sec
+    management_login_attempts = 0
     while time.time() < deadline:
         oauth_url = _read_first_visible_text(page, CPA_OAUTH_URL_SELECTORS)
         if oauth_url:
@@ -212,6 +213,9 @@ def _ensure_cpa_oauth_panel_ready(
             return ""
 
         if login_visible:
+            management_login_attempts += 1
+            if management_login_attempts > 3:
+                raise RuntimeError(f"CPA 管理登录失败，请检查管理密钥或权限 URL={page.url}")
             _login_cpa_management_if_needed(page, management_key, log)
             _safe_page_wait(page, 800)
             continue

test_cpa_oauth.py

@@ -18,6 +18,64 @@ class CallbackUrlTests(unittest.TestCase):
 
 
 class FetchOauthUrlTests(unittest.TestCase):
+    def test_stops_retrying_when_management_login_stays_on_login_page(self):
+        import cpa_oauth
+
+        class _FakeLocator:
+            def __init__(self, page, selector):
+                self.page = page
+                self.selector = selector
+
+            def count(self):
+                if "password" in self.selector:
+                    return 1
+                if 'has-text("Login")' in self.selector:
+                    return 1
+                return 0
+
+            def nth(self, _idx):
+                return self
+
+            def is_visible(self):
+                return True
+
+            def is_enabled(self):
+                return True
+
+            def fill(self, value):
+                self.page.filled.append((self.selector, value))
+
+            def click(self, **_kwargs):
+                self.page.clicked.append(self.selector)
+
+        class _FakePage:
+            def __init__(self):
+                self.url = ""
+                self.filled = []
+                self.clicked = []
+
+            def goto(self, url, **_kwargs):
+                self.url = url.rsplit("#", 1)[0] + "#/login"
+
+            def wait_for_timeout(self, *_args, **_kwargs):
+                pass
+
+            def locator(self, selector):
+                return _FakeLocator(self, selector)
+
+        page = _FakePage()
+
+        with self.assertRaisesRegex(RuntimeError, "CPA 管理登录失败"):
+            cpa_oauth.fetch_cpa_oauth_url(
+                page,
+                cpa_url="https://cpa.example/management.html#/oauth",
+                management_key="bad-key",
+                log=lambda *_: None,
+                timeout_sec=5,
+            )
+
+        self.assertLessEqual(len(page.clicked), 3)
+
     def test_waits_for_oauth_panel_after_management_login(self):
         import cpa_oauth