291 lines
11 KiB
Python
291 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
柏悦WMS - 分阶段脚本
|
|
Phase 1: 登录并保存cookies (需要可见浏览器+手动验证码)
|
|
Phase 2: 加载cookies, headless模式抓取所有页面
|
|
|
|
用法:
|
|
python3 scrape_wms_final.py login # 阶段1: 登录保存cookie
|
|
python3 scrape_wms_final.py scrape # 阶段2: 用cookie抓取
|
|
"""
|
|
import asyncio, json, sys, os, time
|
|
from playwright.async_api import async_playwright
|
|
|
|
BASE = "http://39.101.149.173"
|
|
API_BASE = "https://adtest.bylogistic.cn"
|
|
USERNAME = "admin"
|
|
PASSWORD = "ndui#Emdfj23123"
|
|
COOKIE_FILE = "/tmp/wms_cookies.json"
|
|
OUTPUT_DIR = os.path.expanduser("~/Documents/柏悦")
|
|
os.makedirs(OUTPUT_DIR, exist_ok=True)
|
|
|
|
async def phase_login():
|
|
"""打开浏览器让用户登录，然后保存cookies"""
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch(headless=False, args=["--start-maximized"])
|
|
context = await browser.new_context(viewport={"width": 1440, "height": 900})
|
|
page = await context.new_page()
|
|
|
|
print("=== 打开登录页 ===")
|
|
await page.goto(f"{BASE}/user/login?redirect=%2Fwms%2FByInstoreList", timeout=30000)
|
|
await page.wait_for_timeout(2000)
|
|
|
|
# 自动填充
|
|
inputs_text = page.locator("input[type='text']")
|
|
input_pwd = page.locator("input[type='password']")
|
|
await inputs_text.nth(0).fill(USERNAME)
|
|
await input_pwd.fill(PASSWORD)
|
|
|
|
print("✅ 账号密码已填写")
|
|
print("👉 请在浏览器中输入验证码并点击'确定'")
|
|
print("⏳ 等待登录成功...")
|
|
|
|
try:
|
|
await page.wait_for_function(
|
|
"() => !window.location.pathname.includes('login')",
|
|
timeout=300000 # 等5分钟
|
|
)
|
|
print(f"✅ 登录成功! URL: {page.url}")
|
|
except Exception as e:
|
|
print(f"❌ 等待超时: {e}")
|
|
try:
|
|
await browser.close()
|
|
except:
|
|
pass
|
|
return False
|
|
|
|
# 立即保存cookies（不等页面加载完）
|
|
try:
|
|
await page.wait_for_timeout(3000)
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
cookies = await context.cookies()
|
|
with open(COOKIE_FILE, "w") as f:
|
|
json.dump(cookies, f, indent=2)
|
|
print(f"✅ Cookies已保存到 {COOKIE_FILE} ({len(cookies)} 个)")
|
|
except Exception as ce:
|
|
print(f"⚠️ Cookie保存出错: {ce}")
|
|
|
|
# 截图当前页
|
|
try:
|
|
await page.screenshot(path="/tmp/wms_logged_in.png", full_page=True)
|
|
print("📸 当前页面截图: /tmp/wms_logged_in.png")
|
|
except Exception as se:
|
|
print(f"⚠️ 截图出错: {se}")
|
|
|
|
# 保持浏览器打开
|
|
print("\n浏览器将保持10秒后关闭...")
|
|
try:
|
|
await asyncio.sleep(10)
|
|
except:
|
|
pass
|
|
|
|
try:
|
|
await browser.close()
|
|
except:
|
|
pass
|
|
return True
|
|
|
|
async def phase_scrape():
|
|
"""加载cookies并headless抓取所有页面"""
|
|
|
|
# 读取cookies
|
|
if not os.path.exists(COOKIE_FILE):
|
|
print(f"❌ Cookie文件不存在: {COOKIE_FILE}")
|
|
print("请先运行: python3 scrape_wms_final.py login")
|
|
return
|
|
|
|
with open(COOKIE_FILE) as f:
|
|
saved_cookies = json.load(f)
|
|
print(f"🍪 加载了 {len(saved_cookies)} 个cookies")
|
|
|
|
async with async_playwright() as p:
|
|
browser = await p.chromium.launch(headless=True)
|
|
context = await browser.new_context(viewport={"width": 1440, "height": 900})
|
|
await context.add_cookies(saved_cookies)
|
|
page = await context.new_page()
|
|
|
|
# 监听API
|
|
all_apis = []
|
|
page.on("response", lambda r: all_apis.append({
|
|
"url": r.url, "status": r.status, "method": r.request.method
|
|
}))
|
|
|
|
# 先访问入库单列表（默认首页）
|
|
print("\n=== 访问系统首页 ===")
|
|
await page.goto(f"{BASE}/wms/ByInstoreList", timeout=20000)
|
|
await page.wait_for_timeout(4000)
|
|
|
|
if "login" in page.url.lower():
|
|
print("❌ Cookies已失效，请重新运行 login 命令")
|
|
await browser.close()
|
|
return
|
|
|
|
print(f"✅ 已进入系统: {page.url}")
|
|
|
|
# ===== 抓取菜单结构 =====
|
|
print("\n=== 抓取菜单结构 ===")
|
|
|
|
menu_structure = {}
|
|
|
|
# 一级菜单
|
|
top_menus = page.locator(".ant-menu-submenu-title")
|
|
top_count = await top_menus.count()
|
|
|
|
for i in range(top_count):
|
|
try:
|
|
name = (await top_menus.nth(i).inner_text()).strip().split("\n")[0].strip()
|
|
if not name or len(name) > 20:
|
|
continue
|
|
|
|
# 展开菜单
|
|
await top_menus.nth(i).click()
|
|
await page.wait_for_timeout(600)
|
|
|
|
# 获取子菜单
|
|
sub_items = page.locator(".ant-menu-submenu-open .ant-menu-item")
|
|
subs = []
|
|
for j in range(await sub_items.count()):
|
|
try:
|
|
sn = (await sub_items.nth(j).inner_text()).strip().split("\n")[0]
|
|
if sn and len(sn) < 30:
|
|
subs.append(sn)
|
|
except:
|
|
pass
|
|
|
|
menu_structure[name] = subs
|
|
print(f" 📂 {name}: {subs}")
|
|
|
|
# 收起
|
|
await top_menus.nth(i).click()
|
|
await page.wait_for_timeout(300)
|
|
|
|
except Exception as e:
|
|
pass
|
|
|
|
# 所有要访问的页面路径
|
|
pages_to_visit = [
|
|
("首页", "/wms/Dashboard"),
|
|
("入库单列表", "/wms/ByInstoreList"),
|
|
("入库单测量", "/wms/ByInstoreList4Measured"),
|
|
("测量数据审核", "/wms/ByInstoreList4MeasuredAudit"),
|
|
("接收审核", "/wms/ByInstoreList4ReceiveAudit"),
|
|
("客户管理", "/wms/ByCustomerList"),
|
|
("出库单列表", "/wms/ByOutstoreList"),
|
|
("派送单列表", "/wms/ByDispatchList"),
|
|
("货柜管理", "/wms/ByContainerList"),
|
|
("库存管理", "/wms/ByInventoryList"),
|
|
("财务结算", "/wms/ByFinanceList"),
|
|
("托盘管理", "/wms/ByTrayList"),
|
|
("BOL管理", "/wms/ByBolList"),
|
|
("用户管理", "/system/UserList"),
|
|
("角色管理", "/system/RoleList"),
|
|
("菜单管理", "/system/MenuList"),
|
|
("字典管理", "/system/DictList"),
|
|
("参数配置", "/system/ConfigList"),
|
|
]
|
|
|
|
# ===== 逐页抓取 =====
|
|
print(f"\n=== 逐页抓取 (共{len(pages_to_visit)}页) ===\n")
|
|
pages_data = {}
|
|
|
|
for idx, (name, path) in enumerate(pages_to_visit, 1):
|
|
try:
|
|
url = BASE + path
|
|
print(f"[{idx}/{len(pages_to_visit)}] {name} -> {path}", end=" ... ")
|
|
|
|
await page.goto(url, timeout=15000)
|
|
await page.wait_for_timeout(2500)
|
|
|
|
if "login" in page.url.lower():
|
|
print("❌ 会话过期!")
|
|
break
|
|
|
|
entry = {
|
|
"title": await page.title(),
|
|
"url": page.url,
|
|
}
|
|
|
|
# 表格列头
|
|
headers = []
|
|
ths = page.locator(".ant-table-thead th, thead th, [class*='table'] th")
|
|
for i in range(min(await ths.count(), 25)):
|
|
h = (await ths.nth(i).inner_text()).strip()
|
|
if h and h not in headers:
|
|
headers.append(h)
|
|
|
|
# 搜索条件字段
|
|
fields = []
|
|
labels = page.locator(".ant-form-item-label label, .query-group label, label[for]")
|
|
for i in range(min(await labels.count(), 20)):
|
|
t = (await labels.nth(i).inner_text()).strip()
|
|
if t and t not in fields and len(t) < 30:
|
|
fields.append(t)
|
|
|
|
# 操作按钮
|
|
buttons = []
|
|
btns = page.locator(".ant-btn:not(.ant-pagination-item):not(.ant-pagination-options-quick-jumper):not([class*='link'])")
|
|
for i in range(min(await btns.count(), 15)):
|
|
b = (await btns.nth(i).inner_text()).strip()
|
|
if b and b not in buttons and len(b) < 20 and b not in ["刷 新","关 闭"]:
|
|
buttons.append(b)
|
|
|
|
# 统计卡片数字
|
|
stat_vals = []
|
|
stats = page.locator("[class*='statistic'] [class*='value'], [class*='number-card'] [class*='value']")
|
|
for i in range(min(await stats.count(), 10)):
|
|
s = (await stats.nth(i).inner_text()).strip()
|
|
if s:
|
|
stat_vals.append(s)
|
|
|
|
entry["table_headers"] = headers
|
|
entry["search_fields"] = fields
|
|
entry["buttons"] = buttons
|
|
entry["statistics"] = stat_vals
|
|
|
|
pages_data[name] = entry
|
|
print(f"✅ 表头:{len(headers)} 字段:{len(fields)} 按钮:{len(buttons)}")
|
|
|
|
# 截图每个页面
|
|
safe_name = name.replace("/", "_").replace("\\", "_")
|
|
await page.screenshot(path=f"/tmp/wms_page_{safe_name}.png")
|
|
|
|
except Exception as e:
|
|
print(f"❌ {e}")
|
|
pages_data[name] = {"error": str(e)}
|
|
|
|
# 保存完整数据
|
|
full_data = {
|
|
"system_name": "柏悦物流管理系统 (Baiyue WMS)",
|
|
"base_url": BASE,
|
|
"api_base": API_BASE,
|
|
"scrape_time": time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
"menu_structure": menu_structure,
|
|
"pages": pages_data,
|
|
"api_calls": all_apis[-50:], # 最后50个API调用
|
|
}
|
|
|
|
output_file = os.path.join(OUTPUT_DIR, "wms_system_data.json")
|
|
with open(output_file, "w", encoding="utf-8") as f:
|
|
json.dump(full_data, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"✅ 全部完成!")
|
|
print(f" - 菜单: {len(menu_structure)} 个一级模块")
|
|
print(f" - 页面: {len(pages_data)} 个页面")
|
|
print(f" - 输出: {output_file}")
|
|
print(f"{'='*60}")
|
|
|
|
await browser.close()
|
|
|
|
if __name__ == "__main__":
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else "login"
|
|
if cmd == "login":
|
|
asyncio.run(phase_login())
|
|
elif cmd == "scrape":
|
|
asyncio.run(phase_scrape())
|
|
else:
|
|
print(f"Usage: python3 {sys.argv[0]} [login|scrape]")
|