LogisticsHubIPA/InboundVerify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Anneng (Electron) site: expected-data download + main_router integration

Browse Source 
安能站点第一阶段：把应到数据（运单信息）下载流程接入主路由。

- site_anneng.py：用页面级 CDP 驱动「安能全网门户」Electron 应用（Playwright
  connect_over_cdp 被 Electron 浏览器域 CDP 拦截，故走页面级）。完整应到流程：
  菜单导航 → 进站交接单查询 → 设日期查询（“带小数点的 0”为加载完成判据）
  → 逐条交接单双击进运单信息并核对单号 → 导出（选中字段+→箭头转移全部列）
  → 关闭查询 tab → 导出下载页轮询 → 免对话框下载（x-auth+TGC 直连 GET）
  → 合并为 安能-应到货物数据.xlsx。CDP 端口可配；anneng_ready() 供路由就绪轮询。
- main_router.py：以调试模式启动安能（动态空闲端口，避免端口冲突），经 CDP
  判断首页就绪，菜单加 [10] 应到下载，退出时关闭应用；网页站点逻辑不变。
- config.example.yaml：补充 anneng.query_days 与 anneng.app_path 说明。
- docs/安能门户CDP连接指南.md：连接方案、独立 webContents/tab 处理、免对话框下载。
- .gitignore：忽略 Archive/（早期探查脚本归档）。

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-06-21 19:38:44 +08:00
parent d4a9fb3b4a
commit 9d3d3cbb78
5 changed files with 1476 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
main_router.py
View File

