Validate and log QR capture

This commit is contained in:
2026-01-07 13:56:16 +08:00
parent 6af8f46129
commit ec90404194

View File

@@ -7,6 +7,7 @@ import os
import queue import queue
import threading import threading
import time import time
from io import BytesIO
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
import database import database
@@ -470,6 +471,8 @@ class KDocsUploader:
continue continue
width = box.get("width", 0) width = box.get("width", 0)
height = box.get("height", 0) height = box.get("height", 0)
if width < 160 or height < 160:
continue
if vp_width and vp_height: if vp_width and vp_height:
if width > vp_width * 0.92 and height > vp_height * 0.92: if width > vp_width * 0.92 and height > vp_height * 0.92:
continue continue
@@ -486,6 +489,27 @@ class KDocsUploader:
except Exception: except Exception:
return None return None
def _is_valid_qr_image(self, data: Optional[bytes]) -> bool:
if not data or len(data) < 1024:
return False
try:
from PIL import Image, ImageStat
img = Image.open(BytesIO(data))
width, height = img.size
if width < 120 or height < 120:
return False
ratio = width / float(height or 1)
if ratio < 0.6 or ratio > 1.4:
return False
gray = img.convert("L")
stat = ImageStat.Stat(gray)
if stat.stddev[0] < 5:
return False
return True
except Exception:
return len(data) >= 2048
def _save_login_state(self) -> None: def _save_login_state(self) -> None:
try: try:
storage_state = getattr(config, "KDOCS_LOGIN_STATE_FILE", "data/kdocs_login_state.json") storage_state = getattr(config, "KDOCS_LOGIN_STATE_FILE", "data/kdocs_login_state.json")
@@ -515,14 +539,18 @@ class KDocsUploader:
self._ensure_login_dialog() self._ensure_login_dialog()
qr_image = None qr_image = None
invalid_qr = None
for _ in range(10): for _ in range(10):
self._ensure_login_dialog() self._ensure_login_dialog()
qr_image = self._capture_qr_image() candidate = self._capture_qr_image()
if qr_image: if candidate and self._is_valid_qr_image(candidate):
qr_image = candidate
break break
if candidate:
invalid_qr = candidate
time.sleep(1) time.sleep(1)
if not qr_image: if not qr_image:
self._last_error = "二维码获取失败" self._last_error = "二维码识别异常" if invalid_qr else "二维码获取失败"
try: try:
pages = self._iter_pages() pages = self._iter_pages()
page_urls = [getattr(p, "url", "") for p in pages] page_urls = [getattr(p, "url", "") for p in pages]
@@ -538,6 +566,14 @@ class KDocsUploader:
continue continue
if saved: if saved:
logger.warning(f"[KDocs] 已保存调试截图: {saved}") logger.warning(f"[KDocs] 已保存调试截图: {saved}")
if invalid_qr:
try:
path = f"data/kdocs_invalid_qr_{ts}.png"
with open(path, "wb") as handle:
handle.write(invalid_qr)
logger.warning(f"[KDocs] 已保存无效二维码截图: {path}")
except Exception:
pass
except Exception: except Exception:
pass pass
return {"success": False, "error": self._last_error} return {"success": False, "error": self._last_error}