# coding=utf-8 import asyncio import asyncio import os import sys import time import cv2 from Apps.TeLaiDian.Kit import ( take_screenshot, get_image_content_md5, clean_station_name, setup_logger, detect_price_click_point_cv, read_image, save_image ) from Apps.TeLaiDian.ReadImageKit import ReadImageKit from Apps.TeLaiDian.Service import TeLaiDianService from Apps.TeLaiDian.Config.Setting import ( SCROLL_DISTANCE_RATIO, WAIT_AFTER_SCROLL, MAX_STATIONS_COUNT, SAFE_EXCLUDE_RATIO, BOTTOM_SAFE_EXCLUDE_RATIO, WAIT_DETAIL_PAGE_LOAD, WAIT_BACK_TO_LIST, DETAIL_SCROLL_DISTANCE_RATIO ) from Core.BaseCrawler import BaseCrawler import uiautomator2 as u2 # 项目根目录处理 project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) if project_root not in sys.path: sys.path.append(project_root) # 初始化日志 logger = setup_logger("TeLaiDianCrawler") class TeLaiDianCrawler(BaseCrawler): def __init__(self, service=None): super().__init__(service or TeLaiDianService()) self.read_image_kit = ReadImageKit() async def start(self): """ 实现 BaseCrawler 的启动入口 注意：在 T4 等全流程脚本中，小程序由 Opener 提前打开， 此处直接开始爬取逻辑。 """ d = u2.connect() await self.crawl_list_logic(d) async def open_app(self): """ 打开特来电小程序 """ from Apps.TeLaiDian import Opener return await Opener.open_mini_program() async def crawl_list_logic(self, d): processed_count = 0 last_md5 = None while processed_count < MAX_STATIONS_COUNT: # 1. 截图并分析 screenshot_path = take_screenshot(d, f"tld_list_{int(time.time())}.jpg") # 检测是否滚动到底部 curr_md5 = get_image_content_md5(screenshot_path, top_ratio=SAFE_EXCLUDE_RATIO, bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO) if last_md5 == curr_md5: logger.info("内容无变化，判定已到底部") if os.path.exists(screenshot_path): os.remove(screenshot_path) break last_md5 = curr_md5 stations = await self.read_image_kit.analyze_station_list(screenshot_path) if not stations: logger.info("本页未检测到场站，尝试滑动...") d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO) await asyncio.sleep(WAIT_AFTER_SCROLL) continue for station in stations: if processed_count >= MAX_STATIONS_COUNT: break name = station.get("name") point = station.get("point") if not name or not point: continue logger.info(f"处理场站: {name} (坐标: {point})") # 点击进入详情 d.click(point[0], point[1]) logger.info(f"已点击场站 '{name}'，等待 {WAIT_DETAIL_PAGE_LOAD}s 加载详情页...") await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 截图验证是否进入详情页 (可选：可以加入更复杂的页面特征判断) detail_check_path = take_screenshot(d, f"tld_detail_check_{int(time.time())}.jpg") logger.info(f"详情页快照已保存: {detail_check_path}") # 爬取详情 await self.crawl_detail_logic(d, station) # 返回列表 d.press("back") await asyncio.sleep(WAIT_BACK_TO_LIST) processed_count += 1 # 滑动到下一页 d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO) await asyncio.sleep(WAIT_AFTER_SCROLL) if os.path.exists(screenshot_path): os.remove(screenshot_path) async def crawl_detail_logic(self, d, station_info): """ 在详情页提取价格和状态信息 """ # 1. 识别第一屏的基础信息 (名称、精确地址) first_screen_path = take_screenshot(d, f"tld_detail_basic_{int(time.time())}.jpg") basic_info = await self.read_image_kit.analyze_detail_basic_info(first_screen_path) station_name = basic_info.get("name") or station_info.get("name") address = basic_info.get("address") or station_info.get("address") logger.info(f"详情页基础信息识别完成: {station_name} | {address}") # 2. 向上滑动以露出价格按钮 logger.info(f"执行滑动操作以显示价格按钮 (距离比例: {DETAIL_SCROLL_DISTANCE_RATIO})...") # 从屏幕中间向上滑动 d.swipe_ext("up", scale=DETAIL_SCROLL_DISTANCE_RATIO) await asyncio.sleep(1.5) # 3. 点击“价格信息”区域 (识别橘红色价格 P0) price_button_screen = take_screenshot(d, f"tld_before_price_click_{int(time.time())}.jpg") logger.info("正在通过 CV 寻找橘红色价格区域 (P0)...") click_point = detect_price_click_point_cv(price_button_screen) # 调试：生成点击点标注图 if click_point: debug_flag_path = price_button_screen.replace(".jpg", "_click_debug.jpg") img_debug = read_image(price_button_screen) if img_debug is not None: cv2.circle(img_debug, (click_point[0], click_point[1]), 20, (0, 0, 255), -1) # 红色大圆点 cv2.line(img_debug, (click_point[0]-40, click_point[1]), (click_point[0]+40, click_point[1]), (255, 255, 255), 3) cv2.line(img_debug, (click_point[0], click_point[1]-40), (click_point[0], click_point[1]+40), (255, 255, 255), 3) save_image(debug_flag_path, img_debug) logger.info(f"点击点调试图已保存: {debug_flag_path}") try: if click_point: logger.info(f"CV 成功定位价格区域，点击坐标: {click_point}") d.click(click_point[0], click_point[1]) await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) else: logger.warning("CV 未能定位价格区域，尝试模板匹配兜底...") template_path = os.path.join(project_root, "Apps", "TeLaiDian", "Template", "jgxx.jpg") match = d.image.match(template_path) if match: d.image.click(template_path) else: logger.warning("模板匹配也失败，执行坐标兜底...") w, h = d.window_size() d.click(w // 2, int(h * 0.45)) # 滑动后价格通常在屏幕中上部 await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) except Exception as e: logger.error(f"点击价格区域失败: {e}") finally: if os.path.exists(price_button_screen): os.remove(price_button_screen) # 4. 循环滑动抓取完整分时电价 all_prices = [] last_price_md5 = None price_page_count = 0 max_price_pages = 3 # 分时电价通常不会超过3页 logger.info("开始循环滑动抓取完整分时电价...") while price_page_count < max_price_pages: price_screen_path = take_screenshot(d, f"tld_detail_price_{int(time.time())}.jpg") # 校验页面是否发生滚动变化 curr_md5 = get_image_content_md5(price_screen_path, top_ratio=0.2, bottom_ratio=0.2) if curr_md5 == last_price_md5: logger.info("价格页面内容无变化，判定已触底") if os.path.exists(price_screen_path): os.remove(price_screen_path) break last_price_md5 = curr_md5 logger.info(f"正在分析价格详情页第 {price_page_count + 1} 页: {price_screen_path}") page_prices = await self.read_image_kit.analyze_detail_price(price_screen_path) if page_prices: # 简单去重：根据时段合并 for p in page_prices: if p not in all_prices: all_prices.append(p) # 向上滑动一点点，继续抓取 d.swipe_ext("up", scale=0.6) await asyncio.sleep(1.5) price_page_count += 1 # 清理临时截图 if os.path.exists(price_screen_path): os.remove(price_screen_path) # 5. 保存数据 if all_prices: station_name_clean = clean_station_name(station_name) # 对价格按时间排序 try: all_prices.sort(key=lambda x: x.get('start', '00:00')) except: pass logger.info(f"✅ 场站 {station_name_clean} 共提取到 {len(all_prices)} 条价格信息，准备保存...") await self.service.save_station_data(station_name_clean, address, all_prices) else: logger.warning(f"❌ 未能提取到任何价格信息，请检查页面识别逻辑") # 清理临时截图 for p in [first_screen_path, price_screen_path]: if os.path.exists(p): os.remove(p) async def crawl_list(self): """ 实现 BaseCrawler 的抽象方法 """ d = u2.connect() await self.crawl_list_logic(d) async def crawl_detail(self, station_info): """ 实现 BaseCrawler 的抽象方法 """ # 逻辑已在 crawl_list_logic 中通过 crawl_detail_logic 调用 pass async def main(service=None): crawler = TeLaiDianCrawler(service=service) await crawler.start() if __name__ == "__main__": asyncio.run(main())