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, FIRST_RUN_ONLY_ONE_STATION
)
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")

PRICE_TAB_X_NORM = 220
PRICE_TAB_Y_NORM = 130
PRICE_ENTRY_X_NORM = 230
PRICE_ENTRY_Y_NORM = 380

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):
            round_idx = i + 1
            logger.info(f"[广告清理] 第 {round_idx} 轮：正在截屏分析...")
            ad_screen = take_screenshot(d, f"tld_ad_check_r{round_idx}_{int(time.time())}.jpg")
            
            try:
                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")
                    reason = res.get("reason", "未提供原因")
                    w, h = d.window_size()
                    target_x = int(close_point[0] * w / 1000)
                    target_y = int(close_point[1] * h / 1000)
                    
                    logger.info(f"[广告清理] 第 {round_idx} 轮：VLM 发现广告！原因: {reason}")
                    logger.info(f"[广告清理] 计划点击坐标: ({target_x}, {target_y})，归一化坐标: {close_point}")
                    
                    # 安全校验：绝对不能点击微信小程序的胶囊按钮区 (右上角)
                    if target_x > w * 0.75 and target_y < 150:
                        logger.warning(f"[广告清理] ⚠️ 拒绝点击疑似微信胶囊按钮的区域: ({target_x}, {target_y})，跳过本轮。")
                        continue

                    logger.info(f"[广告清理] 正在执行点击关闭操作...")
                    d.click(target_x, target_y)
                    # 点击后等待一下，让弹窗消失或下一轮广告弹出
                    await asyncio.sleep(2.0)
                else:
                    logger.info(f"[广告清理] 第 {round_idx} 轮：未发现广告弹窗。VLM 理由: {res.get('reason', '无')}")
                    if os.path.exists(ad_screen): os.remove(ad_screen)
                    break
            except Exception as e:
                logger.error(f"[广告清理] 第 {round_idx} 轮检测发生异常: {e}")
            finally:
                if os.path.exists(ad_screen): os.remove(ad_screen)
        
        logger.info("--- [广告清理] 任务结束 ---")

    async def crawl_list_logic(self, d):
        # 1. 启动即清理广告 (已根据要求关闭)
        # await self.clear_ads(d, max_rounds=3)
        
        # [优化] 向下滚动以刷新/校准地理位置
        # 使用更加显式的 swipe 方式：从屏幕 30% 划到 80%
        popup_screen_path = take_screenshot(d, f"tld_detail_popup_{int(time.time())}.jpg")
        logger.info(f"[详情页] 截图用于检测温馨提示弹窗: {popup_screen_path}")

        template_xczs = os.path.join(os.path.dirname(__file__), "Template", "xczs.jpg")
        template_point = None
        if os.path.exists(template_xczs):
            try:
                match_res = d.image.match(template_xczs)
                if match_res:
                    if hasattr(match_res, "point") and match_res.point:
                        template_point = match_res.point
                    elif isinstance(match_res, dict):
                        if match_res.get("point"):
                            template_point = match_res["point"]
                        elif "x" in match_res and "y" in match_res:
                            template_point = (match_res["x"], match_res["y"])
                    elif isinstance(match_res, (list, tuple)) and len(match_res) >= 2:
                        template_point = (match_res[0], match_res[1])
                if template_point:
                    logger.info(f"[详情页] 模板检测到疑似温馨提示弹窗位置: {template_point}")
                else:
                    logger.info("[详情页] 模板未能确定温馨提示弹窗位置")
            except Exception as e:
                logger.error(f"[详情页] 模板检测温馨提示弹窗失败: {e}")
        else:
            logger.info(f"[详情页] 温馨提示模板文件不存在: {template_xczs}")

        vlm_popup = {"has_popup": False}
        try:
            vlm_popup = await self.read_image_kit.check_warm_popup_vlm(popup_screen_path)
        except Exception as e:
            logger.error(f"[详情页] VLM 检测温馨提示弹窗失败: {e}")

        has_vlm_popup = isinstance(vlm_popup, dict) and vlm_popup.get("has_popup")
        if template_point and has_vlm_popup:
            click_x = int(template_point[0])
            click_y = int(template_point[1])
            logger.info(f"[详情页] 模板与 VLM 均确认存在温馨提示弹窗，即将点击关闭按钮: ({click_x}, {click_y})")
            debug_popup_path = popup_screen_path.replace(".jpg", f"_xczs_click_{click_x}_{click_y}.jpg")
            try:
                img_popup = read_image(popup_screen_path)
                if img_popup is not None:
                    cv2.circle(img_popup, (click_x, click_y), 20, (0, 0, 255), -1)
                    save_image(debug_popup_path, img_popup)
                    logger.info(f"[详情页] 已生成温馨提示弹窗点击诊断图片: {debug_popup_path}")
            except Exception as e:
                logger.error(f"[详情页] 生成温馨提示弹窗诊断图片失败: {e}")
            try:
                d.click(click_x, click_y)
                await asyncio.sleep(1.5)
            except Exception as e:
                logger.error(f"[详情页] 点击温馨提示“下次再说”失败: {e}")
        else:
            logger.info(f"[详情页] 温馨提示弹窗未通过双重确认，模板检测: {bool(template_point)} | VLM 检测: {vlm_popup}")

        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(f"等待 {WAIT_AFTER_SCROLL} 秒确保位置校准和列表刷新完成...")
        await asyncio.sleep(WAIT_AFTER_SCROLL)
        
        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)
                d.press("back")
                await asyncio.sleep(WAIT_BACK_TO_LIST)
                processed_count += 1
                if FIRST_RUN_ONLY_ONE_STATION:
                    logger.info("已完成首个场站的全流程采集，根据配置退出爬取任务。")
                    if os.path.exists(screenshot_path):
                        os.remove(screenshot_path)
                    return
            
            # 滑动到下一页
            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):
        """
        在详情页提取价格和状态信息
        """
        first_screen_path = take_screenshot(d, f"tld_detail_basic_{int(time.time())}.jpg")
        station_name = station_info.get("name")
        address = station_info.get("address")
        logger.info(f"[详情页] 进入 crawl_detail_logic，场站: {station_name} | 地址: {address}")
        logger.info(f"[详情页] 已截取首屏截图，准备识别基础信息: {first_screen_path}")

        try:
            basic_info = await self.read_image_kit.analyze_detail_basic_info(first_screen_path)
            if isinstance(basic_info, dict):
                name2 = basic_info.get("name")
                addr2 = basic_info.get("address")
                if name2:
                    station_name = name2
                if addr2:
                    address = addr2
            logger.info(f"[详情页] 同步基础信息识别结果用于写库: {station_name} | {address}")
        except Exception as ex:
            logger.error(f"[详情页] 同步分析详情页基础信息失败: {ex}")
        finally:
            if os.path.exists(first_screen_path):
                try:
                    os.remove(first_screen_path)
                except:
                    pass

        w, h = d.window_size()

        logger.info("[详情页] 根据用户要求：进入页面后多等会，然后向上滑动一整屏，通过 VLM 查找价格入口")
        w, h = d.window_size()
        
        # 1. 增加等待时间，确保页面加载完成
        logger.info(f"[详情页] 等待 {WAIT_DETAIL_PAGE_LOAD + 2}s 确保页面稳定 (根据要求多等会)...")
        await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD + 2)
        
        # 2. 向上滑动一部分屏 (从屏幕底部的 85% 划到顶部的 15%)
        # [优化] 避免滑过头，从全屏改为 0.7 屏幕左右
        logger.info("[详情页] 执行向上滑动操作 (y: 85% -> 15%)")
        d.swipe(w // 2, int(h * 0.85), w // 2, int(h * 0.15), duration=0.8)
        await asyncio.sleep(2.0) # 滑动后再次等待稳定

        # 3. 截图并识别
        final_screen_path = take_screenshot(d, f"tld_detail_vlm_input_{int(time.time())}.jpg")
        logger.info(f"[详情页] 已截取 VLM 识别用图: {final_screen_path}")

        logger.info("[详情页] 优先使用 VLM 识别顶部“价格”标签和价格入口")
        price_tab_screen = final_screen_path
        entrance_clicked = False

        try:
            # 1. 尝试使用 VLM 寻找顶部“价格”标签
            tab_data = await self.read_image_kit.find_price_tab_vlm(final_screen_path)
            tab_x, tab_y = None, None
            
            if tab_data.get("found") and tab_data.get("point"):
                p = tab_data["point"]
                tab_x = int(p[0] * w / 1000)
                tab_y = int(p[1] * h / 1000)
                logger.info(f"[详情页] VLM 成功找到价格标签: 归一化{p} -> 像素({tab_x}, {tab_y})")
            else:
                tab_x = int(PRICE_TAB_X_NORM * w / 1000)
                tab_y = int(PRICE_TAB_Y_NORM * h / 1000)
                logger.warning(f"[详情页] VLM 未找到价格标签，使用固定坐标兜底: 像素({tab_x}, {tab_y})")

            logger.info(f"[详情页] 正在点击价格标签...")
            d.click(tab_x, tab_y)
            await asyncio.sleep(2.0) # 点击标签后多等会
            
            price_tab_screen = take_screenshot(d, f"tld_detail_after_price_tab_{int(time.time())}.jpg")
            logger.info(f"[详情页] 点击价格标签后的界面截图已保存: {price_tab_screen}")

            # 2. 尝试使用 VLM 寻找价格入口 (重点识别“全部时段 >”)
            entry_data = await self.read_image_kit.find_price_entrance_vlm(price_tab_screen)
            entry_x, entry_y = None, None
            
            if entry_data.get("found") and entry_data.get("point"):
                p = entry_data["point"]
                # 安全校验：严禁点击 y > 850 (85%) 的区域，那是底部浮动条
                if p[1] > 850:
                    logger.warning(f"[详情页] VLM 虽返回坐标 {p}，但 Y 轴过大 (疑似底部浮动条)，强制判定为未找到。")
                    entry_x = int(PRICE_ENTRY_X_NORM * w / 1000)
                    entry_y = int(PRICE_ENTRY_Y_NORM * h / 1000)
                else:
                    entry_x = int(p[0] * w / 1000)
                    entry_y = int(p[1] * h / 1000)
                    logger.info(f"[详情页] VLM 成功找到价格入口 (匹配全部时段标识): 归一化{p} -> 像素({entry_x}, {entry_y})，原因: {entry_data.get('reason')}")
            else:
                entry_x = int(PRICE_ENTRY_X_NORM * w / 1000)
                entry_y = int(PRICE_ENTRY_Y_NORM * h / 1000)
                logger.warning(f"[详情页] VLM 未找到价格入口，使用固定坐标兜底: 像素({entry_x}, {entry_y})")

            click_x = max(5, min(w - 5, entry_x))
            click_y = max(5, min(h - 5, entry_y))

            debug_click_path = price_tab_screen.replace(
                ".jpg",
                f"_click_tab_{tab_x}_{tab_y}_entry_{click_x}_{click_y}.jpg"
            )
            try:
                img = read_image(price_tab_screen)
                if img is not None:
                    cv2.circle(img, (tab_x, tab_y), 20, (0, 0, 255), -1)
                    cv2.circle(img, (click_x, click_y), 20, (0, 0, 255), -1)
                    save_image(debug_click_path, img)
                    logger.info(f"[详情页] 已生成价格点击诊断红点图: {debug_click_path}")
            except Exception as e:
                logger.error(f"[详情页] 生成诊断图片失败: {e}")

            logger.info(f"[详情页] 正在点击电价入口...")
            d.click(click_x, click_y)
            entrance_clicked = True
            await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
        except Exception as e:
            logger.error(f"[详情页] 识别或点击价格入口失败: {e}")

        if entrance_clicked:
            entered_price_path = take_screenshot(d, f"tld_detail_price_after_enter_{int(time.time())}.jpg")
            logger.info(f"[电价页] 入口点击后的电价页截图已保存: {entered_price_path}")

            await asyncio.sleep(1.0)

            logger.info("[电价页] 先多次向下滚动，使列表回到起始位置")
            top_last_md5 = None
            top_stable = 0
            max_top_round = 8
            price_top_screen = entered_price_path

            for idx in range(max_top_round):
                price_top_screen = take_screenshot(d, f"tld_detail_price_top_{int(time.time())}_{idx}.jpg")
                logger.info(f"[电价页] 向下滚动前后的截图: {price_top_screen}")
                curr_md5 = get_image_content_md5(price_top_screen, top_ratio=0.2, bottom_ratio=0.2)
                if top_last_md5 is not None and curr_md5 == top_last_md5:
                    top_stable += 1
                    logger.info(f"[电价页] 页面内容连续第 {top_stable} 次无变化，可能已到顶部")
                    if top_stable >= 2:
                        break
                else:
                    top_stable = 0
                top_last_md5 = curr_md5

                d.swipe_ext("down", scale=0.8)
                await asyncio.sleep(1.0)

            all_prices = []
            last_price_md5 = None
            price_page_count = 0
            max_price_pages = 4
            screenshot_tasks = []
            temp_screenshots = []
            price_screen_path = price_top_screen

            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}，启动后台异步分析")
                    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:
                            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:
                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

            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"❌ 未能提取到任何价格信息，请检查页面识别逻辑")
                if address:
                    station_name_clean = clean_station_name(station_name)
                    logger.info(f"[详情页] 虽未获取价格，但已获取地址，尝试仅保存基础信息: {station_name_clean} | {address}")
                    try:
                        await self.service.save_station_profile_only(station_name_clean, address)
                    except Exception as e:
                        logger.error(f"[详情页] 仅保存基础信息失败: {e}")

            if os.path.exists(price_screen_path):
                os.remove(price_screen_path)

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