'commit'

Apps/XinDianTu/FirstPageKit.py

@@ -35,6 +35,8 @@ NON_STATION_KEYWORDS = [
     "去赚钱",
     "综合排序",
     "停车费",
+    "收费停车",
+    "限时免费停车",
     "充电速度",
     "筛选",
     "不限车长",
@@ -64,6 +66,12 @@ TOP_ZONE_STATION_HINT_KEYWORDS = [
     "超充",
 ]
 
+LLM_NON_STATION_NAME_KEYWORDS = [
+    "收费停车",
+    "限时免费停车",
+    "停车费",
+]
+
 
 def _load_image(path):
     if not os.path.exists(path):
@@ -182,11 +190,12 @@ async def run_ocr_rect(image_path, log_path=None):
         "2) anchor_point_norm: 一个对象 {\"x\": number, \"y\": number}，表示该场站名称文字所在行的中心点坐标，取值范围 0-1。\n"
         "并且尽量补充以下可选字段（找不到时可以省略或设为 null）:\n"
         "3) distance_text: 距离字符串，例如 \"6.9km\"、\"500m\"，从对应卡片中的距离行提取。\n"
-        "4) busy_info: 忙闲信息对象，格式为 {\"mode\": \"快|慢|超|普通\", \"idle\": number, \"total\": number}。\n"
+        "4) busy_list: 忙闲信息数组，数组中的每一项是 {\"mode\": \"快|慢|超|普通\", \"idle\": number, \"total\": number}。\n"
         "   例如：\"快 闲24/32\" => {\"mode\": \"快\", \"idle\": 24, \"total\": 32}；\n"
         "         \"慢 闲0/10\"  => {\"mode\": \"慢\", \"idle\": 0,  \"total\": 10}；\n"
         "         \"超 闲1/3\"   => {\"mode\": \"超\", \"idle\": 1,  \"total\": 3}；\n"
-        "         \"闲5/10\"    => {\"mode\": \"慢\", \"idle\": 5,  \"total\": 10}。\n"
+        "         \"闲5/10\"    => {\"mode\": \"慢\", \"idle\": 5,  \"total\": 10}；\n"
+        "   如果同一张卡片上既有“超 闲1/3”又有“快 闲5/10”，就需要在 busy_list 中放入两条记录。\n"
         "额外提示：\n"
         "- 每个场站卡片通常包含一行类似 \"1.4km\"、\"3.6km\" 的距离文本；\n"
         "- 该距离文本所在行的左侧、且在同一卡片中的那一行文字，就是对应的场站标题 station_name；\n"
@@ -251,6 +260,17 @@ async def run_ocr_rect(image_path, log_path=None):
             log_detail(f"LLM item[{idx}] 不是对象类型, 跳过")
             continue
         name = item.get("station_name") or item.get("name")
+        if name:
+            bad = False
+            for kw in LLM_NON_STATION_NAME_KEYWORDS:
+                if kw and kw in name:
+                    log_detail(
+                        f"LLM item[{idx}] station_name 命中非场站关键词 {kw}, 丢弃, name={name}"
+                    )
+                    bad = True
+                    break
+            if bad:
+                continue
         anchor = item.get("anchor_point_norm") or item.get("anchor") or item.get("center_norm")
         if not anchor:
             log_detail(f"LLM item[{idx}] 缺少 anchor 信息, 跳过, content={item}")
@@ -272,22 +292,42 @@ async def run_ocr_rect(image_path, log_path=None):
             py = 0
 
         distance_text = item.get("distance_text") or item.get("distance")
-        busy_raw = item.get("busy_info") or item.get("busy")
+        busy_list_raw = (
-        busy_mode = None
+            item.get("busy_list")
-        busy_idle = None
+            or item.get("busyInfos")
-        busy_total = None
+            or item.get("busy_info")
-        if isinstance(busy_raw, dict):
+            or item.get("busy")
-            busy_mode = busy_raw.get("mode") or busy_raw.get("type")
+        )
+        busy_list = []
+        if isinstance(busy_list_raw, list):
+            for bi in busy_list_raw:
+                if not isinstance(bi, dict):
+                    continue
+                mode = bi.get("mode") or bi.get("type")
+                idle = bi.get("idle")
+                total = bi.get("total")
+                try:
+                    idle = int(idle) if idle is not None else None
+                except Exception:
+                    idle = None
+                try:
+                    total = int(total) if total is not None else None
+                except Exception:
+                    total = None
+                busy_list.append({"mode": mode, "idle": idle, "total": total})
+        elif isinstance(busy_list_raw, dict):
+            mode = busy_list_raw.get("mode") or busy_list_raw.get("type")
+            idle = busy_list_raw.get("idle")
+            total = busy_list_raw.get("total")
             try:
-                if busy_raw.get("idle") is not None:
+                idle = int(idle) if idle is not None else None
-                    busy_idle = int(busy_raw.get("idle"))
             except Exception:
-                busy_idle = None
+                idle = None
             try:
-                if busy_raw.get("total") is not None:
+                total = int(total) if total is not None else None
-                    busy_total = int(busy_raw.get("total"))
             except Exception:
-                busy_total = None
+                total = None
+            busy_list.append({"mode": mode, "idle": idle, "total": total})
 
         stations.append(
             {
@@ -297,20 +337,31 @@ async def run_ocr_rect(image_path, log_path=None):
                 "px": px,
                 "py": py,
                 "distance_text": distance_text,
-                "busy_mode": busy_mode,
+                "busy_list": busy_list,
-                "busy_idle": busy_idle,
-                "busy_total": busy_total,
             }
         )
         log_detail(
             f"LLM anchor 规范化[{len(stations)}] name={name} ax={ax:.4f} ay={ay:.4f} py={py} "
-            f"distance={distance_text} busy=({busy_mode},{busy_idle},{busy_total})"
+            f"distance={distance_text} busy_list={busy_list}"
         )
 
     if not stations:
         log_detail("LLM 解析后没有可用的场站锚点, 结束当前图片处理")
         return
 
+    filtered = []
+    for s in stations:
+        if s.get("busy_list"):
+            filtered.append(s)
+        else:
+            log_detail(
+                f"场站 {s.get('name')} busy_list 为空, 视为信息不完整, 丢弃"
+            )
+    stations = filtered
+    if not stations:
+        log_detail("所有场站的忙闲信息均为空, 本页不画绿框")
+        return
+
     stations.sort(key=lambda s: s["py"])
 
     overlay = img.copy()
@@ -328,7 +379,7 @@ async def run_ocr_rect(image_path, log_path=None):
                 gaps.append(dy)
         if gaps:
             min_gap = min(gaps)
-            max_no_overlap = max(min_gap - 4, 40)
+            max_no_overlap = max(min_gap - 10, 40)
             if max_no_overlap < box_h_conf * 0.6:
                 box_h = box_h_conf
                 log_detail(
@@ -367,7 +418,13 @@ async def run_ocr_rect(image_path, log_path=None):
         px = s["px"]
         py = s["py"]
 
-        y1 = py - int(box_h * 0.35)
+        if py < effective_top:
+            log_detail(
+                f"Station[{idx + 1}] {name} 锚点 py={py} 位于顶部保护区之上(effective_top={effective_top}), 丢弃"
+            )
+            continue
+
+        y1 = py - int(box_h * 0.1)
         y2 = y1 + box_h
         orig_y1 = y1
         orig_y2 = y2
@@ -396,10 +453,23 @@ async def run_ocr_rect(image_path, log_path=None):
             y1 += shift
             y2 += shift
             if y2 > effective_bottom:
-                log_detail(
+                if idx == len(stations) - 1:
-                    f"Station[{idx + 1}] {name} 因避免重叠无法放入有效区域, 被丢弃"
+                    min_height = int(box_h * 0.5)
-                )
+                    new_y1 = prev_y2 + 1
-                continue
+                    new_y2 = effective_bottom
+                    if new_y2 - new_y1 >= min_height:
+                        y1 = new_y1
+                        y2 = new_y2
+                    else:
+                        log_detail(
+                            f"Station[{idx + 1}] {name} 底部剩余空间不足以放置绿框, 被丢弃"
+                        )
+                        continue
+                else:
+                    log_detail(
+                        f"Station[{idx + 1}] {name} 因避免重叠无法放入有效区域, 被丢弃"
+                    )
+                    continue
 
         prev_y2 = y2
 
@@ -422,9 +492,7 @@ async def run_ocr_rect(image_path, log_path=None):
                 "rect": [x1_fixed, y1, x2_fixed, y2],
                 "click_point": [click_x, click_y],
                 "distance_text": s.get("distance_text"),
-                "busy_mode": s.get("busy_mode"),
+                "busy_list": s.get("busy_list"),
-                "busy_idle": s.get("busy_idle"),
-                "busy_total": s.get("busy_total"),
             }
         )
 

