Harden Yunda export dialog with two-layer reset and plain wording

- Add a two-layer reset engine to the export dialog (both expected and
  actual arrival): an outer 3x reopen loop with a field-list presence
  check (交接单号 / 扫描类型 as per-page anchor), and an inner 4x
  submit loop that distinguishes a missed field (re-click 全选) from a
  vanished field list (reopen the whole dialog), recovering from
  empty/ghost-cleared export dialogs
- Keep log/comment wording plain and factual (no hyperbole)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-06-21 12:19:34 +08:00
parent 47d9140dc6
commit d4a9fb3b4a

View File

@@ -41,9 +41,7 @@ def yunda_login(page):
page.locator('button[type="submit"]', has_text="登录").click() page.locator('button[type="submit"]', has_text="登录").click()
page.wait_for_timeout(1000) page.wait_for_timeout(1000)
else: else:
print( print(" -> 未发现登录按钮,判定为已登录,跳过。")
" -> 未发现登录按钮,判定为已登录,跳过。"
)
except Exception as e: except Exception as e:
print(f" ⚠️ 登录检测出错(可能已在工作台内): {e}") print(f" ⚠️ 登录检测出错(可能已在工作台内): {e}")
@@ -72,9 +70,7 @@ def yunda_smart_menu_click(page, menu_path):
title_locator.click() title_locator.click()
page.wait_for_timeout(500) page.wait_for_timeout(500)
else: else:
print( print(f" -> 父菜单 [{item}] 已展开,跳过点击")
f" -> 父菜单 [{item}] 已展开,跳过点击"
)
elif leaf_locator.is_visible(): elif leaf_locator.is_visible():
print(f" -> 点击菜单项 [{item}]") print(f" -> 点击菜单项 [{item}]")
@@ -103,7 +99,6 @@ def yunda_expected_download(page):
yunda_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"]) yunda_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"])
print(">> 正在定位【进站交接单查询】iframe...") print(">> 正在定位【进站交接单查询】iframe...")
# 使用 frame_locator 定位业务 iframe
ws_frame = page.frame_locator("section iframe") ws_frame = page.frame_locator("section iframe")
ws_frame.locator("#startTime").wait_for(state="attached", timeout=15000) ws_frame.locator("#startTime").wait_for(state="attached", timeout=15000)
@@ -154,10 +149,7 @@ def yunda_expected_download(page):
page.wait_for_timeout(800) page.wait_for_timeout(800)
# ====================================================================
# 将 Loading 蒙层与数据判断限制在 #tab-1 内 # 将 Loading 蒙层与数据判断限制在 #tab-1 内
# 避免多标签页命中多个元素 (Strict Mode)
# ====================================================================
loading_mask = ws_frame.locator( loading_mask = ws_frame.locator(
"#tab-1 .fixed-table-loading", has_text="正在努力地加载数据中" "#tab-1 .fixed-table-loading", has_text="正在努力地加载数据中"
).first ).first
@@ -175,18 +167,13 @@ def yunda_expected_download(page):
match_tickets = re.search(r"进站实际票数:(\d+)", sum_text) match_tickets = re.search(r"进站实际票数:(\d+)", sum_text)
if match_tickets and int(match_tickets.group(1)) > 0: if match_tickets and int(match_tickets.group(1)) > 0:
has_data = True has_data = True
print( print(f" ✅ 统计面板已加载,实际票数: [{match_tickets.group(1)}]")
f" ✅ 统计面板已加载,实际票数: [{match_tickets.group(1)}]"
)
if not has_data: if not has_data:
# 空数据提示同样限制在当前 tab 内
if ws_frame.locator( if ws_frame.locator(
"#tab-1 .no-records-found", has_text="没有找到匹配的记录" "#tab-1 .no-records-found", has_text="没有找到匹配的记录"
).first.is_visible(): ).first.is_visible():
print( print(" ⚠️ 确认为空数据,正在关闭当前标签页...")
" ⚠️ 确认为空数据,正在关闭当前标签页..."
)
page.locator(".tags-view-item", has_text="进站交接单查询").locator( page.locator(".tags-view-item", has_text="进站交接单查询").locator(
".el-icon-close" ".el-icon-close"
).click() ).click()
@@ -207,9 +194,8 @@ def yunda_expected_download(page):
current_row = ws_frame.locator("#exampleTable1 tbody tr[data-index]").nth(i) current_row = ws_frame.locator("#exampleTable1 tbody tr[data-index]").nth(i)
raw_no = current_row.locator("td").nth(1).inner_text().strip() raw_no = current_row.locator("td").nth(1).inner_text().strip()
# ====================================================================
# 跳过已绑定的交接单 # 跳过已绑定的交接单
# ====================================================================
bind_status = current_row.locator("td").nth(2).inner_text().strip() bind_status = current_row.locator("td").nth(2).inner_text().strip()
print(f" -> 交接单号: {raw_no} [绑定状态: {bind_status}]") print(f" -> 交接单号: {raw_no} [绑定状态: {bind_status}]")
@@ -217,58 +203,96 @@ def yunda_expected_download(page):
print(" ⏭️ 该交接单已绑定,跳过。") print(" ⏭️ 该交接单已绑定,跳过。")
continue continue
# ====================================================================
current_row.dblclick() current_row.dblclick()
ws_frame.locator("#docSum").wait_for(state="visible", timeout=15000) ws_frame.locator("#docSum").wait_for(state="visible", timeout=15000)
page.wait_for_timeout(500) page.wait_for_timeout(500)
ws_frame.locator('a.btn-info[onclick*="exportFile"]').click() # 导出弹窗双层重试:外层重新打开面板,内层重新提交。
# 区分字段漏选(补点全选)与字段列表消失(重新打开面板)。
ws_frame.locator(".layui-layer-title", has_text="数据导出").wait_for(
state="visible", timeout=15000
)
export_frame = ws_frame.frame_locator('iframe[name="target1"]')
export_frame.locator(".allRight").click()
page.wait_for_timeout(400)
print(" >> 正在建立后台离线任务...")
task_success = False task_success = False
for attempt in range(5): for major_attempt in range(3):
export_frame.locator("#submitbutton", has_text="导出数据").click() print(f" >> 正在打开数据导出面板 (尝试 {major_attempt + 1}/3)...")
confirm_link = export_frame.get_by_role("link", name="确定") ws_frame.locator('a.btn-info[onclick*="exportFile"]').click()
try:
confirm_link.wait_for(state="visible", timeout=6000)
if export_frame.get_by_text("导出任务建立成功").is_visible():
print(" ✅ 已确认导出任务建立成功。")
confirm_link.click()
task_success = True
break
elif export_frame.get_by_text(
"请选择格式相应的导出字段"
).is_visible():
print(
" ⚠️ 检测到未选择字段,重新点击全选..."
)
confirm_link.click()
page.wait_for_timeout(500)
export_frame.locator(".allRight").click()
page.wait_for_timeout(500)
else:
confirm_link.click()
page.wait_for_timeout(1000)
except Exception:
page.wait_for_timeout(1000)
if not task_success: ws_frame.locator(".layui-layer-title", has_text="数据导出").wait_for(
raise RuntimeError( state="visible", timeout=15000
"连续 5 次尝试均未能建立应到数据离线任务。"
) )
ws_frame.locator(".layui-layer-close1").click() export_frame = ws_frame.frame_locator('iframe[name="target1"]')
page.wait_for_timeout(500)
try:
# 校验字段列表是否加载完成(以“交接单号”为标志)
export_frame.get_by_text("交接单号").first.wait_for(
state="visible", timeout=3000
)
except Exception:
print(" ⚠️ 字段列表未加载,关闭面板后重试...")
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(1000)
continue
export_frame.locator(".allRight").click()
page.wait_for_timeout(500)
inner_success = False
needs_reopen = False
print(" -> 正在提交导出任务...")
for attempt in range(4):
export_frame.locator("#submitbutton", has_text="导出数据").click()
confirm_link = export_frame.get_by_role("link", name="确定")
try:
confirm_link.wait_for(state="visible", timeout=6000)
if export_frame.get_by_text("导出任务建立成功").is_visible():
print(
" ✅ 导出任务已建立成功。"
)
confirm_link.click()
inner_success = True
break
elif export_frame.get_by_text(
"请选择格式相应的导出字段"
).is_visible():
confirm_link.click()
page.wait_for_timeout(500)
# 区分:字段漏选 还是 字段列表消失
if export_frame.get_by_text("交接单号").first.is_visible():
print(
" ⚠️ 检测到未选择字段(字段列表仍在),重新点击全选..."
)
export_frame.locator(".allRight").click()
page.wait_for_timeout(500)
else:
print(
" ⚠️ 字段列表异常消失,重新打开导出面板..."
)
needs_reopen = True
break # 跳出内层循环,重新打开面板
else:
confirm_link.click()
page.wait_for_timeout(1000)
except Exception:
page.wait_for_timeout(1000)
if inner_success:
task_success = True
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(500)
break # 跳出外层循环,继续后续步骤
elif needs_reopen:
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(1000)
continue # 重新打开面板
else:
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(1000)
continue
if not task_success:
raise RuntimeError("多次重试后仍未能建立应到数据离线任务。")
ws_frame.locator("#myTab a", has_text="交接单信息").click() ws_frame.locator("#myTab a", has_text="交接单信息").click()
page.wait_for_timeout(800) page.wait_for_timeout(800)
export_times.append(datetime.now()) export_times.append(datetime.now())
@@ -281,9 +305,7 @@ def yunda_expected_download(page):
# 若所有记录都被跳过export_times 为空,直接结束 # 若所有记录都被跳过export_times 为空,直接结束
if not export_times: if not export_times:
print( print(">> ⚠️ 本次未产生任何离线下载任务(无数据或已全部跳过),结束。")
">> ⚠️ 本次未产生任何离线下载任务(无数据或已全部跳过),结束。"
)
return return
_yunda_poll_and_download_tasks( _yunda_poll_and_download_tasks(
@@ -406,49 +428,92 @@ def yunda_actual_download(page):
).click() ).click()
return return
# 5. 执行导出流 # 导出弹窗双层重试:外层重新打开面板,内层重新提交。
# 区分字段漏选(补点全选)与字段列表消失(重新打开面板)。
print(">> 正在发起导出...") print(">> 正在发起导出...")
ws_frame.locator('input[type="button"][id="export"]').click()
ws_frame.locator(".layui-layer-title", has_text="数据导出").wait_for(
state="visible", timeout=15000
)
export_frame = ws_frame.frame_locator('iframe[name="myFrame"]')
export_frame.locator(".allRight").click()
page.wait_for_timeout(400)
print(" >> 正在建立后台离线任务...")
task_success = False task_success = False
for attempt in range(5):
export_frame.locator("#submitbutton", has_text="导出数据").click() for major_attempt in range(3):
confirm_link = export_frame.get_by_role("link", name="确定") print(f" >> 正在打开数据导出面板 (尝试 {major_attempt + 1}/3)...")
ws_frame.locator('input[type="button"][id="export"]').click()
ws_frame.locator(".layui-layer-title", has_text="数据导出").wait_for(
state="visible", timeout=15000
)
export_frame = ws_frame.frame_locator('iframe[name="myFrame"]')
try: try:
confirm_link.wait_for(state="visible", timeout=6000) # 校验字段列表是否加载完成(以“扫描类型”为标志)
if export_frame.get_by_text("导出任务建立成功").is_visible(): export_frame.get_by_text("扫描类型").first.wait_for(
print(" ✅ 已确认导出任务建立成功。") state="visible", timeout=3000
confirm_link.click() )
task_success = True
break
elif export_frame.get_by_text("请选择格式相应的导出字段").is_visible():
print(" ⚠️ 检测到字段未全选,重新点击全选...")
confirm_link.click()
page.wait_for_timeout(500)
export_frame.locator(".allRight").click()
page.wait_for_timeout(500)
else:
confirm_link.click()
page.wait_for_timeout(1000)
except Exception: except Exception:
print(" ⚠️ 字段列表未加载,关闭面板后重试...")
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(1000) page.wait_for_timeout(1000)
continue
export_frame.locator(".allRight").click()
page.wait_for_timeout(500)
inner_success = False
needs_reopen = False
print(" -> 正在提交导出任务...")
for attempt in range(4):
export_frame.locator("#submitbutton", has_text="导出数据").click()
confirm_link = export_frame.get_by_role("link", name="确定")
try:
confirm_link.wait_for(state="visible", timeout=6000)
if export_frame.get_by_text("导出任务建立成功").is_visible():
print(
" ✅ 导出任务已建立成功。"
)
confirm_link.click()
inner_success = True
break
elif export_frame.get_by_text(
"请选择格式相应的导出字段"
).is_visible():
confirm_link.click()
page.wait_for_timeout(500)
# 区分:字段漏选 还是 字段列表消失
if export_frame.get_by_text("扫描类型").first.is_visible():
print(
" ⚠️ 检测到未选择字段(字段列表仍在),重新点击全选..."
)
export_frame.locator(".allRight").click()
page.wait_for_timeout(500)
else:
print(
" ⚠️ 字段列表异常消失,重新打开导出面板..."
)
needs_reopen = True
break
else:
confirm_link.click()
page.wait_for_timeout(1000)
except Exception:
page.wait_for_timeout(1000)
if inner_success:
task_success = True
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(500)
break
elif needs_reopen:
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(1000)
continue
else:
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(1000)
continue
if not task_success: if not task_success:
raise RuntimeError( raise RuntimeError("多次重试后仍未能建立实到数据离线任务。")
"连续 5 次尝试均未能建立实到数据离线任务。"
)
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(500)
export_times.append(datetime.now()) export_times.append(datetime.now())
print(">> 任务提交完成,正在关闭【扫描记录查询】标签页...") print(">> 任务提交完成,正在关闭【扫描记录查询】标签页...")
@@ -457,9 +522,9 @@ def yunda_actual_download(page):
).click() ).click()
page.wait_for_timeout(500) page.wait_for_timeout(500)
# 防线:双重保护 # 若未产生导出任务则结束
if not export_times: if not export_times:
print(">> ⚠️ 本次未产生任何离线下载任务,结束。") print(">> ⚠️ 本次未产生离线下载任务,结束。")
return return
# 6. 轮询并下载 # 6. 轮询并下载
@@ -490,7 +555,7 @@ def _yunda_poll_and_download_tasks(
) )
page.wait_for_timeout(1000) page.wait_for_timeout(1000)
print(">> 开始轮询离线文件队列(定时刷新直到任务齐全)...") print(">> 开始轮询离线任务队列,直到全部完成...")
total_expected = len(export_times) total_expected = len(export_times)
while True: while True:
@@ -538,7 +603,7 @@ def _yunda_poll_and_download_tasks(
).click() ).click()
page.wait_for_timeout(3000) page.wait_for_timeout(3000)
else: else:
print(">> 所有目标离线任务已就绪,开始依次下载...") print(">> 所有离线任务已就绪,开始依次下载...")
break break
downloaded_files = [] downloaded_files = []
@@ -576,7 +641,7 @@ def _yunda_poll_and_download_tasks(
".el-icon-close" ".el-icon-close"
).click() ).click()
print(" ✅ 【导出服务】标签页已关闭。") print(" ✅ 【导出服务】标签页已关闭。")
except Exception: except:
pass pass
if downloaded_files: if downloaded_files:
@@ -587,7 +652,7 @@ def _yunda_poll_and_download_tasks(
df = pd.read_excel(file_path, dtype=str) df = pd.read_excel(file_path, dtype=str)
if not df.empty: if not df.empty:
all_dfs.append(df) all_dfs.append(df)
except Exception: except:
pass pass
if all_dfs: if all_dfs: