270 lines
12 KiB
Python
270 lines
12 KiB
Python
# coding=utf-8
|
|
import asyncio
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
import time
|
|
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 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. 点击“价格信息”区域 (识别橘红色价格 P0)
|
|
price_button_screen = take_screenshot(d, f"tld_before_price_click_{int(time.time())}.jpg")
|
|
logger.info("正在通过 CV 寻找橘红色价格区域 (P0)...")
|
|
|
|
click_point = detect_price_click_point_cv(price_button_screen)
|
|
|
|
# 调试：生成点击点标注图
|
|
if click_point:
|
|
debug_flag_path = price_button_screen.replace(".jpg", "_click_debug.jpg")
|
|
img_debug = read_image(price_button_screen)
|
|
if img_debug is not None:
|
|
cv2.circle(img_debug, (click_point[0], click_point[1]), 20, (0, 0, 255), -1) # 红色大圆点
|
|
cv2.line(img_debug, (click_point[0]-40, click_point[1]), (click_point[0]+40, click_point[1]), (255, 255, 255), 3)
|
|
cv2.line(img_debug, (click_point[0], click_point[1]-40), (click_point[0], click_point[1]+40), (255, 255, 255), 3)
|
|
save_image(debug_flag_path, img_debug)
|
|
logger.info(f"点击点调试图已保存: {debug_flag_path}")
|
|
|
|
try:
|
|
if click_point:
|
|
logger.info(f"CV 成功定位价格区域，点击坐标: {click_point}")
|
|
d.click(click_point[0], click_point[1])
|
|
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
|
|
else:
|
|
logger.warning("CV 未能定位价格区域，尝试模板匹配兜底...")
|
|
template_path = os.path.join(project_root, "Apps", "TeLaiDian", "Template", "jgxx.jpg")
|
|
match = d.image.match(template_path)
|
|
if match:
|
|
d.image.click(template_path)
|
|
else:
|
|
logger.warning("模板匹配也失败，执行坐标兜底...")
|
|
w, h = d.window_size()
|
|
d.click(w // 2, int(h * 0.45)) # 滑动后价格通常在屏幕中上部
|
|
await asyncio.sleep(2.0) # 等待页面加载及小程序可能的自动滚动
|
|
|
|
# 额外处理：小程序可能会自动滚动到当前时段，我们需要手动滚回顶部以抓取完整数据
|
|
logger.info("向上滚动 2 次，确保回到 00:00 时段顶部...")
|
|
for _ in range(2):
|
|
d.swipe_ext("down", scale=0.9) # 向下滑动 = 页面向上滚动
|
|
await asyncio.sleep(0.8)
|
|
except Exception as e:
|
|
logger.error(f"点击价格区域失败: {e}")
|
|
finally:
|
|
if os.path.exists(price_button_screen): os.remove(price_button_screen)
|
|
|
|
# 4. 循环滑动抓取完整分时电价
|
|
all_prices = []
|
|
last_price_md5 = None
|
|
price_page_count = 0
|
|
max_price_pages = 3 # 分时电价通常不会超过3页
|
|
screenshot_tasks = [] # 用于异步分析图片的任务列表
|
|
temp_screenshots = [] # 记录临时文件以便后续清理
|
|
|
|
logger.info("开始循环截图并异步抓取完整分时电价...")
|
|
|
|
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}，加入异步分析队列")
|
|
# 创建异步任务，但不立即 await
|
|
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.2)
|
|
price_page_count += 1
|
|
|
|
# 异步等待所有图片识别任务完成
|
|
if screenshot_tasks:
|
|
logger.info(f"正在异步分析 {len(screenshot_tasks)} 张价格截图...")
|
|
results = await asyncio.gather(*screenshot_tasks)
|
|
|
|
for page_prices in results:
|
|
if page_prices:
|
|
# 深度去重：根据时段 (start, end) 合并
|
|
for p in page_prices:
|
|
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
|
|
# 如果已有条目信息不全，则更新它（比如之前没识别出 PLUS 价格）
|
|
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:
|
|
existing[key] = p[key]
|
|
break
|
|
if not is_duplicate:
|
|
all_prices.append(p)
|
|
|
|
# 清理所有临时截图
|
|
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())
|