Test/TestFirstPage/T_XinDianTu_FirstPageScrollTest.py

@@ -14,7 +14,7 @@ if project_root not in sys.path:
 
 from Apps.XinDianTu.Kit import setup_logger, take_screenshot
 from Apps.XinDianTu.Config.Setting import SCROLL_DISTANCE_RATIO, WAIT_AFTER_SCROLL
-from Apps.XinDianTu.FirstPageKit import run_ocr_rect
+from Apps.XinDianTu.FirstPageKit import run_ocr_rect, run_batch_in_dir
 
 
 LOGS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Logs")
@@ -39,7 +39,7 @@ async def capture_list_page(d, base_dir: str, index: int):
         logger.exception(f"[{index}] 运行 OCR 绿框测试失败: {e}")
 
 
-async def run():
+async def run_scroll_capture():
     logger.info("请先手动打开微信并进入新电途小程序的场站列表第一页。")
 
     d = u2.connect()
@@ -60,17 +60,56 @@ async def run():
                 shutil.rmtree(path)
     os.makedirs(base_dir, exist_ok=True)
 
-    for i in range(1, 11):
+    for i in range(1, 6):
         await capture_list_page(d, base_dir, i)
-        if i < 10:
+        if i < 5:
             logger.info(f"[{i}] 向上滑动，scale={SCROLL_DISTANCE_RATIO}，等待 {WAIT_AFTER_SCROLL} 秒后继续截图。")
             d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO)
             await asyncio.sleep(WAIT_AFTER_SCROLL)
 
 
