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-13 10:57:30 +08:00
parent 7c1864b995
commit 3978123b88
10 changed files with 73 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
Apps/XinDianTu/Crawler.py
View File

@@ -164,9 +164,11 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
image_uuid = str(uuid.uuid4())
screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR)
# 检查是否存在广告 (VL) - 仅处理非兔子广告
# 检查是否存在广告 (VL)
logger.info("检查是否存在广告弹窗...")
ad_res = await ReadImageKit.detect_ad_popup(screenshot_path, device_info=device_info)
ad_top_y_norm = 0.90 # 默认使用一个更宽松的底部边界 (0.90)，防止点到导航栏
if ad_res:
ad_type = ad_res.get("ad_type")
if ad_type != "rabbit":
@@ -178,7 +180,10 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
if os.path.exists(screenshot_path): os.remove(screenshot_path)
screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR)
else:
logger.info(">>> 检测到兔子广告。根据策略不再处理，直接开始识别场站。")
# 发现兔子广告，不再尝试关闭，直接计算它的顶边界 (归一化 0-1)
# 我们取 0.90 作为安全阈值，低于此 Y 坐标的场站认为被遮挡
ad_top_y_norm = 0.90
logger.info(f">>> 发现兔子广告，设定安全边界为 Y_norm < {ad_top_y_norm}")
# 检查是否已经滚动到底部 (排除状态栏后，内容与上次一致)
current_md5 = Kit.get_image_content_md5(
@@ -234,12 +239,19 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
logger.info(f"场站 {station_name} 匹配到已处理记录，跳过。")
continue
# 【策略调整】每页仅处理第一个识别到的新场站，确保其不被底部广告遮挡
logger.info(f">>> 发现新场站 '{station_name}'。根据最新策略，每页仅处理第一个场站以避开广告。")
# 【优化】检查是否被遮挡或太靠近底部
# card["bounds_norm"] 是 {left, top, right, bottom}
card_bottom = card["bounds_norm"]["bottom"]
if card_bottom > ad_top_y_norm:
logger.warning(f"场站 '{station_name}' 被遮挡或太靠近底部 (Bottom {card_bottom:.2f} > {ad_top_y_norm})，留待下次滚动处理。")
continue
# 正常处理新场站
logger.info(f">>> 发现新场站 '{station_name}'，开始处理...")
new_stations_processed += 1
click_x, click_y = card["click_point"]
logger.info(f"准备处理第 {idx + 1} 个场站: {station_name}, 点击坐标: ({click_x}, {click_y})")
logger.info(f"准备处理场站: {station_name}, 点击坐标: ({click_x}, {click_y})")
d.click(int(click_x), int(click_y))
@@ -248,11 +260,26 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
should_back_to_list = True
# 截取二级页面图
detail_uuid = str(uuid.uuid4())
detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR)
# 【新增】二级页面广告检测 (如：免费停车提示)
logger.info("检查详情页是否存在干扰弹窗...")
detail_ad_res = await ReadImageKit.detect_ad_popup(detail_path, device_info=device_info)
if detail_ad_res:
dad_type = detail_ad_res.get("ad_type")
if dad_type == "rabbit":
logger.info(">>> 详情页检测到 rabbit 广告，判定为误报，忽略。")
else:
dx, dy = detail_ad_res["x"], detail_ad_res["y"]
logger.info(f"检测到详情页弹窗: {dad_type}，正在点击关闭 ({dx}, {dy})...")
d.click(dx, dy)
await asyncio.sleep(1.5)
# 重新截图，确保后续流程使用的是清理后的界面
if os.path.exists(detail_path): os.remove(detail_path)
detail_path = take_screenshot(d, detail_uuid, save_dir=TEMP_IMAGE_DIR)
# 【优化】后台解析详情页地址，直接传本地路径，避免等待上传
logger.info(f"已启动后台分析详情页: {station_name}")
task_addr = asyncio.create_task(service.process_station_address(station_name, detail_path, device_info=device_info))
@@ -351,17 +378,13 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
# 从二级页面返回 (仅当确实需要返回时)
if should_back_to_list:
d.press("back")
logger.info(f"等待 {WAIT_BACK_TO_LIST} 秒返回列表...")
await asyncio.sleep(WAIT_BACK_TO_LIST)
logger.info(f"等待 {WAIT_BACK_TO_LIST + 1} 秒返回列表...")
await asyncio.sleep(WAIT_BACK_TO_LIST + 1)
# 记录 Redis 去重 (仅按名称去重)
cleaned = Kit.clean_station_name(station_name)
await redis_kit.set_data(f"crawled:xdt:{cleaned}", "1", expire=REDIS_STATION_EXPIRE)
# 【策略执行】处理完第一个新场站后立即退出循环，触发翻页
logger.info(f"已完成本页首个新场站 '{station_name}' 的处理，准备翻页。")
break
# 清理已完成的后台任务
done_tasks = [t for t in background_tasks if t.done()]
for t in done_tasks:
@@ -379,7 +402,9 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
no_new_data_count = 0
# 6. 翻页
logger.info("执行翻页滑动...")
# 【优化】使用“小步快跑”策略，减小滑动距离以避免跳过场站，并能更平滑地躲避广告
# 步长已在 Setting.py 中统一配置为 SCROLL_DISTANCE_RATIO
logger.info(f"执行翻页滑动 (步长: {SCROLL_DISTANCE_RATIO})...")
d.swipe_ext("up", scale=SCROLL_DISTANCE_RATIO)
await asyncio.sleep(WAIT_AFTER_SCROLL)