新增Dashboard未登录二维码引导与倒计时

robot.py

@@ -1,5 +1,7 @@
 # -*- coding: utf-8 -*-
 import asyncio
+import base64
+import io
 import os
 import threading
 import time
@@ -8,6 +10,7 @@ import traceback
 import uuid
 from collections import deque
 from loguru import logger
+import qrcode
 
 import wechat_ipad
 from base.plugin_common.message_plugin_interface import MessagePluginInterface
@@ -65,6 +68,12 @@ class Robot:
         # 3. 这样主线程就不会再把“线程已启动”误判成“wechat 已成功就绪”。
         self.ipad_startup_event = threading.Event()
         self.ipad_startup_error = None
+        # Dashboard 登录引导态：
+        # 1. 首次部署或登录失效时，后台首页需要知道当前二维码、剩余有效期和最近一次刷新时间；
+        # 2. 这类状态属于“运行时临时信息”，不应该写回配置文件，也不值得额外拉一层服务；
+        # 3. 因此直接挂在 Robot 上，用锁保护跨线程读写，保持实现足够轻。
+        self._ipad_login_qr_lock = threading.Lock()
+        self.ipad_login_qr_state = self._build_empty_ipad_login_qr_state()
         self.wxid = None
         self.nickname = None
         self.alias = None
@@ -317,6 +326,8 @@ class Robot:
                 on_idle_payload=self._handle_runtime_idle_payload,
                 on_logout=self._handle_ipad_logout,
                 on_runtime_state_change=self._handle_runtime_state_change,
+                on_login_qr_update=self._handle_ipad_login_qr_update,
+                on_login_qr_cleared=self._handle_ipad_login_qr_cleared,
             )
 
         except Exception as e:
@@ -413,6 +424,159 @@ class Robot:
             self.LOG.warning(f"读取 TOML 配置失败，将按空配置继续: path={normalized_path}, error={e}")
             return {}
 
