142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
import asyncio
|
||
import json
|
||
import os
|
||
import time
|
||
|
||
import uiautomator2 as u2
|
||
|
||
from Apps.XinDianTu.Kit import setup_logger, take_screenshot, crop_cards_from_image, clear_temp_dir
|
||
from Apps.XinDianTu.ReadImageKit import ReadImageKit
|
||
from Apps.XinDianTu.Config.Setting import SCROLL_DISTANCE_RATIO
|
||
from Config.Config import TEMP_IMAGE_DIR
|
||
|
||
|
||
logger = setup_logger("XinDianTu.ListDebug")
|
||
|
||
|
||
async def _debug_from_list_image(d, device_info, list_image_path: str, step_prefix: str):
|
||
if not os.path.exists(list_image_path):
|
||
logger.error(f"[{step_prefix}] 列表截图不存在: {list_image_path}")
|
||
return
|
||
|
||
logger.info(f"[{step_prefix}] 使用列表截图: {list_image_path}")
|
||
|
||
crop_cards_from_image(list_image_path)
|
||
|
||
json_path = list_image_path.replace(".jpg", ".json")
|
||
vl_img_path = list_image_path.replace(".jpg", "_vl.jpg")
|
||
|
||
if not os.path.exists(json_path) or not os.path.exists(vl_img_path):
|
||
logger.error(f"[{step_prefix}] 未生成 JSON 或 _vl 调试图: {json_path}, {vl_img_path}")
|
||
return
|
||
|
||
logger.info(f"[{step_prefix}] 绿框调试图路径: {vl_img_path}")
|
||
|
||
with open(json_path, "r", encoding="utf-8") as f:
|
||
json_metadata = json.load(f)
|
||
|
||
cards = json_metadata.get("cards") or []
|
||
if not cards:
|
||
logger.error(f"[{step_prefix}] JSON 中未找到任何卡片元数据")
|
||
return
|
||
|
||
stations = await ReadImageKit.parse_vl_image(vl_img_path, json_metadata, device_info=device_info)
|
||
if not stations:
|
||
logger.error(f"[{step_prefix}] VL 未识别出任何场站")
|
||
return
|
||
|
||
target_station = None
|
||
for st in stations:
|
||
if st and st.get("station_name"):
|
||
target_station = st
|
||
break
|
||
|
||
if not target_station:
|
||
logger.error(f"[{step_prefix}] 未找到带有效名称的场站")
|
||
return
|
||
|
||
station_name = target_station.get("station_name")
|
||
click_x = target_station.get("uia_center_x")
|
||
click_y = target_station.get("uia_center_y")
|
||
|
||
if click_x is None or click_y is None:
|
||
logger.error(f"[{step_prefix}] 目标场站缺少坐标信息: {target_station}")
|
||
return
|
||
|
||
logger.info(
|
||
f"[{step_prefix}] 准备点击场站: '{station_name}', 点击坐标: ({click_x}, {click_y})"
|
||
)
|
||
|
||
d.click(int(click_x), int(click_y))
|
||
|
||
await asyncio.sleep(2.0)
|
||
|
||
detail_uuid = f"xdt_debug_{step_prefix}_detail_{int(time.time())}"
|
||
detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR)
|
||
logger.info(f"[{step_prefix}] 二级详情页截图路径: {detail_path}")
|
||
|
||
address_data = await ReadImageKit.parse_address(station_name, detail_path, device_info=device_info)
|
||
full_name = address_data.get("full_station_name", "")
|
||
address = address_data.get("address", "")
|
||
|
||
logger.info(
|
||
f"[{step_prefix}] VLM 地址识别结果:"
|
||
f" full_station_name='{full_name}', address='{address}'"
|
||
)
|
||
|
||
d.press("back")
|
||
await asyncio.sleep(1.0)
|
||
|
||
|
||
async def run_debug():
|
||
logger.info("=== 新电途 列表页兔子与地址识别调试脚本启动 ===")
|
||
logger.info("请先手动打开微信并进入新电途小程序的场站列表第一页。")
|
||
|
||
clear_temp_dir()
|
||
|
||
d = u2.connect()
|
||
w, h = d.window_size()
|
||
|
||
device_info = d.info or {}
|
||
device_info["width"] = w
|
||
device_info["height"] = h
|
||
|
||
logger.info(f"当前设备: {device_info.get('productName')} | 分辨率: {w}x{h}")
|
||
|
||
image_uuid_1 = f"xdt_debug_step1_{int(time.time())}"
|
||
list_path_1 = take_screenshot(d, image_uuid_1, save_dir=TEMP_IMAGE_DIR)
|
||
logger.info(f"[step1] 列表第一页原始截图路径: {list_path_1}")
|
||
|
||
await _debug_from_list_image(d, device_info, list_path_1, step_prefix="step1")
|
||
|
||
logger.info("[step2] 向上滑动，让兔子躲起来，然后 1 秒后截图...")
|
||
d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO)
|
||
await asyncio.sleep(1.0)
|
||
|
||
image_uuid_2 = f"xdt_debug_step2_{int(time.time())}"
|
||
list_path_2 = take_screenshot(d, image_uuid_2, save_dir=TEMP_IMAGE_DIR)
|
||
logger.info(f"[step2] 滑动后 1 秒截图(预计兔子已躲藏): {list_path_2}")
|
||
|
||
await _debug_from_list_image(d, device_info, list_path_2, step_prefix="step2")
|
||
|
||
logger.info("[step3] 再次向上滑动，进入第3页列表，然后 1 秒后截图...")
|
||
d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO)
|
||
await asyncio.sleep(1.0)
|
||
|
||
image_uuid_3 = f"xdt_debug_step3_{int(time.time())}"
|
||
list_path_3 = take_screenshot(d, image_uuid_3, save_dir=TEMP_IMAGE_DIR)
|
||
logger.info(f"[step3] 第3页列表截图路径: {list_path_3}")
|
||
|
||
await _debug_from_list_image(d, device_info, list_path_3, step_prefix="step3")
|
||
|
||
logger.info("=== 调试脚本完成：已按步骤执行三次进入详情并做地址识别，已包含第3页 ===")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
asyncio.run(run_debug())
|
||
except KeyboardInterrupt:
|
||
logger.info("用户手动中断调试脚本。")
|
||
except Exception as e:
|
||
logger.exception(f"调试脚本运行异常: {e}")
|