+async def run_batch_local():
+    base_dir = os.path.dirname(os.path.abspath(__file__))
+    logger.info(f"批量处理目录中的 JPG 文件: {base_dir}")
+    await run_batch_in_dir(base_dir, log_path=LOG_FILE_PATH)
+
+
+async def run_single_image(image_path: str):
+    if not image_path:
+        logger.error("单图模式下需要提供图片路径。")
+        return
+    if not os.path.isabs(image_path):
+        image_path = os.path.abspath(image_path)
+    if not os.path.exists(image_path):
+        logger.error(f"指定的图片不存在: {image_path}")
+        return
+    logger.info(f"单图 OCR 测试，图片路径: {image_path}")
+    await run_ocr_rect(image_path, log_path=LOG_FILE_PATH)
+
+
 if __name__ == "__main__":
+    mode = "scroll"
+    image_path = None
+    if len(sys.argv) >= 2:
+        arg1 = sys.argv[1]
+        lower = arg1.lower()
+        if lower == "batch":
+            mode = "batch"
+        elif lower == "image":
+            mode = "image"
+            if len(sys.argv) >= 3:
+                image_path = sys.argv[2]
+        else:
+            if any(lower.endswith(ext) for ext in [".jpg", ".jpeg", ".png"]):
+                mode = "image"
+                image_path = arg1
     try:
-        asyncio.run(run())
+        if mode == "batch":
+            asyncio.run(run_batch_local())
+        elif mode == "image":
+            asyncio.run(run_single_image(image_path))
+        else:
+            asyncio.run(run_scroll_capture())
     except KeyboardInterrupt:
         logger.info("用户手动中断测试脚本。")
     except Exception as e:

Test/TestFirstPage/T_XinDianTu_VLRectTest.py

@@ -1,42 +0,0 @@
-import asyncio
-import os
-import sys
-
-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.XinDianTu.FirstPageKit import run_ocr_rect
-
-
-async def run_test(image_path, log_path=None):
-    await run_ocr_rect(image_path, log_path=log_path)
-
-
-async def run_batch():
-    base_dir = os.path.dirname(os.path.abspath(__file__))
-    files = sorted(os.listdir(base_dir))
-
-    jpg_files = []
-    for name in files:
-        lower = name.lower()
-        if not lower.endswith(".jpg"):
-            continue
-        if lower.endswith("_ocr_rect.jpg"):
-            continue
-        jpg_files.append(os.path.join(base_dir, name))
-
-    if not jpg_files:
-        print(f"目录中未找到待处理的 JPG 文件: {base_dir}")
-        return
-
-    print(f"即将批量处理 {len(jpg_files)} 张图片:")
-    for p in jpg_files:
-        print(f"  - {p}")
-
-    for p in jpg_files:
-        await run_test(p)
-
-
-if __name__ == "__main__":
-    asyncio.run(run_batch())