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 09:10:03 +08:00
parent 7f62d84049
commit 7c1864b995
9 changed files with 86 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
Apps/XinDianTu/Crawler.py
View File

@@ -164,38 +164,21 @@ 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)
# 1.5 检查是否存在广告 (VL)
# 检查是否存在广告 (VL) - 仅处理非兔子广告
logger.info("检查是否存在广告弹窗...")
ad_res = await ReadImageKit.detect_ad_popup(screenshot_path, device_info=device_info)
if ad_res:
ad_type = ad_res.get("ad_type")
logger.info(f"检测到广告弹窗，准备关闭: {ad_res} [Type: {ad_type}]")
x, y = ad_res['x'], ad_res['y']
if ad_type == "rabbit":
logger.info(">>> 检测到顽固的兔子广告，执行强力双击策略...")
d.double_click(x, y, duration=0.05)
# 处理“误触背景卡片”的逻辑
logger.info(">>> 等待 3s 以确认是否进入了二级页面...")
await asyncio.sleep(3.0)
# 检查是否还在列表页 (新电途列表页通常有“全城”或“综合排序”按钮)
# 如果这些按钮不存在，说明可能进到了详情页
is_list_page = d(text="全城").exists or d(text="综合排序").exists or d(text="搜索场站").exists
if not is_list_page:
logger.info(">>> 检测到当前不在列表页，尝试执行返回操作...")
d.press("back")
await asyncio.sleep(1.5)
else:
logger.info(">>> 当前仍在列表页，无需返回。")
else:
if ad_type != "rabbit":
x, y = ad_res['x'], ad_res['y']
logger.info(f"检测到非兔子广告: {ad_res} [Type: {ad_type}]，正在关闭...")
d.click(x, y)
await asyncio.sleep(1.5)
# 重新截图
if os.path.exists(screenshot_path): os.remove(screenshot_path)
screenshot_path = take_screenshot(d, image_uuid, save_dir=TEMP_IMAGE_DIR)
# 重新截图
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(">>> 检测到兔子广告。根据策略不再处理，直接开始识别场站。")
# 检查是否已经滚动到底部 (排除状态栏后，内容与上次一致)
current_md5 = Kit.get_image_content_md5(
@@ -251,7 +234,10 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
logger.info(f"场站 {station_name} 匹配到已处理记录，跳过。")
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})")
@@ -331,8 +317,6 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
# 2. 与详情页对比 (使用默认排除比例，与 before_click_md5 一致)
check_detail_md5 = Kit.get_image_content_md5(check_back_path)
if os.path.exists(check_back_path): os.remove(check_back_path)
if check_list_md5 == current_md5:
logger.info("检测到已直接返回列表页，跳过后续返回操作。")
should_back_to_list = False
@@ -352,6 +336,9 @@ async def get_station_list(d, service, uploader, max_scrolls=MAX_SCROLLS):
else:
logger.info("VL 未发现'全部时段'按钮，判定已回到列表页(或非详情页)，停止回退以防误退。")
should_back_to_list = False
# 使用完毕后再清理检查用的截图
if os.path.exists(check_back_path): os.remove(check_back_path)
else:
logger.warning(f"点击 '{station_name}' 的 '全部时段' 按钮后页面无明显变化，跳过三级页面处理。")
@@ -364,12 +351,17 @@ 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)
# 记录 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: