diff --git a/Apps/AiTeJiYiChong/Crawler.py b/Apps/AiTeJiYiChong/Crawler.py index 4fed353..55bfeb6 100644 --- a/Apps/AiTeJiYiChong/Crawler.py +++ b/Apps/AiTeJiYiChong/Crawler.py @@ -30,29 +30,15 @@ logger = logging.getLogger("AiTeJiYiChongCrawler") async def is_already_crawled(redis_kit, station_name): """ - 检查场站是否已经爬取过，支持前缀模糊匹配处理截断名称 + 检查场站是否已经爬取过，仅按名称去重 """ cleaned_name = Kit.clean_station_name(station_name) if not cleaned_name: return False - # 1. 检查精确匹配 redis_key = f"crawled:aite:{cleaned_name}" if await redis_kit.get_data(redis_key): return True - - # 2. 如果名称被截断（原名包含 . ），尝试前缀匹配 - # 逻辑：在 Redis 中查找是否有以当前清理后的名称开头的已完成场站 - # 为了效率，我们配合 Redis 的 keys 模式（小规模数据下可用） - if ".." in station_name or station_name.endswith("."): - # 获取所有已爬取的 key 进行内存前缀比对，或者利用 Redis 模糊查询 - # 这里采用一种更高效的方式：在详情页保存时，额外保存一个前缀索引 - prefix_key = f"crawled:aite:prefix:{cleaned_name[:6]}" # 取前6位作为索引前缀 - stored_names = await redis_kit.get_data(prefix_key) - if stored_names and isinstance(stored_names, list): - for full_name in stored_names: - if full_name.startswith(cleaned_name): - return True return False async def get_station_list(d, service, max_scrolls=MAX_SCROLLS): @@ -190,16 +176,9 @@ async def get_station_list(d, service, max_scrolls=MAX_SCROLLS): d.press("back") await asyncio.sleep(0.8) - # 记录 Redis (保存全名并建立前缀索引) + # 记录 Redis (仅按名称去重，不再使用前缀索引) full_name_key = f"crawled:aite:{Kit.clean_station_name(station_name)}" await redis_kit.set_data(full_name_key, "1", expire=REDIS_STATION_EXPIRE) - - # 建立前缀索引 (取前6位，方便模糊查找) - prefix_idx_key = f"crawled:aite:prefix:{Kit.clean_station_name(station_name)[:6]}" - existing_names = await redis_kit.get_data(prefix_idx_key) or [] - if station_name not in existing_names: - existing_names.append(station_name) - await redis_kit.set_data(prefix_idx_key, existing_names, expire=REDIS_STATION_EXPIRE) # 在每一页结束时，清理已完成的任务，防止内存堆积 done_tasks = [t for t in background_tasks if t.done()] diff --git a/Apps/AiTeJiYiChong/Doc/流程说明.md b/Apps/AiTeJiYiChong/Doc/流程说明.md index 4587899..b5d3e4b 100644 --- a/Apps/AiTeJiYiChong/Doc/流程说明.md +++ b/Apps/AiTeJiYiChong/Doc/流程说明.md @@ -44,9 +44,14 @@ 2. **用户体验**: UI 操作（点击、滑动、返回）是连续的，不会因为网络请求波动而卡顿。 - **任务追踪**: 在 `Crawler.py` 中维护 `background_tasks` 列表，在每页结束时清理已完成任务，并在整个爬取结束时使用 `asyncio.gather` 等待残余任务。 -### 3.2 智能去重机制 -- **Redis 模糊匹配**: 针对小程序中场站名称可能被省略号截断（如“万达茂...”）的情况，不仅存储全名，还存储场站名称的前 6 位作为前缀索引。 -- **逻辑**: 先查全名，再查前缀索引。如果前缀匹配成功，则视为已爬取，跳过重复工作。 +### 3.2 场站去重策略 +- **按名称去重**: 仅依据列表显示的场站名称进行去重，不再使用前缀或模糊匹配。 +- **处理规则**: + - 名称先经 `clean_station_name` 去掉尾部省略号与空格，例如 `AAABBB...` → `AAABBB`； + - 首次遇到该名称时处理并写入 Redis 键 `crawled:aite:{name}`； + - 后续再遇到同名（包括省略号截断的同名）直接跳过，不再点击进入详情页。 +- **设计原因**: 逻辑直白、符合业务预期，降低模糊匹配带来的误判风险。 +- **一致性说明**: 新电途项目采用同样的按名称去重策略以保持跨项目行为一致。 ### 3.3 坐标系转换 - **VL 坐标**: 0-1000 的归一化坐标。 diff --git a/Apps/XinDianTu/Crawler.py b/Apps/XinDianTu/Crawler.py index c3cb9e6..f8fe43d 100644 --- a/Apps/XinDianTu/Crawler.py +++ b/Apps/XinDianTu/Crawler.py @@ -23,6 +23,7 @@ from Util.RedisKit import RedisKit from Util.PaddleOCRKit import get_ocr_kit import cv2 from Apps.XinDianTu.Service import XinDianTuService +from Apps.XinDianTu.ReadImageKit import ReadImageKit from Config.Config import ( OBS_TMP_PREFIX, CDN_DOMAIN, TEMP_IMAGE_DIR ) @@ -95,416 +96,188 @@ class XinDianTuCrawler(BaseCrawler): pass +async def is_already_crawled(redis_kit, station_name): + """ + 检查场站是否已经爬取过，仅按名称去重 + """ + cleaned_name = Kit.clean_station_name(station_name) + if not cleaned_name: + return False + + redis_key = f"crawled:xdt:{cleaned_name}" + if await redis_kit.get_data(redis_key): + return True + return False + + +async def analyze_prices_background(service, station_name, image_paths): + """ + 后台异步处理价格图片分析 + """ + try: + # 使用多图解析并去重 + all_rows = await ReadImageKit.parse_price_schedule_multi(station_name, image_paths) + + if all_rows: + # 24小时铺平 + hourly_schedule = ReadImageKit.hourly_full_day(all_rows) + await service.process_price_detail_data(station_name, hourly_schedule) + logger.info(f"后台处理完成: {station_name} 的分时价格已保存。") + else: + logger.warning(f"后台处理失败: {station_name} 未能解析出价格时段。") + + # 处理完成后清理图片 + for path in image_paths: + try: + if os.path.exists(path): + os.remove(path) + except: + pass + + except Exception as e: + logger.error(f"分析价格后台任务异常 ({station_name}): {e}") + return True + + 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}") + logger.info(f"开始爬取列表，设备: {device_info.get('productName')} | 分辨率: {w}x{h}") - max_distance_reached = False - stop_reason = "max_scrolls_reached" # 默认停止原因：达到最大翻页次数 - - # 存储后台异步任务 background_tasks = [] for i in range(max_scrolls + 1): - if max_distance_reached: - logger.info("已达到最大抓取距离，停止翻页。") - stop_reason = "max_distance_reached" - break logger.info(f"正在处理第 {i + 1} 页...") # 1. 拍摄截图 - logger.info(f"Step [1/6] 正在拍摄列表页截图...") - t_shot = time.time() image_uuid = str(uuid.uuid4()) - # 使用相对路径: 基于当前脚本目录下的 Images 文件夹 - base_dir = os.path.dirname(os.path.abspath(__file__)) - 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)") + screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR) + logger.info(f"列表页截图已完成: {screenshot_path}") - # 校验截图实际尺寸 - img_w, img_h = 0, 0 - try: - with Image.open(screenshot_path) as img: - img_w, img_h = img.size - 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}) 不一致，将以截图尺寸为准进行坐标换算。") - 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 + # 2. 执行本地图形学分析，生成 _vl.jpg 和 .json + logger.info("正在执行图形学切片分析...") 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.warning("可能未识别到任何卡片，跳过当前页") + 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() + # 3. 【优化】后台异步上传 _vl.jpg 供以后参考，不再等待上传 + vl_object_key = f"{OBS_TMP_PREFIX}/{image_uuid}_vl.jpg" + loop = asyncio.get_running_loop() + loop.run_in_executor(None, uploader.upload_file, vl_object_key, vl_img_path) + + # 4. 【优化】直接使用本地路径调用 VL 模型识别，避免等待上传 + logger.info("正在调用 VL 模型识别场站信息...") + stations = await service.process_station_list_vl(vl_img_path, json_metadata, device_info=device_info) + logger.info(f"本页识别到 {len(stations)} 个场站") - stations = [] - try: - ocr_kit = get_ocr_kit() + # 5. 匹配几何卡片与 VL 识别结果 (XinDianTu 的 parse_vl_image 已经按顺序返回了) + # 这里的 stations 列表长度应该与 json_metadata["cards"] 对应 + if json_metadata.get("cards") and stations: + for idx, card in enumerate(json_metadata["cards"]): + # 检查索引是否越界 (VL 模型可能返回的数组长度不一致) + if idx < len(stations) and stations[idx]: + st = stations[idx] + station_name = st.get("station_name") + if not station_name: continue - # 读取原始截图进行裁剪 (比读取 _vl.jpg 更干净，没有绿框干扰) - original_img = cv2.imread(screenshot_path) + # 检查 Redis 去重 + if await is_already_crawled(redis_kit, station_name): + logger.info(f"场站 {station_name} 匹配到已处理记录，跳过。") + continue - 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 识别...") + click_x, click_y = card["click_point"] + logger.info(f"准备处理第 {idx + 1} 个场站: {station_name}, 点击坐标: ({click_x}, {click_y})") + + d.click(int(click_x), int(click_y)) - 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 + # 等待二级页面加载 + await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) - # 边界检查 - 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})...") - 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") - - 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) - else: - 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', []))}") - - if not stations: - logger.warning("当前页面未检测到任何有效的场站信息") - else: - logger.info(f"Step [6/6] 准备开始逐一点击场站进入详情页 (共 {len(stations)} 个)...") - 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 - - seen_names.add(name) - all_stations.append(s) - new_stations_found = True - - # 打印基本信息 - price = s.get("price") - pro_price = s.get("pro_price") - piles_str = s.get("piles", "") - distance_str = s.get("distance", "") - - # 检查距离 - dist_log_part = "" - if distance_str: - try: - import re - dist_match = re.search(r"(\d+(\.\d+)?)", distance_str) - if dist_match: - 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。") - 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}") - - if max_distance_reached: - break - - # --- 点击进入详情页记录地址 --- - # 使用相对坐标点击，以适应不同分辨率 - if img_w > 0 and img_h > 0: - rel_x = max(0.0, min(1.0, x / img_w)) - rel_y = max(0.0, min(1.0, y / img_h)) - logger.info(f"👉 正在点击进入详情页: {name} (相对坐标: {rel_x:.3f}, {rel_y:.3f})") - d.click(rel_x, rel_y) - else: - logger.info(f"👉 正在点击进入详情页: {name} (绝对坐标: {x}, {y})") - d.click(x, y) - - await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 等待详情页加载 - - # 尝试获取并保存地址 - address_recorded = False - for retry in range(2): - # 1. 拍摄详情页截图 - t_d_shot = time.time() + # 截取二级页面图 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") + detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR) - # 2. 上传至 OBS - t_d_up = time.time() + # 【优化】后台解析详情页地址，直接传本地路径，避免等待上传 + logger.info(f"已启动后台分析详情页: {station_name}") + task_addr = asyncio.create_task(service.process_station_address(station_name, detail_path, device_info=device_info)) + background_tasks.append(task_addr) + + # 【优化】后台异步上传详情页截图供以后参考 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) - logger.info(f"Step [解析并保存地址-后台] 耗时: {time.time() - t_addr_task:.4f}s") - if addr_res.get("address"): - logger.info(f"成功记录地址: {addr_res.get('address')}") - # 记录到 Redis，设置过期时间 - await redis_kit.set_data(task_redis_key, "1", expire=REDIS_STATION_EXPIRE) - except Exception as e: - logger.error(f"后台处理地址失败 ({task_name}): {e}") - - # 创建并启动任务 - addr_task = asyncio.create_task(_process_address_task(name, detail_url, device_info, redis_key)) - background_tasks.append(addr_task) - logger.info(f"已提交地址解析任务到后台: {name}") - - # 标记为已记录以跳出 retry 循环 - address_recorded = True - - # 清理详情页临时截图 - if not KEEP_SCREENSHOTS and os.path.exists(detail_screenshot_path): - os.remove(detail_screenshot_path) - - 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) - logger.info(f"Step [识别展开按钮] 耗时: {time.time() - t_find:.4f}s") - - # 清理截图 - try: - os.remove(screenshot_path) - except: - pass + loop = asyncio.get_running_loop() + loop.run_in_executor(None, uploader.upload_file, detail_object_key, detail_path) + # --- 抓取价格时段信息 --- + # 识别 "全部时段" 按钮 + pos = Kit.find_expand_button_position(detail_path, debug_dir=TEMP_IMAGE_DIR, debug_filename_prefix=station_name) if pos: - x, y = pos - logger.info(f"通过几何特征找到 '全部时段' 按钮: ({x}, {y})") - d.click(x, y) + ex, ey = pos + logger.info(f"找到 '全部时段' 按钮: ({ex}, {ey})") + d.click(ex, ey) await asyncio.sleep(1.5) - # 抓取3屏截图 - price_image_urls = [] + price_screenshots = [] for p_idx in range(1, 4): - # 使用场站名称的MD5值作为文件名前缀，避免中文和特殊字符导致的URL问题 - t_p_loop = time.time() - name_md5 = hashlib.md5(name.encode('utf-8')).hexdigest() - 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) - if success: - p_url = f"https://{CDN_DOMAIN}/{p_object_key}" - price_image_urls.append(p_url) - logger.info(f"截图上传成功: {p_url}") - else: - logger.warning(f"截图上传失败: {p_path}") - - logger.info(f"Step [价格截图与上传-{p_idx}] 耗时: {time.time() - t_p_loop:.4f}s") - + p_uuid = f"{hashlib.md5(station_name.encode('utf-8')).hexdigest()}_p_{p_idx}" + p_path = take_screenshot(d, p_uuid, save_dir=TEMP_IMAGE_DIR) + price_screenshots.append(p_path) if p_idx < 3: d.swipe_ext("up", scale=0.7) await asyncio.sleep(1.0) + + # 【异步优化】后台处理价格 + logger.info(f"已启动后台分析价格页: {station_name}") + task_price = asyncio.create_task(analyze_prices_background(service, station_name, price_screenshots)) + background_tasks.append(task_price) - # 调用服务解析价格时段 (异步后台处理) - if price_image_urls: - logger.info(f"正在调用接口解析电费时段数据 ({len(price_image_urls)} 张图片) - 转入后台...") + # 从三级页面返回 + d.press("back") + await asyncio.sleep(0.5) - 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) - logger.info(f"Step [解析电费时段数据-后台] 耗时: {time.time() - t_price_task:.4f}s") - logger.info(f"电费时段数据解析完成: {task_name}") - except Exception as e: - logger.error(f"后台处理价格时段失败 ({task_name}): {e}") + # 从二级页面返回 + d.press("back") + await asyncio.sleep(WAIT_BACK_TO_LIST) - price_task = asyncio.create_task(_process_price_task(name, price_image_urls, device_info)) - background_tasks.append(price_task) - else: - logger.warning("未能获取到有效的价格时段截图，跳过解析") - else: - logger.warning(f"未能通过几何特征识别 '全部时段' 按钮") + # 记录 Redis 去重 (仅按名称去重) + cleaned = Kit.clean_station_name(station_name) + await redis_kit.set_data(f"crawled:xdt:{cleaned}", "1", expire=REDIS_STATION_EXPIRE) - except Exception as e: - logger.warning(f"抓取价格时段信息时发生异常: {e}") + # 清理已完成的后台任务 + done_tasks = [t for t in background_tasks if t.done()] + for t in done_tasks: + if t.exception(): + logger.error(f"后台任务异常: {t.exception()}") + background_tasks.remove(t) - # 测试模式：仅处理第一个场站 - # logger.info("测试模式：仅处理第一个场站，即将退出 (Y3)") - # return all_stations + # 6. 翻页 + logger.info("执行翻页滑动...") + d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO) + await asyncio.sleep(WAIT_AFTER_SCROLL) - # 返回列表页 - d.press("back") - await asyncio.sleep(WAIT_BACK_TO_LIST) # 等待列表页重新稳定 - - if not new_stations_found and i > 0: - logger.info("未发现更多新场站，停止下拉。") - # 如果需要强行爬取所有，可以注释掉 break - # break - - # 5. 如果还没到最后一页，执行下拉 - if i < max_scrolls and not max_distance_reached: - logger.info(f"执行下拉翻页 (距离比例: {SCROLL_DISTANCE_RATIO})...") - # 根据配置的比例计算滑动起始和终点 - # 保证滑动在屏幕中心区域进行 - 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) # 等待页面加载和列表稳定 - - # 清理列表页截图 - if not KEEP_SCREENSHOTS: - if os.path.exists(screenshot_path): - os.remove(screenshot_path) - if os.path.exists(vl_img_path): - os.remove(vl_img_path) - if os.path.exists(json_path): - os.remove(json_path) - - logger.info(f"爬取任务结束，正在等待 {len(background_tasks)} 个后台处理任务完成...") + # 等待所有后台任务完成 if background_tasks: - await asyncio.gather(*background_tasks) - logger.info("所有后台任务已完成。") + logger.info(f"正在等待剩余 {len(background_tasks)} 个后台任务完成...") + await asyncio.gather(*background_tasks, return_exceptions=True) - logger.info(f"任务结束，共采集到 {len(all_stations)} 个场站。停止原因: {stop_reason}") - return all_stations + logger.info("采集任务完成") + return True def clean_images_dir(): diff --git a/Apps/XinDianTu/Kit.py b/Apps/XinDianTu/Kit.py index ff0bb94..3a9a161 100644 --- a/Apps/XinDianTu/Kit.py +++ b/Apps/XinDianTu/Kit.py @@ -2,7 +2,8 @@ import logging import os import cv2 import numpy as np -import time +import time +import hashlib from Apps.XinDianTu.Config.Setting import BOTTOM_SAFE_EXCLUDE_RATIO from Config.Config import TEMP_IMAGE_DIR @@ -11,24 +12,40 @@ logger = logging.getLogger(__name__) def read_image(path): - """读取图片，支持中文路径""" + """读取图片，支持中文路径及鲁棒性检查""" try: - return cv2.imdecode(np.fromfile(path, dtype=np.uint8), -1) + if not path or not os.path.exists(path): + return None + if os.path.getsize(path) == 0: + return None + return cv2.imdecode(np.fromfile(path, dtype=np.uint8), cv2.IMREAD_UNCHANGED) except Exception as e: - logger.info(f"Error reading image {path}: {e}") + logger.error(f"读取图片失败 {path}: {e}") return None def save_image(path, img): - """保存图片，支持中文路径""" + """保存图片，支持中文路径及鲁棒性检查""" try: + if img is None: + return False + + # 确保目录存在 + dir_name = os.path.dirname(path) + if dir_name and not os.path.exists(dir_name): + os.makedirs(dir_name, exist_ok=True) + ext = os.path.splitext(path)[1] if not ext: ext = ".jpg" - cv2.imencode(ext, img)[1].tofile(path) - return True + + success, encoded_img = cv2.imencode(ext, img) + if success: + encoded_img.tofile(path) + return True + return False except Exception as e: - logger.error(f"Error saving image {path}: {e}") + logger.error(f"保存图片失败 {path}: {e}") return False @@ -672,6 +689,29 @@ def get_row_stats(gray): return row_means, row_stds, edge_means +def clean_station_name(name): + """ + 清理场站名称，移除结尾的省略号及多余空格，方便比对 + """ + if not name: return "" + name = name.strip() + # 移除结尾的 . (通常是截断标识) + while name.endswith("."): + name = name[:-1] + return name.strip() + + +def get_file_md5(path): + """计算文件的 MD5 值""" + if not os.path.exists(path): + return "" + hash_md5 = hashlib.md5() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(4096), b""): + hash_md5.update(chunk) + return hash_md5.hexdigest() + + def crop_cards_from_image(img_path, output_dir=None, save_debug=True): """ 从图片中裁剪场站卡片 diff --git a/Apps/XinDianTu/Opener.py b/Apps/XinDianTu/Opener.py index a407df5..427981b 100644 --- a/Apps/XinDianTu/Opener.py +++ b/Apps/XinDianTu/Opener.py @@ -9,6 +9,7 @@ import uiautomator2 as u2 from Apps.XinDianTu.Kit import take_screenshot, detect_black_agree_button, detect_any_ad_close, \ detect_bottom_close_circle, is_background_dimmed, click_image_template +from Apps.XinDianTu.ReadImageKit import ReadImageKit from Config.Config import TEMP_IMAGE_DIR # pip install adbutils @@ -87,7 +88,28 @@ async def check_and_close_ad(d): os.remove(screenshot_path) return True - logger.info(f"本地图形学检测完成，未发现已知广告。") + # 3. 视觉大模型检测方案 (VL Model) - 作为兜底 + logger.info("本地图形学检测未发现广告，尝试调用视觉大模型检测...") + try: + window_size = d.window_size() + device_info = { + "width": window_size[0], + "height": window_size[1], + "productName": d.info.get('productName', 'unknown') + } + + ad_result = await ReadImageKit.detect_ad(screenshot_path, device_info=device_info) + if ad_result.get("has_ad") and ad_result.get("uia_center_x") is not None: + x, y = ad_result["uia_center_x"], ad_result["uia_center_y"] + logger.info(f"通过视觉大模型检测到广告关闭按钮: ({x}, {y})") + d.click(x, y) + if os.path.exists(screenshot_path): + os.remove(screenshot_path) + return True + except Exception as e: + logger.error(f"视觉大模型广告检测异常: {e}") + + logger.info(f"本地图形学和 VL 检测完成，未发现已知广告。") # 清理本地截图 if os.path.exists(screenshot_path): diff --git a/Apps/XinDianTu/ReadImageKit.py b/Apps/XinDianTu/ReadImageKit.py index efadee6..a363479 100644 --- a/Apps/XinDianTu/ReadImageKit.py +++ b/Apps/XinDianTu/ReadImageKit.py @@ -7,7 +7,7 @@ import json import aiohttp import logging import base64 -from openai import OpenAI, BadRequestError +from openai import AsyncOpenAI, BadRequestError from Config.Config import ( ALY_LLM_API_KEY, VL_MODEL_NAME, VL_MODEL_NAME_AD ) @@ -20,7 +20,7 @@ logger = logging.getLogger(__name__) class ReadImageKit: - _client = OpenAI( + _client = AsyncOpenAI( api_key=ALY_LLM_API_KEY, base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) @@ -207,11 +207,18 @@ class ReadImageKit: return obj @staticmethod - async def find_all_time_button_coordinate(image_url: str, device_info: dict = None) -> dict: + async def find_all_time_button_coordinate(image_url_or_path: str, device_info: dict = None) -> dict: if device_info is None: logger.warning("未提供动态设备信息，使用通用回退配置。") device_info = ReadImageKit._FALLBACK_DEVICE_INFO + if os.path.exists(image_url_or_path): + with open(image_url_or_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + else: + img_input = {"url": image_url_or_path} + prompt = ( "仅输出JSON对象（不含任何说明文字），请找到图片中带有“全部时段”字样的按钮区域（通常在价格表下方，是一个带有右箭头的文字按钮）。\n" "返回格式示例：\n" @@ -223,27 +230,21 @@ class ReadImageKit: "如果未找到，返回空JSON {}。" ) - loop = asyncio.get_event_loop() - resp = await loop.run_in_executor( - None, - lambda: ReadImageKit._client.chat.completions.create( - model=VL_MODEL_NAME, - messages=[ - { - "role": "user", - "content": [ - { - "type": "image_url", - "image_url": { - "url": image_url - }, - }, - {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, - {"type": "text", "text": prompt}, - ], - }, - ], - ) + resp = await ReadImageKit._client.chat.completions.create( + model=VL_MODEL_NAME, + messages=[ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": img_input, + }, + {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, + {"type": "text", "text": prompt}, + ], + }, + ], ) content = resp.choices[0].message.content or "" @@ -265,11 +266,18 @@ class ReadImageKit: return {} @staticmethod - async def find_station_coordinate_first_page(image_url: str, station_name: str, device_info: dict = None) -> dict: + async def find_station_coordinate_first_page(image_url_or_path: str, station_name: str, device_info: dict = None) -> dict: if device_info is None: logger.warning("未提供动态设备信息，使用通用回退配置。") device_info = ReadImageKit._FALLBACK_DEVICE_INFO + if os.path.exists(image_url_or_path): + with open(image_url_or_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + else: + img_input = {"url": image_url_or_path} + prompt = ( "仅输出JSON对象（不含任何说明文字）。" "请在图片中找到场站名称完全匹配“" + str(station_name) + "”的那一行/卡片，" @@ -280,22 +288,18 @@ class ReadImageKit: "注意：bounds使用0-1000归一化坐标空间；如果找不到，返回空JSON {}。" ) - loop = asyncio.get_event_loop() - resp = await loop.run_in_executor( - None, - lambda: ReadImageKit._client.chat.completions.create( - model=VL_MODEL_NAME, - messages=[ - { - "role": "user", - "content": [ - {"type": "image_url", "image_url": {"url": image_url}}, - {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, - {"type": "text", "text": prompt}, - ], - }, - ], - ) + resp = await ReadImageKit._client.chat.completions.create( + model=VL_MODEL_NAME, + messages=[ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": img_input}, + {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, + {"type": "text", "text": prompt}, + ], + }, + ], ) content = resp.choices[0].message.content or "" @@ -316,32 +320,33 @@ class ReadImageKit: return {} @staticmethod - async def parse_first_level_image_url(image_url: str, device_info: dict = None) -> list: + async def parse_first_level_image_url(image_url_or_path: str, device_info: dict = None) -> list: if device_info is None: logger.warning("未提供动态设备信息，使用通用回退配置。") device_info = ReadImageKit._FALLBACK_DEVICE_INFO - loop = asyncio.get_event_loop() - resp = await loop.run_in_executor( - None, - lambda: ReadImageKit._client.chat.completions.create( - model=VL_MODEL_NAME, - messages=[ - { - "role": "user", - "content": [ - { - "type": "image_url", - "image_url": { - "url": image_url - }, - }, - {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, - {"type": "text", "text": ReadImageKit._prompt}, - ], - }, - ], - ) + if os.path.exists(image_url_or_path): + with open(image_url_or_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + else: + img_input = {"url": image_url_or_path} + + resp = await ReadImageKit._client.chat.completions.create( + model=VL_MODEL_NAME, + messages=[ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": img_input, + }, + {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, + {"type": "text", "text": ReadImageKit._prompt}, + ], + }, + ], ) content = resp.choices[0].message.content or "" @@ -397,7 +402,7 @@ class ReadImageKit: if name: idxs.append(i) tasks.append(asyncio.create_task( - ReadImageKit.find_station_coordinate_first_page(image_url, name, device_info) + ReadImageKit.find_station_coordinate_first_page(image_url_or_path, name, device_info) )) if tasks: results = await asyncio.gather(*tasks, return_exceptions=True) @@ -421,13 +426,20 @@ class ReadImageKit: return await ReadImageKit.get_stations_hybrid(image_path, uploader, cdn_domain) @staticmethod - async def parse_vl_image(vl_image_url, json_metadata, device_info=None): + async def parse_vl_image(image_url_or_path: str, json_metadata, device_info=None): """ 基于 _vl.jpg (带绿框) 和 JSON 元数据进行识别 """ if not json_metadata or "cards" not in json_metadata: return [] + if os.path.exists(image_url_or_path): + with open(image_url_or_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + else: + img_input = {"url": image_url_or_path} + cards_meta = json_metadata["cards"] # 按 id 或 Y 坐标排序，确保顺序一致 (Kit 生成时已经是 top-down) cards_meta.sort(key=lambda x: x["rect"][1]) @@ -450,21 +462,17 @@ class ReadImageKit: "严格返回纯JSON格式。" ) - loop = asyncio.get_event_loop() - resp = await loop.run_in_executor( - None, - lambda: ReadImageKit._client.chat.completions.create( - model=VL_MODEL_NAME, - messages=[ - { - "role": "user", - "content": [ - {"type": "image_url", "image_url": {"url": vl_image_url}}, - {"type": "text", "text": prompt}, - ], - }, - ], - ) + resp = await ReadImageKit._client.chat.completions.create( + model=VL_MODEL_NAME, + messages=[ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": img_input}, + {"type": "text", "text": prompt}, + ], + }, + ], ) content = resp.choices[0].message.content or "" @@ -769,25 +777,21 @@ class ReadImageKit: """ 对切片进行小图 OCR 识别，提取场站详细信息 """ - loop = asyncio.get_event_loop() - resp = await loop.run_in_executor( - None, - lambda: ReadImageKit._client.chat.completions.create( - model=VL_MODEL_NAME, - messages=[ - { - "role": "user", - "content": [ - { - "type": "text", - "text": "请识别图片中的充电站信息，并以 JSON 格式输出：{\"station_name\": \"...\", \"price\": \"...\", \"piles\": \"空闲数/总数\"}。只输出 JSON，不要有其他文字。" - }, - {"type": "image_url", "image_url": {"url": patch_url}} - ] - } - ], - max_tokens=200 - ) + resp = await ReadImageKit._client.chat.completions.create( + model=VL_MODEL_NAME, + messages=[ + { + "role": "user", + "content": [ + { + "type": "text", + "text": "请识别图片中的充电站信息，并以 JSON 格式输出：{\"station_name\": \"...\", \"price\": \"...\", \"piles\": \"空闲数/总数\"}。只输出 JSON，不要有其他文字。" + }, + {"type": "image_url", "image_url": {"url": patch_url}} + ] + } + ], + max_tokens=200 ) content = resp.choices[0].message.content.strip() # 尝试解析 JSON @@ -803,12 +807,12 @@ class ReadImageKit: return {"station_name": "未知"} @staticmethod - async def detect_ad(image_url: str, device_info: dict = None) -> dict: + async def detect_ad(image_url_or_path: str, device_info: dict = None) -> dict: """ 检测图片中是否存在广告弹窗，并定位关闭按钮坐标 参数： - image_url: 图片地址 + image_url_or_path: 图片地址或本地路径 device_info: 设备信息 返回： @@ -822,34 +826,37 @@ class ReadImageKit: logger.warning("未提供动态设备信息，使用通用回退配置。") device_info = ReadImageKit._FALLBACK_DEVICE_INFO + if os.path.exists(image_url_or_path): + with open(image_url_or_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + else: + img_input = {"url": image_url_or_path} + prompt = ( "分析图片中是否存在覆盖在主界面上的广告弹窗（Popup Ad）。" - "如果存在，请找到关闭该弹窗的按钮（通常是一个带有 'X' 的图标，可能在弹窗的右上角、右下角或正下方）。" + "如果存在，请找到关闭该弹窗的按钮（通常是一个带有 'X' 的图标，可能在弹窗的右上角、右下角 or 正下方）。" "仅输出JSON对象（不含任何说明文字），包含以下字段：" "1. has_ad: 布尔值，是否存在广告弹窗；" "2. close_button_bounds: 关闭按钮的像素坐标或归一化坐标(0-1000) {x1,y1,x2,y2}。如果不存在广告则为 null。" "注意：只需识别最明显的那个关闭按钮。严格返回纯JSON。" ) - loop = asyncio.get_event_loop() - resp = await loop.run_in_executor( - None, - lambda: ReadImageKit._client.chat.completions.create( - model=VL_MODEL_NAME_AD, # 使用更强的视觉模型 - messages=[ - { - "role": "user", - "content": [ - { - "type": "image_url", - "image_url": {"url": image_url}, - }, - {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, - {"type": "text", "text": prompt}, - ], - }, - ], - ) + resp = await ReadImageKit._client.chat.completions.create( + model=VL_MODEL_NAME_AD, # 使用更强的视觉模型 + messages=[ + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": img_input, + }, + {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, + {"type": "text", "text": prompt}, + ], + }, + ], ) content = resp.choices[0].message.content or "" @@ -889,32 +896,48 @@ class ReadImageKit: return None @staticmethod - async def parse_price_schedule(station_name: str, image_url: str, device_info: dict = None) -> list: + async def get_image_md5(image_url_or_path: str) -> str: """ - 解析价格时段表（整图），提取每一行的时间区间与费用信息 - - 参数： - image_url: 图片地址（包含“当前时段电费/全部时段电费”弹窗或列表） - device_info: 设备信息，用于提升视觉理解一致性（可选） - - 返回： - 列表，每个元素为： - { - "start": "HH:MM", 开始时间 - "end": "HH:MM", 结束时间 - "price_kwh": float|None, 总价或站点价（元/度） - "electric_fee_kwh": float|None, 电费（元/度） - "service_fee_kwh": float|None 服务费（元/度） - } - 说明： - - 使用 qwen3-vl-flash 进行视觉解析，提示词约束输出为纯 JSON 数组 - - 若某项缺失则返回 None（保持结构统一） + 计算图片的 MD5（用于内容去重），支持 URL 或 本地路径 + """ + if not image_url_or_path: + return "" + + # 如果是本地文件 + if os.path.exists(image_url_or_path): + from Apps.XinDianTu.Kit import get_file_md5 + return get_file_md5(image_url_or_path) + + # 如果是 URL + try: + async with aiohttp.ClientSession() as session: + async with session.get(image_url_or_path, timeout=10) as resp: + if resp.status != 200: + return "" + content = await resp.read() + return hashlib.md5(content).hexdigest() + except Exception as e: + logger.error(f"MD5 获取失败 ({image_url_or_path}): {e}") + return "" + + @staticmethod + async def parse_price_schedule(station_name: str, image_url_or_path: str, device_info: dict = None) -> list: + """ + 解析价格时段表，支持 URL 或 本地路径 """ if device_info is None: logger.warning("未提供动态设备信息，使用通用回退配置。") device_info = ReadImageKit._FALLBACK_DEVICE_INFO - # 视觉解析提示词：约束输出字段与格式，避免模型输出说明文字 + # 处理图片输入 (URL or Local Path) + if os.path.exists(image_url_or_path): + with open(image_url_or_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + else: + img_input = {"url": image_url_or_path} + + # 视觉解析提示词 prompt = ( "仅输出JSON数组（不含任何说明文字）。识别图片中所有时段的价格信息，返回每一行：" "1) start: 开始时间（HH:MM），2) end: 结束时间（HH:MM），" @@ -925,205 +948,46 @@ class ReadImageKit: "注意：如果某行价格信息为空或表示同上，请尝试复用上一行的价格信息。" ) - loop = asyncio.get_event_loop() - - def _do_request(url_val): - return ReadImageKit._client.chat.completions.create( + try: + resp = await ReadImageKit._client.chat.completions.create( model=VL_MODEL_NAME, messages=[ { "role": "user", "content": [ - { - "type": "image_url", - "image_url": {"url": url_val}, - }, + {"type": "image_url", "image_url": img_input}, {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, {"type": "text", "text": prompt}, ], }, ], ) - - try: - # 在线程池中同步调用 OpenAI 兼容接口，避免阻塞事件循环 - resp = await loop.run_in_executor(None, lambda: _do_request(image_url)) - except BadRequestError as e: - # 尝试捕获 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', '') - - # 如果是媒体格式或数据检查错误 - 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}") - base64_str = await ReadImageKit._download_as_base64(image_url) - if base64_str: - # 简单推断格式，默认 jpeg - ext = "jpeg" - lower_url = image_url.lower() - if ".png" in lower_url: - ext = "png" - elif ".webp" in lower_url: - 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: - logger.error("Base64 download failed during fallback.") - return [] - else: - logger.error(f"API BadRequestError: {e}") - return [] + content = resp.choices[0].message.content or "" + raw = ReadImageKit._extract_json(content) + return json.loads(raw) except Exception as e: - logger.error(f"API Unexpected Error: {e}") - return [] - - content = resp.choices[0].message.content or "" - raw = ReadImageKit._extract_json(content) - - try: - # 反序列化模型返回的 JSON - rows = json.loads(raw) - if not isinstance(rows, list): - return [] - norm = [] - for r in rows: - if not isinstance(r, dict): - continue - # 兼容不同来源字段命名，统一为目标键 - start = r.get("start") - end = r.get("end") - price = r.get("price_kwh") - elec = r.get("electric_fee_kwh") - serv = r.get("service_fee_kwh") - # station_name = station_name - # 将字符串数字安全转换为 float；缺失则为 None - norm.append({ - "station_name": station_name, - "start": start, - "end": end, - "price_kwh": float(price) if isinstance(price, (int, float, str)) and str(price) else None, - "electric_fee_kwh": float(elec) if isinstance(elec, (int, float, str)) and str(elec) else None, - "service_fee_kwh": float(serv) if isinstance(serv, (int, float, str)) and str(serv) else None, - }) - - for i in range(1, len(norm)): - curr = norm[i] - prev = norm[i - 1] - for k in ("price_kwh", "electric_fee_kwh", "service_fee_kwh"): - if curr.get(k) is None and prev.get(k) is not None: - curr[k] = prev[k] - - return norm - except Exception as e: - logger.error(f"Error parsing JSON: {e}") - logger.error(f"Raw content: {raw}") + logger.error(f"解析价格失败 ({station_name}): {e}") return [] @staticmethod - def _to_minutes(t: str) -> int: + async def parse_address(station_name: str, image_url_or_path: str, device_info: dict = None) -> dict: """ - 将 "HH:MM" 转为分钟数（0-1440） - 说明： - - 特殊处理 "24:00" -> 1440，方便区间闭合处理 - - 非法格式返回 0 - """ - if not t: - return 0 - try: - h, m = t.split(":") - h = int(h) - m = int(m) - if h == 24 and m == 0: - return 24 * 60 - return h * 60 + m - except Exception: - return 0 - - @staticmethod - def _fmt(t: int) -> str: - """ - 将分钟数格式化为 "HH:MM" - """ - h = t // 60 - m = t % 60 - return f"{h:02d}:{m:02d}" - - @staticmethod - def expand_schedule_to_hourly(rows: list) -> list: - """ - 将时段列表按小时边界拆分 - - 参数： - rows: parse_price_schedule 返回的时段列表 - - 返回： - 每小时一条数据的列表，区间为闭开 [start, end) 的连续小时段 - 说明： - - 例如 05:00-08:00 -> 05:00-06:00, 06:00-07:00, 07:00-08:00 - - 保留每小时的价格、电费、服务费不变 - """ - hourly = [] - for r in rows: - # 起止时间转分钟 - s = ReadImageKit._to_minutes(r.get("start")) - e = ReadImageKit._to_minutes(r.get("end")) - if e <= s: - continue - cur = s - while cur < e: - # 下一小时边界；不超过区间终点 - nxt = min(e, ((cur // 60) + 1) * 60) - hourly.append({ - "start": ReadImageKit._fmt(cur), - "end": ReadImageKit._fmt(nxt), - "price_kwh": r.get("price_kwh"), - "electric_fee_kwh": r.get("electric_fee_kwh"), - "service_fee_kwh": r.get("service_fee_kwh"), - }) - cur = nxt - return hourly - - @staticmethod - async def _fetch_md5(url: str) -> str: - """ - 下载图片并计算 MD5（用于内容去重） - - 返回： - 32位十六进制 MD5 字符串；失败返回空字符串 - """ - try: - async with aiohttp.ClientSession() as session: - async with session.get(url) as resp: - if resp.status != 200: - return "" - content = await resp.read() - return hashlib.md5(content).hexdigest() - except Exception: - return "" - - @staticmethod - async def parse_address(station_name: str, image_url: str, device_info: dict = None) -> dict: - """ - 解析图片中的充电站地址信息，同时识别“全部时段”按钮坐标 - - 参数： - station_name:场站名称 - image_url: 图片地址 - device_info: 设备信息（可选） - - 返回： - 字典，包含 address 字段以及 uia_center_x/uia_center_y (如果找到按钮) + 解析地址信息，支持 URL 或 本地路径 """ if device_info is None: logger.warning("未提供动态设备信息，使用通用回退配置。") device_info = ReadImageKit._FALLBACK_DEVICE_INFO - # 启动寻找“全部时段”按钮的任务，为后续可能的点击做准备 - button_task = asyncio.create_task(ReadImageKit.find_all_time_button_coordinate(image_url, device_info)) + # 处理图片输入 (URL or Local Path) + if os.path.exists(image_url_or_path): + with open(image_url_or_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + else: + img_input = {"url": image_url_or_path} + + # 寻找按钮坐标任务 + button_task = asyncio.create_task(ReadImageKit.find_all_time_button_coordinate(image_url_or_path, device_info)) prompt = ( "仅输出JSON对象（不含任何说明文字）。" @@ -1136,78 +1000,56 @@ class ReadImageKit: "严格返回纯JSON格式。" ) - loop = asyncio.get_event_loop() - resp = await loop.run_in_executor( - None, - lambda: ReadImageKit._client.chat.completions.create( + result = {} + try: + resp = await ReadImageKit._client.chat.completions.create( model=VL_MODEL_NAME, messages=[ { "role": "user", "content": [ - { - "type": "image_url", - "image_url": {"url": image_url}, - }, + {"type": "image_url", "image_url": img_input}, {"type": "text", "text": json.dumps(device_info, ensure_ascii=False)}, {"type": "text", "text": prompt}, ], }, ], ) - ) - - content = resp.choices[0].message.content or "" - raw = ReadImageKit._extract_json(content) - - result = {} - try: - data = json.loads(raw) - if isinstance(data, dict): - result = data + content = resp.choices[0].message.content or "" + raw = ReadImageKit._extract_json(content) + result = json.loads(raw) except Exception as e: - logger.error(f"Error parsing address JSON: {e}") + logger.error(f"解析地址失败 ({station_name}): {e}") - # 等待并合并按钮坐标结果 + # 合并按钮坐标 try: button_result = await button_task if button_result: result.update(button_result) except Exception as e: - logger.error(f"Error in button coordinate task: {e}") + logger.error(f"按钮坐标获取失败: {e}") return result @staticmethod - async def parse_price_schedule_multi(station_name: str, image_urls: list, device_info: dict = None) -> list: + async def parse_price_schedule_multi(station_name: str, image_urls_or_paths: list, device_info: dict = None) -> list: """ - 多图解析（按图片内容 MD5 去重）并合并时段结果 - - 参数： - image_urls: 多张价格表图片的 URL 列表 - device_info: 设备信息（可选） - - 逻辑： - 1. 逐张下载并计算 MD5，去重得到唯一图片集合 - 2. 对每张唯一图片解析价格时段 - 3. 将所有图片的时段行合并为一个列表返回 - - 返回： - 合并后的时段列表（未按小时拆分） + 多图解析并去重合并 """ - if not image_urls: + if not image_urls_or_paths: return [] - # MD5 -> URL 的映射，用于去重 - md5_to_url = {} - for u in image_urls: - m = await ReadImageKit._fetch_md5(u) - if m and m not in md5_to_url: - md5_to_url[m] = u - unique_urls = list(md5_to_url.values()) - # 合并时段结果 + + # MD5 去重 + md5_to_img = {} + for img in image_urls_or_paths: + m = await ReadImageKit.get_image_md5(img) + if m and m not in md5_to_img: + md5_to_img[m] = img + + unique_imgs = list(md5_to_img.values()) combined = [] - for u in unique_urls: - rows = await ReadImageKit.parse_price_schedule(station_name, u, device_info=device_info) + for img in unique_imgs: + rows = await ReadImageKit.parse_price_schedule(station_name, img, device_info=device_info) if rows: combined.extend(rows) return combined diff --git a/Apps/XinDianTu/Service.py b/Apps/XinDianTu/Service.py index 19f10a2..92da17e 100644 --- a/Apps/XinDianTu/Service.py +++ b/Apps/XinDianTu/Service.py @@ -42,11 +42,11 @@ class XinDianTuService: async def close_db(self): await self.db.close() - async def process_station_list_vl(self, vl_image_url, json_metadata, device_info=None) -> list: + async def process_station_list_vl(self, vl_image_url_or_path, json_metadata, device_info=None) -> list: """ 基于 VL 模式处理场站列表 """ - station_list = await ReadImageKit.parse_vl_image(vl_image_url, json_metadata, device_info=device_info) + station_list = await ReadImageKit.parse_vl_image(vl_image_url_or_path, json_metadata, device_info=device_info) if not station_list: return [] @@ -490,6 +490,28 @@ class XinDianTuService: await session.commit() logger.info(f" - Station status recorded for: {name} (Piles: {free}/{total}, Price: {current_price})") + async def process_price_detail_data(self, station_name, hourly_schedule) -> bool: + """ + 直接保存已处理好的小时段价格数据 + """ + if not station_name or not hourly_schedule: + return False + + station_hash = self.get_hash(station_name) + now = datetime.now() + + async with await self.db.get_session() as session: + schedule_id = self.generate_id() + await self.station_price_schedule_model.save( + session=session, + id=schedule_id, + station_hash=station_hash, + schedule_json=hourly_schedule, + valid_start_time=now + ) + await session.commit() + return True + async def process_price_schedule(self, station_name: str, schedule_urls: list, device_info: dict = None): logger.info(f"\n[Service] Processing Price Schedule for '{station_name}'")