import asyncio import logging import uuid import os 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.ReadImageKit import ReadImageKit from Apps.YeLiTe.Service import YiLaiTeService from Apps.YeLiTe.Config.Setting import ( SCROLL_DISTANCE_RATIO, WAIT_AFTER_SCROLL, MAX_STATIONS_COUNT, WAIT_DETAIL_PAGE_LOAD, WAIT_BACK_TO_LIST, TEST_CLEAR_REDIS, SAFE_EXCLUDE_RATIO, BOTTOM_SAFE_EXCLUDE_RATIO ) from Util.RedisKit import RedisKit 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) logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("YiLaiTeCrawler") class YiLaiTeCrawler(BaseCrawler): def __init__(self, service=None): super().__init__(service or YiLaiTeService()) self.read_image_kit = ReadImageKit() self.redis_kit = RedisKit() async def start(self): """ 实现 BaseCrawler 的启动入口 """ await main(self.service) async def open_app(self): """ 打开驿来特小程序 """ from Apps.YeLiTe import Opener return await Opener.open_mini_program() async def crawl_list(self): """ 开始爬取列表页 """ d = u2.connect() return await self.crawl_list_logic(d) async def crawl_detail(self, station_info): """ 爬取详情页 (BaseCrawler 要求，此处逻辑已集成在 crawl_list_logic 中) """ pass async def clean_redis_data(self): """ 清除测试用的 Redis 记录 """ if TEST_CLEAR_REDIS: logger.info("清理 Redis 中的场站处理记录...") pattern = "crawled:ylt:*" keys = await self.redis_kit.client.keys(pattern) if keys: await self.redis_kit.client.delete(*keys) async def crawl_list_logic(self, d): processed_count = 0 no_new_data_count = 0 last_md5 = None background_tasks = [] while processed_count < MAX_STATIONS_COUNT: # 1. 截图并分析 screenshot_path = take_screenshot(d, f"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: no_new_data_count += 1 if no_new_data_count >= 5: logger.info("连续 5 页无新数据，停止") break # 滑动 d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO) await asyncio.sleep(WAIT_AFTER_SCROLL) continue no_new_data_count = 0 new_stations_in_page = 0 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 # 去重 redis_key = f"crawled:ylt:{clean_station_name(name)}" if await self.redis_kit.get_data(redis_key): continue # 点击进入前截图，用于对比是否成功进入二级页 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]) await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD) # 分析详情页 (采用异步后台模式) 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 os.path.exists(before_click_path): os.remove(before_click_path) if before_md5 != after_md5: logger.info(f"成功进入详情页: {name}") # 启动后台任务处理详情页 task = asyncio.create_task(self.analyze_detail_background(name, detail_shot)) 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) else: logger.warning(f"点击场站 {name} 后页面似乎未跳转，跳过返回操作") if os.path.exists(detail_shot): os.remove(detail_shot) # 即使没进去，也记录一下，避免短时间内重复尝试 await self.redis_kit.set_data(redis_key, "1", expire=3600) if new_stations_in_page == 0 and stations: no_new_data_count += 1 if no_new_data_count >= 5: logger.info("连续 5 页均无新场站，停止") break else: no_new_data_count = 0 # 滑动翻页 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) return processed_count async def analyze_detail_background(self, station_name, image_path): """ 后台异步分析详情页 """ 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} 价格分析完成并入库") 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) async def main(service=None): if service is None: service = YiLaiTeService() await service.init_db() crawler = YiLaiTeCrawler(service) d = u2.connect() # 清理 Redis await crawler.clean_redis_data() # 记录任务日志 task_id = str(uuid.uuid4()) await service.log_task_start(task_id) total_count = 0 status = "success" error_msg = None try: total_count = await crawler.crawl_list_logic(d) except Exception as e: logger.error(f"主流程执行失败: {e}") status = "failed" error_msg = str(e) finally: await service.log_task_end(task_id, total_count, status, error_msg) # 如果是内部初始化的 service，则关闭 if service and not isinstance(service, YiLaiTeService): await service.close_db() async def get_image_md5_async(path): return get_image_md5(path) if __name__ == "__main__": asyncio.run(main())