120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
"""
|
|
BOM 表查询
|
|
用法:
|
|
python bom_query.py # 自动登录（从 .env 读取账号）
|
|
python bom_query.py --manual # 手动登录（浏览器由人工操作）
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import datetime
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
from login import get_page, login, login_manual, log, dump_page_state
|
|
|
|
BOM_PAGE_URL = "https://yunmes.tftykj.cn/MaterialBom"
|
|
BOM_API_PATH = "MaterialBom_SearchList_Proxy"
|
|
OUTPUT_DIR = Path(__file__).parent / "output"
|
|
OUTPUT_DIR.mkdir(exist_ok=True)
|
|
|
|
|
|
# ── 导航到 BOM 页面 ───────────────────────────────────────────────────────────
|
|
def navigate_to_bom(page):
|
|
log("INFO", f"导航到 BOM 表查询...")
|
|
page.get(BOM_PAGE_URL)
|
|
# 等待数据表格区域出现（比 title XPath 更可靠）
|
|
table = page.ele("xpath://table | .el-table__body", timeout=15)
|
|
if table:
|
|
log("OK", "BOM 页面已加载")
|
|
else:
|
|
log("WARN", "表格元素未找到，继续执行")
|
|
|
|
|
|
# ── 拦截并获取第 1 页 BOM 数据 ───────────────────────────────────────────────
|
|
def fetch_bom_page1(page) -> dict | None:
|
|
log("INFO", "开启网络监听...")
|
|
page.listen.start(BOM_API_PATH)
|
|
|
|
# 刷新页面触发自动加载请求
|
|
page.refresh()
|
|
|
|
log("INFO", "等待 API 响应...")
|
|
packet = page.listen.wait(timeout=30)
|
|
page.listen.stop()
|
|
|
|
if not packet:
|
|
log("ERR", "超时未收到 BOM 响应")
|
|
dump_page_state(page, "监听超时")
|
|
return None
|
|
|
|
log("OK", f"拦截成功 → HTTP {packet.response.status}")
|
|
|
|
# DrissionPage 已自动反序列化 JSON
|
|
body = packet.response.body
|
|
data = body if isinstance(body, (dict, list)) else json.loads(body)
|
|
|
|
# 摘要
|
|
if isinstance(data, dict):
|
|
result = data.get("result", {})
|
|
items = result.get("items", [])
|
|
total = result.get("totalCount", "?")
|
|
log("INFO", f"本页记录: {len(items)} 条 | 总计: {total} 条")
|
|
|
|
return data
|
|
|
|
|
|
# ── 保存为 JSON ───────────────────────────────────────────────────────────────
|
|
def save_json(data, prefix: str = "bom") -> Path:
|
|
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
path = OUTPUT_DIR / f"{prefix}_{ts}.json"
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
log("OK", f"已保存: {path}")
|
|
return path
|
|
|
|
|
|
# ── 主流程 ────────────────────────────────────────────────────────────────────
|
|
def run(manual: bool = False):
|
|
mode = "手动登录" if manual else "自动登录"
|
|
log("INFO", f"BOM 查询启动 [{mode}]")
|
|
|
|
page = get_page(port=9333)
|
|
try:
|
|
# Step 1: 登录
|
|
ok = login_manual(page) if manual else login(page)
|
|
if not ok:
|
|
log("ERR", "登录失败，退出")
|
|
return
|
|
|
|
# Step 2: 进入 BOM 页面
|
|
navigate_to_bom(page)
|
|
|
|
# Step 3: 获取第 1 页数据
|
|
data = fetch_bom_page1(page)
|
|
if data is None:
|
|
log("ERR", "数据获取失败")
|
|
return
|
|
|
|
# Step 4: 保存 JSON
|
|
save_json(data, prefix="bom_page1")
|
|
log("OK", "完成！文件保存在 output/")
|
|
|
|
input("\n按 Enter 关闭浏览器...\n")
|
|
|
|
except KeyboardInterrupt:
|
|
log("INFO", "用户中断 (Ctrl+C)")
|
|
except Exception as e:
|
|
log("ERR", f"异常: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
page.quit()
|
|
log("INFO", "浏览器已关闭")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run(manual="--manual" in sys.argv)
|