Force KDocs QR fetch and improve login detection

services/kdocs_uploader.py

@@ -91,8 +91,8 @@ class KDocsUploader:
             self._last_error = "上传队列已满"
             return False
 
-    def request_qr(self, timeout: int = 30) -> Dict[str, Any]:
-        return self._submit_command("qr", timeout=timeout)
+    def request_qr(self, timeout: int = 30, *, force: bool = False) -> Dict[str, Any]:
+        return self._submit_command("qr", timeout=timeout, payload={"force": force})
 
     def clear_login(self, timeout: int = 20) -> Dict[str, Any]:
         return self._submit_command("clear_login", timeout=timeout)
@@ -100,11 +100,16 @@ class KDocsUploader:
     def refresh_login_status(self, timeout: int = 20) -> Dict[str, Any]:
         return self._submit_command("status", timeout=timeout)
 
-    def _submit_command(self, action: str, timeout: int = 30) -> Dict[str, Any]:
+    def _submit_command(
+        self,
+        action: str,
+        timeout: int = 30,
+        payload: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         if not self._running:
             self.start()
         resp_queue: queue.Queue = queue.Queue(maxsize=1)
-        self._queue.put({"action": action, "response": resp_queue})
+        self._queue.put({"action": action, "response": resp_queue, "payload": payload or {}})
         try:
             return resp_queue.get(timeout=timeout)
         except queue.Empty:
@@ -122,7 +127,7 @@ class KDocsUploader:
                 if action == "upload":
                     self._handle_upload(task.get("payload") or {})
                 elif action == "qr":
-                    result = self._handle_qr()
+                    result = self._handle_qr(task.get("payload") or {})
                     task.get("response").put(result)
                 elif action == "clear_login":
                     result = self._handle_clear_login()
@@ -138,7 +143,7 @@ class KDocsUploader:
     def _load_system_config(self) -> Dict[str, Any]:
         return database.get_system_config() or {}
 
-    def _ensure_playwright(self) -> bool:
+    def _ensure_playwright(self, *, use_storage_state: bool = True) -> bool:
         if sync_playwright is None:
             self._last_error = "playwright 未安装"
             return False
@@ -150,7 +155,7 @@ class KDocsUploader:
                 self._browser = self._playwright.chromium.launch(headless=headless)
             if self._context is None:
                 storage_state = getattr(config, "KDOCS_LOGIN_STATE_FILE", "data/kdocs_login_state.json")
-                if os.path.exists(storage_state):
+                if use_storage_state and os.path.exists(storage_state):
                     self._context = self._browser.new_context(storage_state=storage_state)
                 else:
                     self._context = self._browser.new_context()
@@ -205,6 +210,24 @@ class KDocsUploader:
                 return False
         except Exception:
             return False
+        try:
+            login_btn = self._page.get_by_role("button", name="登录")
+            if login_btn.is_visible(timeout=1200):
+                return False
+        except Exception:
+            pass
+        try:
+            login_link = self._page.get_by_role("link", name="登录")
+            if login_link.is_visible(timeout=1200):
+                return False
+        except Exception:
+            pass
+        try:
+            wechat_btn = self._page.get_by_role("button", name="微信登录")
+            if wechat_btn.is_visible(timeout=1200):
+                return False
+        except Exception:
+            pass
         return True
 
     def _has_saved_login_state(self) -> bool:
@@ -220,6 +243,22 @@ class KDocsUploader:
         except Exception:
             pass
 
+        try:
+            login_btn = self._page.get_by_role("button", name="登录")
+            if login_btn.is_visible(timeout=1500):
+                login_btn.click()
+                time.sleep(1)
+        except Exception:
+            pass
+
+        try:
+            login_link = self._page.get_by_role("link", name="登录")
+            if login_link.is_visible(timeout=1500):
+                login_link.click()
+                time.sleep(1)
+        except Exception:
+            pass
+
         try:
             wechat_btn = self._page.get_by_role("button", name="微信登录")
             if wechat_btn.is_visible(timeout=3000):
@@ -258,17 +297,20 @@ class KDocsUploader:
         except Exception as e:
             logger.warning(f"[KDocs] 保存登录态失败: {e}")
 
-    def _handle_qr(self) -> Dict[str, Any]:
+    def _handle_qr(self, payload: Dict[str, Any]) -> Dict[str, Any]:
         cfg = self._load_system_config()
         doc_url = (cfg.get("kdocs_doc_url") or "").strip()
         if not doc_url:
             return {"success": False, "error": "未配置金山文档链接"}
-        if not self._ensure_playwright():
+        force = bool(payload.get("force"))
+        if force:
+            self._handle_clear_login()
+        if not self._ensure_playwright(use_storage_state=not force):
             return {"success": False, "error": self._last_error or "浏览器不可用"}
         if not self._open_document(doc_url):
             return {"success": False, "error": self._last_error or "打开文档失败"}
 
-        if self._has_saved_login_state() and self._is_logged_in():
+        if not force and self._has_saved_login_state() and self._is_logged_in():
             self._login_required = False
             self._last_login_ok = True
             self._save_login_state()