+    @staticmethod
+    def _build_empty_ipad_login_qr_state() -> dict:
+        """构造 Dashboard 可直接消费的默认二维码登录态。"""
+        return {
+            "logged_in": False,
+            "active": False,
+            "status": "idle",
+            "status_text": "尚未进入扫码登录流程",
+            "current": {},
+            "history": [],
+            "updated_at": 0,
+        }
+
+    @staticmethod
+    def _build_qr_image_data(scan_url: str) -> str:
+        """把扫码内容生成 base64 图片，供 Dashboard 直接展示。"""
+        normalized_scan_url = str(scan_url or "").strip()
+        if not normalized_scan_url:
+            return ""
+
+        try:
+            # 这里直接在后端生成二维码图片：
+            # 1. 避免首页再额外引入前端二维码依赖，减少静态资源改动；
+            # 2. 即使 provider 没有返回可直接访问的图片 URL，只要有扫码内容也能展示；
+            # 3. 返回 data URI 后，前端只需要普通 `<img>` 即可渲染。
+            qr = qrcode.QRCode(
+                version=1,
+                error_correction=qrcode.constants.ERROR_CORRECT_L,
+                box_size=10,
+                border=4,
+            )
+            qr.add_data(normalized_scan_url)
+            qr.make(fit=True)
+            image = qr.make_image(fill_color="black", back_color="white")
+            buffer = io.BytesIO()
+            image.save(buffer, format="PNG")
+            encoded = base64.b64encode(buffer.getvalue()).decode("utf-8")
+            return f"data:image/png;base64,{encoded}"
+        except Exception:
+            return ""
+
+    def get_ipad_login_qr_state(self) -> dict:
+        """返回当前 Dashboard 可读取的二维码登录态快照。"""
+        with self._ipad_login_qr_lock:
+            login_state_flag = bool(self.ipad_login_qr_state.get("logged_in", False))
+            qr_status = str(self.ipad_login_qr_state.get("status", "idle") or "idle")
+            state = {
+                "logged_in": bool(self.wxid) or login_state_flag or qr_status in {"confirmed", "logged_in"},
+                "active": bool(self.ipad_login_qr_state.get("active", False)),
+                "status": qr_status,
+                "status_text": str(
+                    self.ipad_login_qr_state.get("status_text", "尚未进入扫码登录流程") or "尚未进入扫码登录流程"
+                ),
+                "updated_at": float(self.ipad_login_qr_state.get("updated_at", 0) or 0),
+                "current": dict(self.ipad_login_qr_state.get("current", {}) or {}),
+                "history": [dict(item or {}) for item in (self.ipad_login_qr_state.get("history", []) or [])],
+                "runtime_running": bool(self.ipad_running),
+                "wxid": str(self.wxid or ""),
+                "nickname": str(self.nickname or ""),
+            }
+
+        now_ts = time.time()
+        current = state.get("current", {}) or {}
+        expires_at = float(current.get("expires_at", 0) or 0)
+        if expires_at > 0:
+            current["remaining_seconds"] = max(0, int(expires_at - now_ts))
+        else:
+            current["remaining_seconds"] = int(current.get("remaining_seconds", 0) or 0)
+        state["current"] = current
+        state["server_now"] = now_ts
+        return state
+
+    async def _handle_ipad_login_qr_update(self, payload: dict) -> None:
+        """同步 provider 扫码登录态到 Robot，供 Dashboard 轮询读取。"""
+        now_ts = time.time()
+        uuid_value = str((payload or {}).get("uuid", "") or "").strip()
+        scan_url = str((payload or {}).get("scan_url", "") or "").strip()
+        raw_url = str((payload or {}).get("url", "") or "").strip()
+        status = str((payload or {}).get("status", "waiting") or "waiting").strip() or "waiting"
+        status_text = str((payload or {}).get("status_text", "等待扫码登录") or "等待扫码登录").strip()
+        login_source = str((payload or {}).get("login_source", "fresh_qr") or "fresh_qr").strip()
+        expires_in = (payload or {}).get("expires_in")
+        expires_in = None if expires_in in (None, "") else max(0, int(expires_in))
+        current_record = {
+            "uuid": uuid_value,
+            "scan_url": scan_url,
+            "raw_url": raw_url,
+            "image_data": self._build_qr_image_data(scan_url),
+            "status": status,
+            "status_text": status_text,
+            "login_source": login_source,
+            "updated_at": now_ts,
+            "updated_at_text": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now_ts)),
+        }
+        if expires_in is not None:
+            current_record["remaining_seconds"] = expires_in
+            current_record["expires_at"] = now_ts + expires_in
+
+        with self._ipad_login_qr_lock:
+            history_records = list(self.ipad_login_qr_state.get("history", []) or [])
+            # 二维码历史只在“uuid 发生变化”时追加一条：
+            # 1. 倒计时刷新会非常频繁，如果每次都入历史，前端会被大量重复记录淹没；
+            # 2. 这里把历史理解为“最近几次生成过哪些二维码”，而不是每一秒状态快照；
+            # 3. 这样首页既能展示多个二维码记录，也能保持列表简洁可读。
+            if uuid_value:
+                existing_index = next(
+                    (index for index, item in enumerate(history_records) if str(item.get("uuid", "") or "") == uuid_value),
+                    -1,
+                )
+                history_entry = dict(current_record)
+                if existing_index >= 0:
+                    history_records[existing_index] = history_entry
+                else:
+                    history_records.insert(0, history_entry)
+                history_records = history_records[:3]
+
+            self.ipad_login_qr_state = {
+                "logged_in": False,
+                "active": status != "confirmed",
+                "status": status,
+                "status_text": status_text,
+                "current": current_record,
+                "history": history_records,
+                "updated_at": now_ts,
+            }
+
+    async def _handle_ipad_login_qr_cleared(self, payload: dict | None = None) -> None:
+        """在登录完成或识别到已有登录态后关闭首页二维码引导。"""
+        now_ts = time.time()
+        status = str((payload or {}).get("status", "idle") or "idle").strip() or "idle"
+        status_text = str((payload or {}).get("status_text", "登录流程已结束") or "登录流程已结束").strip()
+        cleared_uuid = str((payload or {}).get("uuid", "") or "").strip()
+
+        with self._ipad_login_qr_lock:
+            history_records = list(self.ipad_login_qr_state.get("history", []) or [])
+            if cleared_uuid:
+                for item in history_records:
+                    if str(item.get("uuid", "") or "") == cleared_uuid:
+                        item["status"] = status
+                        item["status_text"] = status_text
+                        item["updated_at"] = now_ts
+                        item["updated_at_text"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now_ts))
+
+            self.ipad_login_qr_state = {
+                "logged_in": status in {"confirmed", "logged_in"} or bool(self.wxid),
+                "active": False,
+                "status": status,
+                "status_text": status_text,
+                "current": {},
+                "history": history_records[:3],
+                "updated_at": now_ts,
+            }
+
     async def _on_ipad_login_ready(self, login_identity: dict) -> None:
         """处理 provider 登录成功后的项目侧初始化动作。
 
@@ -433,6 +597,12 @@ class Robot:
         self.ipad_bot.alias = self.alias
         self.ipad_bot.phone = self.phone
         self.ipad_bot.signature = self.signature
+        await self._handle_ipad_login_qr_cleared(
+            {
+                "status": "confirmed",
+                "status_text": "微信已登录，二维码弹窗已关闭",
+            }
+        )
         self.LOG.info(
             f"wechat_ipad登录账号信息: wxid: {self.wxid}  昵称: {self.nickname}  微信号: {self.alias}  手机号: {self.phone}"
         )