aiData/Apps/TeLaiDian/Crawler.py

# coding=utf-8
import asyncio
import logging
import os
import sys
import time
from Apps.TeLaiDian.Kit import take_screenshot, get_image_content_md5, clean_station_name, setup_logger
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. 点击“价格信息”按钮 (jgxx.jpg)
        template_path = os.path.join(project_root, "Apps", "TeLaiDian", "Template", "jgxx.jpg")
        logger.info(f"尝试点击价格详情按钮: {template_path}")
        
        try:
            # 使用 uiautomator2 的图像识别点击
            match = d.image.match(template_path)
            if match:
                logger.info(f"找到价格按钮，坐标: {match['point']}")
                d.image.click(template_path)
                await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
            else:
                logger.warning("未找到价格按钮模板，尝试备选方案：直接点击屏幕下方区域")
                # 备选方案：如果模板匹配失败，尝试点击屏幕中下方
                w, h = d.window_size()
                d.click(w // 2, int(h * 0.8))
                await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
        except Exception as e:
            logger.error(f"点击价格按钮失败: {e}")

        # 4. 截图并分析价格表
        price_screen_path = take_screenshot(d, f"tld_detail_price_{int(time.time())}.jpg")
        prices = await self.read_image_kit.analyze_detail_price(price_screen_path)
        
        # 5. 保存数据
        if prices:
            station_name_clean = clean_station_name(station_name)
            logger.info(f"场站 {station_name_clean} 提取到 {len(prices)} 条价格信息，准备保存...")
            await self.service.save_station_data(station_name_clean, address, prices)
        else:
            logger.warning(f"未能从 {price_screen_path} 提取到价格信息")
        
        # 清理临时截图
        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())