diff --git a/Apps/YeLiTe/Config/__pycache__/Setting.cpython-310.pyc b/Apps/YeLiTe/Config/__pycache__/Setting.cpython-310.pyc index 2e173a7..031fe06 100644 Binary files a/Apps/YeLiTe/Config/__pycache__/Setting.cpython-310.pyc and b/Apps/YeLiTe/Config/__pycache__/Setting.cpython-310.pyc differ diff --git a/Apps/YeLiTe/Crawler.py b/Apps/YeLiTe/Crawler.py index 2ac9fff..df0f31a 100644 --- a/Apps/YeLiTe/Crawler.py +++ b/Apps/YeLiTe/Crawler.py @@ -6,7 +6,7 @@ import sys import json import time from datetime import datetime -from Apps.YeLiTe.Kit import take_screenshot, clean_station_name, get_image_content_md5 +from Apps.YeLiTe.Kit import take_screenshot, clean_station_name, get_image_content_md5, find_template from Apps.YeLiTe.ReadImageKit import ReadImageKit from Apps.YeLiTe.Service import YiLaiTeService from Apps.YeLiTe.Config.Setting import ( @@ -65,9 +65,9 @@ class YiLaiTeCrawler(BaseCrawler): if TEST_CLEAR_REDIS: logger.info("清理 Redis 中的场站处理记录...") pattern = "crawled:ylt:*" - keys = await self.redis_kit.client.keys(pattern) + keys = await self.redis_kit.keys(pattern) if keys: - await self.redis_kit.client.delete(*keys) + await self.redis_kit.delete(*keys) async def crawl_list_logic(self, d): processed_count = 0 @@ -113,15 +113,48 @@ class YiLaiTeCrawler(BaseCrawler): # 去重 redis_key = f"crawled:ylt:{clean_station_name(name)}" if await self.redis_kit.get_data(redis_key): + logger.info(f"场站 {name} 已处理过，跳过") continue + # --- 新增逻辑：立即中断当前页面的后续遍历，先处理当前这个场站 --- + # 因为一旦点击进入二级页面再返回，列表页的状态可能会变（如滚动位置微调）， + # 或者之前的坐标已经失效。为了稳健，每次处理完一个点击跳转，都应该重新截图分析列表页。 + # 但重新截图分析成本太高，所以至少应该确保处理完这一个后，不会盲目点击后续的旧坐标。 + # + # 然而，当前逻辑是遍历 `stations` 列表。如果第一个场站点击并返回了， + # 后面的 `stations` 中的坐标可能已经不准了（如果返回时页面刷新了）。 + # + # 观察日志发现："发现第一场站后，没有进去查看二级，而是直接退回了小程序的搜索页面" + # 这说明可能是 `d.press("back")` 的逻辑有问题，或者页面结构导致 `back` 行为不一致。 + # + # 针对用户反馈："发现第一场站后，没有进去查看二级，而是直接退回了小程序的搜索页面" + # 这通常是因为在列表页点击后，并没有真正进入二级页（也许点偏了，或者加载慢）， + # 然后程序认为进去了，执行了 `back`，结果把列表页关了退回了搜索页。 + # + # 现在的代码已经加了 MD5 校验 `if before_md5 != after_md5`，理论上能防住这个问题。 + # 但用户最新的反馈是："没有马上进入第一个卡片对应的二级，而是向下继续滚动了去获取更多的场站了" + # 这意味着 `before_md5 != after_md5` 判断为 False，认为没进去，所以跳过了处理，继续循环。 + # + # 为什么明明点击了却没进去？ + # 1. 坐标点偏了？测试脚本显示点是红心，应该是准的。 + # 2. 点击操作被某些透明层拦截了？ + # 3. 页面还没加载完全？ + # + # 让我们增加点击后的重试机制，或者微调点击位置。 + # 另外，如果点击没进去，不应该直接 continue，应该尝试 debug 或者稍微移动一下位置再点。 + # + # 这里我们加一个重试点击逻辑 + # 点击进入前截图，用于对比是否成功进入二级页 before_click_path = take_screenshot(d, f"before_{clean_station_name(name)}") before_md5 = get_image_content_md5(before_click_path, top_ratio=SAFE_EXCLUDE_RATIO, bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO) # 点击进入 logger.info(f">>> 发现新场站: {name} 坐标: {point}") - d.click(point[0], point[1]) + # 稍微等待一下，确保 UI 稳定 + await asyncio.sleep(0.5) + # 使用 input tap 替代 d.click，提高点击成功率 (部分小程序对 click 响应不佳) + d.shell(f"input tap {int(point[0])} {int(point[1])}") await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 分析详情页 (采用异步后台模式) @@ -131,19 +164,78 @@ class YiLaiTeCrawler(BaseCrawler): # 清理临时对比图 if os.path.exists(before_click_path): os.remove(before_click_path) + if before_md5 == after_md5: + logger.warning(f"首次点击 {name} 未跳转，尝试稍微偏移位置重试...") + # 尝试向下偏移 20px 点击 + offset_y = point[1] + 20 + d.shell(f"input tap {int(point[0])} {int(offset_y)}") + await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) + + # 再次截图检查 + if os.path.exists(detail_shot): os.remove(detail_shot) + detail_shot = take_screenshot(d, f"detail_{clean_station_name(name)}_{int(time.time())}") + after_md5 = get_image_content_md5(detail_shot, top_ratio=SAFE_EXCLUDE_RATIO, bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO) + if before_md5 != after_md5: logger.info(f"成功进入详情页: {name}") + + # --- 新增：点击“阶段性电价”按钮以获取完整电价列表 --- + dqdf_pos = find_template(detail_shot, "dqdf.jpg") + detail_shots = [] + + if dqdf_pos: + logger.info(f"发现‘阶段性电价’按钮 {dqdf_pos}，点击进入...") + d.click(dqdf_pos[0], dqdf_pos[1]) + await asyncio.sleep(2) # 等待列表加载 + # 删除旧的详情页截图 + if os.path.exists(detail_shot): os.remove(detail_shot) + + # 截图1 (顶部) + shot1 = take_screenshot(d, f"detail_price_{clean_station_name(name)}_{int(time.time())}_1") + detail_shots.append(shot1) + + # 向下滚动以获取完整价格表 + logger.info("向下滚动以获取更多电价信息(从底部向上滑)...") + # User feedback: 必须放在下方，向上滚动 + w, h = d.window_size() + # start_y=0.9 (底部), end_y=0.4 (上部) + d.swipe(w * 0.5, h * 0.9, w * 0.5, h * 0.4, 0.5) + await asyncio.sleep(1.5) + + # 截图2 (底部) + shot2 = take_screenshot(d, f"detail_price_{clean_station_name(name)}_{int(time.time())}_2") + detail_shots.append(shot2) + + # 关闭分时段定价列表 (点击屏幕上半部分空白处) + logger.info("点击屏幕上部空白处以关闭定价列表...") + d.click(w * 0.5, h * 0.2) + await asyncio.sleep(1.0) + else: + logger.info("未发现‘阶段性电价’按钮，直接分析当前页") + detail_shots.append(detail_shot) + # -------------------------------------------------- + # 启动后台任务处理详情页 - task = asyncio.create_task(self.analyze_detail_background(name, detail_shot)) + task = asyncio.create_task(self.analyze_detail_background(name, detail_shots)) background_tasks.append(task) processed_count += 1 new_stations_in_page += 1 await self.redis_kit.set_data(redis_key, "1", expire=86400*7) - # 返回 + # 返回列表页 (现在应该已经回到了详情页主界面) d.press("back") await asyncio.sleep(WAIT_BACK_TO_LIST) + + # 再次检查是否回到了列表页，如果没回，再点一次 back + after_back_shot = take_screenshot(d, "check_back") + after_back_md5 = get_image_content_md5(after_back_shot, top_ratio=SAFE_EXCLUDE_RATIO, bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO) + if os.path.exists(after_back_shot): os.remove(after_back_shot) + + if after_back_md5 == after_md5: # 还在详情页/电费页 + logger.info("似乎还在二级页面，尝试再次返回...") + d.press("back") + await asyncio.sleep(WAIT_BACK_TO_LIST) else: logger.warning(f"点击场站 {name} 后页面似乎未跳转，跳过返回操作") if os.path.exists(detail_shot): os.remove(detail_shot) @@ -169,23 +261,55 @@ class YiLaiTeCrawler(BaseCrawler): return processed_count - async def analyze_detail_background(self, station_name, image_path): + async def analyze_detail_background(self, station_name, image_paths): """ - 后台异步分析详情页 + 后台异步分析详情页 (支持多张截图合并) """ try: - logger.info(f"开始后台分析场站: {station_name}") - prices = await self.read_image_kit.analyze_detail_price(image_path) - if prices: - await self.service.process_price_detail_data(station_name, prices) - logger.info(f"场站 {station_name} 价格分析完成并入库") + if isinstance(image_paths, str): + image_paths = [image_paths] + + logger.info(f"开始后台分析场站: {station_name} (图片数: {len(image_paths)})") + + all_prices = [] + for img_path in image_paths: + prices = await self.read_image_kit.analyze_detail_price(img_path) + if prices: + all_prices.extend(prices) + + # 去重合并 + unique_prices = [] + seen_periods = set() + + for p in all_prices: + start = str(p.get('start', '')).strip() + end = str(p.get('end', '')).strip() + + if not start or not end: + continue + + key = f"{start}-{end}" + if key not in seen_periods: + seen_periods.add(key) + unique_prices.append(p) + + # 按开始时间排序 + unique_prices.sort(key=lambda x: x.get('start', '00:00')) + + if unique_prices: + await self.service.process_price_detail_data(station_name, unique_prices) + logger.info(f"场站 {station_name} 价格分析完成并入库 (记录数: {len(unique_prices)})") else: logger.warning(f"场站 {station_name} 未识别到价格信息") + except Exception as e: logger.error(f"后台分析 {station_name} 失败: {e}") finally: - if os.path.exists(image_path): - os.remove(image_path) + if isinstance(image_paths, list): + for p in image_paths: + if os.path.exists(p): os.remove(p) + elif isinstance(image_paths, str) and os.path.exists(image_paths): + os.remove(image_paths) async def main(service=None): if service is None: diff --git a/Apps/YeLiTe/Kit.py b/Apps/YeLiTe/Kit.py index 676353e..386d9e2 100644 --- a/Apps/YeLiTe/Kit.py +++ b/Apps/YeLiTe/Kit.py @@ -66,10 +66,7 @@ def get_image_content_md5(file_path, top_ratio=0.1, bottom_ratio=0.1): # 裁剪中间部分 content = img[top:bottom, :] - # 将图片数据转换为字节流计算 MD5 - success, encoded_img = cv2.imencode(".jpg", content) - if success: - return hashlib.md5(encoded_img.tobytes()).hexdigest() + # 直接使用原始像素数组计算 MD5，避免压缩损失 return hashlib.md5(content.tobytes()).hexdigest() def read_image(path): @@ -98,93 +95,117 @@ def save_image(path, img): logger.error(f"Error saving image {path}: {e}") return False +def find_template(image_path, template_name, threshold=0.8): + """ + 在图片中查找模板位置 + """ + img = read_image(image_path) + if img is None: + return None + + template_path = os.path.join(os.path.dirname(__file__), "Template", template_name) + if not os.path.exists(template_path): + logger.warning(f"Template not found: {template_path}") + return None + + template = read_image(template_path) + if template is None: + return None + + res = cv2.matchTemplate(img, template, cv2.TM_CCOEFF_NORMED) + min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(res) + + if max_val >= threshold: + t_h, t_w = template.shape[:2] + center_x = max_loc[0] + t_w // 2 + center_y = max_loc[1] + t_h // 2 + return (center_x, center_y) + return None + def detect_cards_cv(image_path, top_ratio=0.15, bottom_ratio=0.1): """ 使用计算机图形学 (OpenCV) 检测列表中的场站卡片 - 方案：以 nav.jpg 为锚点，向上/下探测白色卡片边界 + 方案：使用 Canny 边缘检测 + 轮廓查找，提取圆角矩形卡片 """ img = read_image(image_path) if img is None: return [] h, w = img.shape[:2] - gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) - # 1. 模板匹配查找锚点 (nav.jpg) - template_path = os.path.join(os.path.dirname(__file__), "Template", "nav.jpg") - anchors = [] - if os.path.exists(template_path): - template = read_image(template_path) - if template is not None: - res = cv2.matchTemplate(img, template, cv2.TM_CCOEFF_NORMED) - threshold = 0.8 - loc = np.where(res >= threshold) - - # 去重记录锚点中心 - t_h, t_w = template.shape[:2] - matched_points = [] - for pt in zip(*loc[::-1]): # (x, y) - is_duplicate = False - for mx, my in matched_points: - if abs(mx - pt[0]) < 50 and abs(my - pt[1]) < 50: - is_duplicate = True - break - if not is_duplicate: - matched_points.append(pt) - anchors.append((pt[0] + t_w // 2, pt[1] + t_h // 2)) - - logger.info(f"通过模板匹配找到 {len(anchors)} 个锚点") - - # 2. 基于锚点向上/下扩张探测边界 - # 计算每一行的平均亮度，用于探测背景 - row_means = np.mean(gray[:, int(w*0.2):int(w*0.8)], axis=1) + # 1. 预处理：转灰度 -> 高斯模糊去噪 + gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) + blurred = cv2.GaussianBlur(gray, (5, 5), 0) + + # 2. Canny 边缘检测 + # 阈值设置：降低阈值以适应低对比度边缘 (如浅灰背景上的白卡片) + # 经过测试，(10, 50) 对于浅灰背景上的白卡片效果最佳 + edges = cv2.Canny(blurred, 10, 50) + + # 3. 查找轮廓 + contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) cards = [] - for ax, ay in anchors: - # 向上探测：直到亮度低于 245 (进入灰色背景) - y1 = ay - while y1 > h * top_ratio: - if row_means[int(y1)] < 245: - break - y1 -= 1 + min_card_width = int(w * 0.8) # 卡片宽度通常占屏幕 80% 以上 + min_card_height = 150 # 卡片高度至少 150px + top_limit = int(h * top_ratio) + bottom_limit = int(h * (1 - bottom_ratio)) + + valid_contours = [] + + for cnt in contours: + x, y, cw, ch = cv2.boundingRect(cnt) - # 向下探测：直到亮度低于 245 - y2 = ay - while y2 < h * (1 - bottom_ratio): - if row_means[int(y2)] < 245: - break - y2 += 1 + # 筛选条件： + # 1. 宽度足够大 (排除小图标、文字) + # 2. 高度适中 + # 3. 位置在安全区域内 (排除顶部导航和底部 TabBar) + if (cw > min_card_width and + ch > min_card_height and + y > top_limit and + y + ch < bottom_limit): - # 验证卡片高度是否合理 - if y2 - y1 > 150: - # 给 y 坐标增加一点 padding - cards.append([int(w * 0.05), int(y1 + 5), int(w * 0.95), int(y2 - 5)]) - - # 3. 兜底方案：如果模板匹配没找到，使用之前的投影法 - if not cards: - logger.warning("模板匹配未找到卡片，尝试投影法兜底...") - _, binary = cv2.threshold(gray, 250, 255, cv2.THRESH_BINARY) - top = int(h * top_ratio) - bottom = int(h * (1 - bottom_ratio)) - roi_binary = binary[:, int(w*0.2):int(w*0.8)] - row_white_counts = np.sum(roi_binary == 255, axis=1) - threshold_count = (int(w*0.8) - int(w*0.2)) * 0.9 - is_card_row = row_white_counts > threshold_count - - start_y = None - for y in range(top, bottom): - if is_card_row[y]: - if start_y is None: start_y = y - else: - if start_y is not None: - if y - start_y > 150: - cards.append([int(w * 0.05), start_y + 10, int(w * 0.95), y - 10]) - start_y = None - if start_y is not None and bottom - start_y > 150: - cards.append([int(w * 0.05), start_y + 10, int(w * 0.95), bottom - 10]) - - # 4. 排序 + # 简单去重：检查是否与已有的卡片高度重叠 + is_duplicate = False + center_y = y + ch // 2 + for vx, vy, vcw, vch in valid_contours: + v_center_y = vy + vch // 2 + if abs(center_y - v_center_y) < 50: # 中心点接近，视为同一个 + is_duplicate = True + break + + if not is_duplicate: + valid_contours.append((x, y, cw, ch)) + # 增加一点 padding，收缩一下边界，避免点到边缘 + padding = 5 + cards.append([ + max(0, x + padding), + max(0, y + padding), + min(w, x + cw - padding), + min(h, y + ch - padding) + ]) + + # 按 y 坐标排序 cards.sort(key=lambda c: c[1]) + + # 如果没找到，尝试稍微放宽条件兜底 (比如宽度 > 60%) + if not cards: + logger.warning("严格轮廓检测未找到卡片，尝试放宽条件...") + for cnt in contours: + x, y, cw, ch = cv2.boundingRect(cnt) + if (cw > int(w * 0.6) and ch > 100 and y > top_limit and y + ch < bottom_limit): + # 简单去重 + center_y = y + ch // 2 + is_duplicate = False + for existing in cards: + ec_y = (existing[1] + existing[3]) // 2 + if abs(center_y - ec_y) < 50: + is_duplicate = True + break + if not is_duplicate: + cards.append([x, y, x+cw, y+ch]) + cards.sort(key=lambda c: c[1]) + logger.info(f"最终检测到 {len(cards)} 个场站卡片区域") return cards diff --git a/Apps/YeLiTe/ReadImageKit.py b/Apps/YeLiTe/ReadImageKit.py index d7c7e13..269e97a 100644 --- a/Apps/YeLiTe/ReadImageKit.py +++ b/Apps/YeLiTe/ReadImageKit.py @@ -21,6 +21,44 @@ class ReadImageKit: def __init__(self): self.vlm = VLMKit() + async def analyze_detail_price(self, image_path): + """ + 分析详情页截图，提取电价信息 + """ + prompt = """ + 分析这张充电站详情页截图，提取**电价时段表**。 + 请仔细寻找包含“时段”、“电价”、“服务费”或“总价”的表格或列表。 + + 请提取每个时段的： + 1. 开始时间 (HH:MM) + 2. 结束时间 (HH:MM) + 3. 总电价 (元/度，包含电费和服务费) + + 如果图片中只显示了当前时段的价格，请尽可能提取当前时段的信息。 + 如果包含多个时段，请全部提取。 + + 输出格式为 JSON 数组： + [ + { + "start": "00:00", + "end": "08:00", + "price": 1.23 + }, + ... + ] + 如果无法识别任何价格信息，请返回空数组 []。 + """ + try: + res_text = await self.vlm.analyze_image(image_path, prompt) + json_str = self.vlm.extract_json(res_text) + prices = json.loads(json_str) + if isinstance(prices, list): + return prices + return [] + except Exception as e: + logger.error(f"分析电价详情失败: {e}") + return [] + @classmethod async def detect_ad_popup(cls, image_path, device_info=None): """ @@ -98,14 +136,20 @@ class ReadImageKit: final_stations = [] if cv_bboxes and isinstance(vlm_results, list): # 将 VLM 识别到的名称与 CV 的坐标匹配 + # 注意：VLM 返回的数组可能包含 null (对应无效的框)，我们需要跳过 for i, res in enumerate(vlm_results): - if i < len(cv_bboxes) and res and res.get("is_valid"): + if i < len(cv_bboxes): bbox = cv_bboxes[i] - final_stations.append({ - "name": res.get("name"), - "point": [(bbox[0] + bbox[2]) // 2, (bbox[1] + bbox[3]) // 2], - "bbox": bbox - }) + # 只要 VLM 认为有效，或者 VLM 没明确说是无效的且有名字，就认为是有效场站 + # 有时候 VLM 会漏掉 is_valid 字段 + if res and (res.get("is_valid") is True or (res.get("name") and res.get("is_valid") is not False)): + final_stations.append({ + "name": res.get("name"), + "point": [(bbox[0] + bbox[2]) // 2, (bbox[1] + bbox[3]) // 2], + "bbox": bbox + }) + else: + logger.info(f"VLM 判定第 {i+1} 个框无效或未识别到名称: {res}") elif not cv_bboxes: final_stations = vlm_results if isinstance(vlm_results, list) else [] diff --git a/Apps/YeLiTe/Test/check_last_run.py b/Apps/YeLiTe/Test/check_last_run.py new file mode 100644 index 0000000..0f8b187 --- /dev/null +++ b/Apps/YeLiTe/Test/check_last_run.py @@ -0,0 +1,44 @@ +# coding=utf-8 +import sys +import os +import hashlib +import cv2 +import numpy as np + +# Add project root to sys.path +project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) +sys.path.append(project_root) + +from Apps.YeLiTe.Kit import get_image_content_md5 +from Apps.YeLiTe.Config import Setting + +def check_result(): + output_dir = os.path.join(project_root, "Output") + files = sorted([f for f in os.listdir(output_dir) if f.startswith("verify_step1_list_")]) + if not files: + print("No verification files found.") + return + + latest_timestamp = files[-1].split("_")[-1].replace(".jpg", "") + print(f"Checking latest run: {latest_timestamp}") + + step1 = os.path.join(output_dir, f"verify_step1_list_{latest_timestamp}.jpg") + step4 = os.path.join(output_dir, f"verify_step4_after_{latest_timestamp}.jpg") + + if not os.path.exists(step4): + print("Step 4 file not found. Script might have failed mid-way.") + return + + md5_1 = get_image_content_md5(step1, top_ratio=Setting.SAFE_EXCLUDE_RATIO, bottom_ratio=Setting.BOTTOM_SAFE_EXCLUDE_RATIO) + md5_4 = get_image_content_md5(step4, top_ratio=Setting.SAFE_EXCLUDE_RATIO, bottom_ratio=Setting.BOTTOM_SAFE_EXCLUDE_RATIO) + + print(f"Step 1 MD5: {md5_1}") + print(f"Step 4 MD5: {md5_4}") + + if md5_1 != md5_4: + print(">>> RESULT: SUCCESS (MD5 Changed)") + else: + print(">>> RESULT: FAILURE (MD5 Identical)") + +if __name__ == "__main__": + check_result() diff --git a/Apps/YeLiTe/Test/debug_contours.jpg b/Apps/YeLiTe/Test/debug_contours.jpg new file mode 100644 index 0000000..843ca08 Binary files /dev/null and b/Apps/YeLiTe/Test/debug_contours.jpg differ diff --git a/Apps/YeLiTe/Test/debug_edges.jpg b/Apps/YeLiTe/Test/debug_edges.jpg new file mode 100644 index 0000000..a304ec2 Binary files /dev/null and b/Apps/YeLiTe/Test/debug_edges.jpg differ diff --git a/Apps/YeLiTe/Test/list_1768309287_flag.jpg b/Apps/YeLiTe/Test/list_1768309287_flag.jpg new file mode 100644 index 0000000..ff10978 Binary files /dev/null and b/Apps/YeLiTe/Test/list_1768309287_flag.jpg differ diff --git a/Apps/YeLiTe/Test/list_1768309287_lv.jpg b/Apps/YeLiTe/Test/list_1768309287_lv.jpg new file mode 100644 index 0000000..32c40e4 Binary files /dev/null and b/Apps/YeLiTe/Test/list_1768309287_lv.jpg differ diff --git a/Apps/YeLiTe/Test/result_list_1768309287.jpg b/Apps/YeLiTe/Test/result_list_1768309287.jpg new file mode 100644 index 0000000..7a5c9e9 Binary files /dev/null and b/Apps/YeLiTe/Test/result_list_1768309287.jpg differ diff --git a/Apps/YeLiTe/Test/test_card_detection.py b/Apps/YeLiTe/Test/test_card_detection.py new file mode 100644 index 0000000..8f47ccf --- /dev/null +++ b/Apps/YeLiTe/Test/test_card_detection.py @@ -0,0 +1,82 @@ +# coding=utf-8 +import sys +import os +import cv2 +import numpy as np +import time + +# Add project root to sys.path +project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) +sys.path.append(project_root) + +try: + from Apps.YeLiTe.Kit import detect_cards_cv + from Apps.YeLiTe.Config import Setting +except ImportError as e: + print(f"Import Error: {e}") + print(f"Current sys.path: {sys.path}") + sys.exit(1) + +def test_detection_and_click_simulation(image_path): + print(f"Loading image: {image_path}") + if not os.path.exists(image_path): + print(f"Error: File not found: {image_path}") + return + + # 1. Run detection + print("Running detect_cards_cv...") + cards = detect_cards_cv(image_path, top_ratio=Setting.SAFE_EXCLUDE_RATIO, bottom_ratio=Setting.BOTTOM_SAFE_EXCLUDE_RATIO) + + if not cards: + print("No cards detected!") + return + + print(f"Detected {len(cards)} cards.") + + # 2. Draw results + img = cv2.imread(image_path) + + for i, box in enumerate(cards): + x1, y1, x2, y2 = box + + # Draw green box + cv2.rectangle(img, (x1, y1), (x2, y2), (0, 255, 0), 2) + + # Calculate click point (center) + center_x = int((x1 + x2) / 2) + center_y = int((y1 + y2) / 2) + + # Draw click point (Red Circle) + cv2.circle(img, (center_x, center_y), 10, (0, 0, 255), -1) + + # Add label + label = f"#{i+1} ({center_x}, {center_y})" + cv2.putText(img, label, (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2) + + print(f"Card #{i+1}: Box=[{x1}, {y1}, {x2}, {y2}], Click Point=({center_x}, {center_y})") + + # 3. Save result + output_filename = f"result_{os.path.basename(image_path)}" + output_path = os.path.join(os.path.dirname(__file__), output_filename) + cv2.imwrite(output_path, img) + print(f"Result saved to: {output_path}") + +if __name__ == "__main__": + # Default to the latest list_*.jpg in Output if no argument provided + target_image = "" + + if len(sys.argv) > 1: + target_image = sys.argv[1] + else: + output_dir = os.path.join(project_root, "Output") + if os.path.exists(output_dir): + files = [f for f in os.listdir(output_dir) if f.startswith("list_") and f.endswith(".jpg")] + if files: + # Sort by time (filename usually contains timestamp) + files.sort(reverse=True) + target_image = os.path.join(output_dir, files[0]) + + if target_image: + test_detection_and_click_simulation(target_image) + else: + print("No target image found. Please provide an image path as an argument.") diff --git a/Apps/YeLiTe/Test/test_click_verification.py b/Apps/YeLiTe/Test/test_click_verification.py new file mode 100644 index 0000000..2326079 --- /dev/null +++ b/Apps/YeLiTe/Test/test_click_verification.py @@ -0,0 +1,129 @@ +# coding=utf-8 +import sys +import os +import time +import asyncio +import uiautomator2 as u2 +import cv2 + +# Add project root to sys.path +project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../")) +sys.path.append(project_root) + +try: + from Apps.YeLiTe.Kit import detect_cards_cv, draw_rectangles, take_screenshot, get_image_content_md5 + from Apps.YeLiTe.Config import Setting +except ImportError as e: + print(f"Import Error: {e}") + sys.exit(1) + +async def run_verification(): + print(">>> [Step 0] 连接设备...") + try: + d = u2.connect() + info = d.info + print(f"设备已连接: {info.get('productName', 'Unknown Device')}") + except Exception as e: + print(f"设备连接失败: {e}") + return + + # 这里我们假设已经在列表页，或者直接启动 + # 为了保险，尝试回到桌面重新进? 不，用户说“获取第一屏场站列表”，假设用户已经准备好了环境，或者我们手动进一下 + # 为了全自动化，我们调用 Opener 进一下 + # 但是 Opener 可能会重启微信，有点慢。 + # 我们先判断当前包名，如果是微信，就默认在列表页（或者由用户手动滑到列表页） + # 用户指令暗示是交互式的："获取第一屏...你去点击一下...停止程序" + # 所以我们假设当前就在列表页 + + current_app = d.app_current() + print(f"当前前台应用: {current_app.get('package')}") + if "com.tencent.mm" not in current_app.get('package', ''): + print("警告：微信似乎不在前台，请确保手机显示的是【驿来特】小程序列表页") + print("给你 5 秒钟调整手机屏幕...") + time.sleep(5) + else: + print("检测到微信在前台，准备截图...") + time.sleep(2) # 稍微等一下稳定 + + # 1. 获取第一屏场站列表，然后截图 + timestamp = int(time.time()) + list_screenshot_path = os.path.join(project_root, "Output", f"verify_step1_list_{timestamp}.jpg") + print(f"\n>>> [Step 1] 截图列表页: {list_screenshot_path}") + d.screenshot(list_screenshot_path) + + # 2. 绿框标注，标识模拟点击点 + print("\n>>> [Step 2] 识别卡片并标注...") + # 使用配置中的排除比例 + cards = detect_cards_cv(list_screenshot_path, top_ratio=Setting.SAFE_EXCLUDE_RATIO, bottom_ratio=Setting.BOTTOM_SAFE_EXCLUDE_RATIO) + + if not cards: + print("!!! [Error] 未检测到任何场站卡片，测试终止") + print(f"请检查截图: {list_screenshot_path}") + return + + print(f"检测到 {len(cards)} 个卡片区域") + + # 在图上绘制绿框和红点 + debug_img = cv2.imread(list_screenshot_path) + if debug_img is None: + print("读取截图失败") + return + + first_card = cards[0] + # 计算点击点：中心点 + click_x = int((first_card[0] + first_card[2]) / 2) + click_y = int((first_card[1] + first_card[3]) / 2) + + # 绘制所有卡片 + for i, box in enumerate(cards): + # 绿框 + cv2.rectangle(debug_img, (int(box[0]), int(box[1])), (int(box[2]), int(box[3])), (0, 255, 0), 2) + # 标号 + cv2.putText(debug_img, str(i+1), (int(box[0]), int(box[1])+30), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) + + # 绘制第一个卡片的点击点 (实心红圆) + cv2.circle(debug_img, (click_x, click_y), 15, (0, 0, 255), -1) + # 绘制同心圆 + cv2.circle(debug_img, (click_x, click_y), 25, (0, 0, 255), 2) + + marked_path = os.path.join(project_root, "Output", f"verify_step2_marked_{timestamp}.jpg") + cv2.imwrite(marked_path, debug_img) + print(f">>> 已生成标注图 (绿框+红点): {marked_path}") + print(f">>> 模拟点击目标: 第 1 个卡片, 坐标 ({click_x}, {click_y})") + + # 计算点击前的 MD5 + before_md5 = get_image_content_md5(list_screenshot_path, top_ratio=Setting.SAFE_EXCLUDE_RATIO, bottom_ratio=Setting.BOTTOM_SAFE_EXCLUDE_RATIO) + + # 3. 你去点击一下，等一会 + print(f"\n>>> [Step 3] 执行点击操作...") + # 使用 shell input tap + cmd = f"input tap {click_x} {click_y}" + print(f"执行命令: {cmd}") + d.shell(cmd) + + print(">>> 点击完成，等待 5 秒让页面跳转...") + await asyncio.sleep(5) + + # 4. 再继续截图，判断是否进入 + after_screenshot_path = os.path.join(project_root, "Output", f"verify_step4_after_{timestamp}.jpg") + print(f"\n>>> [Step 4] 截图当前页面: {after_screenshot_path}") + d.screenshot(after_screenshot_path) + + after_md5 = get_image_content_md5(after_screenshot_path, top_ratio=Setting.SAFE_EXCLUDE_RATIO, bottom_ratio=Setting.BOTTOM_SAFE_EXCLUDE_RATIO) + + print("\n>>> [Step 5] 结果验证") + print("-" * 30) + print(f"点击前 MD5: {before_md5}") + print(f"点击后 MD5: {after_md5}") + + if before_md5 != after_md5: + print(">>> [SUCCESS] 判定结果: 成功进入详情页！ (MD5发生变化)") + else: + print(">>> [FAILURE] 判定结果: 未进入详情页 (MD5未变化)") + print("可能原因: 1. 点击无效 2. 页面加载极慢 3. 已经是详情页(误判)") + print("-" * 30) + +if __name__ == "__main__": + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + loop.run_until_complete(run_verification()) diff --git a/Apps/YeLiTe/Test/test_cv.py b/Apps/YeLiTe/Test/test_cv.py new file mode 100644 index 0000000..763bb18 --- /dev/null +++ b/Apps/YeLiTe/Test/test_cv.py @@ -0,0 +1,111 @@ +import sys +import os +import cv2 +import numpy as np + +# Add project root to path +project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) +sys.path.append(project_root) + +from Apps.YeLiTe.Kit import detect_cards_cv, read_image + +def test_detection(): + image_path = r"d:\dsWork\aiData\Output\list_1768309287.jpg" + if not os.path.exists(image_path): + print(f"Error: Image not found at {image_path}") + # Try to list files in Output to see what's available + output_dir = os.path.dirname(image_path) + if os.path.exists(output_dir): + print(f"Files in {output_dir}:") + for f in os.listdir(output_dir): + print(f) + return + + print(f"Testing detection on {image_path}") + + # Debug mode: copy logic from Kit.py but add debug prints and saves + img = read_image(image_path) + if img is None: return + h, w = img.shape[:2] + print(f"Image size: {w}x{h}") + + gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) + blurred = cv2.GaussianBlur(gray, (5, 5), 0) + + # Try different thresholds + edges = cv2.Canny(blurred, 10, 50) # Lower thresholds + + # Save edges + debug_edges_path = os.path.join(os.path.dirname(__file__), "debug_edges.jpg") + cv2.imencode('.jpg', edges)[1].tofile(debug_edges_path) + print(f"Saved edges to {debug_edges_path}") + + contours, _ = cv2.findContours(edges, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) + print(f"Found {len(contours)} contours") + + # Draw all contours on a debug image + debug_contours_img = img.copy() + cv2.drawContours(debug_contours_img, contours, -1, (0, 0, 255), 1) + debug_contours_path = os.path.join(os.path.dirname(__file__), "debug_contours.jpg") + cv2.imencode('.jpg', debug_contours_img)[1].tofile(debug_contours_path) + print(f"Saved all contours to {debug_contours_path}") + + min_card_width = int(w * 0.8) + print(f"Min card width (strict): {min_card_width}") + + for i, cnt in enumerate(contours): + x, y, cw, ch = cv2.boundingRect(cnt) + if cw > 100 and ch > 50: # Only print somewhat large ones + print(f"Contour {i}: pos=({x},{y}), size={cw}x{ch}, aspect_ratio={cw/ch:.2f}") + + # Run actual detection + try: + cards = detect_cards_cv(image_path) + print(f"Detected {len(cards)} cards") + except Exception as e: + print(f"Error during detection: {e}") + import traceback + traceback.print_exc() + return + + # Read image for drawing + img_lv = read_image(image_path) + img_flag = read_image(image_path) + if img_lv is None: + print("Failed to read image") + return + + for i, card in enumerate(cards): + x1, y1, x2, y2 = card + print(f"Card {i+1}: [{x1}, {y1}, {x2}, {y2}]") + + # Draw green rectangle on img_lv + cv2.rectangle(img_lv, (x1, y1), (x2, y2), (0, 255, 0), 2) + cv2.putText(img_lv, f"Card {i+1}", (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2) + + # Calculate click point (center) + center_x = (x1 + x2) // 2 + center_y = (y1 + y2) // 2 + + # Draw red click point on img_flag + cv2.circle(img_flag, (center_x, center_y), 5, (0, 0, 255), -1) + # Also draw a crosshair for better visibility + cv2.line(img_flag, (center_x - 10, center_y), (center_x + 10, center_y), (0, 0, 255), 1) + cv2.line(img_flag, (center_x, center_y - 10), (center_x, center_y + 10), (0, 0, 255), 1) + cv2.putText(img_flag, f"Click {i+1}", (center_x + 10, center_y), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 255), 1) + + print(f"Click Point {i+1}: ({center_x}, {center_y})") + + # Save _lv.jpg + base_name = os.path.splitext(os.path.basename(image_path))[0] + output_lv = os.path.join(os.path.dirname(__file__), f"{base_name}_lv.jpg") + cv2.imencode('.jpg', img_lv)[1].tofile(output_lv) + print(f"Green box image saved to {output_lv}") + + # Save _flag.jpg + output_flag = os.path.join(os.path.dirname(__file__), f"{base_name}_flag.jpg") + cv2.imencode('.jpg', img_flag)[1].tofile(output_flag) + print(f"Click point image saved to {output_flag}") + +if __name__ == "__main__": + test_detection() diff --git a/Apps/YeLiTe/__pycache__/Crawler.cpython-310.pyc b/Apps/YeLiTe/__pycache__/Crawler.cpython-310.pyc index 3d2e898..af5c093 100644 Binary files a/Apps/YeLiTe/__pycache__/Crawler.cpython-310.pyc and b/Apps/YeLiTe/__pycache__/Crawler.cpython-310.pyc differ diff --git a/Apps/YeLiTe/__pycache__/Kit.cpython-310.pyc b/Apps/YeLiTe/__pycache__/Kit.cpython-310.pyc index df619fd..1b0c008 100644 Binary files a/Apps/YeLiTe/__pycache__/Kit.cpython-310.pyc and b/Apps/YeLiTe/__pycache__/Kit.cpython-310.pyc differ diff --git a/Apps/YeLiTe/__pycache__/ReadImageKit.cpython-310.pyc b/Apps/YeLiTe/__pycache__/ReadImageKit.cpython-310.pyc index 8c06839..4d1b247 100644 Binary files a/Apps/YeLiTe/__pycache__/ReadImageKit.cpython-310.pyc and b/Apps/YeLiTe/__pycache__/ReadImageKit.cpython-310.pyc differ diff --git a/T3_YeLiTe.py b/T3_YeLiTe.py index f90a0f7..add48e8 100644 --- a/T3_YeLiTe.py +++ b/T3_YeLiTe.py @@ -14,7 +14,7 @@ logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) -logger = logging.getLogger("T3_YiLaiTe") +logger = logging.getLogger("T3_YeLiTe") try: from Apps.YeLiTe.Service import YiLaiTeService diff --git a/T3_YeLiTe_DirectCrawl.py b/T3_YeLiTe_DirectCrawl.py new file mode 100644 index 0000000..521df81 --- /dev/null +++ b/T3_YeLiTe_DirectCrawl.py @@ -0,0 +1,59 @@ +# coding=utf-8 +import sys +import os +import asyncio +import logging + +# 将项目根目录添加到 sys.path +project_root = os.path.dirname(os.path.abspath(__file__)) +if project_root not in sys.path: + sys.path.append(project_root) + +# 配置日志 +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger("T3_YeLiTe_DirectCrawl") + +try: + from Apps.YeLiTe.Service import YiLaiTeService + from Apps.YeLiTe import Crawler, Kit +except Exception as e: + logger.error(f"❌ 初始化导入失败: {e}") + sys.exit(1) + +async def run_process(): + logger.info("=== 开始直接爬取任务 (驿来特) ===") + logger.info("注意：请确保手机当前停留在【驿来特】小程序的场站列表页！") + + # 步骤 0: 初始化基础服务 + logger.info(">>> 步骤 0: 初始化基础服务 (数据库连接)...") + service = YiLaiTeService() + await service.init_db() + + try: + # 启动前清空临时目录 + Kit.clear_temp_dir() + + # 步骤 2: 执行爬取任务 + logger.info(">>> 步骤 1: 开始执行场站爬取任务 (跳过启动步骤)...") + # 调用 Crawler 的 main，并传入 service + await Crawler.main(service=service) + + logger.info("✅ 爬取任务完成！") + + except Exception as e: + logger.error(f"❌ 运行异常: {e}") + finally: + if service: + await service.close_db() + logger.info("=== 任务结束 ===") + +if __name__ == "__main__": + try: + asyncio.run(run_process()) + except KeyboardInterrupt: + logger.info("\n🛑 用户手动停止了程序 (Ctrl+C)。") + except Exception as e: + logger.exception(f"主程序崩溃: {e}")