'commit'

Test/TestFirstPage/T_TeLaiDian_FirstPageScrollTest.py

@@ -0,0 +1,147 @@
+import asyncio
+import os
+import sys
+import time
+import shutil
+
+import uiautomator2 as u2
+import cv2
+
+
+project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+if project_root not in sys.path:
+    sys.path.append(project_root)
+
+
+from Apps.TeLaiDian.Kit import setup_logger, take_screenshot
+from Apps.TeLaiDian.Config.Setting import SCROLL_DISTANCE_RATIO, WAIT_AFTER_SCROLL
+from Apps.TeLaiDian.FirstPageKit import run_ocr_rect, run_batch_in_dir
+
+
+OUTPUT_DIR = os.path.join(project_root, "Output")
+LOGS_DIR = os.path.join(project_root, "Logs")
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+os.makedirs(LOGS_DIR, exist_ok=True)
+LOG_FILE_PATH = os.path.join(LOGS_DIR, "T_TeLaiDian_FirstPageScrollTest.log")
+
+logger = setup_logger(
+    "TeLaiDian.FirstPageScrollTest", log_file=LOG_FILE_PATH, clear_old_log=True
+)
+
+
+async def capture_list_page(d, base_dir: str, index: int):
+    ts = int(time.time())
+    image_uuid = f"tld_firstpage_{index}_{ts}"
+    list_path = take_screenshot(d, image_uuid, save_dir=base_dir)
+    logger.info(f"[{index}] 列表截图路径: {list_path}")
+
+    try:
+        await run_ocr_rect(list_path, log_path=LOG_FILE_PATH)
+        logger.info(f"[{index}] 已根据 OCR+LLM 算法生成首屏绿框调试图。")
+    except Exception as e:
+        logger.exception(f"[{index}] 运行 OCR 绿框测试失败: {e}")
+
+
+async def run_capture_sequence(d, base_dir: str, pages: int = 5):
+    if os.path.exists(base_dir):
+        shutil.rmtree(base_dir, ignore_errors=True)
+    os.makedirs(base_dir, exist_ok=True)
+
+    for i in range(1, pages + 1):
+        await capture_list_page(d, base_dir, i)
+
+        if i < pages:
+            w, h = d.window_size()
+            start_x, start_y = w // 2, int(h * 0.8)
+            end_x, end_y = start_x, int(h * (1 - SCROLL_DISTANCE_RATIO))
+            logger.info(f"[{i}] 执行翻页滑动: ({start_x},{start_y}) -> ({end_x},{end_y})")
+            d.swipe(start_x, start_y, end_x, end_y, 0.3)
+            await asyncio.sleep(WAIT_AFTER_SCROLL)
+
+
+async def offline_validate(base_dir: str):
+    img_files = []
+    for name in os.listdir(base_dir):
+        if name.lower().endswith(".jpg"):
+            img_files.append(os.path.join(base_dir, name))
+    img_files.sort()
+
+    if not img_files:
+        logger.info(f"目录下未找到图片: {base_dir}")
+        return
+
+    await run_batch_in_dir(base_dir, log_file=LOG_FILE_PATH)
+
+    total = len(img_files)
+    ok_count = 0
+    for path in img_files:
+        logger.info(f"开始离线校验: {path}")
+        img = cv2.imread(path)
+        if img is None:
+            logger.error(f"无法读取图片: {path}")
+            continue
+        h, w = img.shape[:2]
+
+        log_path = LOG_FILE_PATH
+        stations = await run_ocr_rect(path, log_path=log_path)
+        if not stations:
+            logger.error(f"未识别到任何场站: {path}")
+            continue
+
+        page_ok = True
+        for st in stations:
+            rect = st.get("rect") or []
+            if not isinstance(rect, (list, tuple)) or len(rect) < 4:
+                logger.error(f"矩形数据不合法: {st}")
+                page_ok = False
+                break
+            x1, y1, x2, y2 = rect
+            if not (0 <= x1 < x2 <= w and 0 <= y1 < y2 <= h):
+                logger.error(f"矩形越界: {rect}, 图片尺寸=({w},{h})")
+                page_ok = False
+                break
+            busy_list = st.get("busy_list") or []
+            if not busy_list:
+                logger.error(f"场站忙闲信息为空: {st}")
+                page_ok = False
+                break
+            click_point = st.get("click_point") or st.get("click") or []
+            if isinstance(click_point, (list, tuple)) and len(click_point) >= 2:
+                cx, cy = click_point[0], click_point[1]
+                if not (x1 <= cx <= x2 and y1 <= cy <= y2):
+                    logger.error(f"点击点不在矩形内部: click={click_point}, rect={rect}")
+                    page_ok = False
+                    break
+        if page_ok:
+            ok_count += 1
+            logger.info(f"离线校验通过: {path}")
+        else:
+            logger.error(f"离线校验未通过: {path}")
+    if ok_count != total:
+        raise AssertionError(f"离线首屏 OCR+LLM 校验未全部通过: {ok_count}/{total}")
+    logger.info(f"离线首屏 OCR+LLM 校验全部通过: {ok_count}/{total}")
+
+
+async def main():
+    if os.path.exists(LOGS_DIR):
+        for name in os.listdir(LOGS_DIR):
+            if not name.startswith("T_TeLaiDian_"):
+                continue
+            path = os.path.join(LOGS_DIR, name)
+            if os.path.isfile(path) or os.path.islink(path):
+                try:
+                    os.remove(path)
+                except PermissionError:
+                    pass
+            elif os.path.isdir(path):
+                shutil.rmtree(path, ignore_errors=True)
+    else:
+        os.makedirs(LOGS_DIR, exist_ok=True)
+    d = u2.connect()
+    base_dir = OUTPUT_DIR
+    await run_capture_sequence(d, base_dir, pages=3)
+    await offline_validate(base_dir)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())