Files
aiData/Apps/TeLaiDian/Crawler.py
HuangHai ed37643ffe 'commit'
2026-01-14 14:26:03 +08:00

511 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# 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")
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
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}")
async def analyze_basic_info_background(image_path, fallback_name, fallback_address):
try:
basic_info = await self.read_image_kit.analyze_detail_basic_info(image_path)
name2 = basic_info.get("name") or fallback_name
addr2 = basic_info.get("address") or fallback_address
logger.info(f"[异步] 详情页基础信息识别完成: {name2} | {addr2}")
except Exception as ex:
logger.error(f"[异步] 分析详情页基础信息失败: {ex}")
finally:
if os.path.exists(image_path):
try:
os.remove(image_path)
except:
pass
asyncio.create_task(analyze_basic_info_background(first_screen_path, station_name, address))
w, h = d.window_size()
logger.info("[详情页] 先执行一次较大的向上滑动,将详情内容上移一屏左右")
d.swipe(w * 0.5, h * 0.8, w * 0.5, h * 0.3, 0.5)
await asyncio.sleep(1.0)
template_jgxx = os.path.join(os.path.dirname(__file__), "Template", "jgxx.jpg")
entrance_point = None
if os.path.exists(template_jgxx):
logger.info(f"[详情页] 使用模板匹配价格入口: {template_jgxx},第 1 轮(只匹配一次)")
try:
match_res = d.image.match(template_jgxx)
except Exception as e:
match_res = None
logger.error(f"[详情页] 模板匹配价格入口失败: {e}")
if match_res:
logger.info(f"[详情页] 第 1 轮模板匹配返回结果: {match_res}")
if hasattr(match_res, "point") and match_res.point:
entrance_point = match_res.point
elif isinstance(match_res, dict):
if "point" in match_res and match_res["point"]:
entrance_point = match_res["point"]
elif "x" in match_res and "y" in match_res:
entrance_point = (match_res["x"], match_res["y"])
elif isinstance(match_res, (list, tuple)) and len(match_res) >= 2:
entrance_point = (match_res[0], match_res[1])
if not entrance_point:
logger.error("[详情页] 未能通过 jgxx.jpg 找到价格入口,本次流程直接中止,需人工检查详情页布局或滚动距离。")
raise RuntimeError("TeLaiDian: 价格入口模板匹配失败")
before_price_path = take_screenshot(d, f"tld_detail_price_before_{int(time.time())}.jpg")
before_md5 = get_image_content_md5(
before_price_path,
top_ratio=SAFE_EXCLUDE_RATIO,
bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO,
)
ex, ey = int(entrance_point[0]), int(entrance_point[1])
click_x = ex
offset_y = int(h * 0.1)
click_y = min(h - 10, ey + offset_y)
logger.info(f"[详情页] 通过 jgxx.jpg 成功找到价格入口,模板中心: ({ex}, {ey}),调整后点击坐标: ({click_x}, {click_y}),即将进入电价页")
d.click(click_x, click_y)
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
after_price_path = take_screenshot(d, f"tld_detail_price_after_{int(time.time())}.jpg")
after_md5 = get_image_content_md5(
after_price_path,
top_ratio=SAFE_EXCLUDE_RATIO,
bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO,
)
entered_price_page = False
if before_md5 and after_md5 and before_md5 != after_md5:
entered_price_page = True
logger.info("[电价页] 模板点击后页面内容发生变化,判定已进入电价详情页。")
else:
logger.warning("[电价页] 模板点击后页面内容无明显变化,尝试使用 VLM 兜底寻找价格入口。")
try:
vlm_res = await self.read_image_kit.find_price_entrance_vlm(before_price_path)
except Exception as e:
vlm_res = {}
logger.error(f"[电价页] VLM 寻找价格入口失败: {e}")
point = vlm_res.get("point") if isinstance(vlm_res, dict) else None
if vlm_res.get("found") and point and len(point) >= 2:
w, h = d.window_size()
vx = int(point[0] * w / 1000)
vy = int(point[1] * h / 1000)
if vy < int(h * SAFE_EXCLUDE_RATIO):
vy = int(h * SAFE_EXCLUDE_RATIO)
if vy > int(h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO)):
vy = int(h * (1 - BOTTOM_SAFE_EXCLUDE_RATIO))
logger.info(f"[电价页] 使用 VLM 兜底点击价格入口: ({vx}, {vy}),原因: {vlm_res.get('reason')}")
d.click(vx, vy)
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
vlm_after_path = take_screenshot(d, f"tld_detail_price_after_vlm_{int(time.time())}.jpg")
vlm_after_md5 = get_image_content_md5(
vlm_after_path,
top_ratio=SAFE_EXCLUDE_RATIO,
bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO,
)
if before_md5 and vlm_after_md5 and before_md5 != vlm_after_md5:
entered_price_page = True
logger.info("[电价页] VLM 兜底点击后页面内容发生变化,判定已进入电价详情页。")
else:
logger.error(f"[电价页] VLM 未能找到可靠的价格入口: {vlm_res}")
if not entered_price_page:
logger.error("[电价页] 多轮尝试后页面仍未变化,疑似点击未生效或入口不可用。")
raise RuntimeError("TeLaiDian: 价格入口点击后页面未变化")
logger.info("[电价页] 已成功进入电价详情页,调试模式下在此主动停止,供人工检查页面内容。")
raise RuntimeError("TeLaiDian: 调试模式已在电价页入口处停止")
# 4. 进入分时电价页面后的处理
try:
logger.info("[电价页] 已点击价格入口,等待小程序自动滚动定位完成 (4秒)...")
await asyncio.sleep(4.0)
logger.info("[电价页] 开始执行两次大幅向下拉动,目标是回到 00:00 时段顶部")
for idx in range(2):
logger.info(f"[电价页] 第 {idx+1} 次向下滑动(d.swipe_ext(\"down\", scale=0.9))")
d.swipe_ext("down", scale=0.9)
await asyncio.sleep(1.0)
except Exception as e:
logger.error(f"[电价页] 处理分时电价页面初始状态失败: {e}")
# 5. 循环滑动抓取完整分时电价
all_prices = []
last_price_md5 = None
price_page_count = 0
max_price_pages = 4
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},启动后台异步分析")
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
# 6. 保存数据
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 os.path.exists(price_screen_path):
os.remove(price_screen_path)
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
# 6. 保存数据
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 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())