From c36ab00318cd5d691b3806e5a7398962b16544cb Mon Sep 17 00:00:00 2001 From: HuangHai <10402852@qq.com> Date: Tue, 13 Jan 2026 11:55:08 +0800 Subject: [PATCH] 'commit' --- Apps/AiTeJiYiChong/Crawler.py | 289 +++++++------- Apps/AiTeJiYiChong/Kit.py | 6 + Apps/AiTeJiYiChong/Opener.py | 45 +++ Apps/AiTeJiYiChong/ReadImageKit.py | 366 ++++++++---------- Apps/AiTeJiYiChong/Service.py | 140 +------ .../__pycache__/Setting.cpython-310.pyc | Bin 533 -> 533 bytes .../__pycache__/Crawler.cpython-310.pyc | Bin 13499 -> 13457 bytes 7 files changed, 352 insertions(+), 494 deletions(-) diff --git a/Apps/AiTeJiYiChong/Crawler.py b/Apps/AiTeJiYiChong/Crawler.py index d181818..a0fdb12 100644 --- a/Apps/AiTeJiYiChong/Crawler.py +++ b/Apps/AiTeJiYiChong/Crawler.py @@ -69,38 +69,6 @@ async def get_station_list(d, service, max_scrolls=MAX_SCROLLS): image_uuid = str(uuid.uuid4()) screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR) - # 1.5 检查是否存在广告 (VL) - logger.info("检查是否存在广告弹窗...") - ad_res = await ReadImageKit.detect_ad_popup(screenshot_path, device_info=device_info) - if ad_res: - ad_type = ad_res.get("ad_type") - logger.info(f"检测到广告弹窗，准备关闭: {ad_res} [Type: {ad_type}]") - - x, y = ad_res['x'], ad_res['y'] - - if ad_type == "rabbit": - logger.info(">>> 检测到顽固的兔子广告，执行强力双击策略...") - d.double_click(x, y, duration=0.05) - - # 处理“误触背景卡片”的逻辑 - logger.info(">>> 等待 3s 以确认是否进入了二级页面...") - await asyncio.sleep(3.0) - - # 爱特极特征元素检测 (列表页特征) - is_list_page = d(text="筛选").exists or d(text="排序").exists or d(text="搜索").exists - if not is_list_page: - logger.info(">>> 检测到当前不在列表页，尝试执行返回操作...") - d.press("back") - await asyncio.sleep(1.5) - else: - logger.info(">>> 当前仍在列表页，无需返回。") - else: - d.click(x, y) - await asyncio.sleep(1.5) - # 重新截图 - if os.path.exists(screenshot_path): os.remove(screenshot_path) - screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR) - # 检查是否已经滚动到底部 (排除状态栏后，内容与上次一致) current_md5 = Kit.get_image_content_md5( screenshot_path, @@ -115,141 +83,152 @@ async def get_station_list(d, service, max_scrolls=MAX_SCROLLS): logger.info(f"列表页截图已完成: {screenshot_path}") - # 2. 调用混合模式识别 (内部包含图形学切片 + VL 模型) - logger.info("正在执行混合模式识别 (切片 + VL)...") - stations = await service.process_station_list_hybrid(screenshot_path, device_info=device_info) - logger.info(f"本页识别到 {len(stations)} 个有效场站卡片") + # 2. 执行本地图形学分析，生成 _vl.jpg 和 .json + logger.info("正在执行图形学切片分析...") + json_metadata = Kit.crop_cards_from_image(screenshot_path, save_debug=True) + + vl_img_path = screenshot_path.replace(".jpg", "_vl.jpg") - # 3. 遍历处理本页所有场站 - for card_idx, station in enumerate(stations): - station_name = station.get("station_name") - if not station_name: - continue - - # 检查 Redis 去重 - if await is_already_crawled(redis_kit, station_name): - logger.info(f"场站 {station_name} 匹配到已处理记录，跳过。") - continue + if not json_metadata or not json_metadata.get("cards"): + logger.warning("未识别到任何卡片，跳过当前页") + continue - # 使用识别结果中自带的点击坐标 - click_x = station.get("uia_center_x") - click_y = station.get("uia_center_y") - - if click_x is None or click_y is None: - logger.warning(f"场站 {station_name} 缺失点击坐标，跳过。") - continue + # 3. 后台异步上传 _vl.jpg 供以后参考 (如果需要，目前仅本地处理) + + # 4. 调用 VL 模型识别场站信息 (整页识别) + logger.info("正在调用 VL 模型识别场站信息 (整页识别)...") + # 优先使用带绿框的图片进行识别 + img_to_process = vl_img_path if os.path.exists(vl_img_path) else screenshot_path + stations = await service.process_station_list_vl(img_to_process, json_metadata, device_info=device_info) + logger.info(f"本页识别到 {len(stations)} 个有效场站") - logger.info(f"准备处理第 {card_idx + 1} 个场站: {station_name}, 点击坐标: ({click_x}, {click_y})") - d.click(click_x, click_y) - - # 等待二级页面加载 - await asyncio.sleep(3) - - should_back_to_list = True + ad_top_y_norm = 0.78 # 默认点击边界 - # 截取二级页面图 - detail_uuid = f"detail_{station_name}_{image_uuid}" - detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR) - - # 后台分析详情页基础数据 - logger.info(f"已启动后台分析详情页: {station_name}") - - async def detail_task_wrapper(p, name): - try: - await service.process_station_detail(p, station_name=name) - if os.path.exists(p): os.remove(p) - except Exception as ex: - logger.error(f"详情页分析异常 ({name}): {ex}") - - task_detail = asyncio.create_task(detail_task_wrapper(detail_path, station_name)) - background_tasks.append(task_detail) - - # 4. 寻找分时价格入口并进入三级页面 - # 优先使用 VL 模型精准识别，避免误判导致退回微信 - res = await ReadImageKit.find_time_price_button_coordinate(detail_path, device_info=device_info) - pos = (res.get("uia_center_x"), res.get("uia_center_y")) if res.get("uia_center_x") else None - - entered_price_page = False - if pos and pos[0] is not None and pos[1] is not None: - ex, ey = pos - logger.info(f"找到 '分时价格' 按钮: ({ex}, {ey})") - - # 记录点击前的页面特征，用于验证是否成功进入三级页面 - before_click_md5 = Kit.get_image_content_md5(detail_path) - - d.click(ex, ey) - await asyncio.sleep(2) - - # 检查页面是否真的变化了 - after_click_path = take_screenshot(d, f"check_{detail_uuid}", save_dir=TEMP_IMAGE_DIR) - after_click_md5 = Kit.get_image_content_md5(after_click_path) - - if before_click_md5 != after_click_md5: - entered_price_page = True - # 抓取三级页面图片 - price_screenshots = [] - price_screenshots.append(after_click_path) + # 5. 遍历处理本页所有场站 + new_stations_processed = 0 + if json_metadata.get("cards") and stations: + for card_idx, card in enumerate(json_metadata["cards"]): + # 检查索引是否越界 (VL 模型可能返回的数组长度不一致) + if card_idx < len(stations) and stations[card_idx]: + station = stations[card_idx] + station_name = station.get("station_name") + if not station_name: continue - # 滑动处理 - last_price_md5 = after_click_md5 - for scroll_idx in range(2): - d.swipe(w // 2, int(h * 0.7), w // 2, int(h * 0.3), duration=0.4) - await asyncio.sleep(1) - scroll_path = take_screenshot(d, f"price_scroll_{scroll_idx}_{station_name}", save_dir=TEMP_IMAGE_DIR) - curr_price_md5 = Kit.get_image_content_md5( - scroll_path, - top_ratio=SAFE_EXCLUDE_RATIO, - bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO - ) - if curr_price_md5 == last_price_md5: break - price_screenshots.append(scroll_path) - last_price_md5 = curr_price_md5 + # 检查 Redis 去重 + if await is_already_crawled(redis_kit, station_name): + logger.info(f"场站 {station_name} 匹配到已处理记录，跳过。") + continue - # 后台处理价格图片 - logger.info(f"已启动后台分析价格页: {station_name}，共 {len(price_screenshots)} 张图") - task_price = asyncio.create_task(analyze_prices_background(service, station_name, price_screenshots)) - background_tasks.append(task_price) + # 检查是否被遮挡或太靠近底部 + card_bottom = card["bounds_norm"]["bottom"] + if card_bottom > ad_top_y_norm: + logger.warning(f"场站 '{station_name}' 被遮挡或太靠近底部 (Bottom {card_bottom:.2f} > {ad_top_y_norm})，留待下次滚动处理。") + continue + + # 正常处理新场站 + click_x, click_y = card["click_point"] + logger.info(f"准备处理第 {card_idx + 1} 个场站: {station_name}, 点击坐标: ({click_x}, {click_y})") + new_stations_processed += 1 + d.click(click_x, click_y) - # 立即返回 - d.press("back") - await asyncio.sleep(0.5) - - # 检查返回后的页面状态 - check_back_path = take_screenshot(d, f"check_back_{detail_uuid}", save_dir=TEMP_IMAGE_DIR) + # 等待二级页面加载 + await asyncio.sleep(3) - check_list_md5 = Kit.get_image_content_md5( - check_back_path, - top_ratio=SAFE_EXCLUDE_RATIO, - bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO - ) - check_detail_md5 = Kit.get_image_content_md5(check_back_path) + should_back_to_list = True + + # 截取二级页面图 + detail_uuid = f"detail_{station_name}_{image_uuid}" + detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR) - if os.path.exists(check_back_path): os.remove(check_back_path) + # 后台分析详情页基础数据 + logger.info(f"已启动后台分析详情页: {station_name}") + + async def detail_task_wrapper(p, name): + try: + await service.process_station_detail(p, station_name=name) + if os.path.exists(p): os.remove(p) + except Exception as ex: + logger.error(f"详情页分析异常 ({name}): {ex}") - if check_list_md5 == current_md5: - logger.info("检测到已直接返回列表页，跳过后续返回操作。") - should_back_to_list = False - elif check_detail_md5 == before_click_md5: - logger.info("检测到返回详情页，需执行第二次返回。") - should_back_to_list = True - else: - logger.warning(f"返回后状态未知 (ListMD5:{check_list_md5[:8]} vs {current_md5[:8]}, DetailMD5:{check_detail_md5[:8]} vs {before_click_md5[:8]})，默认执行返回。") - should_back_to_list = True + task_detail = asyncio.create_task(detail_task_wrapper(detail_path, station_name)) + background_tasks.append(task_detail) - else: - logger.warning(f"点击 '{station_name}' 的 '分时价格' 按钮后页面无明显变化，跳过三级页面处理。") - if os.path.exists(after_click_path): os.remove(after_click_path) - else: - logger.warning(f"场站 {station_name} 未找到分时价格入口。") + # 4. 寻找分时价格入口并进入三级页面 + # 优先使用 VL 模型精准识别，避免误判导致退回微信 + res = await ReadImageKit.find_time_price_button_coordinate(detail_path, device_info=device_info) + pos = (res.get("uia_center_x"), res.get("uia_center_y")) if res.get("uia_center_x") else None + + entered_price_page = False + if pos and pos[0] is not None and pos[1] is not None: + ex, ey = pos + logger.info(f"找到 '分时价格' 按钮: ({ex}, {ey})") + + # 记录点击前的页面特征，用于验证是否成功进入三级页面 + before_click_md5 = Kit.get_image_content_md5(detail_path) + + d.click(ex, ey) + await asyncio.sleep(2) + + # 检查页面是否真的变化了 + after_click_path = take_screenshot(d, f"check_{detail_uuid}", save_dir=TEMP_IMAGE_DIR) + after_click_md5 = Kit.get_image_content_md5(after_click_path) + + if before_click_md5 != after_click_md5: + entered_price_page = True + # 抓取三级页面图片 + price_screenshots = [] + price_screenshots.append(after_click_path) + + # 滑动处理 + last_price_md5 = after_click_md5 + for scroll_idx in range(2): + d.swipe(w // 2, int(h * 0.7), w // 2, int(h * 0.3), duration=0.4) + await asyncio.sleep(1) + scroll_path = take_screenshot(d, f"price_scroll_{scroll_idx}_{station_name}", save_dir=TEMP_IMAGE_DIR) + curr_price_md5 = Kit.get_image_content_md5( + scroll_path, + top_ratio=SAFE_EXCLUDE_RATIO, + bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO + ) + if curr_price_md5 == last_price_md5: break + price_screenshots.append(scroll_path) + last_price_md5 = curr_price_md5 - # 从二级页面返回列表页 - if should_back_to_list: - d.press("back") - await asyncio.sleep(0.8) - - # 记录 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) + # 后台处理价格图片 + logger.info(f"已启动后台分析价格页: {station_name}，共 {len(price_screenshots)} 张图") + task_price = asyncio.create_task(analyze_prices_background(service, station_name, price_screenshots)) + background_tasks.append(task_price) + + # 立即返回 + d.press("back") + await asyncio.sleep(0.5) + + # 检查返回后的页面状态 + check_back_path = take_screenshot(d, f"check_back_{detail_uuid}", save_dir=TEMP_IMAGE_DIR) + + check_list_md5 = Kit.get_image_content_md5( + check_back_path, + top_ratio=SAFE_EXCLUDE_RATIO, + bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO + ) + check_detail_md5 = Kit.get_image_content_md5(check_back_path) + + if os.path.exists(check_back_path): os.remove(check_back_path) + + if check_list_md5 == current_md5: + logger.info("检测到已直接返回列表页，跳过后续返回操作。") + should_back_to_list = False + elif check_detail_md5 == before_click_md5: + logger.info("检测到返回详情页，需执行第二次返回。") + + # 从二级页面返回列表页 + if should_back_to_list: + d.press("back") + await asyncio.sleep(0.8) + + # 记录 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) # 在每一页结束时，清理已完成的任务 done_tasks = [t for t in background_tasks if t.done()] diff --git a/Apps/AiTeJiYiChong/Kit.py b/Apps/AiTeJiYiChong/Kit.py index e4e9f19..406e243 100644 --- a/Apps/AiTeJiYiChong/Kit.py +++ b/Apps/AiTeJiYiChong/Kit.py @@ -399,6 +399,12 @@ def crop_cards_from_image(img_path, output_dir=None, save_debug=True): json_data["cards"].append({ "id": idx + 1, "rect": [int(x1), int(y1), int(x2), int(y2)], + "bounds_norm": { + "left": round(float(x1) / w, 4), + "top": round(float(y1) / h, 4), + "right": round(float(x2) / w, 4), + "bottom": round(float(y2) / h, 4) + }, "click_point": [int(click_x), int(click_y)] }) diff --git a/Apps/AiTeJiYiChong/Opener.py b/Apps/AiTeJiYiChong/Opener.py index 2f52624..26b1d3f 100644 --- a/Apps/AiTeJiYiChong/Opener.py +++ b/Apps/AiTeJiYiChong/Opener.py @@ -4,7 +4,9 @@ import logging import os import time import uiautomator2 as u2 +import uuid from Apps.AiTeJiYiChong.Kit import click_image_template, take_screenshot +from Apps.AiTeJiYiChong.ReadImageKit import ReadImageKit from Config.Config import TEMP_IMAGE_DIR logging.basicConfig( @@ -15,6 +17,45 @@ logger = logging.getLogger("OpenAiTeJiYiChong") BASE_DIR = os.path.dirname(os.path.abspath(__file__)) +async def check_and_close_ad(d): + """ + 检测并关闭广告弹窗（循环检测直至无广告） + """ + max_loops = 3 # 最大检测 3 轮，防止死循环 + device_info = d.info + w, h = d.window_size() + device_info['displayWidth'] = w + device_info['displayHeight'] = h + + for loop_idx in range(max_loops): + logger.info(f"开始第 {loop_idx + 1} 轮广告检测...") + + image_uuid = f"ad_check_{int(time.time())}" + screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR) + + ad_result = await ReadImageKit.detect_ad_popup(screenshot_path, device_info=device_info) + + if ad_result: + x, y = ad_result['x'], ad_result['y'] + ad_type = ad_result.get("ad_type", "unknown") + logger.info(f"检测到广告 [{ad_type}]，坐标: ({x}, {y})，执行关闭操作...") + + if ad_type == "rabbit": + # 针对“兔子”广告的特殊处理 + d.double_click(x, y, duration=0.05) + else: + d.click(x, y) + + await asyncio.sleep(2.0) + + # 清理截图并继续下一轮检测 + if os.path.exists(screenshot_path): os.remove(screenshot_path) + else: + logger.info("未检测到广告弹窗。") + if os.path.exists(screenshot_path): os.remove(screenshot_path) + break + return True + async def open_mini_program(): """ 异步形式的进入微信小程序: 艾特吉易充 @@ -48,6 +89,10 @@ async def open_mini_program(): logger.info("已点击搜索结果第一项") await asyncio.sleep(8) + + # 5. 进入后的广告检测 (只在进入时执行一次) + await check_and_close_ad(d) + return True if __name__ == "__main__": diff --git a/Apps/AiTeJiYiChong/ReadImageKit.py b/Apps/AiTeJiYiChong/ReadImageKit.py index ef55e72..7b77a4f 100644 --- a/Apps/AiTeJiYiChong/ReadImageKit.py +++ b/Apps/AiTeJiYiChong/ReadImageKit.py @@ -48,39 +48,6 @@ class ReadImageKit: "productName": "generic" } - _prompt = ( - "仅输出JSON数组（不含任何说明文字），按从左到右、从上到下的顺序识别图片中由【绿色方框】标识的充电站区域。识别规则如下：\n" - "1. 必须是图中用绿色实线方框圈出的区域。\n" - "2. 每一个卡片区域必须同时具备以下所有要素，否则严禁识别：\n" - " - 场站名称 (station_name)；\n" - " - 距离信息 (distance, 例如 '2.82km' 或 '90m')，通常位于卡片右侧蓝色胶囊区域内；\n" - " - 金额/电费 (price，例如 '1.2500')，通常以红色字体显示；\n" - " - **充电枪信息 (piles)**: 必须包含类型('快充'、'慢充'、'超充')、总枪数和空闲枪数。识别规则如下：\n" - " - 如果明确标有'快'、'慢'或'超'，则对应为'快充'、'慢充'、'超充'。\n" - " - 如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢充\"`。\n" - " - 必须以 JSON 对象列表形式输出，例如：`[{\"type\": \"快充\", \"free\": 4, \"total\": 4}, {\"type\": \"慢充\", \"free\": 1, \"total\": 2}]`。\n" - "3. 如果绿色方框内缺少上述任何一项要素，说明它不是真正的场站卡片，请直接跳过。\n" - "\n" - "JSON对象字段要求：\n" - "1. b_use: 状态标识（1或0）。如果场站名称为灰色或带有“暂停使用”等标签，则为0，否则为1。\n" - "2. station_name: 场站名称；\n" - "3. price: 一度电的价格（数字，如 1.2500）；\n" - "4. piles: 充电枪列表 [{type: '快'|'慢'|'超', free: 4, total: 4}]；\n" - "5. parking: 停车费用描述（通常在蓝色'P'图标后，例如 '免费停车三小时'）；\n" - "6. distance: 距离信息字符串（例如 '2.82km' 或 '90m'）；\n" - "7. bounds: {x1,y1,x2,y2} 区域像素坐标（0-1000）；\n" - "8. bounds_norm: {left,top,right,bottom} 归一化坐标(0-1)；\n" - "9. station_name_bounds: 场站名称文字区域坐标 {x1,y1,x2,y2}（0-1000）；\n" - "10. station_name_bounds_norm: 场站名称文字归一化坐标(0-1)。\n" - "\n" - "重要约束：\n" - "A. 严禁识别未被绿色方框圈出的区域。如顶部的“长春市”选择、顶部的搜索框、以及中间的“推荐站点”等标签。\n" - "B. 真正的场站卡片在绿色方框内包含：场站名称、金额（红色文字）、距离（右侧蓝色背景内）、充电枪状态（绿色或蓝色徽章，格式为 闲x/x 或 x/x）。\n" - "C. 严禁将顶部的功能图标（如我的订单、收藏站点等）误认为场站卡片。\n" - "\n" - "严格返回纯JSON格式。" - ) - @staticmethod def _extract_json(text: str) -> str: if not text: @@ -126,6 +93,79 @@ class ReadImageKit: "- 严禁输出 Markdown 代码块标签，严格返回纯JSON对象。" ) + @classmethod + async def detect_ad_popup(cls, image_path: str, device_info=None): + """ + 使用 VL 模型检测是否存在弹窗广告，并返回关闭按钮坐标 + """ + if not os.path.exists(image_path): + return None + + with open(image_path, "rb") as image_file: + encoded_image = base64.b64encode(image_file.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded_image}"} + + # 构建提示词 + prompt = ( + "请仔细检查这张图片中是否存在**弹窗广告**或**悬浮广告**。\n" + "广告可能有以下几种形式：\n" + "1. **屏幕中央的大型弹窗广告**：通常遮挡了页面内容，内容多为优惠券、活动推广、或“免费停车”类提示。\n" + "2. **悬浮广告**：一个小型的可点击图标，带有明显的关闭按钮。\n" + "3. **底部的横幅广告**：带有明显的关闭按钮。\n\n" + "请返回关闭按钮的中心坐标。\n\n" + "请以纯 JSON 格式输出：\n" + "{\n" + " \"has_ad\": true/false,\n" + " \"ad_type\": \"center\" | \"bottom\" | \"other\",\n" + " \"close_point\": [x, y] // 0-1000 归一化坐标\n" + "}\n" + "如果没有广告，请返回 `{\"has_ad\": false}`。" + ) + + try: + resp = await cls._client.chat.completions.create( + model=VL_MODEL_NAME, + timeout=30, + messages=[ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": img_input}, + {"type": "text", "text": prompt}, + ], + } + ], + ) + content = resp.choices[0].message.content.strip() + raw_json = cls._extract_json(content) + result = json.loads(raw_json) + + if result.get("has_ad") and result.get("close_point"): + norm_point = result["close_point"] + + # 转换坐标 + if len(norm_point) == 2: + if not device_info: + device_info = cls._FALLBACK_DEVICE_INFO + + w = device_info.get("displayWidth", FALLBACK_WIDTH) + h = device_info.get("displayHeight", FALLBACK_HEIGHT) + + x = int(norm_point[0] / 1000 * w) + y = int(norm_point[1] / 1000 * h) + + return { + "x": x, + "y": y, + "ad_type": result.get("ad_type", "unknown") + } + + return None + + except Exception as e: + logger.error(f"Failed to detect ad popup: {e}") + return None + @staticmethod def _to_minutes(t_str: str) -> int: """HH:MM -> 分钟数""" @@ -309,202 +349,116 @@ class ReadImageKit: return None @staticmethod - async def recognize_card_with_vl(image_path: str): + async def parse_vl_image(image_path: str, json_metadata, device_info=None): """ - 对卡片切片进行 VL 识别 + 基于 _vl.jpg (带绿框) 和 JSON 元数据进行整页识别 (节约 TOKEN) """ - if not os.path.exists(image_path): - return None - - with open(image_path, "rb") as f: - encoded = base64.b64encode(f.read()).decode("utf-8") - img_input = {"url": f"data:image/jpeg;base64,{encoded}"} - - prompt = ( - "你是一个专业的充电桩信息提取助手。请识别图片中充电站卡片的核心信息。\n" - "注意以下识别规则：\n" - "1. **场站名称 (station_name)**: 场站名称通常位于卡片的第一行（左上角），字体最大且加粗。严禁提取'距离优先'、'空闲较多'、'价格最低'等筛选标签。严禁将卡片中下部的描述信息（如'导航到...'、'停车说明...'、'距您...'等）识别为名称。名称通常是一个具体的地点或品牌名。\n" - "2. **价格 (price)**: 提取数字部分，通常以'元/度'结尾，字体颜色通常为红色或加粗。\n" - "3. **桩数 (piles)**: 识别所有类型的充电桩。请提取'快'、'慢'或'超'字样及其对应的'空闲/总数'（如 '快 4/4'）。\n" - " - 如果明确标有'快'、'慢'或'超'，则对应为'快充'、'慢充'、'超充'。\n" - " - 如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢充\"`。\n" - " - 必须以 JSON 对象列表形式输出，例如：`[{\"type\": \"快充\", \"free\": 4, \"total\": 4}]`。\n" - "4. **距离 (distance)**: 提取如 '2.5km' 格式，通常在蓝色胶囊背景内。\n" - "5. **停车 (parking)**: 简要描述停车收费信息，通常在蓝色'P'图标后。\n\n" - "请以 JSON 格式输出：\n" - "{\n" - " \"station_name\": \"场站名称\",\n" - " \"price\": \"价格\",\n" - " \"piles\": [\n" - " {\"type\": \"快\", \"free\": 4, \"total\": 4}\n" - " ],\n" - " \"distance\": \"距离\",\n" - " \"parking\": \"停车信息\"\n" - "}\n" - "只输出 JSON，不要有任何解释文字。" - ) - - try: - resp = await ReadImageKit._client.chat.completions.create( - model=VL_MODEL_NAME, - timeout=30, - messages=[ - { - "role": "user", - "content": [ - {"type": "text", "text": prompt}, - {"type": "image_url", "image_url": img_input} - ] - } - ], - max_tokens=300, - temperature=0.01 - ) - content = resp.choices[0].message.content.strip() - raw = ReadImageKit._extract_json(content) - return json.loads(raw) - except Exception as e: - logger.error(f"VL recognize_card_with_vl error: {e}") - return None - - @staticmethod - async def get_stations_hybrid(image_path, device_info=None): - """ - 混合识别模式：图形学切片 + VL 模型识别 - """ - if device_info is None: - device_info = ReadImageKit._FALLBACK_DEVICE_INFO - - # 1. 使用 Kit 中的图形学算法识别卡片区域 - json_data = Kit.crop_cards_from_image(image_path, save_debug=True) - if not json_data or not json_data.get("cards"): - logger.warning("No cards detected by graphical slicing.") + if not json_metadata or "cards" not in json_metadata: return [] - # 2. 对每个卡片区域进行 VL 识别 - tasks = [] - card_metas = [] - - # 为了使用 VL，我们需要将每个卡片切出来保存到临时文件 - temp_dir = os.path.join(os.path.dirname(image_path), "Patches") - os.makedirs(temp_dir, exist_ok=True) - - img = Image.open(image_path).convert("RGB") - - for card in json_data["cards"]: - rect = card["rect"] # [x1, y1, x2, y2] - patch_path = os.path.join(temp_dir, f"patch_{card['id']}_{uuid.uuid4().hex[:8]}.jpg") - - # 裁剪并保存 - patch = img.crop((rect[0], rect[1], rect[2], rect[3])) - patch.save(patch_path) - - card_metas.append({ - "meta": card, - "patch_path": patch_path - }) - tasks.append(ReadImageKit.recognize_card_with_vl(patch_path)) - - # 并行执行 VL 识别 - results = await asyncio.gather(*tasks) - - final_stations = [] - for i, res in enumerate(results): - meta = card_metas[i]["meta"] - patch_path = card_metas[i]["patch_path"] - - if res and isinstance(res, dict) and res.get("station_name"): - # 注入点击坐标和原始区域信息 - res["uia_center_x"] = meta["click_point"][0] - res["uia_center_y"] = meta["click_point"][1] - res["rect"] = meta["rect"] - - # 转换 bounds 到 0-1000 空间 - w, h = img.size - rect = meta["rect"] - res["bounds"] = [ - int(rect[0] * 1000 / w), - int(rect[1] * 1000 / h), - int(rect[2] * 1000 / w), - int(rect[3] * 1000 / h) - ] - - final_stations.append(res) - logger.info(f"卡片 {meta['id']} 识别成功: {res['station_name']}") - else: - logger.warning(f"卡片 {meta['id']} 识别失败或无名称") - - # 清理临时切片文件 - try: - if os.path.exists(patch_path): - os.remove(patch_path) - except: - pass - - # 3. 后置过滤 - if final_stations: - processed_excluded_titles = {str(x).replace(" ", "").strip() for x in STATION_EXCLUDED_TITLES} - filtered = [] - for item in final_stations: - name = item.get("station_name") - normalized_name = str(name).replace(" ", "").strip() - if normalized_name in processed_excluded_titles: - continue - if any(kw in normalized_name for kw in STATION_BLACKLIST_KEYWORDS): - continue - filtered.append(item) - final_stations = filtered - - return final_stations - - @classmethod - async def get_stations_from_image(cls, image_path: str, device_info=None): - """ - 使用 Qwen-VL 模型从截图中识别充电站列表 - """ - if device_info is None: - device_info = cls._FALLBACK_DEVICE_INFO - if not os.path.exists(image_path): logger.error(f"Image not found: {image_path}") return [] - # 将图片转换为 Base64 - with open(image_path, "rb") as image_file: - encoded_image = base64.b64encode(image_file.read()).decode("utf-8") + with open(image_path, "rb") as f: + encoded = base64.b64encode(f.read()).decode("utf-8") + img_input = {"url": f"data:image/jpeg;base64,{encoded}"} + + cards_meta = json_metadata["cards"] + # 按 id 或 Y 坐标排序，确保顺序一致 + cards_meta.sort(key=lambda x: x["rect"][1]) + + prompt = ( + "图片中用绿色矩形框标记了若干个充电站卡片区域。\n" + "请按从上到下的顺序，依次识别每个绿色框内的场站信息，并返回一个JSON数组。\n" + "数组中元素的顺序必须与图片中绿色框从上到下的顺序严格一致。\n" + "如果某个框内不是有效的场站卡片（例如是广告或推荐标签），请返回null或空对象，不要跳过顺序。\n" + "\n" + "每个JSON对象包含以下字段：\n" + "1. b_use: 状态标识（1或0）。如果场站名称为灰色或带有“暂停使用”等标签，则为0，否则为1。\n" + "2. station_name: 场站名称；\n" + "3. price: 价格（数字，如 1.2500）；\n" + "4. piles: 充电枪列表 [{type: '快充'|'慢充'|'超充', free: 4, total: 4}]。如果图中只显示数字（如 '34/42'）而没有明确标明类型，请统一识别为 `type: \"慢充\"`；\n" + "5. parking: 停车费用描述（通常在蓝色'P'图标后，例如 '免费停车三小时'）；\n" + "6. distance: 距离信息字符串（例如 '2.82km' 或 '90m'）。\n" + "\n" + "严格返回纯JSON格式。" + ) try: - response = await cls._client.chat.completions.create( + resp = await ReadImageKit._client.chat.completions.create( model=VL_MODEL_NAME, - timeout=30, messages=[ { "role": "user", "content": [ - {"type": "text", "text": cls._prompt}, - { - "type": "image_url", - "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}, - }, + {"type": "image_url", "image_url": img_input}, + {"type": "text", "text": prompt}, ], - } + }, ], - max_tokens=2000, - temperature=0.01 ) - content = response.choices[0].message.content - logger.info(f"VL Model Response: {content}") + content = resp.choices[0].message.content or "" + raw = ReadImageKit._extract_json(content) + llm_data = json.loads(raw) - json_str = cls._extract_json(content) - stations = json.loads(json_str) + final_stations = [] + if isinstance(llm_data, list): + # 尝试与 cards_meta 对齐 + count = min(len(llm_data), len(cards_meta)) + for i in range(count): + item = llm_data[i] + meta = cards_meta[i] - # 后处理：如果 bounds 是归一化的，则转换为像素坐标（如果需要） - # 或者如果 bounds 是 0-1000 的，则保持原样或按需转换 - return stations + if not item or not isinstance(item, dict): + continue + + name = item.get("station_name") + if not name or name == "未知": + continue + + # 注入元数据中的点击坐标 + click_pt = meta.get("click_point", [0, 0]) + item["uia_center_x"] = click_pt[0] + item["uia_center_y"] = click_pt[1] + item["rect"] = meta.get("rect") + + # 转换 bounds 到 0-1000 空间 (用于匹配) + img_w = 1080 + img_h = 2400 + if device_info: + img_w = device_info.get("displayWidth") or device_info.get("width") or 1080 + img_h = device_info.get("displayHeight") or device_info.get("height") or 2400 + + rect = meta["rect"] + item["bounds"] = [ + int(rect[0] * 1000 / img_w), + int(rect[1] * 1000 / img_h), + int(rect[2] * 1000 / img_w), + int(rect[3] * 1000 / img_h) + ] + + final_stations.append(item) + + # 后置过滤 + if final_stations: + processed_excluded_titles = {str(x).replace(" ", "").strip() for x in STATION_EXCLUDED_TITLES} + filtered = [] + for item in final_stations: + name = item.get("station_name") + normalized_name = str(name).replace(" ", "").strip() + if normalized_name in processed_excluded_titles: + continue + if any(kw in normalized_name for kw in STATION_BLACKLIST_KEYWORDS): + continue + filtered.append(item) + final_stations = filtered + + return final_stations except Exception as e: - logger.error(f"Error calling VL model: {e}") + logger.error(f"Error calling VL model for page recognition: {e}") return [] @classmethod diff --git a/Apps/AiTeJiYiChong/Service.py b/Apps/AiTeJiYiChong/Service.py index 1b021b2..eb1f0c1 100644 --- a/Apps/AiTeJiYiChong/Service.py +++ b/Apps/AiTeJiYiChong/Service.py @@ -144,141 +144,15 @@ class AiTeJiYiChongService: logger.info(f"场站详情处理完成: {name}") return detail - async def process_station_list_hybrid(self, image_path, device_info=None) -> list: + async def process_station_list_vl(self, image_path, json_metadata, device_info=None) -> list: """ - 基于混合模式处理场站列表 (图形学切片 + VL 模型) - """ - station_list = await ReadImageKit.get_stations_hybrid(image_path, device_info=device_info) - if not station_list: - return [] - - processed_stations = [] - async with await self.db.get_session() as session: - for station in station_list: - name = station.get("station_name") - if not name: - continue - - station_hash = self.get_hash(name) - now = datetime.now() - station["station_hash"] = station_hash - - # 1. 保存 Profile - profile_id = self.generate_id() - await self.station_profile_model.save( - session=session, - id=profile_id, - station_hash=station_hash, - operator=self.operator, - station_name=name, - valid_start_time=now - ) - station["profile_id"] = profile_id - station["valid_start_time"] = now.isoformat() - - # 2. 保存 Status (解析价格和电桩) - status_id = self.generate_id() - - # 处理 piles 字段 - piles_data = station.get("piles") - total, free = 0, 0 - standardized_piles = [] - if isinstance(piles_data, list): - for idx, p in enumerate(piles_data): - try: - t = int(p.get("total", 0)) - f = int(p.get("free", 0)) - total += t - free += f - standardized_piles.append({ - "pile_no": f"G{idx+1}", - "type": p.get("type", "未知"), - "power": "", - "status_text": f"空闲{f}/总{t}", - "remark": "列表页VL" - }) - except: - pass - else: - # Fallback: 如果不是列表（可能是字符串） - piles_str = str(piles_data) if piles_data else "0/0" - - nums = re.findall(r"\d+", piles_str) - if len(nums) >= 2: - # 假设最后两个数字是 free 和 total - v1 = int(nums[-2]) - v2 = int(nums[-1]) - if v2 >= v1: - free = v1 - total = v2 - else: - free = v2 - total = v1 - - # 构造一条记录用于 Excel 展示 - standardized_piles.append({ - "pile_no": "-", - "type": "未知", - "status_text": piles_str, - "remark": "原始识别文本" - }) - else: - # Fallback: 如果不是列表（可能是字符串） - piles_str = str(piles_data) if piles_data else "0/0" - - nums = re.findall(r"\d+", piles_str) - if len(nums) >= 2: - # 假设最后两个数字是 free 和 total - v1 = int(nums[-2]) - v2 = int(nums[-1]) - if v2 >= v1: - free = v1 - total = v2 - else: - free = v2 - total = v1 - - # 构造一条记录用于 Excel 展示 - standardized_piles.append({ - "pile_no": "-", - "type": "未知", - "status_text": piles_str, - "remark": "原始识别文本" - }) - - await self.station_status_model.save( - session=session, - id=status_id, - station_hash=station_hash, - total_piles=total, - free_piles=free, - piles_detail_json=standardized_piles if standardized_piles else None, - current_price=self._to_float(station.get("price")), - parking_info=station.get("parking", ""), - distance=station.get("distance", ""), - valid_start_time=now - ) - station["status_id"] = status_id - - processed_stations.append(station) - - await session.commit() - - return processed_stations - - async def process_station_list_vl(self, image_path, device_info=None) -> list: - """ - 基于 VL 模式处理场站列表 + 基于 VL 模式处理场站列表 (整页识别) """ # 优先使用带绿框的 _vl.jpg 图片进行识别 vl_img_path = image_path.replace(".jpg", "_vl.jpg") - if os.path.exists(vl_img_path): - logger.info(f"使用带绿框的图片进行识别: {vl_img_path}") - image_to_process = vl_img_path - else: - image_to_process = image_path + img_to_process = vl_img_path if os.path.exists(vl_img_path) else image_path - station_list = await ReadImageKit.get_stations_from_image(image_to_process, device_info=device_info) + station_list = await ReadImageKit.parse_vl_image(img_to_process, json_metadata, device_info=device_info) if not station_list: return [] @@ -328,8 +202,8 @@ class AiTeJiYiChongService: "remark": "列表页VL" }) except: - pass - + continue + await self.station_status_model.save( session=session, id=status_id, @@ -345,7 +219,7 @@ class AiTeJiYiChongService: station["status_id"] = status_id processed_stations.append(station) - + await session.commit() return processed_stations diff --git a/Apps/XinDianTu/Config/__pycache__/Setting.cpython-310.pyc b/Apps/XinDianTu/Config/__pycache__/Setting.cpython-310.pyc index c9f476b08f178d92e7881d654190c60d0e6ae801..76ed9d8b3bca92f72852f6f741419c7509b0fdc6 100644 GIT binary patch delta 27 hcmbQrGL?llpO=@50SJEYNzJ@4kvEaiW@G*`MgU|C2rB>p delta 27 hcmbQrGL?llpO=@50SH!XOU=AEkvEYsWn=y_MgU+22m1g3 diff --git a/Apps/XinDianTu/__pycache__/Crawler.cpython-310.pyc b/Apps/XinDianTu/__pycache__/Crawler.cpython-310.pyc index ecf1b1d4bbae677106465abede3247885117be75..fa3dc224789880eae273481e6e8141d6b0e1b1ca 100644 GIT binary patch delta 1845 zcmY*aOK)366rLG7O&rILZZ=>8G%Sn4EyH06 z@RvAeuTH(`T^y*Qg&>5~!V#9+?_@^k%8#6xjRf%0UJJBf<2u_A3L)S3M#JF{6T)mV zw`*6eSijC7f(d zX+3UutY&3%##zys6BsaSH8zIcd-3%=uFr<8sEtOi)eZ^hC<&9!YrB)|py3(P?whsy zgOy-GHM=lRj)Kso9OWmC7&b#U^gvQ>)+a*wOPT^_^F+kxptO3?Lu04(`x3lQj{Bl3 z{YDZatDIH`HXO}T0Y?aTkqT{&lm@uPeJ4Wr(%amr} zJB(m7w8b)G7$ZL^>RBA)#C5aC^IsZ$=*0M{*vr7RhpkY(e$%xW!Z?u_wn-=;w!=t> z!Dxt?2~41G*;y2YYMt(w#_qrbjITRk0)2FH=SH=nuR|zdOlr(Swm)nGF~KOYc9;q= zBHv-!-D{lIuo+a6O;L)#rB0I}NGokF@=;9Q1~(d#KW)@SgnJAMDE6 zDxOo$NyUehii_gPH~-V|Bl4O*8MD>fXZCAXk76&8e+>I{BKDafsS_9fyEZA6yre9Mnu~ zolz|t%@NJuDRsN3h$A%4%3m9Y`92wJiVa+bj5)l>%yFX%1ngfGbL{wtIbn=gqZqfw zfYL8cDUo`b1m!6FfGj%>&%i-Av?fTvP1bgJw#04WP^TUFcXiyu@~fuQ z$`RcTg?a7)Mzsy*Rn{A#(!41noNdJ^BVx3Hy4BN0*rGX!bOxs}K#g)1?Dyb#<&o;p zl{<_xII4HzI^t2iLlVjDrFjsJ;q-2B!SUCqG?c^%3~ZU5#F;I#6F9nMc51_=X)D3L zAV1z7SpiJj4Ivt!gS~h^m=eE$m+sN3yaPm?u$T0sLpEnQmnH>T*{AIbP{EFA?3%`s zUBL>lFW`InwgMg7QN_Wa_jRb%#Nxcp(?0nTrhfA8ml lZ2N(bV delta 1829 zcmYjSUu;uV7{BMVV{1qMcWdvuuIoC=+HUK*jSbkktvGcBtRR~y}?%bzjcbXMVJ^dq5bPs}@xL?Nt;JK{i=_Ayb!YsPybx(G4 z={rZ}6mYOARgCqjLaFZ6YoxItI4&!!0bDqS@)be{D24~TP*b4%wmnE5@a=a0MlCQ^ z4|Xl6HL%7DCcPG{OC%Qq{7zFblgh&4xMrKO8?|6B5R^yxHitUd1ork*1&okD zp+URR$eP)@0@Z!xc@*dVvL>I@w6DDKi`B_-)Z=IC#p#(%S{Qt+g~eE$-*QCM**TPo zcMN}8GX|Cb8;((Kbw)@lZ+C{;+O!x17YRz+^CqE%5xoKGupUX&Z_eF>pcu6{)UhO= zcLo=}haiMh1*PSg9)>XDJFtFD(1bqyxhrIWkgeS9NmeFcJ#IGh5l`>Ir_iJ+Hv&m? z7U%N2tw4^@EJ2!gu&4DDbYOihK?ft4Tg$Adq*&3_D@>0^@02#Si^jpkbmBO~Z+lYY z8SeCk+kj>C)P$nP7UZaNXEszoyuz04KytKQIf9g#xJSWm7F=LM~ngta0yofqaH7i?- zz1AmtuWzOI`n7;j4}HkHH1tC)mdl(|rQR2WM$AJryBE^XC3m87XTRLZ;La$n{cIfi zw=fGk@Fu1PgE$Dat5$~~y=rw3x>l`T%v+?}f`cz{qb9g;6b4EKK{t*{V=u256ZaQo zOs>(batovDh<#1789osPQL+~4)M3fiBPC9kvE34TQDSjk#!^;lsENF(SqLGD3O~n- zYa=UVRhT`hZDQ0X!kgrWYGaux7!+A$(^@}!<$`IYD%g$e7|Xt)fMr-&B>E5xVZD8N z1~NvvAUggox>v`J^Y3Z{hkwSKp(5wRmLr&7F{_r%CwiB%6HtxzX3A|@&2rNfh;hih zlfqLBpgv3SwkhE?Z2n3)WE<%BTuOf#btG{isXy;!zb5}_xbUbAqO#^u@fi0RcGh}x4C$=Ab>(SD4QJH?*V7L%H)3Ru@#+bD69wNg!kQ1sAz F|38lF4$uGq