huanghai/aiData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2026-01-15 09:33:38 +08:00
parent b49dd9bc1b
commit fe27bc6884
6 changed files with 147 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
Apps/TeLaiDian/Crawler.py
View File

@@ -14,9 +14,11 @@ 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
WAIT_BACK_TO_LIST, DETAIL_SCROLL_DISTANCE_RATIO, FIRST_RUN_ONLY_ONE_STATION,
REDIS_STATION_EXPIRE
)
from Core.BaseCrawler import BaseCrawler
from Util.RedisKit import RedisKit
import uiautomator2 as u2
# 项目根目录处理
@@ -36,6 +38,7 @@ class TeLaiDianCrawler(BaseCrawler):
def __init__(self, service=None):
super().__init__(service or TeLaiDianService())
self.read_image_kit = ReadImageKit()
self.redis_kit = RedisKit()
async def start(self):
"""
@@ -166,7 +169,6 @@ class TeLaiDianCrawler(BaseCrawler):
processed_count = 0
last_md5 = None
processed_station_names = set() # 用于列表级去重
while processed_count < MAX_STATIONS_COUNT:
# 1. 截图并分析
@@ -205,8 +207,11 @@ class TeLaiDianCrawler(BaseCrawler):
if not name or not point:
continue
if name in processed_station_names:
logger.info(f"跳过已处理场站: {name}")
# [优化] 使用 Redis 进行跨运行去重
cleaned_name = clean_station_name(name)
redis_key = f"crawled:tld:{cleaned_name}"
if await self.redis_kit.get_data(redis_key):
logger.info(f"跳过已处理场站 (Redis): {name}")
continue
logger.info(f"处理场站: {name} (坐标: {point})")
@@ -234,7 +239,9 @@ class TeLaiDianCrawler(BaseCrawler):
# 爬取详情
await self.crawl_detail_logic(d, station)
processed_station_names.add(name)
# 标记为已处理
await self.redis_kit.set_data(redis_key, "1", ex=REDIS_STATION_EXPIRE)
d.press("back")
await asyncio.sleep(WAIT_BACK_TO_LIST)
@@ -270,19 +277,29 @@ class TeLaiDianCrawler(BaseCrawler):
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")
total_piles = None
free_piles = None
piles_detail = None
parking_info = None
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}")
if basic_info.get("name"):
station_name = basic_info.get("name")
if basic_info.get("address"):
address = basic_info.get("address")
# 提取电桩信息
total_piles = basic_info.get("total_piles")
free_piles = basic_info.get("free_piles")
piles_detail = basic_info.get("piles_detail")
parking_info = basic_info.get("parking_info")
logger.info(f"[详情页] 基础信息识别结果: {station_name} | {address} | 桩数: {total_piles}/{free_piles} | 停车费: {parking_info}")
except Exception as ex:
logger.error(f"[详情页] 同步分析详情页基础信息失败: {ex}")
finally:
@@ -298,81 +315,94 @@ class TeLaiDianCrawler(BaseCrawler):
w, h = d.window_size()
# 1. 增加等待时间，确保页面加载完成
logger.info(f"[详情页] 等待 {WAIT_DETAIL_PAGE_LOAD + 1}s 确保页面稳定...")
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD + 1)
logger.info(f"[详情页] 等待 {WAIT_DETAIL_PAGE_LOAD}s 确保页面稳定...")
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
# 2. 向上滑动大一些，确保“全部时段”露出来
logger.info("[详情页] 执行大范围向上滑动 (y: 90% -> 10%)，确保显示全部时段")
d.swipe(w // 2, int(h * 0.9), w // 2, int(h * 0.1), duration=1.0)
await asyncio.sleep(2.5) # 滑动后等待稳定
# --- [优化] 小步快跑滚动查找逻辑 ---
# 目标：通过多次小幅度滚动 + OCR 实时探测，精准捕获“全部时段”入口，避免滑过头
max_scroll_attempts = 6 # 最大滚动尝试次数
scroll_step_ratio = 0.3 # 每次滚动的步长（屏幕高度的 30%）
found_entry = None
# 3. 截图并识别 (纯 OCR 流程)
final_screen_path = take_screenshot(d, f"tld_detail_ocr_input_{int(time.time())}.jpg")
logger.info(f"[详情页] 已截取 OCR 识别用图: {final_screen_path}")
logger.info(f"[详情页] 开始“小步快跑”滚动查找‘全部时段’入口 (最多 {max_scroll_attempts} 次)...")
for i in range(max_scroll_attempts):
# 1. 截图识别当前屏
curr_screen = take_screenshot(d, f"tld_scroll_ocr_{i}_{int(time.time())}.jpg")
# 2. 尝试 OCR 识别“全部时段”
entry_data = await self.read_image_kit.find_price_entrance_ocr(curr_screen)
if entry_data.get("found"):
logger.info(f"[详情页] 第 {i+1} 次尝试：成功探测到‘全部时段’入口！")
found_entry = {
"screen": curr_screen,
"point": entry_data["point"]
}
break
# 3. 如果没找到，小幅向上滚动一段距离
if i < max_scroll_attempts - 1:
logger.info(f"[详情页] 第 {i+1} 次尝试未找到，小幅向上滚动 (步长: {scroll_step_ratio*100}%)...")
d.swipe(w // 2, int(h * 0.7), w // 2, int(h * (0.7 - scroll_step_ratio)), duration=0.5)
await asyncio.sleep(1.5) # 滚动后短暂停留
# 清理过程截图
if os.path.exists(curr_screen):
try: os.remove(curr_screen)
except: pass
entrance_clicked = False
try:
# 1. 使用 OCR 寻找顶部“价格”标签
tab_data = await self.read_image_kit.find_price_tab_ocr(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"[详情页] OCR 成功找到价格标签: 归一化{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"[详情页] OCR 未找到价格标签，使用固定坐标兜底: 像素({tab_x}, {tab_y})")
logger.info(f"[详情页] 正在点击价格标签...")
d.click(tab_x, tab_y)
await asyncio.sleep(2.5) # 点击标签后等待
price_tab_screen = take_screenshot(d, f"tld_detail_after_price_tab_{int(time.time())}.jpg")
logger.info(f"[详情页] 点击价格标签后的界面截图已保存: {price_tab_screen}")
# 2. 寻找并点击“全部时段”入口 (纯 OCR 识别)
entry_data = await self.read_image_kit.find_price_entrance_ocr(price_tab_screen)
entry_x, entry_y = None, None
if entry_data.get("found") and entry_data.get("point"):
p = entry_data["point"]
# 安全校验：严禁点击 y > 900 的区域，那是底部浮动条
if p[1] > 900:
logger.warning(f"[详情页] OCR 返回坐标 {p} 偏低 (疑似底部浮动条)，尝试使用固定坐标兜底")
entry_x = int(PRICE_ENTRY_X_NORM * w / 1000)
entry_y = int(PRICE_ENTRY_Y_NORM * h / 1000)
if found_entry:
price_tab_screen = found_entry["screen"]
p = found_entry["point"]
# 1. 先点击顶部“价格”标签 (确保切到价格页，虽然滚动前可能已经点击，但这里做双保险)
# 先 OCR 找一次标签
tab_data = await self.read_image_kit.find_price_tab_ocr(price_tab_screen)
tab_x, tab_y = None, None
if tab_data.get("found"):
p_tab = tab_data["point"]
tab_x, tab_y = int(p_tab[0] * w / 1000), int(p_tab[1] * h / 1000)
logger.info(f"[详情页] 点击顶部价格标签: ({tab_x}, {tab_y})")
d.click(tab_x, tab_y)
await asyncio.sleep(1.0)
else:
entry_x = int(p[0] * w / 1000)
entry_y = int(p[1] * h / 1000)
logger.info(f"[详情页] OCR 成功找到价格入口: 归一化{p} -> 像素({entry_x}, {entry_y})")
else:
entry_x = int(PRICE_ENTRY_X_NORM * w / 1000)
entry_y = int(PRICE_ENTRY_Y_NORM * h / 1000)
logger.warning(f"[详情页] OCR 未找到价格入口，使用固定坐标兜底: 像素({entry_x}, {entry_y})")
# 按照用户要求：找不到文字时输出日志、截图并停止程序
fail_screen = take_screenshot(d, f"tld_ocr_tab_fail_{int(time.time())}.jpg")
logger.error(f"❌ [OCR失败] 在页面中未找到‘价格’标签文字！")
logger.error(f"❌ [OCR失败] 最终截图已保存至: {fail_screen}")
logger.error("❌ [OCR失败] 程序将停止运行，请检查页面内容或识别逻辑。")
sys.exit(1)
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_diag.jpg")
try:
# 2. 点击“全部时段”入口
entry_x = int(p[0] * w / 1000)
entry_y = int(p[1] * h / 1000)
# 安全校验
if entry_y > h * 0.9:
logger.warning(f"[详情页] 入口坐标偏低 ({entry_y})，可能在底部遮罩，尝试微调。")
# 绘制最终点击诊断图
debug_click_path = price_tab_screen.replace(".jpg", "_final_click.jpg")
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, 255, 0), -1)
cv2.circle(img, (entry_x, entry_y), 25, (0, 255, 0), -1)
save_image(debug_click_path, img)
logger.info(f"[详情页] 已生成 OCR 点击诊断图 (红点:价格标签, 绿点:入口): {debug_click_path}")
except Exception as e:
logger.error(f"[详情页] 生成诊断图片失败: {e}")
logger.info(f"[详情页] 已生成最终点击诊断图: {debug_click_path}")
logger.info(f"[详情页] 正在点击电价入口...")
d.click(click_x, click_y)
entrance_clicked = True
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD + 1)
logger.info(f"[详情页] 正在点击电价入口: ({entry_x}, {entry_y})")
d.click(entry_x, entry_y)
entrance_clicked = True
await asyncio.sleep(WAIT_DETAIL_PAGE_LOAD)
else:
# 按照用户要求：找不到文字时输出日志、截图并停止程序
fail_screen = take_screenshot(d, f"tld_ocr_fail_{int(time.time())}.jpg")
logger.error(f"❌ [OCR失败] 经过 {max_scroll_attempts} 次滚动仍未在页面中找到‘全部时段’文字！")
logger.error(f"❌ [OCR失败] 最终截图已保存至: {fail_screen}")
logger.error("❌ [OCR失败] 程序将停止运行，请检查页面内容或识别逻辑。")
sys.exit(1) # 停止程序
except Exception as e:
logger.error(f"[详情页] 识别或点击价格入口失败: {e}")
@@ -475,7 +505,15 @@ class TeLaiDianCrawler(BaseCrawler):
except:
pass
logger.info(f"✅ 场站 {station_name_clean} 共提取到 {len(all_prices)} 条价格信息，准备保存...")
await self.service.save_station_data(station_name_clean, address, all_prices)
await self.service.save_station_data(
station_name_clean,
address,
all_prices,
total_piles=total_piles,
free_piles=free_piles,
piles_detail=piles_detail,
parking_info=parking_info
)
else:
logger.warning(f"❌ 未能提取到任何价格信息，请检查页面识别逻辑")
if address: