diff --git a/Apps/TeLaiDian/Crawler.py b/Apps/TeLaiDian/Crawler.py index 723863b..793a49e 100644 --- a/Apps/TeLaiDian/Crawler.py +++ b/Apps/TeLaiDian/Crawler.py @@ -120,6 +120,15 @@ class TeLaiDianCrawler(BaseCrawler): detail_check_path = take_screenshot(d, f"tld_detail_check_{int(time.time())}.jpg") logger.info(f"详情页快照已保存: {detail_check_path}") + # 简单验证:如果标题包含 "我的卡券"、"优惠券" 等,说明点错了 + is_wrong_page = await self.check_wrong_page(d, detail_check_path) + if is_wrong_page: + logger.warning(f"检测到进入了错误页面,尝试返回列表...") + d.press("back") + await asyncio.sleep(WAIT_BACK_TO_LIST) + if os.path.exists(detail_check_path): os.remove(detail_check_path) + continue + # 爬取详情 await self.crawl_detail_logic(d, station) @@ -136,6 +145,31 @@ class TeLaiDianCrawler(BaseCrawler): if os.path.exists(screenshot_path): os.remove(screenshot_path) + async def check_wrong_page(self, d, image_path): + """ + 检查是否误触进入了错误的页面(如:我的卡券、活动页等) + """ + prompt = """ + 分析这张截图,判断这是否是一个“充电站详情页”。 + 特征:详情页通常包含场站名称、电价列表、电桩状态等。 + 如果页面显示的是“我的卡券”、“优惠券”、“个人中心”或“活动广告”,则不是详情页。 + + 输出格式为 JSON: + { + "is_detail_page": true/false, + "page_type": "detail" / "coupons" / "ad" / "other", + "reason": "判断依据" + } + """ + try: + res_text = await self.read_image_kit.vlm.analyze_image(image_path, prompt) + json_str = self.read_image_kit.vlm.extract_json(res_text) + data = json.loads(json_str) + return not data.get("is_detail_page", True) + except Exception as e: + logger.error(f"检查页面类型失败: {e}") + return False + async def crawl_detail_logic(self, d, station_info): """ 在详情页提取价格和状态信息 diff --git a/Apps/TeLaiDian/Kit.py b/Apps/TeLaiDian/Kit.py index 6377f22..908fcca 100644 --- a/Apps/TeLaiDian/Kit.py +++ b/Apps/TeLaiDian/Kit.py @@ -169,19 +169,22 @@ def detect_cards_cv(image_path, top_ratio=0.40, bottom_ratio=0.12): # 限制检测范围 top_limit = int(h * top_ratio) bottom_limit = int(h * (1 - bottom_ratio)) + logger.info(f"CV检测卡片: h={h}, w={w}, top_limit={top_limit}, bottom_limit={bottom_limit}") # 使用自适应阈值 thresh = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, 11, 2) # 闭运算:连接断开的边缘 - kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (w // 2, 1)) + # 减小水平核宽度,增加一点垂直连接,以适应卡片边缘不连续的情况 + kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (w // 4, 3)) closed = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, kernel) # 寻找轮廓 contours, _ = cv2.findContours(closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) cards = [] - min_card_width = int(w * 0.7) + # 降低最小宽度要求,有些卡片边缘可能没完全闭合 + min_card_width = int(w * 0.6) temp_boxes = [] for cnt in contours: @@ -189,11 +192,18 @@ def detect_cards_cv(image_path, top_ratio=0.40, bottom_ratio=0.12): # 增加最小高度验证 MIN_CARD_HEIGHT if (cw > min_card_width and - ch > MIN_CARD_HEIGHT and + ch > MIN_CARD_HEIGHT * 0.8 and # 稍微放宽高度限制 y > top_limit and y + ch < bottom_limit): + logger.info(f"✅ 找到候选卡片: y={y}, h={ch}, w={cw}") temp_boxes.append((x, y, cw, ch)) + else: + if cw > min_card_width: + logger.debug(f"❌ 排除候选(不符条件): y={y}, h={ch}, w={cw} (limit: {top_limit}-{bottom_limit}, min_h: {MIN_CARD_HEIGHT*0.8})") + + if not temp_boxes: + logger.warning(f"⚠️ CV 未能在指定范围 ({top_limit}-{bottom_limit}) 内检测到任何场站卡片") # 按 Y 轴排序 temp_boxes.sort(key=lambda b: b[1]) diff --git a/Apps/TeLaiDian/ReadImageKit.py b/Apps/TeLaiDian/ReadImageKit.py index 957f86d..fe96caa 100644 --- a/Apps/TeLaiDian/ReadImageKit.py +++ b/Apps/TeLaiDian/ReadImageKit.py @@ -189,7 +189,11 @@ class ReadImageKit: else: prompt = """ 分析这张充电站列表截图,提取所有充电站卡片信息。 - 忽略顶部的筛选栏,仅提取下方重复出现的场站卡片。 + + **重要限制**: + 1. 忽略页面上半部分(坐标 y < 500)的所有内容,包括顶部的搜索框、广告 Banner 和“PLUS会员/我的卡券”等图标。 + 2. 仅提取下方重复出现的、包含“充电站”名称的矩形卡片。 + 输出格式为 JSON 数组,每个对象包含: - "name": 场站名称 - "address": 场站地址 @@ -215,7 +219,22 @@ class ReadImageKit: "bbox": bbox }) elif not cv_bboxes: - final_stations = vlm_results if isinstance(vlm_results, list) else [] + vlm_list = vlm_results if isinstance(vlm_results, list) else [] + w, h = 0, 0 + for res in vlm_list: + p = res.get("point") + if p and len(p) == 2: + # 如果坐标小于 1000,认为是归一化坐标,需要转换 + if p[0] <= 1000 and p[1] <= 1000: + if w == 0: + import cv2 + img = cv2.imread(image_path) + if img is not None: + h, w = img.shape[:2] + else: + w, h = 1080, 2400 # 兜底 + res["point"] = [int(p[0] * w / 1000), int(p[1] * h / 1000)] + final_stations.append(res) return final_stations except Exception as e: