# coding=utf-8 import asyncio import logging import uuid import os import sys import json import time import hashlib from PIL import Image # 将项目根目录添加到 sys.path 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) import uiautomator2 as u2 from Core.BaseCrawler import BaseCrawler from Apps.XinDianTu import Kit from Apps.XinDianTu.Kit import take_screenshot from Util.ObsUtil import ObsUploader from Util.RedisKit import RedisKit import cv2 from Apps.XinDianTu.Service import XinDianTuService from Apps.XinDianTu.ReadImageKit import ReadImageKit from Config.Config import ( OBS_TMP_PREFIX, CDN_DOMAIN, TEMP_IMAGE_DIR ) from Apps.XinDianTu.Config.Setting import ( SCROLL_DISTANCE_RATIO, MAX_SCROLLS, REDIS_STATION_EXPIRE, WAIT_DETAIL_PAGE_LOAD, WAIT_BACK_TO_LIST, WAIT_AFTER_SCROLL, MAX_CRAWL_DISTANCE, SAFE_EXCLUDE_RATIO, BOTTOM_SAFE_EXCLUDE_RATIO ) # --- 用户配置区域 --- # 是否保留截图文件（True=保留备查，False=随用随删） KEEP_SCREENSHOTS = True # [Testing] 是否在启动时清除 Redis 中的场站处理记录 # True: 每次运行前清除记录，方便反复测试同一个场站 # False: 生产模式，保留记录以避免重复爬取 TEST_CLEAR_REDIS = True # 配置说明： # SCROLL_DISTANCE_RATIO 控制翻页时的滑动距离（在 Config.py 中修改）。 # 对于华为 Mate 20x 等分辨率较低或屏幕较小的手机，如果发现翻页时跳过了某些场站， # 请尝试减小 SCROLL_DISTANCE_RATIO（例如设置为 0.4 或 0.3）。 # 这样每次滑动的距离变短，可以确保所有场站都能被完整显示并识别。 # 配置日志输出 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger("StationList") # 强制设置所有相关模块的日志级别为 INFO，防止被第三方库干扰 logging.getLogger("OpenXinDianTu").setLevel(logging.INFO) logging.getLogger("FullProcess").setLevel(logging.INFO) class XinDianTuCrawler(BaseCrawler): """ 新电途小程序爬虫实现 """ def __init__(self, service=None): super().__init__(service) # 初始化配置参数 self.max_scrolls = MAX_SCROLLS self.uploader = ObsUploader() self.redis_kit = RedisKit() async def start(self): """ 实现 BaseCrawler 的启动入口 """ # 兼容旧逻辑，直接调用 main await main(self.service) async def open_app(self): # 实际逻辑在 Opener.py，此处可作为封装层 pass async def crawl_list(self): # 实际逻辑在 get_station_list，此处可作为封装层 pass async def crawl_detail(self, station_info): pass async def is_already_crawled(redis_kit, station_name): """ 检查场站是否已经爬取过，仅按名称去重 """ cleaned_name = Kit.clean_station_name(station_name) if not cleaned_name: return False redis_key = f"crawled:xdt:{cleaned_name}" if await redis_kit.get_data(redis_key): return True return False async def analyze_prices_background(service, station_name, image_paths, device_info=None): """ 后台异步处理价格图片分析 """ try: # 使用多图解析并去重 all_rows = await ReadImageKit.parse_price_schedule_multi(station_name, image_paths, device_info=device_info) if all_rows: # 24小时铺平 hourly_schedule = ReadImageKit.hourly_full_day(all_rows) await service.process_price_detail_data(station_name, hourly_schedule) logger.info(f"后台处理完成: {station_name} 的分时价格已保存。") else: logger.warning(f"后台处理失败: {station_name} 未能解析出价格时段。") # 处理完成后清理图片 for path in image_paths: try: if os.path.exists(path): os.remove(path) except: pass except Exception as e: logger.error(f"分析价格后台任务异常 ({station_name}): {e}") return True async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS): """ 获取场站列表并处理翻页 (异步优化版) """ redis_kit = RedisKit() window_size = d.window_size() w, h = window_size[0], window_size[1] device_info = d.info device_info['width'] = w device_info['height'] = h logger.info(f"开始爬取列表，设备: {device_info.get('productName')} | 分辨率: {w}x{h}") background_tasks = [] last_list_md5 = None no_new_data_count = 0 for i in range(max_scrolls + 1): logger.info(f"正在处理第 {i + 1} 页...") # 1. 拍摄截图 image_uuid = str(uuid.uuid4()) screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR) # 检查是否存在广告 (VL) logger.info("检查是否存在广告弹窗...") ad_res = await ReadImageKit.detect_ad_popup(screenshot_path, device_info=device_info) ad_top_y_norm = 0.78 # 恢复默认的点击边界 (0.78)，不再需要为了避开兔子而设为 0.90 if ad_res: ad_type = ad_res.get("ad_type") if ad_type == "rabbit": logger.info(">>> 检测到兔子广告。尝试使用图片识别关闭...") try: # 使用基于图片的关闭方案 template_path = os.path.join(os.path.dirname(__file__), "Template", "X.jpg") if d.image.click(template_path, timeout=3.0): logger.info(">>> 成功通过图片识别关闭兔子广告。") 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) else: logger.warning(">>> 图片识别未找到兔子广告关闭按钮。") except Exception as ex: logger.error(f">>> 图片识别关闭兔子广告异常: {ex}") else: x, y = ad_res['x'], ad_res['y'] logger.info(f"检测到非兔子广告: {ad_res} [Type: {ad_type}]，正在关闭...") 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, top_ratio=SAFE_EXCLUDE_RATIO, bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO ) if last_list_md5 and current_md5 == last_list_md5: logger.info("检测到列表核心内容无变化，已到达底部或加载完成，提前结束。") if os.path.exists(screenshot_path): os.remove(screenshot_path) break last_list_md5 = current_md5 logger.info(f"列表页截图已完成: {screenshot_path}") # 2. 执行本地图形学分析，生成 _vl.jpg 和 .json logger.info("正在执行图形学切片分析...") Kit.crop_cards_from_image(screenshot_path) json_path = screenshot_path.replace(".jpg", ".json") vl_img_path = screenshot_path.replace(".jpg", "_vl.jpg") if not os.path.exists(json_path) or not os.path.exists(vl_img_path): logger.warning("未识别到任何卡片，跳过当前页") continue with open(json_path, 'r', encoding='utf-8') as f: json_metadata = json.load(f) # 3. 【优化】后台异步上传 _vl.jpg 供以后参考，不再等待上传 vl_object_key = f"{OBS_TMP_PREFIX}/{image_uuid}_vl.jpg" loop = asyncio.get_running_loop() loop.run_in_executor(None, uploader.upload_file, vl_object_key, vl_img_path) # 4. 【优化】直接使用本地路径调用 VL 模型识别，避免等待上传 logger.info("正在调用 VL 模型识别场站信息...") stations = await service.process_station_list_vl(vl_img_path, json_metadata, device_info=device_info) logger.info(f"本页识别到 {len(stations)} 个场站") # 5. 匹配几何卡片与 VL 识别结果 (XinDianTu 的 parse_vl_image 已经按顺序返回了) # 这里的 stations 列表长度应该与 json_metadata["cards"] 对应 new_stations_processed = 0 if json_metadata.get("cards") and stations: for idx, card in enumerate(json_metadata["cards"]): # 检查索引是否越界 (VL 模型可能返回的数组长度不一致) if idx < len(stations) and stations[idx]: st = stations[idx] station_name = st.get("station_name") if not station_name: continue # 检查 Redis 去重 if await is_already_crawled(redis_kit, station_name): logger.info(f"场站 {station_name} 匹配到已处理记录，跳过。") continue # 【优化】检查是否被遮挡或太靠近底部 # card["bounds_norm"] 是 {left, top, right, bottom} 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 # 正常处理新场站 logger.info(f">>> 发现新场站 '{station_name}'，开始处理...") new_stations_processed += 1 click_x, click_y = card["click_point"] logger.info(f"准备处理场站: {station_name}, 点击坐标: ({click_x}, {click_y})") d.click(int(click_x), int(click_y)) # 等待二级页面加载 await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) should_back_to_list = True # 截取二级页面图 detail_uuid = str(uuid.uuid4()) detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR) # 【新增】二级页面广告检测 (如：免费停车提示) logger.info("检查详情页是否存在干扰弹窗...") detail_ad_res = await ReadImageKit.detect_ad_popup(detail_path, device_info=device_info) if detail_ad_res: dad_type = detail_ad_res.get("ad_type") if dad_type == "rabbit": logger.info(">>> 详情页检测到 rabbit 广告。尝试使用图片识别关闭...") try: template_path = os.path.join(os.path.dirname(__file__), "Template", "X.jpg") if d.image.click(template_path, timeout=3.0): logger.info(">>> 详情页成功通过图片识别关闭兔子广告。") await asyncio.sleep(1.5) # 重新截图 if os.path.exists(detail_path): os.remove(detail_path) detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR) else: logger.warning(">>> 详情页图片识别未找到兔子广告关闭按钮，跳过。") except Exception as ex: logger.error(f">>> 详情页图片识别关闭兔子广告异常: {ex}") else: dx, dy = detail_ad_res["x"], detail_ad_res["y"] logger.info(f"检测到详情页弹窗: {dad_type}，正在点击关闭 ({dx}, {dy})...") d.click(dx, dy) await asyncio.sleep(1.5) # 重新截图，确保后续流程使用的是清理后的界面 if os.path.exists(detail_path): os.remove(detail_path) detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR) # 【优化】后台解析详情页地址，直接传本地路径，避免等待上传 logger.info(f"已启动后台分析详情页: {station_name}") task_addr = asyncio.create_task(service.process_station_address(station_name, detail_path, device_info=device_info)) background_tasks.append(task_addr) # 【优化】后台异步上传详情页截图供以后参考 detail_object_key = f"{OBS_TMP_PREFIX}/{detail_uuid}.jpg" loop = asyncio.get_running_loop() loop.run_in_executor(None, uploader.upload_file, detail_object_key, detail_path) # --- 抓取价格时段信息 --- # 【优化】使用模板匹配点击 "全部时段" 按钮，节省 VL Token 且更稳定 template_qbsd = os.path.join(os.path.dirname(__file__), "Template", "qbsd.jpg") logger.info(f"尝试通过图片识别点击 '全部时段' 按钮: {template_qbsd}") # 记录点击前的页面特征，用于验证是否成功进入三级页面 before_click_md5 = Kit.get_image_content_md5(detail_path) entered_price_page = False try: if d.image.click(template_qbsd, timeout=5.0): logger.info("成功通过图片识别点击 '全部时段' 按钮") await asyncio.sleep(1.5) # 检查页面是否真的变化了 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 = [] for p_idx in range(1, 4): p_uuid = f"{hashlib.md5(station_name.encode('utf-8')).hexdigest()}_p_{p_idx}" p_path = take_screenshot(d, p_uuid, save_dir=TEMP_IMAGE_DIR) price_screenshots.append(p_path) if p_idx < 3: d.swipe_ext("up", scale=0.7) await asyncio.sleep(1.0) # 【异步优化】后台处理价格 logger.info(f"已启动后台分析价格页: {station_name}") task_price = asyncio.create_task(analyze_prices_background(service, station_name, price_screenshots, device_info=device_info)) 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) # 1. 与列表页对比 (使用列表页的排除比例) check_list_md5 = Kit.get_image_content_md5( check_back_path, top_ratio=SAFE_EXCLUDE_RATIO, bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO ) # 2. 与详情页对比 (使用默认排除比例，与 before_click_md5 一致) check_detail_md5 = Kit.get_image_content_md5(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("检测到返回详情页，需执行第二次返回。") 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]})。") logger.info("使用图片识别进一步确认当前页面状态...") # 使用图片识别检查当前页面是否存在 '全部时段' 按钮 if d.image.match(template_qbsd): logger.info("图片识别确认当前仍处于详情页 (发现了'全部时段'按钮)，执行返回。") should_back_to_list = True else: logger.info("图片识别未发现'全部时段'按钮，判定已回到列表页(或非详情页)，停止回退以防误退。") should_back_to_list = False # 使用完毕后再清理检查用的截图 if os.path.exists(check_back_path): os.remove(check_back_path) else: logger.warning(f"点击 '{station_name}' 的 '全部时段' 按钮后页面无明显变化，跳过三级页面处理。") # 清理检查用的截图 if os.path.exists(after_click_path): os.remove(after_click_path) else: logger.info(f"场站 {station_name} 通过图片识别未发现 '全部时段' 按钮。") except Exception as e: logger.error(f"通过图片识别点击 '全部时段' 按钮异常: {e}") # 从二级页面返回 (仅当确实需要返回时) if should_back_to_list: d.press("back") logger.info(f"等待 {WAIT_BACK_TO_LIST + 1} 秒返回列表...") await asyncio.sleep(WAIT_BACK_TO_LIST + 1) # 记录 Redis 去重 (仅按名称去重) cleaned = Kit.clean_station_name(station_name) await redis_kit.set_data(f"crawled:xdt:{cleaned}", "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) if stations and new_stations_processed == 0: no_new_data_count += 1 logger.info(f"本页所有场站均已处理过，连续 {no_new_data_count} 页无新数据。") # 【优化】由于滑动步长已调大 (0.5)，连续多页重复的可能性降低 if no_new_data_count >= 5: logger.info("连续 5 页无新数据，判定为已到底或重复循环，提前结束。") break else: no_new_data_count = 0 # 6. 翻页 # 【优化】使用“小步快跑”策略，减小滑动距离以避免跳过场站，并能更平滑地躲避广告 # 步长已在 Setting.py 中统一配置为 SCROLL_DISTANCE_RATIO logger.info(f"执行翻页滑动 (步长: {SCROLL_DISTANCE_RATIO})...") d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO) await asyncio.sleep(WAIT_AFTER_SCROLL) # 等待所有后台任务完成 if background_tasks: logger.info(f"正在等待剩余 {len(background_tasks)} 个后台任务完成...") await asyncio.gather(*background_tasks, return_exceptions=True) logger.info("采集任务完成") return True def clean_images_dir(): """清理 Images 目录下除保留文件外的所有文件""" base_dir = os.path.dirname(os.path.abspath(__file__)) images_dir = os.path.join(base_dir, "Images") if not os.path.exists(images_dir): return keep_files = {'1.jpg', '2.jpg', '3.jpg', '4.jpg'} logger.info(f"正在清理 Images 目录 (保留: {keep_files})...") deleted_count = 0 for filename in os.listdir(images_dir): if filename in keep_files: continue file_path = os.path.join(images_dir, filename) try: if os.path.isfile(file_path): os.remove(file_path) deleted_count += 1 except Exception as e: logger.warning(f"删除文件失败 {filename}: {e}") logger.info(f"清理完成，共删除 {deleted_count} 个文件。") async def clean_redis_data(redis_kit): """清除 Redis 中的场站处理记录""" if TEST_CLEAR_REDIS: logger.info("测试模式开启：正在清除 Redis 中的场站处理记录 (processed_station:*)...") keys = await redis_kit.keys("processed_station:*") if keys: count = await redis_kit.delete(*keys) logger.info(f"已清除 {count} 条场站处理记录") else: logger.info("未发现旧的场站处理记录") async def main(service=None, do_cleanup=True): # 连接设备 d = u2.connect() # 初始化服务 should_close_service = False if service is None: service = XinDianTuService() should_close_service = True # 初始化数据库连接（XinDianTuService 需要） await service.init_db() uploader = ObsUploader() redis_kit = RedisKit() try: if do_cleanup: # 清理图片目录 clean_images_dir() # [Testing] 如果配置为测试模式，启动时清除 Redis 记录 await clean_redis_data(redis_kit) # 清理过期数据 # await service.cleanup_old_data() # 获取场站列表 stations = await get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS) if stations: logger.info("场站列表采集完成。") else: logger.warning("未采集到任何场站信息。") return True except Exception as e: logger.exception(f"运行过程中出现异常: {e}") return False finally: # 关闭数据库连接 if should_close_service: await service.close_db() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logger.info("程序被用户中断.") except Exception as e: logger.exception(f"程序崩溃: {e}")