Files
aiData/Apps/TeLaiDian/Crawler.py
HuangHai 3e9fc3737e 'commit'
2026-01-15 07:29:51 +08:00

500 lines
23 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")
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):
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%
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("[详情页] 根据用户策略: 多次大幅向上滑动,直到页面基本不再变化")
last_md5 = None
stable_count = 0
max_round = 30
final_screen_path = None
from Apps.TeLaiDian.Kit import get_image_content_md5
for idx in range(max_round):
start_x = int(w * 0.9)
start_y = int(h * 0.85)
end_y = int(h * 0.25)
logger.info(f"[详情页] 第 {idx + 1} 轮大幅向上滑动: ({start_x}, {start_y}) -> ({start_x}, {end_y})")
d.swipe(start_x, start_y, start_x, end_y, 0.25)
await asyncio.sleep(1.0)
screen_path = take_screenshot(d, f"tld_detail_scan_{int(time.time())}_{idx}.jpg")
logger.info(f"[详情页] 第 {idx + 1} 轮滑动后的截图: {screen_path}")
curr_md5 = get_image_content_md5(
screen_path,
top_ratio=SAFE_EXCLUDE_RATIO,
bottom_ratio=BOTTOM_SAFE_EXCLUDE_RATIO,
)
if last_md5 is not None and curr_md5 == last_md5:
stable_count += 1
logger.info(f"[详情页] 页面内容连续第 {stable_count} 次无变化,可能已到稳定区域")
else:
stable_count = 0
last_md5 = curr_md5
final_screen_path = screen_path
if stable_count >= 2:
logger.info("[详情页] 检测到页面多次无变化,认为已到达顶部固定区域,提前结束扫描。")
break
if not final_screen_path:
final_screen_path = take_screenshot(d, f"tld_detail_scan_final_{int(time.time())}.jpg")
logger.info(f"[详情页] 扫描流程未生成截图,使用兜底截图: {final_screen_path}")
else:
logger.info(f"[详情页] 使用最终稳定区域截图作为价格页识别输入: {final_screen_path}")
logger.info("[详情页] 使用固定归一化坐标点击顶部“价格”标签,并点击左侧当前价数字进入全时段电价")
price_tab_screen = final_screen_path
entrance_clicked = False
try:
tab_x = int(PRICE_TAB_X_NORM * w / 1000)
tab_y = int(PRICE_TAB_Y_NORM * h / 1000)
logger.info(f"[详情页] 固定坐标点击价格标签: 归一化({PRICE_TAB_X_NORM}, {PRICE_TAB_Y_NORM}) -> 像素({tab_x}, {tab_y})")
d.click(tab_x, tab_y)
await asyncio.sleep(1.0)
price_tab_screen = take_screenshot(d, f"tld_detail_after_price_tab_{int(time.time())}.jpg")
logger.info(f"[详情页] 点击价格标签后的界面截图已保存: {price_tab_screen}")
entry_x = int(PRICE_ENTRY_X_NORM * w / 1000)
entry_y = int(PRICE_ENTRY_Y_NORM * h / 1000)
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_x}_{tab_y}_price_{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}")
else:
logger.warning(f"[详情页] 加载价格页截图失败,无法绘制诊断红点: {price_tab_screen}")
except Exception as e:
logger.error(f"[详情页] 生成价格入口点击诊断图片失败: {e}")
logger.info(f"[详情页] 点击当前价入口: 像素({click_x}, {click_y}),屏幕大小: ({w}, {h})")
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())