aiData/Apps/TeLaiDian/Crawler.py

# coding=utf-8
import asyncio
import os
import sys
import time
import json
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 clear_ads(self, d, max_rounds=3):
        """
        清理页面上的广告弹窗
        """
        logger.info(f"开始清理广告弹窗，最多尝试 {max_rounds} 轮...")
        for i in range(max_rounds):
            ad_screen = take_screenshot(d, f"tld_ad_check_{int(time.time())}.jpg")
            res = await self.read_image_kit.find_close_button_vlm(ad_screen)
            
            if res.get("has_ad") and res.get("close_point"):
                close_point = res.get("close_point")
                w, h = d.window_size()
                target_x = int(close_point[0] * w / 1000)
                target_y = int(close_point[1] * h / 1000)
                
                # 安全校验：绝对不能点击微信小程序的胶囊按钮区 (右上角)
                # 胶囊按钮通常在 x > 80% 宽度 且 y < 100 像素(或 8% 高度) 的区域
                if target_x > w * 0.75 and target_y < 150:
                    logger.warning(f"⚠️ 拒绝点击疑似微信胶囊按钮的区域: ({target_x}, {target_y})")
                    continue

                logger.info(f"第 {i+1} 轮发现广告: {res.get('reason')}，点击关闭: ({target_x}, {target_y})")
                d.click(target_x, target_y)
                await asyncio.sleep(1.5)
            else:
                logger.info(f"第 {i+1} 轮未发现明显广告，清理结束。")
                if os.path.exists(ad_screen): os.remove(ad_screen)
                break
            
            if os.path.exists(ad_screen): os.remove(ad_screen)

    async def crawl_list_logic(self, d):
        # [临时禁用] 广告清理逻辑误触严重，先解决主流程问题
        # await self.clear_ads(d)
        
        # [优化] 向下滚动以刷新/校准地理位置
        # 使用更加显式的 swipe 方式：从屏幕 30% 划到 80%
        w, h = d.window_size()
        logger.info(f"执行显式下拉刷新操作: (x={w//2}, y1={int(h*0.3)} -> y2={int(h*0.8)})")
        d.swipe(w // 2, int(h * 0.3), w // 2, int(h * 0.8), duration=0.5)
        
        logger.info("等待 2.5 秒确保位置校准和列表刷新完成...")
        await asyncio.sleep(2.5)
        
        processed_count = 0
        last_md5 = None
        processed_station_names = set() # 用于列表级去重
        
        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:
                # 检查是否意外退出了小程序
                is_wrong_page = await self.check_wrong_page(d, screenshot_path)
                if is_wrong_page:
                    logger.error("检测到已退出详情列表页（可能回到了搜索页），尝试重新进入...")
                    await self.open_app()
                    await asyncio.sleep(5)
                    continue

                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
                
                if name in processed_station_names:
                    logger.info(f"跳过已处理场站: {name}")
                    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}")

                # 简单验证：如果标题包含 "我的卡券"、"优惠券"、"新人福利" 等，说明点错了
                is_wrong_page = await self.check_wrong_page(d, detail_check_path)
                if is_wrong_page:
                    logger.warning(f"检测到进入了错误页面，尝试返回列表...")
                    # 尝试点击左上角的返回箭头，如果没有，则执行系统 back
                    d.click(40, 70) # 特来电通常左上角有返回箭头
                    await asyncio.sleep(1.0)
                    d.press("back")
                    await asyncio.sleep(WAIT_BACK_TO_LIST)
                    if os.path.exists(detail_check_path): os.remove(detail_check_path)
                    continue
                
                # 爬取详情
                await self.crawl_detail_logic(d, station)
                processed_station_names.add(name)
                
                # 返回列表
                d.press("back")
                await asyncio.sleep(WAIT_BACK_TO_LIST)
                # 返回列表后也检查一下是否有新广告弹出
                await self.clear_ads(d, max_rounds=1)
                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 check_wrong_page(self, d, image_path):
        """
        检查是否误触进入了错误的页面（如：我的卡券、活动页等）
        """
        data = await self.read_image_kit.check_wrong_page_vlm(image_path)
        is_detail = data.get("is_detail_page", True)
        if not is_detail:
            logger.warning(f"⚠️ 确认进入错误页面: {data.get('page_type')} ({data.get('reason')})")
        return not is_detail

    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. 小步快跑寻找价格入口 (结合 CV 和 VLM)
        found_entrance = False
        entrance_point = None
        max_search_steps = 4
        
        logger.info(f"开始“小步快跑”策略寻找价格入口，最多尝试 {max_search_steps} 次小幅度滑动...")
        
        for step in range(max_search_steps):
            current_screen = take_screenshot(d, f"tld_search_price_step_{step}.jpg")
            logger.info(f"--- 寻找入口 第 {step+1} 步 ---")
            
            # 优先使用 CV 快速识别橘红色价格 P0
            logger.info("尝试 CV 识别价格卡片 (P0)...")
            cv_point = detect_price_click_point_cv(current_screen)
            
            if cv_point:
                logger.info(f"✅ CV 在第 {step+1} 步成功定位入口: {cv_point}")
                entrance_point = cv_point
                found_entrance = True
            else:
                # CV 没找到，使用 VLM 进行深度语义检查
                logger.info("CV 未找到，启动 VLM 深度语义识别...")
                vlm_res = await self.read_image_kit.find_price_entrance_vlm(current_screen)
                if vlm_res.get("found"):
                    norm_point = vlm_res.get("point") # [x, y] in 0-1000
                    if norm_point and len(norm_point) == 2:
                        w, h = d.window_size()
                        entrance_point = [int(norm_point[0] * w / 1000), int(norm_point[1] * h / 1000)]
                        logger.info(f"✅ VLM 在第 {step+1} 步成功定位入口: {entrance_point} ({vlm_res.get('reason')})")
                        found_entrance = True
                else:
                    logger.info(f"第 {step+1} 步未发现入口: {vlm_res.get('reason', '未知原因')}")

            # 如果找到入口，进行标注并点击
            if found_entrance and entrance_point:
                debug_flag_path = current_screen.replace(".jpg", "_entrance_found.jpg")
                img_debug = read_image(current_screen)
                if img_debug is not None:
                    cv2.circle(img_debug, (entrance_point[0], entrance_point[1]), 25, (0, 255, 0), 5) # 绿色大圆圈
                    save_image(debug_flag_path, img_debug)
                    logger.info(f"入口位置标注图已保存: {debug_flag_path}")
                
                logger.info(f"正在点击价格入口: {entrance_point}")
                d.click(entrance_point[0], entrance_point[1])
                await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
                
                # 清理临时截图
                if os.path.exists(current_screen): os.remove(current_screen)
                break
            
            # 没找到，小步向上滚动
            if step < max_search_steps - 1:
                scroll_scale = 0.35
                logger.info(f"未发现入口，执行小幅度向上滑动 (scale={scroll_scale})...")
                d.swipe_ext("up", scale=scroll_scale)
                await asyncio.sleep(1.2)
            
            # 清理临时截图
            if os.path.exists(current_screen): os.remove(current_screen)

        if not found_entrance:
            logger.warning("“小步快跑”策略未能找到价格入口，尝试坐标兜底...")
            w, h = d.window_size()
            d.click(w // 2, int(h * 0.45))
            await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)

        # 3. 进入分时电价页面后的处理
        try:
            # 1. 延长等待时间，等待小程序自动定位到当前时段的滚动完成
            logger.info("已点击进入价格详情，等待小程序自动滚动定位完成 (4秒)...")
            await asyncio.sleep(4.0) 
            
            # 2. 回到 00:00 原点：要看到上面的内容，需要“向下拉动”页面（即向上滚动列表）
            logger.info("执行向下拉动，尝试回到 00:00 时段顶部...")
            for i in range(3):
                # swipe_ext("down") 是手指从上往下划，动作是“向下”，结果是页面“向上”滚动
                d.swipe_ext("down", scale=0.8)
                await asyncio.sleep(0.5)
        except Exception as e:
            logger.error(f"处理分时电价页面初始状态失败: {e}")

        # 4. 循环滑动抓取完整分时电价
        all_prices = []
        last_price_md5 = None
        price_page_count = 0
        max_price_pages = 4 # 增加到4页确保覆盖 00:00-24:00
        screenshot_tasks = [] # 用于异步分析图片的任务列表
        temp_screenshots = [] # 记录临时文件以便后续清理

        logger.info("开始循环截图（UI操作优先，后台并行分析）...")
        
        try:
            while price_page_count < max_price_pages:
                price_screen_path = take_screenshot(d, f"tld_detail_price_{int(time.time())}_{price_page_count}.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
                temp_screenshots.append(price_screen_path)

                logger.info(f"已截取价格详情页第 {price_page_count + 1} 页: {price_screen_path}，启动后台异步分析")
                # 使用 asyncio.create_task 立即在后台开始执行分析
                task = asyncio.create_task(self.read_image_kit.analyze_detail_price(price_screen_path))
                screenshot_tasks.append(task)
                
                # 向上滚动列表（手指向上划），看后面的时段
                logger.info("向上滚动列表，准备截取下一屏价格...")
                d.swipe_ext("up", scale=0.8)
                await asyncio.sleep(1.0)
                price_page_count += 1

            # 等待所有后台分析任务完成
            if screenshot_tasks:
                logger.info(f"UI 操作已完成，等待 {len(screenshot_tasks)} 个后台分析任务结束...")
                results = await asyncio.gather(*screenshot_tasks, return_exceptions=True)
                
                for res in results:
                    if isinstance(res, Exception):
                        logger.error(f"后台分析任务出错: {res}")
                        continue
                    if res:
                        # 深度去重：根据时段 (start, end) 合并
                        for p in res:
                            is_duplicate = False
                            for existing in all_prices:
                                if p.get('start') == existing.get('start') and p.get('end') == existing.get('end'):
                                    is_duplicate = True
                                    # 字段补全逻辑
                                    for key in ['price', 'plus_price', 'market_price', 'elec_price', 'service_price']:
                                        if p.get(key) is not None and (existing.get(key) is None or existing.get(key) == 0):
                                            existing[key] = p[key]
                                    break
                            if not is_duplicate:
                                all_prices.append(p)
        except Exception as e:
            logger.error(f"抓取价格详情过程中发生异常: {e}")
        finally:
            # 无论是否异常，都要确保清理未完成的任务，避免 "never awaited" 警告
            for task in screenshot_tasks:
                if not task.done():
                    task.cancel()
            
            # 清理所有临时截图
            for path in temp_screenshots:
                if os.path.exists(path):
                    try:
                        os.remove(path)
                    except:
                        pass

        # 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())