import os import re import glob from datetime import datetime import pandas as pd from playwright.sync_api import sync_playwright def task_expected_goods_download(page): """模块一：应到货物数据下载""" print("\n▶ 开始执行【应到货物数据下载】任务...") # 初始化并创建下载目录 download_dir = os.path.join(os.getcwd(), "downloads") if not os.path.exists(download_dir): os.makedirs(download_dir) print(f">> 已创建专属下载文件夹: {download_dir}") export_times = [] target_task_timestamps = [] try: # 1. 导航与页面加载判断 print(">> 正在进入【车辆点到】界面...") page.locator("span.ant-pro-menu-item-title:has-text('派件管理')").click() page.locator("div.ant-pro-menu-item:has-text('车辆点到')").click() page.get_by_role("button", name="点到").wait_for(state="visible") page.get_by_role("button", name="打印交接单").wait_for(state="visible") page.get_by_role("button", name="强卸").wait_for(state="visible") print("✅ 车辆点到界面加载完毕") # 2. 筛选条件：1天、状态=已发 -> 已到、查询 print(">> 正在设置筛选条件: 选择【1天】...") page.get_by_role("radio", name="1天").check() print(">> 正在展开【状态】下拉菜单...") # 匹配包含 已发 或 已到 的下拉选项 page.locator( ".ant-select-selection-item", has_text=re.compile(r"已发|已到") ).click() print(">> 正在选择状态为【已到】...") page.locator(".ant-select-item-option", has_text="已到").click() print(">> 正在点击【查询】按钮...") page.get_by_role("button", name="search 查询").click() page.wait_for_timeout(2000) # 3. 获取所有【运单列表】按钮并循环处理 waybill_btns = page.get_by_role("button", name="运单列表") count = waybill_btns.count() print(f">> 共发现 {count} 个班次需要导出。") for i in range(count): print(f" ⏳ 正在处理第 {i+1}/{count} 个班次...") waybill_btns.nth(i).click() page.locator("label[title='运单查询']").wait_for(state="visible") # 4. 执行导出流程 page.get_by_role("button", name="export 导出").click() page.locator( "span.ant-transfer-list-header-title:has-text('待选导出列')" ).wait_for(state="visible") page.locator(".ant-transfer-list").first.locator( ".ant-transfer-list-header label" ).click() page.get_by_label("导出").get_by_role("button", name="right").click() page.get_by_role("button", name="export 导出数据").click() export_times.append(datetime.now()) page.locator("text=任务添加成功!").wait_for(state="visible") page.get_by_role("button", name="知道了").click() page.get_by_role("tab", name="车辆点到").click() page.wait_for_timeout(500) if count > 0: print("✅ 所有班次的导出任务已成功提交！") else: print("⚠️ 未发现任何运单列表，直接跳转至下载环节。") # 6. 前往数据导出页面去下载 print(">> 正在前往【数据导出】界面...") page.locator("a[href='/dataExport']").click() page.get_by_role("columnheader", name="任务标题").wait_for(state="visible") page.wait_for_timeout(2000) page.get_by_role("button", name="search 查询").click() page.wait_for_timeout(2000) # 7. 轮询任务状态 print(">> 列表中已渲染，开始匹配并检查后端处理状态...") while True: rows = page.locator(".ant-table-tbody > tr.ant-table-row") row_count = rows.count() pending_tasks = 0 current_ready_timestamps = [] for i in range(row_count): tds = rows.nth(i).locator("td") if tds.count() < 9: continue submit_time_str = tds.nth(2).inner_text().strip() title_str = tds.nth(5).inner_text().strip() status_str = tds.nth(6).inner_text().strip() if title_str == "发车管理运单列表": try: row_time = datetime.strptime( submit_time_str, "%Y-%m-%d %H:%M:%S" ) matched = any( abs((row_time - et).total_seconds()) <= 60 for et in export_times ) if matched: if status_str != "执行完成": pending_tasks += 1 if submit_time_str not in current_ready_timestamps: print( f" ⏳ 任务 [{submit_time_str}] 状态为【{status_str}】，数据生成中..." ) else: if submit_time_str not in current_ready_timestamps: current_ready_timestamps.append(submit_time_str) except Exception as e: print(f" ⚠️ 解析时间时出错: {e}") if pending_tasks > 0: print( f">> 共有 {pending_tasks} 个匹配任务还在处理中，等待 5 秒后刷新..." ) page.wait_for_timeout(5000) page.get_by_role("button", name="search 查询").click() page.wait_for_timeout(2000) else: target_task_timestamps = current_ready_timestamps if len(target_task_timestamps) > 0: print(">> ✅ 所有目标任务已就绪！开始并行下载...") break # 8. 下载逻辑 downloaded_files = [] for time_str in target_task_timestamps: try: target_row = page.locator(".ant-table-tbody > tr.ant-table-row").filter( has=page.locator(f"td:nth-child(3):has-text('{time_str}')") ) print(f" 🎯 触发下载 -> 任务 [{time_str}] ...") with page.expect_download() as download_info: target_row.locator("td").nth(8).locator( "button", has_text=re.compile(r"下\s*载") ).click() download = download_info.value save_path = os.path.join(download_dir, download.suggested_filename) download.save_as(save_path) downloaded_files.append(save_path) print(f" ⬇️ 文件已落盘: downloads/{download.suggested_filename}") except Exception as e: print(f" ❌ 下载任务 [{time_str}] 失败: {e}") # 9. 合并数据 if downloaded_files: print("\n>> 🧪 正在开始执行扁平数据高能合并流程...") all_data_frames = [] for file_path in downloaded_files: try: df = pd.read_excel(file_path) if not df.empty: all_data_frames.append(df) except Exception as e: pass if all_data_frames: combined_df = pd.concat(all_data_frames, ignore_index=True) final_output_path = os.path.join(download_dir, "应到货物数据.xlsx") combined_df.to_excel(final_output_path, index=False) print(f"====================================================") print(f" 🎉 恭喜！合并成功！最终输出路径: {final_output_path}") print(f"====================================================") for file_path in downloaded_files: os.remove(file_path) print("✅ 临时数据清理完毕。") print("\n🎉 【应到货物数据下载】全流程测试完毕！") except Exception as e: print(f"\n❌ 任务执行过程中发生异常: {e}") def task_actual_goods_download(page): """模块二：实到货物数据下载""" print("\n▶ 开始执行【实到货物数据下载】任务...") # 初始化并创建下载目录 download_dir = os.path.join(os.getcwd(), "downloads") if not os.path.exists(download_dir): os.makedirs(download_dir) export_times = [] target_task_timestamps = [] try: # 1. 导航与页面加载 print(">> 正在进入【卸车扫描记录】界面...") page.locator("span.ant-pro-menu-item-title:has-text('派件管理')").click() page.locator("div.ant-pro-menu-item:has-text('卸车扫描记录')").click() page.get_by_role("radio", name="1天").wait_for(state="visible") # page.locator("button:has-text('查询')").wait_for(state="visible") print("✅ 卸车扫描记录界面加载完毕") # 2. 筛选条件：1天、查询 print(">> 正在设置筛选条件: 选择【1天】...") page.get_by_role("radio", name="1天").check() # page.pause() print(">> 正在点击【查询】按钮...") page.get_by_role("button", name="search 查询").click() page.wait_for_timeout(2000) # 3. 直接发起全局导出 print(">> 正在发起导出请求...") page.get_by_role("button", name="export 导出").click() page.locator( "span.ant-transfer-list-header-title:has-text('待选导出列')" ).wait_for(state="visible") page.locator(".ant-transfer-list").first.locator( ".ant-transfer-list-header label" ).click() page.get_by_label("导出").get_by_role("button", name="right").click() page.get_by_role("button", name="export 导出数据").click() export_times.append(datetime.now()) page.locator("text=任务添加成功!").wait_for(state="visible") page.get_by_role("button", name="知道了").click() print("✅ 卸车扫描记录导出任务已成功提交！") # 4. 前往数据导出页面去下载 print(">> 正在前往【数据导出】界面...") page.locator("a[href='/dataExport']").click() # page.pause() page.get_by_role("columnheader", name="任务标题").wait_for(state="visible") page.wait_for_timeout(2000) page.get_by_role("button", name="search 查询").click() page.wait_for_timeout(2000) # 5. 轮询任务状态 print(">> 列表中已渲染，开始匹配并检查后端处理状态...") while True: rows = page.locator(".ant-table-tbody > tr.ant-table-row") row_count = rows.count() pending_tasks = 0 current_ready_timestamps = [] for i in range(row_count): tds = rows.nth(i).locator("td") if tds.count() < 9: continue submit_time_str = tds.nth(2).inner_text().strip() title_str = tds.nth(5).inner_text().strip() status_str = tds.nth(6).inner_text().strip() if title_str.startswith("卸车扫描记录"): try: row_time = datetime.strptime( submit_time_str, "%Y-%m-%d %H:%M:%S" ) matched = any( abs((row_time - et).total_seconds()) <= 60 for et in export_times ) if matched: if status_str != "执行完成": pending_tasks += 1 if submit_time_str not in current_ready_timestamps: print( f" ⏳ 任务 [{submit_time_str}] 状态为【{status_str}】，数据生成中..." ) else: if submit_time_str not in current_ready_timestamps: current_ready_timestamps.append(submit_time_str) except Exception as e: print(f" ⚠️ 解析时间时出错: {e}") if pending_tasks > 0: print( f">> 共有 {pending_tasks} 个匹配任务还在处理中，等待 5 秒后刷新..." ) page.wait_for_timeout(5000) page.get_by_role("button", name="search 查询").click() page.wait_for_timeout(2000) else: target_task_timestamps = current_ready_timestamps if len(target_task_timestamps) > 0: print(">> ✅ 目标任务已就绪！开始下载...") break # 6. 下载逻辑 downloaded_files = [] for time_str in target_task_timestamps: try: target_row = page.locator(".ant-table-tbody > tr.ant-table-row").filter( has=page.locator(f"td:nth-child(3):has-text('{time_str}')") ) print(f" 🎯 触发下载 -> 任务 [{time_str}] ...") with page.expect_download() as download_info: target_row.locator("td").nth(8).locator( "button", has_text=re.compile(r"下\s*载") ).click() download = download_info.value save_path = os.path.join(download_dir, download.suggested_filename) download.save_as(save_path) downloaded_files.append(save_path) print(f" ⬇️ 文件已落盘: downloads/{download.suggested_filename}") except Exception as e: print(f" ❌ 下载任务 [{time_str}] 失败: {e}") # 7. 合并数据 if downloaded_files: print("\n>> 🧪 正在开始执行数据归档整理...") all_data_frames = [] for file_path in downloaded_files: try: df = pd.read_excel(file_path) if not df.empty: all_data_frames.append(df) except Exception as e: pass if all_data_frames: combined_df = pd.concat(all_data_frames, ignore_index=True) final_output_path = os.path.join(download_dir, "实到货物数据.xlsx") combined_df.to_excel(final_output_path, index=False) print(f"====================================================") print(f" 🎉 恭喜！处理成功！最终输出路径: {final_output_path}") print(f"====================================================") for file_path in downloaded_files: os.remove(file_path) print("✅ 临时数据清理完毕。") print("\n🎉 【实到货物数据下载】全流程测试完毕！") except Exception as e: print(f"\n❌ 任务执行过程中发生异常: {e}") def task_process_undelivered_data(): """模块三：对下载的应到与实到数据进行高级清洗与比对，抽取出应到未到的异常运单""" print("\n▶ 开始执行【应到未到数据处理】任务...") download_dir = os.path.join(os.getcwd(), "downloads") expected_path = os.path.join(download_dir, "应到货物数据.xlsx") actual_path = os.path.join(download_dir, "实到货物数据.xlsx") output_path = os.path.join(download_dir, "应到未到货物数据.xlsx") # 基础校验：确保源文件全部存在 if not os.path.exists(expected_path): print(f"❌ 错误：找不到【应到货物数据】主文档：{expected_path}") print("💡 请先执行菜单选项 [1] 进行提取下载。") return if not os.path.exists(actual_path): print(f"❌ 错误：找不到【实到货物数据】主文档：{actual_path}") print("💡 请先执行菜单选项 [2] 进行提取下载。") return try: print(">> 正在载入本地 Excel 文档...") df_expected = pd.read_excel(expected_path) df_actual = pd.read_excel(actual_path) # 字段安全检查 if "运单号" not in df_expected.columns: print( "❌ 核心资产校验失败：应到货物数据中缺失【运单号】字段，请检查系统导出配置。" ) return if "运单号" not in df_actual.columns: print( "❌ 核心资产校验失败：实到货物数据中缺失【运单号】字段，请检查系统导出配置。" ) return print(">> 正在启动多维数据集比对引擎...") # 核心比对清洗逻辑：在应到中剔除已经实到的运单号 (Left Anti-Join 机制) df_undelivered = df_expected[~df_expected["运单号"].isin(df_actual["运单号"])] # 按需求抽取指定的输出列：班次号、交接单号、运单号 target_columns = ["班次号", "交接单号", "运单号"] # 确保列名完全匹配 available_columns = [ col for col in target_columns if col in df_undelivered.columns ] df_output = df_undelivered[available_columns] print(f">> 筛选完毕！共捕获到异常【应到未到】货物数据: {len(df_output)} 条。") # 输出最终汇总报告 df_output.to_excel(output_path, index=False) print(f"====================================================") print(f" 🎉 异常比对流完成！独立数据已安全输出。") print(f" 📁 成果归档成果路径: {output_path}") print(f"====================================================") except Exception as e: print(f"❌ 数据处理引擎在执行连接和输出时发生致命异常: {e}") def run_daemon_automation(): with sync_playwright() as p: browser = p.chromium.launch(headless=False) context = browser.new_context() page = context.new_page() target_url = "https://sxne.sxjdfreight.com" print(f"正在打开登录页面: {target_url}") page.goto(target_url) print("====================================================") print("【等待人工介入】请手动完成登录。") print("====================================================") try: page.wait_for_selector('h1:has-text("盟商门户网")', timeout=120000) print("🎉 登录成功！系统已接管浏览器。") page.wait_for_timeout(1000) # 跳过初始一些弹窗干扰 page.locator("a").nth(4).click() page.wait_for_timeout(1000) page.get_by_role("button", name="Close").click() page.wait_for_timeout(1000) page.get_by_role("button", name="不再询问").click() page.wait_for_timeout(1000) while True: print("\n==============================") print(" 物流数据自动提取系统 ") print("==============================") print("1. 执行【应到货物数据下载】") print("2. 执行【实到货物数据下载】") print("3. 执行【应到未到数据处理】 (★新)") print("4. 退出程序") print("==============================") choice = input("请输入任务编号并回车: ") if choice == "1": task_expected_goods_download(page) elif choice == "2": task_actual_goods_download(page) elif choice == "3": task_process_undelivered_data() elif choice == "4": print("\n准备退出程序...") break else: print("\n⚠️ 无效输入，请重新选一下。") except Exception as e: print(f"❌ 运行发生致命错误: {e}") finally: browser.close() print("浏览器已安全关闭。") if __name__ == "__main__": run_daemon_automation()