@@ -1,7 +1,10 @@
# main_router.py
import os
import socket
import subprocess
import time
import urllib.request
import yaml
import pandas as pd
from playwright.sync_api import sync_playwright
@@ -13,6 +16,7 @@ import site_shunxin
import site_baishi
import site_zto
import site_yunda
import site_anneng
# 定义网点及对应的初始登录 URL
SITES_CONFIG = {
@@ -32,6 +36,49 @@ READY_SELECTORS = {
"韵达": '.el-menu-item:has-text("首页")',
}
# 安能是 Electron 桌面应用（不是 Playwright 打开的网页），需要单独启动。
APP_SITES = {"安能"}
def _find_free_port():
"""让操作系统分配一个空闲端口，避免固定端口冲突。"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
def _wait_cdp_up(port, timeout=60.0):
"""轮询直到 CDP 调试端口就绪（应用启动需要时间）。"""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
try:
with urllib.request.urlopen(
f"http://localhost:{port}/json/version", timeout=2
) as resp:
if resp.status == 200:
return True
except Exception:
pass
time.sleep(1)
return False
def launch_anneng(app_path):
"""以调试模式启动安能 Electron 应用（自动选取空闲端口），返回子进程对象。
启动后请在应用内手动登录；就绪状态由 run_multi_site_daemon 的就绪轮询判断。
"""
port = _find_free_port()
print(f">> 以调试模式启动【安能】应用（端口 {port}）：{app_path}")
proc = subprocess.Popen([app_path, f"--remote-debugging-port={port}"])
site_anneng.set_cdp_port(port)
if not _wait_cdp_up(port):
raise RuntimeError(
f"安能应用调试端口 {port} 未就绪——可能应用已在运行（单实例），"
"请先关闭已有的安能窗口再试"
)
return proc
def task_process_undelivered_data(site_name="顺心"):
"""应到未到比对：找出应到但未实到的运单 (按站点前缀)"""
@@ -178,7 +225,9 @@ def _print_test_report(results):
print("====================================================")
for i, (site_name, label, status, elapsed, err) in enumerate(results, start=1):
mark = "✅" if status == "PASS" else "❌"
print(f" {i:>2}. {mark} {status} {site_name} - {label} (耗时 {elapsed:.1f}s)")
print(
f" {i:>2}. {mark} {status} {site_name} - {label} (耗时 {elapsed:.1f}s)"
)
if err:
note = err if len(err) <= 60 else err[:57] + "..."
print(f" 说明: {note}")
@@ -197,22 +246,37 @@ def run_multi_site_daemon():
# 1. 读取配置文件
debug_mode = False
debug_target = ""
anneng_app_path = ""
try:
if os.path.exists(CONFIG_PATH):
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
config = yaml.safe_load(f)
debug_mode = config.get("debug", {}).get("enabled", False)
debug_target = config.get("debug", {}).get("target_site", "")
config = yaml.safe_load(f) or {}
debug_mode = (config.get("debug", {}) or {}).get("enabled", False)
debug_target = (config.get("debug", {}) or {}).get("target_site", "")
anneng_app_path = (config.get("anneng", {}) or {}).get("app_path", "")
except Exception as e:
print(f"⚠️ 读取 config.yaml 异常，将使用全量模式启动: {e}")
# 动态确定需要挂载启动的网页
active_sites = {}
if debug_mode and debug_target in SITES_CONFIG:
print(f"\n🛠️ 【调试模式】仅加载目标站点: [{debug_target}]")
active_sites = {debug_target: SITES_CONFIG[debug_target]}
# 动态确定需要挂载启动的网页站点；安能（Electron 应用）单独标记
anneng_active = False
if debug_mode:
if debug_target in SITES_CONFIG:
print(f"\n🛠️ 【调试模式】仅加载目标站点: [{debug_target}]")
active_sites = {debug_target: SITES_CONFIG[debug_target]}
elif debug_target == "安能":
print(f"\n🛠️ 【调试模式】仅加载目标站点: [安能]")
active_sites = {}
anneng_active = True
else:
active_sites = dict(SITES_CONFIG)
anneng_active = True
else:
active_sites = SITES_CONFIG
active_sites = dict(SITES_CONFIG)
anneng_active = True
if anneng_active and not anneng_app_path:
print("⚠️ 已启用安能但 config.yaml 未配置 anneng.app_path，将跳过安能。")
anneng_active = False
with sync_playwright() as p:
browser = p.chromium.launch(headless=False)
@@ -230,6 +294,16 @@ def run_multi_site_daemon():
page.goto(url)
pages_map[site_name] = page
# 安能：以调试模式启动 Electron 应用（非 Playwright 网页）
anneng_proc = None
if anneng_active:
try:
anneng_proc = launch_anneng(anneng_app_path)
pages_map["安能"] = True # 哨兵：表示已启动（无 Playwright page 对象）
except Exception as e:
print(f"⚠️ 启动安能应用失败，已跳过安能：{e}")
anneng_active = False
print("\n====================================================")
print("【登录检测】正在准备各站点登录...")
print("====================================================")
@@ -247,27 +321,38 @@ def run_multi_site_daemon():
# 自动识别各站点登录完成状态，无需手动回车
# ====================================================================
ready_status = {site: False for site in active_sites.keys()}
if anneng_active:
ready_status["安能"] = False
print("\n>> 正在轮询各站点就绪状态 (自动登录或手动登录均可)...")
while not all(ready_status.values()):
for site_name, page in pages_map.items():
if not ready_status[site_name]:
try:
for site_name in list(ready_status.keys()):
if ready_status[site_name]:
continue
try:
if site_name == "安能":
# 安能走 CDP 判断主页是否就绪（连不上返回 False，不抛异常）
ok = site_anneng.anneng_ready()
else:
# 0.5 秒轻量探测，避免阻塞主循环
if page.locator(READY_SELECTORS[site_name]).is_visible(
timeout=500
):
ready_status[site_name] = True
print(f" ✅ 【{site_name}】已检测到主页，登录就绪。")
except Exception:
pass
ok = (
pages_map[site_name]
.locator(READY_SELECTORS[site_name])
.is_visible(timeout=500)
)
if ok:
ready_status[site_name] = True
print(f" ✅ 【{site_name}】已检测到主页，登录就绪。")
except Exception:
pass
pending_sites = [s for s, ready in ready_status.items() if not ready]
if pending_sites:
print(
f" ⏳ 等待以下站点完成登录: [{', '.join(pending_sites)}] ... (请在浏览器中操作)"
f" ⏳ 等待以下站点完成登录: [{', '.join(pending_sites)}] ... "
"(请在浏览器/应用中操作)"
)
page.wait_for_timeout(3000) # 等待 3 秒后进行下一轮检查
time.sleep(3) # 等待 3 秒后进行下一轮检查
print("\n====================================================")
print("【准备】所有站点已就绪，正在清理初始弹窗...")
@@ -326,6 +411,9 @@ def run_multi_site_daemon():
if "韵达" in pages_map:
print(" ✅ 【韵达】已就绪。")
if "安能" in pages_map:
print(" ✅ 【安能】已就绪。")
def is_site_ready(site_name):
if site_name not in pages_map:
print(f"\n🚫 站点 [{site_name}] 未加载（当前为调试模式），已跳过。")
@@ -353,6 +441,9 @@ def run_multi_site_daemon():
print(" [6] 执行 - 应到货物数据下载")
print(" [7] 执行 - 实到货物数据下载")
print("-" * 52)
print(" 模块五：【安能】数据处理流（Electron 应用）")
print(" [10] 执行 - 应到货物数据下载（运单信息）")
print("-" * 52)
print(" 自动化测试")
print(" [8] 执行 - 全站点下载流程自动化测试 (交叉跑通校验)")
print("-" * 52)
@@ -386,6 +477,8 @@ def run_multi_site_daemon():
elif choice == "7" and is_site_ready("韵达"):
pages_map["韵达"].bring_to_front()
site_yunda.yunda_actual_download(pages_map["韵达"])
elif choice == "10" and is_site_ready("安能"):
site_anneng.anneng_expected_download()
elif choice == "8":
run_automation_test(pages_map)
elif choice == "9":
@@ -399,12 +492,30 @@ def run_multi_site_daemon():
print("\n正在关闭浏览器并退出...")
break
else:
if choice not in ["1", "2", "3", "4", "5", "6", "7", "8", "9", "0"]:
if choice not in [
"1",
"2",
"3",
"4",
"5",
"6",
"7",
"8",
"9",
"10",
"0",
]:
print("\n⚠️ 无效输入，请查证后回车。")
except Exception as e:
print(f"❌ 任务调度异常: {e}")
browser.close()
if anneng_proc is not None:
try:
anneng_proc.terminate()
print("已关闭安能应用。")
except Exception:
pass
print("程序已退出。")