diff --git a/Apps/AiTeJiYiChong/Crawler.py b/Apps/AiTeJiYiChong/Crawler.py index b0d9a27..f2a94db 100644 --- a/Apps/AiTeJiYiChong/Crawler.py +++ b/Apps/AiTeJiYiChong/Crawler.py @@ -76,124 +76,107 @@ async def get_station_list(d, service, max_scrolls=MAX_SCROLLS): logger.info(f"列表页截图已完成: {screenshot_path}") - # 2. 执行图形学分析，生成 _flag.jpg, _vl.jpg 和 .json - logger.info("正在执行图形学切片分析...") - json_data = Kit.crop_cards_from_image(screenshot_path) - - # 3. 调用混合模式识别 (图形学切片 + 本地 OCR) + # 2. 调用混合模式识别 (内部包含图形学切片 + VL 模型) + logger.info("正在执行混合模式识别 (切片 + VL)...") stations = await service.process_station_list_hybrid(screenshot_path, device_info=device_info) - logger.info(f"本页识别到 {len(stations)} 个场站") + logger.info(f"本页识别到 {len(stations)} 个有效场站卡片") - # 4. 匹配几何卡片与识别结果 (混合模式下已经包含在 stations 中，但为了兼容旧逻辑进行填充) - if json_data and json_data.get("cards") and stations: - for card in json_data["cards"]: - card_rect = card["rect"] # [x1, y1, x2, y2] - for st in stations: - st_bounds = st.get("bounds") # [x1, y1, x2, y2] (0-1000) - if not st_bounds: continue - - # 转换 VL 坐标到像素坐标 - st_y1_px = st_bounds[1] * h / 1000 - st_y2_px = st_bounds[3] * h / 1000 - - # 计算 y 轴重叠 - overlap = min(card_rect[3], st_y2_px) - max(card_rect[1], st_y1_px) - if overlap > 50: # 有显著重叠 - card["station_name"] = st.get("station_name") - logger.info(f"匹配成功: 几何卡片 {card_rect} -> 场站 {card['station_name']}") - break + # 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 - # 5. 遍历处理本页所有场站 - if json_data and json_data.get("cards"): - for card_idx, card in enumerate(json_data["cards"]): - station_name = card.get("station_name") - if not station_name: - logger.warning(f"第 {card_idx + 1} 个几何卡片未匹配到场站名称，跳过。") - 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 + + logger.info(f"准备处理第 {card_idx + 1} 个场站: {station_name}, 点击坐标: ({click_x}, {click_y})") + d.click(click_x, click_y) + + # 等待二级页面加载 + await asyncio.sleep(3) + + # 截取二级页面图 + 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. 寻找 timePrice.jpg 图标并进入三级页面 + time_price_template = os.path.join(os.path.dirname(__file__), "BiaoShi", "timePrice.jpg") + coords = Kit.find_template_coords(detail_path, time_price_template) + + if coords: + cx, cy, conf = coords + logger.info(f"找到分时价格图标，进入三级页面...") + d.click(cx, cy) + await asyncio.sleep(2) - # 检查 Redis 去重 (使用新的模糊匹配逻辑) - if await is_already_crawled(redis_kit, station_name): - logger.info(f"场站 {station_name} 匹配到已处理记录，跳过。") - continue - - click_x, click_y = card["click_point"] - logger.info(f"准备处理第 {card_idx + 1} 个场站: {station_name}, 点击坐标: ({click_x}, {click_y})") - d.click(click_x, click_y) + # 抓取三级页面图片 + price_screenshots = [] + price_detail_uuid = f"price_detail_{station_name}_{image_uuid}" + price_path = take_screenshot(d, price_detail_uuid, save_dir=TEMP_IMAGE_DIR) + price_screenshots.append(price_path) - # 等待二级页面加载 - await asyncio.sleep(3) - - # 截取二级页面图 - detail_uuid = f"detail_{station_name}_{image_uuid}" - detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR) - - # 【异步优化】后台解析详情页基础数据，不阻塞 UI 流程 - 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}") + # 滑动处理 + last_price_md5 = Kit.get_file_md5(price_path) + 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_file_md5(scroll_path) + if curr_price_md5 == last_price_md5: break + price_screenshots.append(scroll_path) + last_price_md5 = curr_price_md5 - task_detail = asyncio.create_task(detail_task_wrapper(detail_path, station_name)) - background_tasks.append(task_detail) - - # 6. 寻找 timePrice.jpg 图标并进入三级页面 (分时价格页) - time_price_template = os.path.join(os.path.dirname(__file__), "BiaoShi", "timePrice.jpg") - coords = Kit.find_template_coords(detail_path, time_price_template) + # 后台处理价格图片 + 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) - if coords: - cx, cy, conf = coords - logger.info(f"找到分时价格图标，进入三级页面...") - d.click(cx, cy) - await asyncio.sleep(2) # 等待加载 - - # 抓取三级页面图片（包括可能的滑动） - price_screenshots = [] - price_detail_uuid = f"price_detail_{station_name}_{image_uuid}" - price_path = take_screenshot(d, price_detail_uuid, save_dir=TEMP_IMAGE_DIR) - price_screenshots.append(price_path) - - # 如果需要滑动，先完成所有截图动作 - last_md5 = Kit.get_file_md5(price_path) - for scroll_idx in range(2): # 减少滑动次数以加快速度，通常1-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_md5 = Kit.get_file_md5(scroll_path) - if curr_md5 == last_md5: break - price_screenshots.append(scroll_path) - last_md5 = curr_md5 - - # 【异步优化】后台处理所有价格图片，不阻塞 UI 返回 - 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) - else: - logger.warning(f"场站 {station_name} 未找到分时价格入口。") - - # 从二级页面返回列表页 + # 立即返回 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) + await asyncio.sleep(0.5) + else: + logger.warning(f"场站 {station_name} 未找到分时价格入口。") - # 在每一页结束时，清理已完成的任务，防止内存堆积 + # 从二级页面返回列表页 + 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()] for t in done_tasks: if t.exception(): logger.error(f"后台任务执行异常: {t.exception()}") background_tasks.remove(t) - # 7. 列表页翻页滑动 + # 5. 列表页翻页滑动 logger.info("执行翻页滑动...") start_x, start_y = w // 2, int(h * 0.8) end_x, end_y = w // 2, int(h * (0.8 - SCROLL_DISTANCE_RATIO)) diff --git a/Apps/AiTeJiYiChong/Doc/流程说明.md b/Apps/AiTeJiYiChong/Doc/流程说明.md index b5d3e4b..bcea7ad 100644 --- a/Apps/AiTeJiYiChong/Doc/流程说明.md +++ b/Apps/AiTeJiYiChong/Doc/流程说明.md @@ -13,10 +13,10 @@ ### 2.1 第一级：场站列表页 - **操作**: 自动滚动列表，每屏截取一张大图。 - **识别方法**: - - **图形学切片**: 使用 OpenCV 识别图片中的白色卡片区域（几何卡片），确定点击位置。 - - **VL 模型分析**: 使用 `Qwen-VL` 模型识别本页所有场站的名称、基础价格和归一化坐标。 - - **坐标匹配**: 将几何卡片的像素坐标与 VL 模型的归一化坐标进行交叉比对，确定每个卡片对应的场站名称。 -- **关键点**: 通过几何卡片确保点击的准确性，通过 VL 模型确保数据的语义准确性。 + - **图形学切片**: 使用 OpenCV 识别图片中的白色卡片区域（几何卡片），将其裁剪为独立的小图。 + - **VL 并行识别**: 对裁剪出的每个场站小图，并行调用 `Qwen-VL` 模型提取名称、价格、枪数等信息。 + - **坐标映射**: 将 VL 识别结果直接映射回原始几何卡片的点击坐标。 +- **关键点**: 通过几何切片降低 VL 模型的识别负担，通过并行调用提高整体处理速度。 ### 2.2 第二级：场站详情页 - **进入**: 点击列表页卡片。 diff --git a/Apps/AiTeJiYiChong/ReadImageKit.py b/Apps/AiTeJiYiChong/ReadImageKit.py index 4a30318..f57d598 100644 --- a/Apps/AiTeJiYiChong/ReadImageKit.py +++ b/Apps/AiTeJiYiChong/ReadImageKit.py @@ -8,6 +8,7 @@ import aiohttp import logging import base64 import cv2 +import uuid from openai import AsyncOpenAI, BadRequestError from Config.Config import ( ALY_LLM_API_KEY, VL_MODEL_NAME, VL_MODEL_NAME_AD @@ -16,7 +17,6 @@ from Apps.AiTeJiYiChong.Config.Setting import ( SAFE_EXCLUDE_RATIO, FALLBACK_WIDTH, FALLBACK_HEIGHT, BOTTOM_SAFE_EXCLUDE_RATIO ) -from Util.PaddleOCRKit import get_ocr_kit from Apps.AiTeJiYiChong import Kit logger = logging.getLogger(__name__) @@ -290,51 +290,107 @@ class ReadImageKit: logger.error(f"Error calling VL model for detail: {e}") return None - @classmethod - async def get_stations_hybrid(cls, image_path: str, device_info=None): + @staticmethod + async def recognize_card_with_vl(image_path: str): """ - 混合识别模式：图形学切片 + 本地 PaddleOCR 识别 + 对卡片切片进行 VL 识别 """ if not os.path.exists(image_path): - logger.error(f"Image not found: {image_path}") - return [] + 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 = ( + "请识别图片中的充电站卡片信息，并以 JSON 格式输出：\n" + "{\n" + " \"station_name\": \"场站名称\",\n" + " \"price\": \"价格(数字)\",\n" + " \"piles\": \"空闲/总数(如 4/4)\",\n" + " \"distance\": \"距离(如 2.5km)\",\n" + " \"parking\": \"停车描述\"\n" + "}\n" + "只输出 JSON，不要有其他文字。" + ) + + try: + resp = await ReadImageKit._client.chat.completions.create( + model=VL_MODEL_NAME, + 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 中的图形学算法识别卡片区域 - # Kit.crop_cards_from_image 会生成 .json, _flag.jpg, _vl.jpg - # 我们主要需要它返回的 json_data 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.") 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") - ocr_kit = get_ocr_kit() - final_stations = [] - - # 2. 对每个卡片区域进行 OCR 识别 - # 注意：PaddleOCR 识别过程较快，且通常不涉及网络请求，可以根据需要选择并行或串行 - # 这里使用串行以保证日志输出整齐，如果追求极致性能可改用 asyncio.to_thread 并行 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) - # 转换为 ndarray 供 PaddleOCR 使用 - patch_cv = cv2.cvtColor(np.array(patch), cv2.COLOR_RGB2BGR) + 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"] - # OCR 识别 - logger.info(f"正在识别卡片 {card['id']}: {rect}") - res = ocr_kit.recognize(patch_cv) - - if res and res.get("station_name"): + if res and isinstance(res, dict) and res.get("station_name"): # 注入点击坐标和原始区域信息 - res["uia_center_x"] = card["click_point"][0] - res["uia_center_y"] = card["click_point"][1] - res["rect"] = rect + res["uia_center_x"] = meta["click_point"][0] + res["uia_center_y"] = meta["click_point"][1] + res["rect"] = meta["rect"] - # 转换 bounds 到 0-1000 空间（保持与 VL 模式兼容） + # 转换 bounds 到 0-1000 空间 w, h = img.size + rect = meta["rect"] res["bounds"] = [ int(rect[0] * 1000 / w), int(rect[1] * 1000 / h), @@ -343,9 +399,16 @@ class ReadImageKit: ] final_stations.append(res) - logger.info(f"卡片 {card['id']} 识别成功: {res['station_name']}") + logger.info(f"卡片 {meta['id']} 识别成功: {res['station_name']}") else: - logger.warning(f"卡片 {card['id']} 识别失败或无名称") + logger.warning(f"卡片 {meta['id']} 识别失败或无名称") + + # 清理临时切片文件 + try: + if os.path.exists(patch_path): + os.remove(patch_path) + except: + pass return final_stations diff --git a/Apps/AiTeJiYiChong/Service.py b/Apps/AiTeJiYiChong/Service.py index 7bf3baf..6253d01 100644 --- a/Apps/AiTeJiYiChong/Service.py +++ b/Apps/AiTeJiYiChong/Service.py @@ -127,7 +127,7 @@ class AiTeJiYiChongService: async def process_station_list_hybrid(self, image_path, device_info=None) -> list: """ - 基于混合模式处理场站列表 (图形学切片 + 本地 OCR) + 基于混合模式处理场站列表 (图形学切片 + VL 模型) """ station_list = await ReadImageKit.get_stations_hybrid(image_path, device_info=device_info) if not station_list: diff --git a/Apps/AiTeJiYiChong/__pycache__/Crawler.cpython-310.pyc b/Apps/AiTeJiYiChong/__pycache__/Crawler.cpython-310.pyc index 6d62598..dd733f6 100644 Binary files a/Apps/AiTeJiYiChong/__pycache__/Crawler.cpython-310.pyc and b/Apps/AiTeJiYiChong/__pycache__/Crawler.cpython-310.pyc differ diff --git a/Apps/XinDianTu/Crawler.py b/Apps/XinDianTu/Crawler.py index f8fe43d..b4074c2 100644 --- a/Apps/XinDianTu/Crawler.py +++ b/Apps/XinDianTu/Crawler.py @@ -20,7 +20,6 @@ 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 Apps.XinDianTu.ReadImageKit import ReadImageKit @@ -60,7 +59,6 @@ logging.basicConfig( logger = logging.getLogger("StationList") # 强制设置所有相关模块的日志级别为 INFO，防止被第三方库干扰 -logging.getLogger("PaddleOCRKit").setLevel(logging.INFO) logging.getLogger("OpenXinDianTu").setLevel(logging.INFO) logging.getLogger("FullProcess").setLevel(logging.INFO) diff --git a/Apps/XinDianTu/ReadImageKit.py b/Apps/XinDianTu/ReadImageKit.py index a363479..0eebf83 100644 --- a/Apps/XinDianTu/ReadImageKit.py +++ b/Apps/XinDianTu/ReadImageKit.py @@ -534,14 +534,14 @@ class ReadImageKit: @staticmethod async def get_stations_hybrid(image_path, uploader, cdn_domain): """ - 混合识别模式：图形学切片 + 大模型小图 OCR + 混合识别模式：图形学切片 + 大模型小图 VL 识别 """ # 1. 图形学切片 segments = ReadImageKit.get_card_segments(image_path) if not segments: return [] - # 2. 对每个切片并行进行 OCR + # 2. 对每个切片并行进行 VL 识别 tasks = [] for seg in segments: # 上传切片 @@ -552,9 +552,9 @@ class ReadImageKit: # 记录上传后的 URL 供识别使用 seg["patch_url"] = patch_url - tasks.append(ReadImageKit.recognize_card_text(patch_url)) + tasks.append(ReadImageKit.recognize_card_with_vl(patch_url)) - # 等待所有 OCR 完成 + # 等待所有 VL 完成 results = await asyncio.gather(*tasks) # 3. 组装结果 @@ -773,9 +773,9 @@ class ReadImageKit: return [] @staticmethod - async def recognize_card_text(patch_url): + async def recognize_card_with_vl(patch_url): """ - 对切片进行小图 OCR 识别，提取场站详细信息 + 对切片进行小图 VL 识别，提取场站详细信息 """ resp = await ReadImageKit._client.chat.completions.create( model=VL_MODEL_NAME, @@ -802,7 +802,7 @@ class ReadImageKit: if json_match: return json.loads(json_match.group()) except Exception: - logger.warning(f"OCR 结果解析 JSON 失败: {content}") + logger.warning(f"VL 结果解析 JSON 失败: {content}") return {"station_name": "未知"} diff --git a/Apps/XinDianTu/Service.py b/Apps/XinDianTu/Service.py index 92da17e..abe30fc 100644 --- a/Apps/XinDianTu/Service.py +++ b/Apps/XinDianTu/Service.py @@ -161,7 +161,7 @@ class XinDianTuService: async def process_station_list_hybrid(self, local_image_path: str, uploader, cdn_domain) -> list: """ - 混合模式处理场站列表 (图形学切片 + OCR) + 混合模式处理场站列表 (图形学切片 + VL) """ station_list = await ReadImageKit.parse_hybrid_image(local_image_path, uploader, cdn_domain) if not station_list: @@ -207,9 +207,10 @@ class XinDianTuService: "type": "未知", "power": "", "status_text": f"空闲{free}/总{total}", - "remark": "混合模式解析" + "remark": "列表页VL" }) - except: pass + except: + pass price_str = station.get("price", "") current_price = None @@ -431,7 +432,7 @@ class XinDianTuService: "type": p_type, "power": "", "status_text": f"空闲{f_int}/总{t_int}", - "remark": "列表页OCR" + "remark": "列表页VL" }) else: # 模式 2: 无类型，仅 "2/10" @@ -445,7 +446,7 @@ class XinDianTuService: "type": "未知", "power": "", "status_text": f"空闲{f_int}/总{t_int}", - "remark": "列表页OCR" + "remark": "列表页VL" }) if not standardized_piles: @@ -459,7 +460,7 @@ class XinDianTuService: "type": "未知", "power": "", "status_text": f"空闲{free}/总{total}", - "remark": "列表页OCR兜底" + "remark": "列表页VL兜底" }) # 解析价格 diff --git a/Util/OcrParser.py b/Util/OcrParser.py deleted file mode 100644 index f26ab09..0000000 --- a/Util/OcrParser.py +++ /dev/null @@ -1,126 +0,0 @@ -# coding=utf-8 -import re - -class OcrParser: - @staticmethod - def parse(ocr_lines): - """ - 解析 OCR 文本行，提取场站结构化数据。 - 不依赖 LLM，使用规则和正则匹配。 - """ - result = { - "station_name": "", - "distance": "", - "price": None, - "piles": [], - "parking": "", - "tags": [] - } - - # 预处理：清洗无关字符 - clean_lines = [] - for line in ocr_lines: - # 去除 "行 X: " 前缀 - text = re.sub(r"^行\s*\d+:\s*", "", line) - # 去除 "(置信度: ...)" 后缀 - text = re.sub(r"\s*\(置信度:.*?\)\s*$", "", text) - text = text.strip() - if text: - clean_lines.append(text) - - # 1. 场站名称 (Station Name) - # 策略：通常是第一行，且不包含特定关键词（如km, /度, ￥等） - for line in clean_lines: - if len(line) < 2: continue - if line.startswith("（") or line.startswith("(") or line == "…": continue - - # 排除明显属性行 - if any(k in line.lower() for k in ["km", "/度", "￥", "停车", "积分", "组团"]): - continue - - # 排除纯数字 - if re.match(r"^\d+$", line): continue - - # 排除可能是充电枪信息的行 (例如 "闲2/4", "快", "慢") - if re.search(r"\d+/\d+", line): continue - if line in ["快", "慢", "超", "快充", "慢充", "超充"]: continue - - result["station_name"] = line - break - - # 2. 距离 (Distance) - for line in clean_lines: - # 匹配 "7.4km", "17.4km", "90m" - m = re.search(r"(\d+(\.\d+)?)\s*(km|m)", line, re.IGNORECASE) - if m: - result["distance"] = m.group(0) - break - - # 3. 价格 (Price) - for line in clean_lines: - # 匹配 "0.7111/度", "1.2/度" - m = re.search(r"(\d+\.\d+)(?=/度)", line) - if m: - try: - result["price"] = float(m.group(1)) - break - except: pass - - # 4. 充电枪信息 (Piles) - # 规则：查找 "快", "慢", "超" 以及 "x/y" 格式 - # 示例： "快", "闲3/4" (跨行) 或 "快充 3/4" (同行) - current_type = "未知" - - # 辅助：先尝试在单行内同时匹配类型和数量 - # 如果单行内有 "快" 和 "3/4"，则直接提取 - # 如果单行只有类型，更新 current_type - # 如果单行只有 "3/4"，使用 current_type - - for line in clean_lines: - # 更新当前类型上下文 - if "快" in line: current_type = "快" - elif "慢" in line: current_type = "慢" - elif "超" in line: current_type = "超" - - # 匹配 "3/4" 或 "闲3/4" - # 排除价格行 (包含小数点，如 0.7111/度 这里的 / 前后也可能是数字，需小心) - if "/度" in line: continue - - # 匹配分数结构：整数/整数 - # (?:闲)? 可选的前缀 "闲" - m = re.search(r"(?:闲)?(\d+)/(\d+)", line) - if m: - # 再次确认不是价格（有时候 OCR 会把 . 漏掉，变成 07111/度，但通常会有 /度） - # 简单校验：总数通常不会特别大，比如不会是 2024/12 (日期) - try: - free = int(m.group(1)) - total = int(m.group(2)) - - # 过滤可能的日期或异常值 - if total > 500: continue - - result["piles"].append({ - "type": current_type, - "free": free, - "total": total - }) - except: pass - - # 5. 停车信息 (Parking) - for line in clean_lines: - if "停车" in line: - # 清理前缀符号，如 "·收费停车" -> "收费停车" - clean_text = re.sub(r"^[·\.\sP]+", "", line) - result["parking"] = clean_text - break - - # 6. 标签 (Tags) - # 收集除上述已解析字段外的简短关键词 - known_keywords = ["快", "慢", "超", "P", "组团", "积分"] - for line in clean_lines: - # 简单匹配一些常见 Tag - if "积分" in line: result["tags"].append(line) - elif "组团" in line: result["tags"].append("组团") - elif line.strip() == "P": result["tags"].append("P") - - return result diff --git a/Util/PaddleOCRKit.py b/Util/PaddleOCRKit.py deleted file mode 100644 index e67e880..0000000 --- a/Util/PaddleOCRKit.py +++ /dev/null @@ -1,123 +0,0 @@ -# coding=utf-8 -import os -import sys -import logging -import cv2 -import numpy as np -from paddleocr import PaddleOCR -from Util.OcrParser import OcrParser - -# Initialize logger -logger = logging.getLogger("PaddleOCRKit") - -class PaddleOCRKit: - _instance = None - _ocr = None - - def __new__(cls): - if cls._instance is None: - cls._instance = super(PaddleOCRKit, cls).__new__(cls) - cls._instance._initialize_model() - return cls._instance - - def _initialize_model(self): - """ - Initialize the PaddleOCR model once. - Using PP-OCRv4 mobile model for speed. - """ - try: - # 1. 抑制 PaddlePaddle 的日志输出 - os.environ['GLOG_minloglevel'] = '3' # 抑制 GLOG (C++) 日志 - os.environ['FLAGS_allocator_strategy'] = 'auto_growth' # 减少显存占用警告 - - # 2. 抑制 Python 层的 paddle 日志 - import logging as py_logging - py_logging.getLogger('paddle').setLevel(py_logging.ERROR) - py_logging.getLogger('ppocr').setLevel(py_logging.ERROR) - - logger.info("Initializing PaddleOCR (PP-OCRv4 Mobile)...") - # use_angle_cls is deprecated, using use_textline_orientation instead - # ocr_version='PP-OCRv4' defaults to mobile model - self._ocr = PaddleOCR( - use_textline_orientation=True, - lang="ch", - ocr_version='PP-OCRv4', - show_log=False # 关键：关闭 PaddleOCR 内部日志打印 - ) - logger.info("PaddleOCR initialized successfully.") - except Exception as e: - logger.error(f"Failed to initialize PaddleOCR: {e}") - self._ocr = None - - def recognize(self, image_input): - """ - Recognize text from an image and parse it into structured data. - - :param image_input: Image path (str) or OpenCV image object (numpy.ndarray) - :return: Parsed structured data (dict) or None if failed - """ - if self._ocr is None: - logger.error("PaddleOCR model is not initialized.") - return None - - try: - # PaddleOCR handles both path and ndarray - # predict() is recommended over ocr() in newer versions, but ocr() is the stable high-level API - # We use ocr() as in the test script which worked well. - # cls=True was causing issues in test, but use_textline_orientation=True in init handles it? - # actually in test script: ocr = PaddleOCR(use_textline_orientation=True, ...) - # result = ocr.ocr(image_path) -> this worked, but gave a warning about predict. - # Let's stick to ocr() for now as it returns the structure we expect. - - result = self._ocr.ocr(image_input) - - ocr_text_lines = [] - - # Handle result structure - if not result: - logger.warning("OCR returned empty result.") - else: - res = result[0] - if res is None: - logger.warning("OCR result[0] is None.") - elif isinstance(res, dict) and 'rec_texts' in res: - # New PaddleOCR / Paddlex format - ocr_text_lines = res['rec_texts'] - elif isinstance(res, list): - # Legacy PaddleOCR format - for line in res: - # line structure: [[points], (text, confidence)] - if len(line) >= 2 and isinstance(line[1], (tuple, list)): - ocr_text_lines.append(line[1][0]) - - # --- 关键：打印原始识别内容 --- - if ocr_text_lines: - logger.info(f"🔍 OCR 原始识别文字 ({len(ocr_text_lines)} 行):") - for i, line in enumerate(ocr_text_lines): - logger.info(f" [{i+1}] {line}") - else: - logger.info("No text detected.") - return {} - - # Parse with Regex Parser - parsed_data = OcrParser.parse(ocr_text_lines) - - # --- 关键：打印解析后的结构化数据 --- - if parsed_data: - logger.info("📦 OCR 解析结果 (JSON):") - logger.info(f" {parsed_data}") - - return parsed_data - - except Exception as e: - logger.error(f"Error during OCR recognition: {e}") - return None - -# Global instance for easy access -_kit_instance = None - -def get_ocr_kit(): - global _kit_instance - if _kit_instance is None: - _kit_instance = PaddleOCRKit() - return _kit_instance diff --git a/Util/__pycache__/PaddleOCRKit.cpython-310.pyc b/Util/__pycache__/PaddleOCRKit.cpython-310.pyc index 1a508c2..8454d6e 100644 Binary files a/Util/__pycache__/PaddleOCRKit.cpython-310.pyc and b/Util/__pycache__/PaddleOCRKit.cpython-310.pyc differ