diff --git a/Apps/XinDianTu/Config/__pycache__/Setting.cpython-310.pyc b/Apps/XinDianTu/Config/__pycache__/Setting.cpython-310.pyc new file mode 100644 index 0000000..23fe7e2 Binary files /dev/null and b/Apps/XinDianTu/Config/__pycache__/Setting.cpython-310.pyc differ diff --git a/Apps/XinDianTu/Crawler.py b/Apps/XinDianTu/Crawler.py index 2d6e428..c3cb9e6 100644 --- a/Apps/XinDianTu/Crawler.py +++ b/Apps/XinDianTu/Crawler.py @@ -16,23 +16,26 @@ if project_root not in sys.path: import uiautomator2 as u2 from Core.BaseCrawler import BaseCrawler -from Util import Kit -from Util.Kit import take_screenshot +from Apps.XinDianTu import Kit +from Apps.XinDianTu.Kit import take_screenshot from Util.ObsUtil import ObsUploader from Util.RedisKit import RedisKit from Util.PaddleOCRKit import get_ocr_kit import cv2 from Apps.XinDianTu.Service import XinDianTuService from Config.Config import ( - OBS_TMP_PREFIX, CDN_DOMAIN, SCROLL_DISTANCE_RATIO, + OBS_TMP_PREFIX, CDN_DOMAIN, TEMP_IMAGE_DIR +) +from Apps.XinDianTu.Config.Setting import ( + SCROLL_DISTANCE_RATIO, MAX_SCROLLS, REDIS_STATION_EXPIRE, WAIT_DETAIL_PAGE_LOAD, WAIT_BACK_TO_LIST, WAIT_AFTER_SCROLL, - MAX_CRAWL_DISTANCE, TEMP_IMAGE_DIR + MAX_CRAWL_DISTANCE ) # --- 用户配置区域 --- # 是否保留截图文件（True=保留备查，False=随用随删） -KEEP_SCREENSHOTS = True +KEEP_SCREENSHOTS = True # [Testing] 是否在启动时清除 Redis 中的场站处理记录 # True: 每次运行前清除记录，方便反复测试同一个场站 @@ -57,14 +60,15 @@ logger = logging.getLogger("StationList") # 强制设置所有相关模块的日志级别为 INFO，防止被第三方库干扰 logging.getLogger("PaddleOCRKit").setLevel(logging.INFO) -logging.getLogger("Util.Kit").setLevel(logging.INFO) logging.getLogger("OpenXinDianTu").setLevel(logging.INFO) logging.getLogger("FullProcess").setLevel(logging.INFO) + class XinDianTuCrawler(BaseCrawler): """ 新电途小程序爬虫实现 """ + def __init__(self, service=None): super().__init__(service) # 初始化配置参数 @@ -90,6 +94,7 @@ class XinDianTuCrawler(BaseCrawler): async def crawl_detail(self, station_info): pass + async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): """ 获取场站列表并处理翻页 @@ -97,21 +102,21 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): all_stations = [] seen_names = set() redis_kit = RedisKit() - + # 确保在“找桩”或者首页列表界面 window_size = d.window_size() w, h = window_size[0], window_size[1] - + # 获取设备信息以用于坐标计算 device_info = d.info device_info['width'] = w device_info['height'] = h - + logger.info(f"设备信息: {device_info.get('productName')} | 窗口分辨率: {w}x{h}") - + max_distance_reached = False stop_reason = "max_scrolls_reached" # 默认停止原因：达到最大翻页次数 - + # 存储后台异步任务 background_tasks = [] @@ -121,7 +126,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): stop_reason = "max_distance_reached" break logger.info(f"正在处理第 {i + 1} 页...") - + # 1. 拍摄截图 logger.info(f"Step [1/6] 正在拍摄列表页截图...") t_shot = time.time() @@ -131,7 +136,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): save_dir = TEMP_IMAGE_DIR screenshot_path = take_screenshot(d, image_uuid, save_dir=save_dir) logger.info(f"Step [1/6] 列表页截图已完成: {screenshot_path} (耗时: {time.time() - t_shot:.2f}s)") - + # 校验截图实际尺寸 img_w, img_h = 0, 0 try: @@ -140,115 +145,119 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): logger.info(f"截图实际尺寸: {img_w}x{img_h}") # 如果截图尺寸与窗口尺寸不符，更新 device_info 以适配坐标换算 if img_w != w or img_h != h: - logger.warning(f"检测到截图尺寸 ({img_w}x{img_h}) 与窗口尺寸 ({w}x{h}) 不一致，将以截图尺寸为准进行坐标换算。") + logger.warning( + f"检测到截图尺寸 ({img_w}x{img_h}) 与窗口尺寸 ({w}x{h}) 不一致，将以截图尺寸为准进行坐标换算。") device_info['width'] = img_w device_info['height'] = img_h except Exception as e: logger.warning(f"无法获取截图尺寸信息: {e}") - + # 2. 使用 Kit.crop_cards_from_image 进行切片和生成 _vl.jpg logger.info("Step [2/6] 正在执行本地图形学切片分析，识别场站卡片并生成绿色方框图...") t_crop = time.time() # 这会生成 _vl.jpg 和 .json Kit.crop_cards_from_image(screenshot_path) logger.info(f"Step [2/6] 图形学切片分析完成 (耗时: {time.time() - t_crop:.2f}s)") - + # 读取生成的元数据 json_path = screenshot_path.replace(".jpg", ".json") vl_img_path = screenshot_path.replace(".jpg", "_vl.jpg") - + if not os.path.exists(json_path) or not os.path.exists(vl_img_path): - logger.error(f"❌ Step [2/6] 失败: 未生成有效的 JSON 或 VL 图片 (JSON: {os.path.exists(json_path)}, VL: {os.path.exists(vl_img_path)})") + logger.error( + f"❌ Step [2/6] 失败: 未生成有效的 JSON 或 VL 图片 (JSON: {os.path.exists(json_path)}, VL: {os.path.exists(vl_img_path)})") logger.warning("可能未识别到任何卡片，跳过当前页") continue - + logger.info(f"Step [3/6] 绿色方框图已生成: {vl_img_path}") - + with open(json_path, 'r', encoding='utf-8') as f: json_metadata = json.load(f) - + # --- New Logic: Local OCR via PaddleOCRKit --- logger.info("Step [4/6] 正在开始本地 OCR 识别流程，将识别出的卡片逐一交给 OCR...") t_ocr_total = time.time() - + stations = [] try: ocr_kit = get_ocr_kit() - + # 读取原始截图进行裁剪 (比读取 _vl.jpg 更干净，没有绿框干扰) original_img = cv2.imread(screenshot_path) - + if "cards" in json_metadata and original_img is not None: h_img, w_img = original_img.shape[:2] num_cards = len(json_metadata['cards']) logger.info(f"检测到 {num_cards} 个场站卡片，准备开始逐一 OCR 识别...") - + for idx, card in enumerate(json_metadata["cards"]): # card: {"id": 1, "rect": [x1, y1, x2, y2], "click_point": [cx, cy]} rect = card.get("rect") if not rect: continue x1, y1, x2, y2 = rect - + # 边界检查 x1 = max(0, min(x1, w_img)) x2 = max(0, min(x2, w_img)) y1 = max(0, min(y1, h_img)) y2 = max(0, min(y2, h_img)) - + if x2 <= x1 or y2 <= y1: continue - + # 裁剪卡片 cropped_card = original_img[y1:y2, x1:x2] - + # 识别 - logger.info(f"正在识别第 {idx+1}/{num_cards} 个卡片 (区域: {rect})...") + logger.info(f"正在识别第 {idx + 1}/{num_cards} 个卡片 (区域: {rect})...") t_card_start = time.time() parsed_data = ocr_kit.recognize(cropped_card) t_card_end = time.time() - + # Log detailed OCR result - logger.info(f"卡片 {idx+1} OCR 识别耗时: {t_card_end - t_card_start:.2f}s") - + logger.info(f"卡片 {idx + 1} OCR 识别耗时: {t_card_end - t_card_start:.2f}s") + if parsed_data and parsed_data.get("station_name"): - # 格式化数据以匹配原有结构 - piles_list = parsed_data.get("piles", []) - piles_str_parts = [] - for p in piles_list: - p_type = p.get("type", "") - p_free = p.get("free", 0) - p_total = p.get("total", 0) - piles_str_parts.append(f"{p_type}:{p_free}/{p_total}") - - piles_str = " ".join(piles_str_parts) - - station_info = { - "station_name": parsed_data.get("station_name"), - "price": str(parsed_data.get("price")) if parsed_data.get("price") is not None else "", - "pro_price": "", # 暂未解析 - "piles": piles_str, - "distance": parsed_data.get("distance", ""), - "uia_center_x": card["click_point"][0], - "uia_center_y": card["click_point"][1], - "tags": parsed_data.get("tags", []), - "parking_info": parsed_data.get("parking", "") - } - logger.info(f"✅ 成功解析场站: {station_info['station_name']} | 距离: {station_info['distance']}") - - # 立即记录场站基础状态信息 (电桩、价格等) - try: - await service.record_station_status(station_info) - except Exception as e: - logger.error(f"❌ 记录场站状态失败: {e}") - - stations.append(station_info) + # 格式化数据以匹配原有结构 + piles_list = parsed_data.get("piles", []) + piles_str_parts = [] + for p in piles_list: + p_type = p.get("type", "") + p_free = p.get("free", 0) + p_total = p.get("total", 0) + piles_str_parts.append(f"{p_type}:{p_free}/{p_total}") + + piles_str = " ".join(piles_str_parts) + + station_info = { + "station_name": parsed_data.get("station_name"), + "price": str(parsed_data.get("price")) if parsed_data.get("price") is not None else "", + "pro_price": "", # 暂未解析 + "piles": piles_str, + "distance": parsed_data.get("distance", ""), + "uia_center_x": card["click_point"][0], + "uia_center_y": card["click_point"][1], + "tags": parsed_data.get("tags", []), + "parking_info": parsed_data.get("parking", "") + } + logger.info( + f"✅ 成功解析场站: {station_info['station_name']} | 距离: {station_info['distance']}") + + # 立即记录场站基础状态信息 (电桩、价格等) + try: + await service.record_station_status(station_info) + except Exception as e: + logger.error(f"❌ 记录场站状态失败: {e}") + + stations.append(station_info) else: - logger.warning(f"⚠️ 卡片 {idx+1} 未能识别出有效场站名称") - + logger.warning(f"⚠️ 卡片 {idx + 1} 未能识别出有效场站名称") + except Exception as e: logger.error(f"❌ Local OCR processing failed: {e}", exc_info=True) - - logger.info(f"Step [5/6] 本地 OCR 解析完成 (总耗时: {time.time() - t_ocr_total:.2f}s) | 成功识别: {len(stations)}/{len(json_metadata.get('cards', []))}") - + + logger.info( + f"Step [5/6] 本地 OCR 解析完成 (总耗时: {time.time() - t_ocr_total:.2f}s) | 成功识别: {len(stations)}/{len(json_metadata.get('cards', []))}") + if not stations: logger.warning("当前页面未检测到任何有效的场站信息") else: @@ -256,25 +265,25 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): new_stations_found = False for s_idx, s in enumerate(stations): name = s.get("station_name") - + # 过滤掉明显的非场站名称 if not name: continue - + # --- 短期去重：检查 Redis 缓存 --- # 2分钟内如果处理过该场站，则跳过 redis_key = f"processed_station:{name}" if await redis_kit.exists(redis_key): logger.info(f"场站 {name} 在 2 分钟内已处理过，跳过") continue - + if name in seen_names: continue # 使用从 JSON 元数据中恢复的点击坐标 x = s.get("uia_center_x") y = s.get("uia_center_y") - + if x is None or y is None: logger.warning(f"场站 {name} 缺少坐标信息，无法进入详情页") continue @@ -282,7 +291,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): seen_names.add(name) all_stations.append(s) new_stations_found = True - + # 打印基本信息 price = s.get("price") pro_price = s.get("pro_price") @@ -299,20 +308,22 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): dist_val = float(dist_match.group(1)) is_km = "km" in distance_str.lower() dist_km = dist_val if is_km else dist_val / 1000.0 - + # 将距离解析结果和限制检查合并到日志中 dist_log_part = f" (解析: {dist_km:.2f}km / 限: {MAX_CRAWL_DISTANCE}km)" if dist_km >= MAX_CRAWL_DISTANCE: - logger.info(f"当前场站 {name} 距离 {dist_km}km (原始: {distance_str}) 已达到或超过限制 {MAX_CRAWL_DISTANCE}km。") + logger.info( + f"当前场站 {name} 距离 {dist_km}km (原始: {distance_str}) 已达到或超过限制 {MAX_CRAWL_DISTANCE}km。") max_distance_reached = True except Exception as e: logger.warning(f"解析距离出错: {distance_str}, {e}") - + pro_info = f" | PRO会员价: {pro_price}" if pro_price else "" # 确保距离始终显示，如果为空则显示 "未知" display_distance = distance_str if distance_str else "未知" - logger.info(f"正在处理第 {s_idx+1} 个场站: {name} | 价格: {price}{pro_info} | 枪: {piles_str} | 距离: {display_distance}{dist_log_part}") + logger.info( + f"正在处理第 {s_idx + 1} 个场站: {name} | 价格: {price}{pro_info} | 枪: {piles_str} | 距离: {display_distance}{dist_log_part}") if max_distance_reached: break @@ -328,7 +339,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): logger.info(f"👉 正在点击进入详情页: {name} (绝对坐标: {x}, {y})") d.click(x, y) - await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 等待详情页加载 + await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 等待详情页加载 # 尝试获取并保存地址 address_recorded = False @@ -338,21 +349,23 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): detail_uuid = str(uuid.uuid4()) detail_screenshot_path = take_screenshot(d, detail_uuid, save_dir=save_dir) logger.info(f"Step [详情页截图] 耗时: {time.time() - t_d_shot:.4f}s") - + # 2. 上传至 OBS t_d_up = time.time() detail_object_key = f"{OBS_TMP_PREFIX}/{detail_uuid}.jpg" up_success, _ = uploader.upload_file(detail_object_key, detail_screenshot_path) logger.info(f"Step [上传详情页截图] 耗时: {time.time() - t_d_up:.4f}s") - + if up_success: detail_url = f"https://{CDN_DOMAIN}/{detail_object_key}" # 3. 解析并保存地址 (异步后台处理) + # 定义异步任务函数 async def _process_address_task(task_name, task_url, task_dev_info, task_redis_key): t_addr_task = time.time() try: - addr_res = await service.process_station_address(task_name, task_url, device_info=task_dev_info) + addr_res = await service.process_station_address(task_name, task_url, + device_info=task_dev_info) logger.info(f"Step [解析并保存地址-后台] 耗时: {time.time() - t_addr_task:.4f}s") if addr_res.get("address"): logger.info(f"成功记录地址: {addr_res.get('address')}") @@ -368,40 +381,41 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): # 标记为已记录以跳出 retry 循环 address_recorded = True - + # 清理详情页临时截图 if not KEEP_SCREENSHOTS and os.path.exists(detail_screenshot_path): os.remove(detail_screenshot_path) - - break # 直接跳出 retry 循环 + + break # 直接跳出 retry 循环 # --- 抓取价格时段信息 (New Feature) --- # 仅在测试模式下或特定需求下启用 try: logger.info(f"开始抓取价格时段信息: {name}") - + # 使用几何特征识别 "全部时段" 按钮 # 临时截图 temp_uuid = "temp_find_expand" screenshot_path = take_screenshot(d, temp_uuid, save_dir=TEMP_IMAGE_DIR) - + # 尝试识别，将调试图片保存到 Images 目录 t_find = time.time() - pos = Kit.find_expand_button_position(screenshot_path, debug_dir=save_dir, debug_filename_prefix=name) + pos = Kit.find_expand_button_position(screenshot_path, debug_dir=save_dir, + debug_filename_prefix=name) logger.info(f"Step [识别展开按钮] 耗时: {time.time() - t_find:.4f}s") - + # 清理截图 try: os.remove(screenshot_path) except: pass - + if pos: x, y = pos logger.info(f"通过几何特征找到 '全部时段' 按钮: ({x}, {y})") d.click(x, y) await asyncio.sleep(1.5) - + # 抓取3屏截图 price_image_urls = [] for p_idx in range(1, 4): @@ -411,7 +425,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): p_uuid = f"{name_md5}_price_{p_idx}" p_path = take_screenshot(d, p_uuid, save_dir=save_dir) logger.info(f"已保存价格时段截图 {p_idx} (Station: {name}, MD5: {name_md5}): {p_path}") - + # 上传到 OBS p_object_key = f"{OBS_TMP_PREFIX}/{p_uuid}.jpg" success, _ = uploader.upload_file(p_object_key, p_path) @@ -427,15 +441,16 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): if p_idx < 3: d.swipe_ext("up", scale=0.7) await asyncio.sleep(1.0) - + # 调用服务解析价格时段 (异步后台处理) if price_image_urls: logger.info(f"正在调用接口解析电费时段数据 ({len(price_image_urls)} 张图片) - 转入后台...") - + async def _process_price_task(task_name, task_urls, task_dev_info): t_price_task = time.time() try: - await service.process_price_schedule(task_name, task_urls, device_info=task_dev_info) + await service.process_price_schedule(task_name, task_urls, + device_info=task_dev_info) logger.info(f"Step [解析电费时段数据-后台] 耗时: {time.time() - t_price_task:.4f}s") logger.info(f"电费时段数据解析完成: {task_name}") except Exception as e: @@ -447,7 +462,7 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): logger.warning("未能获取到有效的价格时段截图，跳过解析") else: logger.warning(f"未能通过几何特征识别 '全部时段' 按钮") - + except Exception as e: logger.warning(f"抓取价格时段信息时发生异常: {e}") @@ -457,8 +472,8 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): # 返回列表页 d.press("back") - await asyncio.sleep(WAIT_BACK_TO_LIST) # 等待列表页重新稳定 - + await asyncio.sleep(WAIT_BACK_TO_LIST) # 等待列表页重新稳定 + if not new_stations_found and i > 0: logger.info("未发现更多新场站，停止下拉。") # 如果需要强行爬取所有，可以注释掉 break @@ -472,17 +487,17 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): start_y = 0.5 + (SCROLL_DISTANCE_RATIO / 2) end_y = 0.5 - (SCROLL_DISTANCE_RATIO / 2) d.swipe(w * 0.5, h * start_y, w * 0.5, h * end_y, duration=0.5) - await asyncio.sleep(WAIT_AFTER_SCROLL) # 等待页面加载和列表稳定 - + await asyncio.sleep(WAIT_AFTER_SCROLL) # 等待页面加载和列表稳定 + # 清理列表页截图 if not KEEP_SCREENSHOTS: if os.path.exists(screenshot_path): - os.remove(screenshot_path) + os.remove(screenshot_path) if os.path.exists(vl_img_path): - os.remove(vl_img_path) + os.remove(vl_img_path) if os.path.exists(json_path): - os.remove(json_path) - + os.remove(json_path) + logger.info(f"爬取任务结束，正在等待 {len(background_tasks)} 个后台处理任务完成...") if background_tasks: await asyncio.gather(*background_tasks) @@ -491,22 +506,23 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): logger.info(f"任务结束，共采集到 {len(all_stations)} 个场站。停止原因: {stop_reason}") return all_stations + def clean_images_dir(): """清理 Images 目录下除保留文件外的所有文件""" base_dir = os.path.dirname(os.path.abspath(__file__)) images_dir = os.path.join(base_dir, "Images") - + if not os.path.exists(images_dir): return keep_files = {'1.jpg', '2.jpg', '3.jpg', '4.jpg'} logger.info(f"正在清理 Images 目录 (保留: {keep_files})...") - + deleted_count = 0 for filename in os.listdir(images_dir): if filename in keep_files: continue - + file_path = os.path.join(images_dir, filename) try: if os.path.isfile(file_path): @@ -514,9 +530,10 @@ def clean_images_dir(): deleted_count += 1 except Exception as e: logger.warning(f"删除文件失败 {filename}: {e}") - + logger.info(f"清理完成，共删除 {deleted_count} 个文件。") + async def clean_redis_data(redis_kit): """清除 Redis 中的场站处理记录""" if TEST_CLEAR_REDIS: @@ -528,10 +545,11 @@ async def clean_redis_data(redis_kit): else: logger.info("未发现旧的场站处理记录") + async def main(service=None, do_cleanup=True): # 连接设备 d = u2.connect() - + # 初始化服务 should_close_service = False if service is None: @@ -542,28 +560,28 @@ async def main(service=None, do_cleanup=True): uploader = ObsUploader() redis_kit = RedisKit() - + try: if do_cleanup: # 清理图片目录 clean_images_dir() - + # [Testing] 如果配置为测试模式，启动时清除 Redis 记录 await clean_redis_data(redis_kit) # 清理过期数据 # await service.cleanup_old_data() - + # 获取场站列表 stations = await get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS) - + if stations: logger.info("场站列表采集完成。") else: logger.warning("未采集到任何场站信息。") - + return True - + except Exception as e: logger.exception(f"运行过程中出现异常: {e}") return False @@ -572,6 +590,7 @@ async def main(service=None, do_cleanup=True): if should_close_service: await service.close_db() + if __name__ == "__main__": try: asyncio.run(main()) diff --git a/Apps/XinDianTu/Kit.py b/Apps/XinDianTu/Kit.py index 67aa0cd..2d5cdb7 100644 --- a/Apps/XinDianTu/Kit.py +++ b/Apps/XinDianTu/Kit.py @@ -3,11 +3,13 @@ import os import cv2 import numpy as np import time -from Config.Config import BOTTOM_SAFE_EXCLUDE_RATIO, TEMP_IMAGE_DIR +from Apps.XinDianTu.Config.Setting import BOTTOM_SAFE_EXCLUDE_RATIO +from Config.Config import TEMP_IMAGE_DIR logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) + def read_image(path): """读取图片，支持中文路径""" try: @@ -16,6 +18,7 @@ def read_image(path): logger.info(f"Error reading image {path}: {e}") return None + def save_image(path, img): """保存图片，支持中文路径""" try: @@ -28,13 +31,15 @@ def save_image(path, img): logger.error(f"Error saving image {path}: {e}") return False + # 截图 -def take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR): - path = f"{save_dir}/{image_uuid}.jpg" +def take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR): + path = f"{save_dir}/{image_uuid}.jpg" os.makedirs(save_dir, exist_ok=True) - d.screenshot(path) + d.screenshot(path) return path + def click_image_template(d, template_path, timeout=5.0, threshold=0.8): """ 使用 OpenCV 模板匹配查找并点击图片 @@ -52,22 +57,22 @@ def click_image_template(d, template_path, timeout=5.0, threshold=0.8): if template is None: logger.info(f"Failed to load template: {template_path}") return False - + t_h, t_w = template.shape[:2] - + start_time = time.time() best_val_overall = 0.0 - + while time.time() - start_time < timeout: # 临时截图 temp_uuid = "temp_click_check" screenshot_path = take_screenshot(d, temp_uuid, save_dir=TEMP_IMAGE_DIR) - + target = read_image(screenshot_path) if target is None: time.sleep(0.5) continue - + # 多尺度匹配 found = None # 缩放比例从 0.5 到 1.5，步长 0.1 @@ -75,43 +80,44 @@ def click_image_template(d, template_path, timeout=5.0, threshold=0.8): # 调整模板大小 resized_template = cv2.resize(template, (int(t_w * scale), int(t_h * scale))) r_h, r_w = resized_template.shape[:2] - + # 如果模板比目标还大，跳过 if r_h > target.shape[0] or r_w > target.shape[1]: continue - + result = cv2.matchTemplate(target, resized_template, cv2.TM_CCOEFF_NORMED) min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result) - + if max_val > best_val_overall: best_val_overall = max_val - + if max_val >= threshold: found = (max_val, max_loc, r_w, r_h) - break # 找到满足阈值的即停止当前截图的搜索 - + break # 找到满足阈值的即停止当前截图的搜索 + # 清理临时文件 try: os.remove(screenshot_path) except: pass - + if found: max_val, max_loc, r_w, r_h = found # 计算中心点 top_left = max_loc center_x = top_left[0] + r_w // 2 center_y = top_left[1] + r_h // 2 - + logger.info(f"Found image at ({center_x}, {center_y}) with confidence {max_val:.2f}") d.click(center_x, center_y) return True - + time.sleep(1.0) - + logger.info(f"Image not found after {timeout}s (Best confidence: {best_val_overall:.2f})") return False + def detect_black_agree_button(image_path, debug_dir=None): """ 通过计算机图形学检测黑色的"同意"按钮 (Image 1 场景) @@ -121,79 +127,80 @@ def detect_black_agree_button(image_path, debug_dir=None): """ if not os.path.exists(image_path): return None - + img = read_image(image_path) if img is None: return None - + h, w = img.shape[:2] - + # 转换为HSV颜色空间，因为黑色更容易过滤 hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV) - + # 定义黑色的范围 # 黑色：V (Brightness) 很低 lower_black = np.array([0, 0, 0]) - upper_black = np.array([180, 255, 40]) # V < 40 认为是黑色 - + upper_black = np.array([180, 255, 40]) # V < 40 认为是黑色 + mask = cv2.inRange(hsv, lower_black, upper_black) - + # 限制搜索区域：通常在屏幕下半部分 roi_top = int(h * 0.4) roi_bottom = int(h * 0.8) roi_mask = np.zeros_like(mask) roi_mask[roi_top:roi_bottom, :] = mask[roi_top:roi_bottom, :] - + # 形态学操作：去除噪点，连接断开的区域 kernel = np.ones((5, 5), np.uint8) roi_mask = cv2.morphologyEx(roi_mask, cv2.MORPH_CLOSE, kernel) roi_mask = cv2.morphologyEx(roi_mask, cv2.MORPH_OPEN, kernel) - + # 查找轮廓 contours, _ = cv2.findContours(roi_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) - + best_cnt = None max_area = 0 - + for cnt in contours: area = cv2.contourArea(cnt) x, y, cw, ch = cv2.boundingRect(cnt) aspect_ratio = float(cw) / ch - + # 过滤条件 # 1. 面积要够大 (例如 > 屏幕面积的 1%) if area < (w * h * 0.01): continue - + # 2. 宽高比：通常按钮是扁长条，例如 > 2.0 if aspect_ratio < 2.0 or aspect_ratio > 10.0: continue - + # 3. 宽度：通常占据屏幕宽度的 50% 以上 if cw < (w * 0.5): continue - + if area > max_area: max_area = area best_cnt = cnt - + if best_cnt is not None: x, y, cw, ch = cv2.boundingRect(best_cnt) center_x = x + cw // 2 center_y = y + ch // 2 logger.info(f"Found Black Agree Button at ({center_x}, {center_y}), Size: {cw}x{ch}") - + if debug_dir: os.makedirs(debug_dir, exist_ok=True) debug_img = img.copy() - cv2.rectangle(debug_img, (x, y), (x+cw, y+ch), (0, 0, 255), 2) + cv2.rectangle(debug_img, (x, y), (x + cw, y + ch), (0, 0, 255), 2) cv2.circle(debug_img, (center_x, center_y), 5, (0, 255, 0), -1) save_image(os.path.join(debug_dir, "debug_agree_btn.jpg"), debug_img) - + return (center_x, center_y) - + return None + def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7): """ 通过模板匹配检测"关闭(X)"按钮 (Image 2 场景) @@ -206,139 +213,145 @@ def detect_ad_close_x(image_path, template_path, debug_dir=None, threshold=0.7): """ if not os.path.exists(image_path): return None - + if not os.path.exists(template_path): logger.warning(f"Template not found: {template_path}") return None target = read_image(image_path) template = read_image(template_path) - + if target is None or template is None: return None - + # 转换为灰度图进行匹配，减少颜色干扰 target_gray = cv2.cvtColor(target, cv2.COLOR_BGR2GRAY) template_gray = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY) - + t_h, t_w = template_gray.shape[:2] - + best_match = None - + # 多尺度匹配: 缩放模板 # 假设模板可能比实际大，也可能小。范围 0.5 - 1.5 scales = np.linspace(0.5, 1.5, 20) - + for scale in scales: # 计算缩放后的模板尺寸 new_w = int(t_w * scale) new_h = int(t_h * scale) - + # 确保缩放后的模板不大于目标图像 if new_w > target_gray.shape[1] or new_h > target_gray.shape[0]: continue - + resized_template = cv2.resize(template_gray, (new_w, new_h)) - + # 匹配 result = cv2.matchTemplate(target_gray, resized_template, cv2.TM_CCOEFF_NORMED) min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result) - + if best_match is None or max_val > best_match[0]: best_match = (max_val, max_loc, scale, new_w, new_h) - + if best_match is None: return None - + max_val, max_loc, best_scale, best_w, best_h = best_match - + # 获取目标图像尺寸 target_h, target_w = target_gray.shape[:2] - + top_left = max_loc center_x = top_left[0] + best_w // 2 center_y = top_left[1] + best_h // 2 - + # --- 位置启发式过滤 --- # 计算相对位置 rel_x = center_x / target_w rel_y = center_y / target_h - - logger.info(f"Ad Close Button Match: Confidence={max_val:.4f}, Scale={best_scale:.2f}, Pos=({center_x}, {center_y}), Rel=({rel_x:.2f}, {rel_y:.2f})") - + + logger.info( + f"Ad Close Button Match: Confidence={max_val:.4f}, Scale={best_scale:.2f}, Pos=({center_x}, {center_y}), Rel=({rel_x:.2f}, {rel_y:.2f})") + is_valid_pos = True - + # 规则1: 过滤掉屏幕正中央偏上的区域 (通常是广告标题、图标或内容) # 范围: X在 [0.3, 0.7] 且 Y在 [0.15, 0.5] if 0.3 < rel_x < 0.7 and 0.15 < rel_y < 0.5: logger.warning(f"Ignored match at ({center_x}, {center_y}) - likely Ad Content/Title (Center-Top area).") is_valid_pos = False - + # 规则2: 如果置信度不是特别高 (>0.9), 强制要求在典型区域 (右上角 或 底部中间) # 右上角: X > 0.7, Y < 0.5 # 底部中间: Y > 0.6 - if is_valid_pos and max_val < 0.98: # 如果置信度非常高(0.98+)，可能是极其标准的X，暂时放过(除非在禁区) + if is_valid_pos and max_val < 0.98: # 如果置信度非常高(0.98+)，可能是极其标准的X，暂时放过(除非在禁区) if not ((rel_x > 0.7 and rel_y < 0.5) or (rel_y > 0.6)): - logger.warning(f"Ignored match at ({center_x}, {center_y}) - not in typical Close Button regions (Top-Right or Bottom).") - is_valid_pos = False + logger.warning( + f"Ignored match at ({center_x}, {center_y}) - not in typical Close Button regions (Top-Right or Bottom).") + is_valid_pos = False if max_val >= threshold: if is_valid_pos: # [Safety Check] 底部安全区排除 # 如果检测到的关闭按钮位于屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 区域内，认为是误判（如误触底部功能按钮） if center_y > (target_h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): - logger.warning(f"Ignored Ad Close Button at ({center_x}, {center_y}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}%).") + logger.warning( + f"Ignored Ad Close Button at ({center_x}, {center_y}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}%).") # 也可以保存一下调试图 if debug_dir: os.makedirs(debug_dir, exist_ok=True) debug_img = target.copy() - cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 0, 128), 2) # Dark Red for Safety Ignored - cv2.putText(debug_img, f"SAFETY IGNORED", + cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 0, 128), + 2) # Dark Red for Safety Ignored + cv2.putText(debug_img, f"SAFETY IGNORED", (top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 128), 1) save_image(os.path.join(debug_dir, "debug_ad_close_safety_ignored.jpg"), debug_img) return None logger.info(f"Found Ad Close Button at ({center_x}, {center_y})") - + if debug_dir: os.makedirs(debug_dir, exist_ok=True) debug_img = target.copy() cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 0, 255), 2) cv2.circle(debug_img, (center_x, center_y), 5, (0, 255, 0), -1) - cv2.putText(debug_img, f"Conf: {max_val:.2f}, Sc: {best_scale:.2f}", + cv2.putText(debug_img, f"Conf: {max_val:.2f}, Sc: {best_scale:.2f}", (top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1) save_image(os.path.join(debug_dir, "debug_ad_close_x.jpg"), debug_img) - + return (center_x, center_y) else: - # 虽然置信度高，但是位置不对，保存为 False Positive 供调试 - if debug_dir: + # 虽然置信度高，但是位置不对，保存为 False Positive 供调试 + if debug_dir: os.makedirs(debug_dir, exist_ok=True) debug_img = target.copy() - cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 165, 255), 2) # Orange for ignored - cv2.putText(debug_img, f"IGNORED Pos", + cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 165, 255), + 2) # Orange for ignored + cv2.putText(debug_img, f"IGNORED Pos", (top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 165, 255), 1) save_image(os.path.join(debug_dir, "debug_ad_close_ignored.jpg"), debug_img) - + # 如果没找到，但有一定置信度，也保存一下调试图以便分析 if max_val > 0.4 and debug_dir: os.makedirs(debug_dir, exist_ok=True) debug_img = target.copy() top_left = max_loc cv2.rectangle(debug_img, top_left, (top_left[0] + best_w, top_left[1] + best_h), (0, 255, 255), 2) - cv2.putText(debug_img, f"Failed Conf: {max_val:.2f}", + cv2.putText(debug_img, f"Failed Conf: {max_val:.2f}", (top_left[0], top_left[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 255), 1) save_image(os.path.join(debug_dir, "debug_ad_close_fail.jpg"), debug_img) return None + def detect_any_ad_close(image_path, template_dir, debug_dir=None): """ 遍历模板目录下的所有 ad_close*.jpg 进行匹配 """ if not os.path.exists(template_dir): return None - + for filename in os.listdir(template_dir): if filename.startswith("ad_close") and filename.endswith(".jpg"): template_path = os.path.join(template_dir, filename) @@ -348,6 +361,7 @@ def detect_any_ad_close(image_path, template_dir, debug_dir=None): return pos return None + def detect_bottom_close_circle(image_path, debug_dir=None): """ 通过几何特征检测底部的圆形关闭按钮 (常见于插屏广告) @@ -359,79 +373,81 @@ def detect_bottom_close_circle(image_path, debug_dir=None): """ if not os.path.exists(image_path): return None - + img = read_image(image_path) if img is None: return None - + h, w = img.shape[:2] - + # 1. 提取感兴趣区域 (ROI): 屏幕底部 25% (缩小范围，避开列表内容区) roi_top = int(h * 0.75) roi_h = h - roi_top roi = img[roi_top:h, :] - + # 转灰度 gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY) # 高斯模糊降噪 gray_blurred = cv2.GaussianBlur(gray, (9, 9), 2) - + # 2. 霍夫圆变换检测圆形 # dp=1.2 (累加器分辨率), minDist=w/5 (圆心最小距离), param1=100 (Canny高阈值), param2=30 (圆心累加阈值), minR=w*0.04, maxR=w*0.1 # 缩小最大半径范围，防止识别到过大的按钮（如扫码充电按钮） - circles = cv2.HoughCircles(gray_blurred, cv2.HOUGH_GRADIENT, dp=1.2, minDist=w/5, - param1=100, param2=30, minRadius=int(w*0.04), maxRadius=int(w*0.09)) - + circles = cv2.HoughCircles(gray_blurred, cv2.HOUGH_GRADIENT, dp=1.2, minDist=w / 5, + param1=100, param2=30, minRadius=int(w * 0.04), maxRadius=int(w * 0.09)) + if circles is not None: circles = np.round(circles[0, :]).astype("int") - + best_circle = None min_dist_to_center = float('inf') - + for (cx, cy, r) in circles: # 还原到全图坐标 global_cy = roi_top + cy global_cx = cx - + # 过滤1: 必须在水平中心附近 (容差 15%，收紧限制) - if abs(global_cx - w//2) > (w * 0.15): + if abs(global_cx - w // 2) > (w * 0.15): continue - + # 过滤2: 垂直方向限制 (必须在屏幕 80% - 95% 之间) # 这样可以避开位于 60%-80% 区域的列表项卡片 rel_y = global_cy / h if rel_y < 0.80 or rel_y > 0.95: continue - - dist = abs(global_cx - w//2) + + dist = abs(global_cx - w // 2) if dist < min_dist_to_center: min_dist_to_center = dist best_circle = (global_cx, global_cy, r) - + if best_circle: cx, cy, r = best_circle # 确保转换为标准的 Python int，否则 uiautomator2 click 可能会报错 (JSON serializable error) cx, cy, r = int(cx), int(cy), int(r) - + # [Safety Check] 底部安全区排除 # 如果检测到的圆形按钮位于屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 区域内，认为是误判（如误触底部扫码充电等） if cy > (h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): - logger.warning(f"Ignored Bottom Circle at ({cx}, {cy}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}%).") + logger.warning( + f"Ignored Bottom Circle at ({cx}, {cy}) - in Bottom Safety Zone ({int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}%).") return None logger.info(f"Found Bottom Circle Button via Hough: ({cx}, {cy}), r={r}") - + if debug_dir: os.makedirs(debug_dir, exist_ok=True) debug_img = img.copy() cv2.circle(debug_img, (cx, cy), r, (0, 255, 0), 2) cv2.circle(debug_img, (cx, cy), 2, (0, 0, 255), 3) save_image(os.path.join(debug_dir, "debug_bottom_circle.jpg"), debug_img) - + return (cx, cy) - + return None + def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefix=None): """ 通过几何特征识别"全部时段"按钮的位置 @@ -443,101 +459,101 @@ def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefi """ if not os.path.exists(image_path): return None - + img = read_image(image_path) if img is None: return None - + h, w = img.shape[:2] - + # 转灰度 gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) - + # 边缘检测 edges = cv2.Canny(gray, 50, 150) - + # 定义区域 left_w = int(w * 0.35) right_w = int(w * 0.65) - + # 提取各部分边缘 left_part = edges[:, :left_w] right_part = edges[:, right_w:] center_part = edges[:, left_w:right_w] - + # 计算每一行的边缘点数量 row_sum_left = np.sum(left_part, axis=1) / 255.0 row_sum_right = np.sum(right_part, axis=1) / 255.0 row_sum_center = np.sum(center_part, axis=1) / 255.0 - + # 筛选条件：左右边缘点很少，中间边缘点较多 # 阈值可以根据实际情况调整 # 允许少量噪点，所以不是严格的0，而是小于某个较小值（例如宽度的1%） - noise_threshold = 2 # 允许2个像素的噪点 - content_threshold = 5 # 中间至少有5个像素的边缘 - + noise_threshold = 2 # 允许2个像素的噪点 + content_threshold = 5 # 中间至少有5个像素的边缘 + candidates = (row_sum_left <= noise_threshold) & \ (row_sum_right <= noise_threshold) & \ (row_sum_center >= content_threshold) - + # 找到连续的候选行 y_indices = np.where(candidates)[0] - + if len(y_indices) == 0: return None - + # 将连续行分组 segments = [] if len(y_indices) > 0: start = y_indices[0] prev = y_indices[0] for y in y_indices[1:]: - if y > prev + 5: # 允许5像素断裂 + if y > prev + 5: # 允许5像素断裂 segments.append((start, prev)) start = y prev = y segments.append((start, prev)) - + # 筛选最合适的段 best_segment = None - + # 我们期望按钮在屏幕中下部，且高度适中（例如 20-100px） # 且通常是在价格表下方。假设价格表占据了屏幕上部。 # 我们可以简单地取符合条件的段中，Y值最大的那个（最靠下的），或者最符合"中间有字"特征的。 # 考虑到页面底部可能有其他干扰，取"中下部"的一个。 - + valid_segments = [] for start, end in segments: height = end - start mid_y = (start + end) // 2 - + # 过滤掉太高或太矮的区域 if height < 20 or height > 150: continue - + # 过滤掉顶部的区域（可能是标题栏误判） if mid_y < h * 0.3: continue - + # 过滤掉底部的区域（可能是底部按钮） if mid_y > h * 0.9: continue - + valid_segments.append((start, end)) - + if not valid_segments: return None - + # 如果有多个，通常"全部时段"是在价格表之后，紧接着的一个 # 这里我们取第一个（最靠上的）符合条件的段，因为它紧跟在价格表下方 # 或者取所有段中，中间内容最"紧凑"的？ # 让我们简单点，取第一个符合条件的段。 best_segment = valid_segments[0] - + start, end = best_segment center_y = (start + end) // 2 center_x = w // 2 - + if debug_dir: os.makedirs(debug_dir, exist_ok=True) debug_img = img.copy() @@ -545,7 +561,7 @@ def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefi cv2.rectangle(debug_img, (0, start), (w, end), (0, 255, 0), 2) # 画出红点 cv2.circle(debug_img, (center_x, center_y), 10, (0, 0, 255), -1) - + # 保存 flag 图片 if debug_filename_prefix: # Sanitize filename: remove invalid chars @@ -555,11 +571,11 @@ def find_expand_button_position(image_path, debug_dir=None, debug_filename_prefi else: timestamp = time.strftime("%Y%m%d_%H%M%S") debug_name = f"{timestamp}_flag_expand.jpg" - + debug_path = os.path.join(debug_dir, debug_name) save_image(debug_path, debug_img) logger.info(f"Saved debug image to {debug_path}") - + return (int(center_x), int(center_y)) @@ -568,22 +584,23 @@ def get_row_stats(gray): 计算每一行的统计特征 """ h, w = gray.shape - + # 中央区域 (用于检测内容) center_x = w // 2 strip_w = 100 - center_strip = gray[:, center_x - 50 : center_x + 50] - + center_strip = gray[:, center_x - 50: center_x + 50] + # 边缘区域 (用于检测背景/边距) # 假设边距至少有 10px edge_strip = gray[:, 0:20] - + row_means = np.mean(center_strip, axis=1) row_stds = np.std(center_strip, axis=1) edge_means = np.mean(edge_strip, axis=1) - + return row_means, row_stds, edge_means + def crop_cards_from_image(img_path, output_dir=None, save_debug=True): """ 从图片中裁剪场站卡片 @@ -604,53 +621,53 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) h, w = gray.shape - + row_means, row_stds, edge_means = get_row_stats(gray) - + # 参数定义 # 背景灰度值范围 (根据 Analyze2.py 的输出，背景约 242) BG_GRAY_MIN = 230 BG_GRAY_MAX = 250 - + # 间隙判定：中央区域也是均匀的灰度 # 真正的背景灰度约 242。卡片内的伪背景(不够白)约 246-248。 # 所以降低 GAP_MEAN_MAX 以区分 Gap 和 Dirty White。 GAP_MEAN_MIN = 235 - GAP_MEAN_MAX = 244 + GAP_MEAN_MAX = 244 GAP_STD_MAX = 10.0 - + segments = [] is_in_card = False start_y = 0 - + # 记录原始的行状态，用于后续可能的合并优化 row_is_card = [] - + for y in range(h): rm = row_means[y] rs = row_stds[y] em = edge_means[y] - + # 1. 判定当前行是否可能是卡片的一部分 # 条件A: 边缘是灰色的 (说明有边距，排除了全宽的Header/Footer) has_gray_margin = (BG_GRAY_MIN <= em <= BG_GRAY_MAX) - + # 条件B: 中央不是均匀的背景灰 (说明有内容，或者是白色底) # 如果是 Gap，则 Mean 在 Gap范围 且 Std 很小 is_gap = (GAP_MEAN_MIN <= rm <= GAP_MEAN_MAX) and (rs < GAP_STD_MAX) - + is_card_row = has_gray_margin and not is_gap row_is_card.append(is_card_row) - + # 简单的形态学闭运算：填补卡片内部的小裂缝 # 如果一个 False (Gap) 的上下都是 True (Card)，且 Gap 长度很短，则将其视为 Card # 允许的裂缝最大长度 # [优化] 减小填补阈值，避免把两个卡片中间的真 Gap 填补了导致粘连 # 之前是 10，现在改为 3。 - MAX_GAP_FILL = 3 - + MAX_GAP_FILL = 3 + cleaned_row_is_card = row_is_card[:] - + # 查找连续的 False 区域 i = 0 while i < h: @@ -661,18 +678,18 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): i += 1 gap_end = i gap_len = gap_end - gap_start - + # Check context prev_is_card = (gap_start > 0) and cleaned_row_is_card[gap_start - 1] next_is_card = (gap_end < h) and cleaned_row_is_card[gap_end] - + if prev_is_card and next_is_card and gap_len <= MAX_GAP_FILL: # Fill the gap for k in range(gap_start, gap_end): cleaned_row_is_card[k] = True else: i += 1 - + # 根据 cleaned_row_is_card 生成 segments is_in_card = False start_y = 0 @@ -686,11 +703,11 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): is_in_card = False end_y = y height = end_y - start_y - + # 过滤太矮的区域 - if height > 100: # 稍微放宽一点，150 -> 100 + if height > 100: # 稍微放宽一点，150 -> 100 segments.append((start_y, end_y)) - + # 处理最后一个 segment if is_in_card: end_y = h @@ -700,7 +717,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): # [新增] 触底过滤：过滤掉延伸到图片底部的 Segment # 底部通常是导航栏或被截断的卡片 - BOTTOM_MARGIN = 50 # 距离底部多少像素内算触底 + BOTTOM_MARGIN = 50 # 距离底部多少像素内算触底 valid_segments_scan = [] for y1, y2 in segments: if y2 >= h - BOTTOM_MARGIN: @@ -710,7 +727,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): segments = valid_segments_scan logger.info(f" Found {len(segments)} potential segments based on vertical scan.") - + # 1.5 Group Segments by Large Gaps (to separate Header / List / Footer) # The filter bar usually creates a large gap (e.g. > 60px). SECTION_GAP_MIN = 60 @@ -718,19 +735,19 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): if segments: current_group = [segments[0]] for i in range(1, len(segments)): - prev_end = segments[i-1][1] + prev_end = segments[i - 1][1] curr_start = segments[i][0] gap = curr_start - prev_end - + if gap > SECTION_GAP_MIN: groups.append(current_group) current_group = [] - + current_group.append(segments[i]) groups.append(current_group) - + logger.info(f" Found {len(groups)} segment groups.") - + # Select the Target Group # Strategy: Select the LAST group that has at least one 'substantial' segment (H > 150). # This assumes the Station List is the main content and usually at the bottom (scrolling area). @@ -742,55 +759,55 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): if (y2 - y1) > 150: has_content = True break - + if has_content: target_group = g break - + if not target_group and segments: # Fallback to all segments if no substantial group found target_group = segments - + logger.info(f" Selected group with {len(target_group)} segments.") - + # 2. 确定每个卡片的左右边界 (Width Refinement) # 收集所有 segment 的建议边界，然后统一 - + candidate_x1 = [] candidate_x2 = [] - + temp_valid_segments = [] for y1, y2 in target_group: # 取中间一段行来分析 mid_y = (y1 + y2) // 2 sample_h = min(10, y2 - y1) - sample_rows = gray[mid_y - sample_h//2 : mid_y + sample_h//2, :] + sample_rows = gray[mid_y - sample_h // 2: mid_y + sample_h // 2, :] col_means = np.mean(sample_rows, axis=0) - + # 从左向右找第一个非灰像素 x1 = 0 for x in range(w // 2): if not (BG_GRAY_MIN <= col_means[x] <= BG_GRAY_MAX): x1 = x break - + # 从右向左找第一个非灰像素 x2 = w for x in range(w - 1, w // 2, -1): if not (BG_GRAY_MIN <= col_means[x] <= BG_GRAY_MAX): x2 = x break - + if x1 >= x2 or (x2 - x1) < w * 0.5: - # 如果这个 segment 无法确定宽度，可能不是有效卡片，暂不参与宽度投票 - # 但为了不漏掉，暂时先记录，用默认值填充 - logger.warning(f" Warning: Segment {y1}-{y2} has weird width {x2-x1}.") - pass + # 如果这个 segment 无法确定宽度，可能不是有效卡片，暂不参与宽度投票 + # 但为了不漏掉，暂时先记录，用默认值填充 + logger.warning(f" Warning: Segment {y1}-{y2} has weird width {x2 - x1}.") + pass else: candidate_x1.append(x1) candidate_x2.append(x2) - + temp_valid_segments.append((y1, y2)) # 计算统一宽度 @@ -804,13 +821,13 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): # 通常卡片是对齐的，所以 x1 应该几乎一样。 final_x1 = int(np.median(candidate_x1)) final_x2 = int(np.median(candidate_x2)) - + # 稍微加点 Padding (但不要超过图片边界) final_x1 = max(0, final_x1 - 5) final_x2 = min(w, final_x2 + 5) - + logger.info(f" Unified Width: X={final_x1}-{final_x2}, W={final_x2 - final_x1}") - + # 3. 过滤高度异常的卡片 (Height Filtering) # 计算所有潜在卡片的高度 final_cards = [] @@ -819,19 +836,20 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): else: heights = [y2 - y1 for y1, y2 in temp_valid_segments] if not heights: - logger.info(" No heights to calculate.") + logger.info(" No heights to calculate.") else: max_h = max(heights) median_h = np.median(heights) - + # 策略：如果高度差异较大，丢弃过小的卡片 # 阈值设定为 Median 的 75% (允许一定程度的偏差，如 220 vs 258 是允许的，但 150 vs 250 应被过滤) # 使用 Median 比 Max 更稳健，避免被单个超高卡片(如广告)带偏 threshold_h = median_h * 0.70 - + for (y1, y2), h in zip(temp_valid_segments, heights): if h < threshold_h: - logger.info(f" Filtering out segment Y={y1}-{y2} (H={h}) because it's too short (Threshold={threshold_h:.1f}).") + logger.info( + f" Filtering out segment Y={y1}-{y2} (H={h}) because it's too short (Threshold={threshold_h:.1f}).") else: final_cards.append((y1, y2, final_x1, final_x2)) logger.info(f" Card: Y={y1}-{y2}, X={final_x1}-{final_x2}, H={h}") @@ -839,15 +857,15 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): # 4. 保存结果 if output_dir is None: output_dir = os.path.dirname(img_path) - + base_name = os.path.basename(img_path) stem, ext = os.path.splitext(base_name) - + # 准备可视化标记图 debug_img = img.copy() - + results = [] - + # 准备 JSON 数据结构 json_data = { "image": base_name, @@ -855,10 +873,10 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): "height": h, "cards": [] } - + # 准备 _vl.jpg (只画框，不画红点) vl_img = img.copy() - + logger.info(f" Step [2.1/VL] 准备在 VL 图片上绘制 {len(final_cards)} 个场站的绿色方框...") for idx, (y1, y2, x1, x2) in enumerate(final_cards): @@ -868,7 +886,7 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): h_card = y2 - y1 click_x = int(x1 + w_card * 0.15) click_y = int(y1 + h_card * 0.20) - + # [修改] 不再保存单张子图，只记录元数据 # card = img[y1:y2, x1:x2] # 文件名添加坐标: _ClickX_ClickY @@ -876,17 +894,17 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): # out_path = os.path.join(output_dir, out_name) # cv2.imwrite(out_path, card) # logger.info(f" Saved {out_path}") - + # results.append((out_path, (click_x, click_y))) - + # 在标记图上画红点 (实心圆, 半径10, 红色BGR) cv2.circle(debug_img, (click_x, click_y), 10, (0, 0, 255), -1) # [修改] 必须画绿框，因为后续视觉模型依赖这个框来识别范围 cv2.rectangle(debug_img, (x1, y1), (x2, y2), (0, 255, 0), 2) - + # 在 _vl 图上只画绿框 cv2.rectangle(vl_img, (x1, y1), (x2, y2), (0, 255, 0), 2) - + # 收集 JSON 数据 card_info = { "id": idx + 1, @@ -894,29 +912,29 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): "click_point": [click_x, click_y] } json_data["cards"].append(card_info) - + # 记录区域信息供调用者使用 (如果需要) # 格式: (None, (click_x, click_y), (x1, y1, x2, y2)) results.append((None, (click_x, click_y), (x1, y1, x2, y2))) - + # [删除] 之前生成的单张 _for_vl.jpg 逻辑已移除 - + # 保存标记图 (_flag.jpg) if save_debug: flag_out_path = os.path.join(output_dir, f"{stem}_flag{ext}") save_image(flag_out_path, debug_img) logger.info(f" Saved Debug Image: {flag_out_path}") - + # 保存 _vl.jpg vl_out_path = os.path.join(output_dir, f"{stem}_vl{ext}") save_image(vl_out_path, vl_img) logger.info(f" Step [2.2/VL] 已保存带有绿色方框的图片: {vl_out_path}") - + # 保存 .json import json json_out_path = os.path.join(output_dir, f"{stem}.json") with open(json_out_path, 'w', encoding='utf-8') as f: json.dump(json_data, f, ensure_ascii=False, indent=4) logger.info(f" Step [2.3/JSON] 已保存场站坐标元数据: {json_out_path}") - + return results diff --git a/Apps/XinDianTu/Opener.py b/Apps/XinDianTu/Opener.py index 1656b28..814ccdd 100644 --- a/Apps/XinDianTu/Opener.py +++ b/Apps/XinDianTu/Opener.py @@ -1,14 +1,15 @@ # coding=utf-8 import asyncio import logging +import os import time import uuid -import os + import uiautomator2 as u2 -from Util.Kit import take_screenshot, detect_black_agree_button, click_image_template, detect_ad_close_x, detect_any_ad_close, detect_bottom_close_circle -from Util.ObsUtil import ObsUploader -from Util.XinDianTuReadImageKit import XinDianTuReadImageKit -from Config.Config import OBS_TMP_PREFIX, CDN_DOMAIN, TEMP_IMAGE_DIR + +from Apps.XinDianTu.Kit import take_screenshot, detect_black_agree_button, detect_any_ad_close, \ + detect_bottom_close_circle +from Config.Config import TEMP_IMAGE_DIR # pip install adbutils # 配置日志输出，方便调试和监控 diff --git a/Apps/XinDianTu/Run.py b/Apps/XinDianTu/Run.py index ecb1e89..acf63ae 100644 --- a/Apps/XinDianTu/Run.py +++ b/Apps/XinDianTu/Run.py @@ -12,8 +12,6 @@ if project_root not in sys.path: from Apps.XinDianTu.Service import XinDianTuService from Apps.XinDianTu import Opener from Apps.XinDianTu import Crawler -from Apps.XinDianTu import Coord as Y4_Coord -from Apps.XinDianTu import GetCookie as T2_GetAmapCookie # 配置日志 logging.basicConfig( @@ -59,35 +57,6 @@ async def main(): except Exception as e: logger.error(f"❌ 步骤 2 执行异常: {e}") return - - # 步骤 3: 补全坐标 (调用 Y4_Coord) - logger.info(">>> 步骤 3: 启动高德地图坐标回填任务...") - try: - # 首次尝试，开启 Cookie 检查 - await Y4_Coord.main(check_cookie_first=True) - except Y4_Coord.CookieInvalidError: - logger.warning("⚠️ 检测到高德地图 Cookie 失效或缺失。") - logger.info(">>> 正在启动 Cookie 获取工具 (T2)...") - logger.info("请在弹出的浏览器窗口中完成登录，然后在控制台按回车确认。") - - # 调用同步的 T2 工具 - # 由于 T2 包含 input() 阻塞，且这是客户端脚本，直接调用即可 - try: - # 传入 force=True 跳过“文件已存在”的询问，因为我们确定它无效 - # run_in_executor 用于避免完全阻塞事件循环（虽然在这里影响不大） - loop = asyncio.get_event_loop() - success = await loop.run_in_executor(None, T2_GetAmapCookie.run_login_tool, True) - - if success: - logger.info("✅ Cookie 更新成功，正在重新启动坐标回填任务...") - # 再次尝试执行，这次可以跳过检查（刚获取完）或者继续检查 - await Y4_Coord.main(check_cookie_first=False) - else: - logger.error("❌ Cookie 获取失败或用户取消，跳过坐标回填步骤。") - except Exception as e: - logger.error(f"❌ 调用 T2 工具失败: {e}") - except Exception as e: - logger.error(f"❌ 步骤 3 执行异常: {e}") finally: # 确保数据库连接关闭 diff --git a/Apps/XinDianTu/Service.py b/Apps/XinDianTu/Service.py index d5b0505..3cc1402 100644 --- a/Apps/XinDianTu/Service.py +++ b/Apps/XinDianTu/Service.py @@ -1,19 +1,17 @@ import hashlib +import logging import os +import re import sys import uuid -import logging -import json -import re -from datetime import datetime, timedelta +from datetime import datetime # Ensure sys path includes root for imports if not already 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 sqlalchemy import text -from Util.XinDianTuReadImageKit import XinDianTuReadImageKit +from Apps.XinDianTu.XinDianTuReadImageKit import XinDianTuReadImageKit from DbKit.Db import Db from Config.Config import DB_URL from Model.StationProfile import StationProfile diff --git a/Apps/XinDianTu/XinDianTuReadImageKit.py b/Apps/XinDianTu/XinDianTuReadImageKit.py index edbfcee..64bd3bc 100644 --- a/Apps/XinDianTu/XinDianTuReadImageKit.py +++ b/Apps/XinDianTu/XinDianTuReadImageKit.py @@ -9,13 +9,16 @@ import logging import base64 from openai import OpenAI, BadRequestError from Config.Config import ( - ALY_LLM_API_KEY, VL_MODEL_NAME, VL_MODEL_NAME_AD, + ALY_LLM_API_KEY, VL_MODEL_NAME, VL_MODEL_NAME_AD +) +from Apps.XinDianTu.Config.Setting import ( SAFE_EXCLUDE_RATIO, FALLBACK_WIDTH, FALLBACK_HEIGHT, BOTTOM_SAFE_EXCLUDE_RATIO ) logger = logging.getLogger(__name__) + class XinDianTuReadImageKit: _client = OpenAI( api_key=ALY_LLM_API_KEY, @@ -105,7 +108,7 @@ class XinDianTuReadImageKit: # 获取显示宽高，增加容错 display_width = float(device_info.get("displayWidth") or device_info.get("width") or 1080) display_height = float(device_info.get("displayHeight") or device_info.get("height") or 2400) - + bounds = obj.get("bounds") bn = obj.get("bounds_norm") text_bounds = obj.get("station_name_bounds") @@ -116,7 +119,7 @@ class XinDianTuReadImageKit: def get_pixel_coords(b_data, d_w, d_h): if not b_data: return None - + # Extract values based on dict or list if isinstance(b_data, list) and len(b_data) == 4: v1, v2, v3, v4 = b_data @@ -126,7 +129,7 @@ class XinDianTuReadImageKit: v2 = b_data.get("top") if b_data.get("top") is not None else b_data.get("y1") v3 = b_data.get("right") if b_data.get("right") is not None else b_data.get("x2") v4 = b_data.get("bottom") if b_data.get("bottom") is not None else b_data.get("y2") - + if any(v is None for v in (v1, v2, v3, v4)): return None else: @@ -138,9 +141,9 @@ class XinDianTuReadImageKit: return None max_v = max(v1, v2, v3, v4) - + # 1. 0-1 归一化坐标 - if max_v <= 1.05: + if max_v <= 1.05: x1, y1, x2, y2 = v1 * d_w, v2 * d_h, v3 * d_w, v4 * d_h # 2. 0-1000 归一化坐标 (Qwen-VL 常用) elif max_v <= 1005: @@ -148,20 +151,24 @@ class XinDianTuReadImageKit: # 3. 绝对像素坐标 else: x1, y1, x2, y2 = v1, v2, v3, v4 - + return min(x1, x2), min(y1, y2), max(x1, x2), max(y1, y2) # 优先使用场站名称区域进行点击 (最安全) - coords = get_pixel_coords(text_bn, display_width, display_height) or get_pixel_coords(text_bounds, display_width, display_height) + coords = get_pixel_coords(text_bn, display_width, display_height) or get_pixel_coords(text_bounds, + display_width, + display_height) if coords: x1, y1, x2, y2 = coords uia_x = int(x1 + (x2 - x1) / 2) uia_y = int(y1 + (y2 - y1) / 2) - logger.info(f"坐标计算: 使用文字区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}") - + logger.info( + f"坐标计算: 使用文字区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}") + # 备选：使用整个卡片区域 if uia_x is None: - coords = get_pixel_coords(bn, display_width, display_height) or get_pixel_coords(bounds, display_width, display_height) + coords = get_pixel_coords(bn, display_width, display_height) or get_pixel_coords(bounds, display_width, + display_height) if coords: x1, y1, x2, y2 = coords w, h = x2 - x1, y2 - y1 @@ -172,24 +179,27 @@ class XinDianTuReadImageKit: # 默认中心点，但稍微偏上一点，避开底部可能的按钮 uia_x = int(x1 + w / 2) uia_y = int(y1 + h * 0.4) - logger.info(f"坐标计算: 使用卡片区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}") + logger.info( + f"坐标计算: 使用卡片区域 -> ({uia_x}, {uia_y}) | 区域: {coords} | 屏幕: {display_width}x{display_height}") # --- 安全过滤：过滤掉屏幕顶部的点击坐标 (通常是广告或菜单) --- if uia_y is not None: # 如果点击点在屏幕顶部 SAFE_EXCLUDE_RATIO 范围内，极大概率是误触广告位，将其排除 if uia_y < (display_height * SAFE_EXCLUDE_RATIO): - logger.warning(f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO*100)}% 区域，疑似广告或菜单，已忽略。") + logger.warning( + f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似广告或菜单，已忽略。") return None # 如果点击点在屏幕底部 BOTTOM_SAFE_EXCLUDE_RATIO 范围内，极大概率是误触底部功能区（如扫码充电），将其排除 if uia_y > (display_height * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): - logger.warning(f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}% 区域，疑似底部功能区，已忽略。") + logger.warning( + f"安全排除: 坐标 ({uia_x}, {uia_y}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似底部功能区，已忽略。") return None if uia_x is not None and uia_y is not None: obj["uia_center_x"] = uia_x obj["uia_center_y"] = uia_y - + # Clean up temporary fields for k in ["bounds", "bounds_norm", "station_name_bounds", "station_name_bounds_norm"]: if k in obj: @@ -261,13 +271,13 @@ class XinDianTuReadImageKit: device_info = XinDianTuReadImageKit._FALLBACK_DEVICE_INFO prompt = ( - "仅输出JSON对象（不含任何说明文字）。" - "请在图片中找到场站名称完全匹配“" + str(station_name) + "”的那一行/卡片，" - "返回该场站名称文字本身的区域坐标（优先返回归一化坐标）。" - "返回格式示例：" - "{\"station_name_bounds\": {\"x1\": 100, \"y1\": 200, \"x2\": 300, \"y2\": 240}, " - "\"station_name_bounds_norm\": {\"left\": 0.1, \"top\": 0.2, \"right\": 0.3, \"bottom\": 0.24}}" - "注意：bounds使用0-1000归一化坐标空间；如果找不到，返回空JSON {}。" + "仅输出JSON对象（不含任何说明文字）。" + "请在图片中找到场站名称完全匹配“" + str(station_name) + "”的那一行/卡片，" + "返回该场站名称文字本身的区域坐标（优先返回归一化坐标）。" + "返回格式示例：" + "{\"station_name_bounds\": {\"x1\": 100, \"y1\": 200, \"x2\": 300, \"y2\": 240}, " + "\"station_name_bounds_norm\": {\"left\": 0.1, \"top\": 0.2, \"right\": 0.3, \"bottom\": 0.24}}" + "注意：bounds使用0-1000归一化坐标空间；如果找不到，返回空JSON {}。" ) loop = asyncio.get_event_loop() @@ -293,7 +303,9 @@ class XinDianTuReadImageKit: try: data = json.loads(raw) - if isinstance(data, dict) and (data.get("station_name_bounds") or data.get("station_name_bounds_norm") or data.get("bounds") or data.get("bounds_norm")): + if isinstance(data, dict) and ( + data.get("station_name_bounds") or data.get("station_name_bounds_norm") or data.get( + "bounds") or data.get("bounds_norm")): data = XinDianTuReadImageKit._add_click_point(data, device_info, anchor="station_text") return { "uia_center_x": data.get("uia_center_x"), @@ -339,7 +351,8 @@ class XinDianTuReadImageKit: data = json.loads(raw) if isinstance(data, list): # 过滤掉 None (即被 _add_click_point 排除掉的顶部项) - data = [XinDianTuReadImageKit._add_click_point(x, device_info, anchor="station_text") for x in data if x.get("b_use", 1) == 1] + data = [XinDianTuReadImageKit._add_click_point(x, device_info, anchor="station_text") for x in data if + x.get("b_use", 1) == 1] data = [x for x in data if x is not None] elif isinstance(data, dict): if data.get("b_use", 1) == 1: @@ -389,7 +402,8 @@ class XinDianTuReadImageKit: if tasks: results = await asyncio.gather(*tasks, return_exceptions=True) for i, r in zip(idxs, results): - if isinstance(r, dict) and r.get("uia_center_x") is not None and r.get("uia_center_y") is not None: + if isinstance(r, dict) and r.get("uia_center_x") is not None and r.get( + "uia_center_y") is not None: data[i]["uia_center_x"] = r.get("uia_center_x") data[i]["uia_center_y"] = r.get("uia_center_y") @@ -418,11 +432,11 @@ class XinDianTuReadImageKit: """ if not json_metadata or "cards" not in json_metadata: return [] - + cards_meta = json_metadata["cards"] # 按 id 或 Y 坐标排序，确保顺序一致 (Kit 生成时已经是 top-down) - cards_meta.sort(key=lambda x: x["rect"][1]) - + cards_meta.sort(key=lambda x: x["rect"][1]) + prompt = ( "图片中用绿色矩形框标记了若干个充电站卡片区域。\n" "请按从上到下的顺序，依次识别每个绿色框内的场站信息，并返回一个JSON数组。\n" @@ -440,7 +454,7 @@ class XinDianTuReadImageKit: "\n" "严格返回纯JSON格式。" ) - + loop = asyncio.get_event_loop() resp = await loop.run_in_executor( None, @@ -457,10 +471,10 @@ class XinDianTuReadImageKit: ], ) ) - + content = resp.choices[0].message.content or "" raw = XinDianTuReadImageKit._extract_json(content) - + final_stations = [] try: llm_data = json.loads(raw) @@ -470,46 +484,48 @@ class XinDianTuReadImageKit: for i in range(count): item = llm_data[i] meta = cards_meta[i] - + if not item or not isinstance(item, dict): continue - + name = item.get("station_name") if not name: continue - + # 注入元数据中的点击坐标 click_pt = meta.get("click_point", [0, 0]) cx, cy = click_pt - + # 安全检查 img_h = 2400 if device_info: img_h = float(device_info.get("displayHeight") or device_info.get("height") or 2400) else: img_h = json_metadata.get("height", 2400) - + # 顶部安全排除 if cy < (img_h * SAFE_EXCLUDE_RATIO): - logger.warning(f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO*100)}% 区域，已忽略。") + logger.warning( + f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕顶部 {int(SAFE_EXCLUDE_RATIO * 100)}% 区域，已忽略。") continue - + # 底部安全排除 if cy > (img_h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)): - logger.warning(f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO*100)}% 区域，疑似底部功能区，已忽略。") + logger.warning( + f"VL安全排除: 坐标 ({cx}, {cy}) 位于屏幕底部 {int(BOTTOM_SAFE_EXCLUDE_RATIO * 100)}% 区域，疑似底部功能区，已忽略。") continue item["uia_center_x"] = cx item["uia_center_y"] = cy item["rect"] = meta.get("rect") - + final_stations.append(item) else: logger.warning(f"LLM return format error: expected list, got {type(llm_data)}") - + except Exception as e: logger.error(f"Error parsing VL response: {e}") - + return final_stations @staticmethod @@ -530,7 +546,7 @@ class XinDianTuReadImageKit: remote_path = f"tmp/patches/{os.path.basename(patch_path)}" uploader.upload_file(patch_path, remote_path) patch_url = f"{cdn_domain}/{remote_path}" - + # 记录上传后的 URL 供识别使用 seg["patch_url"] = patch_url tasks.append(XinDianTuReadImageKit.recognize_card_text(patch_url)) @@ -547,7 +563,7 @@ class XinDianTuReadImageKit: res["x"] = seg["center_x"] res["y"] = seg["center_y"] final_stations.append(res) - + # 4. 后置过滤 (复用原有过滤逻辑) if final_stations: processed_excluded_titles = {str(x).replace(" ", "").strip() for x in STATION_EXCLUDED_TITLES} @@ -580,40 +596,40 @@ class XinDianTuReadImageKit: img_data = np.array(img) logger.info(f"页面分析 - 分辨率: {width}x{height}") - + # 1. 行特征分析：识别分割线 (Divider) # 分割线特征：横贯全屏，亮度均匀，通常比纯白(255)稍暗，比文字内容亮 # 典型值：Mean=242-247, Std<10 - - row_types = [] # 0: Unknown/Content, 1: Divider, 2: White/Empty - + + row_types = [] # 0: Unknown/Content, 1: Divider, 2: White/Empty + # 采样点：左中右 l_x, m_x, r_x = int(width * 0.05), int(width * 0.5), int(width * 0.95) - + debug_rows = [] for y in range(height): # 避开顶部和底部导航栏 (15% - 85%) - 扩大排除范围以避免误识别 Header/Footer if y < height * 0.15 or y > height * 0.85: - row_types.append(1) # 视为无关区域 (标记为 Divider 以防止形成 Segment) + row_types.append(1) # 视为无关区域 (标记为 Divider 以防止形成 Segment) continue row = img_data[y] row_mean = np.mean(row) row_std = np.std(row) - + # 判定逻辑： # 1. 纯白行 (卡片底色) -> mean > 252 (允许微小噪点) # 2. 分割线 (Divider) -> 230 < mean < 252 且 std < 15 (颜色均匀) # 3. 内容行 (Content) -> mean <= 230 或 (mean > 230 且 std >= 15) (有文字/图片导致方差大) - + if row_mean > 252: - r_type = 2 # White/Empty + r_type = 2 # White/Empty elif 230 < row_mean <= 252 and row_std < 15: - r_type = 1 # Divider + r_type = 1 # Divider else: - r_type = 0 # Content - + r_type = 0 # Content + row_types.append(r_type) # Debug log sampling @@ -626,33 +642,33 @@ class XinDianTuReadImageKit: # 2. 聚合连续的 Content 区域 (Type 0 or Type 2 sandwiched by Type 0) # 实际上，卡片是由 Divider 分隔开的区域。 # 我们寻找两个 Divider 之间的区域，且该区域必须包含 Content (Type 0)。 - + segments = [] - + # 简化状态机： # 寻找非 Divider 的连续段 in_segment = False seg_start = -1 has_content = False - + raw_blocks = [] - + for y, r_type in enumerate(row_types): - if r_type != 1: # Not Divider + if r_type != 1: # Not Divider if not in_segment: in_segment = True seg_start = y has_content = False - + if r_type == 0: has_content = True - else: # Is Divider + else: # Is Divider if in_segment: # 结束一段 - if has_content: # 只有包含内容的段才算 + if has_content: # 只有包含内容的段才算 raw_blocks.append((seg_start, y)) in_segment = False - + # 处理最后一段 if in_segment and has_content: raw_blocks.append((seg_start, len(row_types))) @@ -660,52 +676,52 @@ class XinDianTuReadImageKit: # 3. 过滤和后处理 # 合并距离很近的块？或者过滤太小的块 base_name = os.path.splitext(os.path.basename(image_path))[0] - + valid_segments = [] logger.info(f"Initial raw blocks count: {len(raw_blocks)}") - + for i, (y1, y2) in enumerate(raw_blocks): h = y2 - y1 # 过滤太小的块 (可能是杂噪或单纯的文字行) - if h < 50: + if h < 50: # logger.debug(f"Block {i} too small: {h}") continue # 过滤太大的块 (可能是全屏错误) if h > 1000: continue - + # 再次确认内部是否有足够的“暗像素”（内容） # 避免切出纯白的空隙 region = img_data[y1:y2] region_mean = np.mean(region) - if region_mean > 254.5: # 整体太白 (放宽阈值，避免误杀大面积白色的卡片) + if region_mean > 254.5: # 整体太白 (放宽阈值，避免误杀大面积白色的卡片) logger.info(f" [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 整体太白 (Mean={region_mean:.1f} > 254.5)") continue # --- 优化：边缘背景检查与宽度裁剪 (Margin Check & Crop) --- # 不再直接拒绝白边，而是尝试计算内容的有效宽度 # 假设：有效内容行的方差较高，或者亮度显著不同于背景 - + # 简单策略：保留全宽，但增加高度限制 - if h < 60: # 稍微降低阈值以测试 + if h < 60: # 稍微降低阈值以测试 logger.info(f" [-] 忽略区域 {i}: Y({y1}-{y2}), H={h}, 高度不足 (<60)") continue - + # 计算左右边距的平均亮度，辅助判断（仅记录日志，不强行过滤） margin_w = max(5, int(width * 0.05)) l_margin = img_data[y1:y2, 0:margin_w] - r_margin = img_data[y1:y2, width-margin_w:width] + r_margin = img_data[y1:y2, width - margin_w:width] l_mean = np.mean(l_margin) r_mean = np.mean(r_margin) - + # 如果是全宽卡片，左右边缘可能是白色的。 # 之前的逻辑: if l_mean > 252 or r_mean > 252: continue (导致漏检) # 现在移除该逻辑。 - + # --- 新增：自动裁剪水平宽度 (Auto Horizontal Crop) --- # 尝试找到内容的左右边界 (基于列的方差或亮度差异) x1, x2 = 0, width - + # 从左向右扫描 for x in range(0, int(width * 0.4), 2): col = img_data[y1:y2, x] @@ -713,22 +729,22 @@ class XinDianTuReadImageKit: if np.std(col) > 5 or np.mean(col) < 245: x1 = x break - + # 从右向左扫描 for x in range(width - 1, int(width * 0.6), -2): col = img_data[y1:y2, x] if np.std(col) > 5 or np.mean(col) < 245: x2 = x + 1 break - + # 增加一点 padding x1 = max(0, x1 - 10) x2 = min(width, x2 + 10) - + # 如果裁剪后宽度太小，可能不是有效卡片 if (x2 - x1) < width * 0.5: - logger.info(f" [-] 忽略区域 {i}: 裁剪后宽度过小 ({x2-x1})") - continue + logger.info(f" [-] 忽略区域 {i}: 裁剪后宽度过小 ({x2 - x1})") + continue # 保存 patch = img.crop((x1, y1, x2, y2)) @@ -743,8 +759,9 @@ class XinDianTuReadImageKit: "center_y": center_y, "y_range": (y1, y2) }) - logger.info(f" [+] 发现卡片 {i}: Y({y1}-{y2}), H={h}, Crop X({x1}-{x2}), 边缘(L={l_mean:.1f}, R={r_mean:.1f}), 已保存") - + logger.info( + f" [+] 发现卡片 {i}: Y({y1}-{y2}), H={h}, Crop X({x1}-{x2}), 边缘(L={l_mean:.1f}, R={r_mean:.1f}), 已保存") + logger.info(f"分析完成：识别到 {len(valid_segments)} 个区域") return valid_segments @@ -767,7 +784,7 @@ class XinDianTuReadImageKit: "role": "user", "content": [ { - "type": "text", + "type": "text", "text": "请识别图片中的充电站信息，并以 JSON 格式输出：{\"station_name\": \"...\", \"price\": \"...\", \"piles\": \"空闲数/总数\"}。只输出 JSON，不要有其他文字。" }, {"type": "image_url", "image_url": {"url": patch_url}} @@ -787,7 +804,7 @@ class XinDianTuReadImageKit: return json.loads(json_match.group()) except Exception: logger.warning(f"OCR 结果解析 JSON 失败: {content}") - + return {"station_name": "未知"} @staticmethod @@ -940,8 +957,8 @@ class XinDianTuReadImageKit: # 尝试捕获 DataInspection 错误并进行 Base64 回退 err_code = getattr(e, 'code', '') or '' if not err_code and hasattr(e, 'body') and isinstance(e.body, dict): - err_code = e.body.get('code', '') - + err_code = e.body.get('code', '') + # 如果是媒体格式或数据检查错误 if 'InvalidParameter.DataInspection' in str(err_code) or 'media format' in str(e).lower(): logger.warning(f"Image URL rejected ({err_code}). Attempting Base64 fallback: {image_url}") @@ -956,7 +973,7 @@ class XinDianTuReadImageKit: ext = "webp" elif ".jpg" in lower_url or ".jpeg" in lower_url: ext = "jpeg" - + data_uri = f"data:image/{ext};base64,{base64_str}" resp = await loop.run_in_executor(None, lambda: _do_request(data_uri)) else: @@ -966,8 +983,8 @@ class XinDianTuReadImageKit: logger.error(f"API BadRequestError: {e}") return [] except Exception as e: - logger.error(f"API Unexpected Error: {e}") - return [] + logger.error(f"API Unexpected Error: {e}") + return [] content = resp.choices[0].message.content or "" raw = XinDianTuReadImageKit._extract_json(content) @@ -1147,7 +1164,7 @@ class XinDianTuReadImageKit: content = resp.choices[0].message.content or "" raw = XinDianTuReadImageKit._extract_json(content) - + result = {} try: data = json.loads(raw) @@ -1155,7 +1172,7 @@ class XinDianTuReadImageKit: result = data except Exception as e: logger.error(f"Error parsing address JSON: {e}") - + # 等待并合并按钮坐标结果 try: button_result = await button_task @@ -1284,7 +1301,7 @@ async def test3(station_name: str): if __name__ == "__main__": - #asyncio.run(test1()) + # asyncio.run(test1()) station_name = '吉林省看守所充电站' - #asyncio.run(test2(station_name)) + # asyncio.run(test2(station_name)) asyncio.run(test3(station_name)) diff --git a/Apps/XinDianTu/__pycache__/Crawler.cpython-310.pyc b/Apps/XinDianTu/__pycache__/Crawler.cpython-310.pyc index f042670..4b4cf41 100644 Binary files a/Apps/XinDianTu/__pycache__/Crawler.cpython-310.pyc and b/Apps/XinDianTu/__pycache__/Crawler.cpython-310.pyc differ diff --git a/Apps/XinDianTu/__pycache__/Kit.cpython-310.pyc b/Apps/XinDianTu/__pycache__/Kit.cpython-310.pyc new file mode 100644 index 0000000..ea864fe Binary files /dev/null and b/Apps/XinDianTu/__pycache__/Kit.cpython-310.pyc differ diff --git a/Apps/XinDianTu/__pycache__/Opener.cpython-310.pyc b/Apps/XinDianTu/__pycache__/Opener.cpython-310.pyc index 66a0299..b10d6d2 100644 Binary files a/Apps/XinDianTu/__pycache__/Opener.cpython-310.pyc and b/Apps/XinDianTu/__pycache__/Opener.cpython-310.pyc differ diff --git a/Apps/XinDianTu/__pycache__/Run.cpython-310.pyc b/Apps/XinDianTu/__pycache__/Run.cpython-310.pyc index c5e0116..da0a0e5 100644 Binary files a/Apps/XinDianTu/__pycache__/Run.cpython-310.pyc and b/Apps/XinDianTu/__pycache__/Run.cpython-310.pyc differ diff --git a/Apps/XinDianTu/__pycache__/Service.cpython-310.pyc b/Apps/XinDianTu/__pycache__/Service.cpython-310.pyc index 78238e7..686bdb1 100644 Binary files a/Apps/XinDianTu/__pycache__/Service.cpython-310.pyc and b/Apps/XinDianTu/__pycache__/Service.cpython-310.pyc differ diff --git a/Apps/XinDianTu/__pycache__/XinDianTuReadImageKit.cpython-310.pyc b/Apps/XinDianTu/__pycache__/XinDianTuReadImageKit.cpython-310.pyc new file mode 100644 index 0000000..75d15a2 Binary files /dev/null and b/Apps/XinDianTu/__pycache__/XinDianTuReadImageKit.cpython-310.pyc differ