179 lines
7.5 KiB
Python
179 lines
7.5 KiB
Python
"""
|
||
ERP 登录模块 - DrissionPage
|
||
"""
|
||
|
||
import os
|
||
import time
|
||
import datetime
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
from DrissionPage import ChromiumPage, ChromiumOptions
|
||
|
||
# ── 加载 .env ─────────────────────────────────────────────────────────────────
|
||
load_dotenv(Path(__file__).parent / ".env")
|
||
|
||
ERP_URL = os.getenv("ERP_URL", "https://yunmes.tftykj.cn/#")
|
||
ERP_TENANT = os.getenv("ERP_TENANT", "")
|
||
ERP_USERNAME = os.getenv("ERP_USERNAME", "")
|
||
ERP_PASSWORD = os.getenv("ERP_PASSWORD", "")
|
||
|
||
|
||
# ── 日志 ──────────────────────────────────────────────────────────────────────
|
||
def log(level: str, msg: str):
|
||
icons = {"INFO": "ℹ️ ", "OK": "✅", "WARN": "⚠️ ", "ERR": "❌"}
|
||
ts = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||
print(f"[{ts}] {icons.get(level, ' ')} [{level}] {msg}", flush=True)
|
||
|
||
|
||
def dump_page_state(page: ChromiumPage, label: str = ""):
|
||
"""仅在出错时调用，打印页面快照用于排查。"""
|
||
tag = f" ({label})" if label else ""
|
||
log("WARN", f"─── 页面快照{tag} ───")
|
||
log("WARN", f" URL : {page.url}")
|
||
log("WARN", f" 标题 : {page.title}")
|
||
# 错误提示（timeout=0 不阻塞）
|
||
for sel in [".el-message--error", ".error-msg"]:
|
||
el = page.ele(sel, timeout=0)
|
||
if el and el.text.strip():
|
||
log("WARN", f" 页面提示: {el.text.strip()}")
|
||
log("WARN", "─────────────────────────────")
|
||
|
||
|
||
# ── 浏览器 ────────────────────────────────────────────────────────────────────
|
||
def get_page(headless: bool = False, port: int = 9222) -> ChromiumPage:
|
||
co = ChromiumOptions()
|
||
# 强制在生产环境下使用无头模式
|
||
is_docker = os.path.exists('/.dockerenv')
|
||
if headless or is_docker:
|
||
co.set_argument("--headless=new")
|
||
co.set_argument("--disable-blink-features=AutomationControlled")
|
||
co.set_argument("--no-sandbox")
|
||
co.set_argument("--disable-dev-shm-usage") # 增加这个参数防止容器内存不足
|
||
co.set_argument("--window-size=1440,900")
|
||
co.set_local_port(port)
|
||
return ChromiumPage(co)
|
||
|
||
|
||
# ── Vue 表单专用输入（JS setter + 模拟键盘） ──────────────────────────────────
|
||
def set_input_value(page: ChromiumPage, ele, value: str):
|
||
"""
|
||
Vue 双向绑定下 clear()+input() 会留残值，必须用 JS 原生 setter 清空
|
||
再通过模拟键盘输入触发 keydown/change 事件。
|
||
"""
|
||
page.run_js("""
|
||
var el = arguments[0];
|
||
Object.getOwnPropertyDescriptor(
|
||
window.HTMLInputElement.prototype, 'value'
|
||
).set.call(el, '');
|
||
el.dispatchEvent(new Event('input', {bubbles: true}));
|
||
""", ele)
|
||
time.sleep(0.05) # 等 Vue 处理 clear 事件
|
||
ele.input(value)
|
||
|
||
|
||
# ── 自动登录（使用 .env 账号） ────────────────────────────────────────────────
|
||
def login(page: ChromiumPage) -> bool:
|
||
"""自动填写 .env 中的账号信息并登录。"""
|
||
log("INFO", f"打开登录页: {ERP_URL}")
|
||
page.get(ERP_URL)
|
||
|
||
# 先用较短 timeout 等登录表单；若不出现则检测是否已处于登录状态
|
||
tenant_input = page.ele("@placeholder=租户代码", timeout=5)
|
||
if not tenant_input:
|
||
# 没有登录表单 → 判断是否已登录（主界面无密码框）
|
||
has_pwd = bool(page.ele("@placeholder=密码", timeout=0))
|
||
if not has_pwd:
|
||
log("OK", f"检测到已登录状态，跳过登录 → {page.url}")
|
||
return True
|
||
log("ERR", "未找到登录表单，页面可能未加载")
|
||
dump_page_state(page, "找不到表单")
|
||
return False
|
||
|
||
log("INFO", "填写登录信息...")
|
||
set_input_value(page, tenant_input, ERP_TENANT)
|
||
set_input_value(page, page.ele("@placeholder=账号"), ERP_USERNAME)
|
||
set_input_value(page, page.ele("@placeholder=密码"), ERP_PASSWORD)
|
||
time.sleep(0.2) # 等所有字段的 change/blur 事件结算
|
||
|
||
# 点击登录按钮
|
||
login_btn = page.ele(
|
||
"xpath://*[@id='app']/div/div[3]/div/div[2]/div[2]/div[5]/button",
|
||
timeout=5
|
||
)
|
||
if not login_btn:
|
||
log("ERR", "未找到登录按钮（XPath 失效，请更新选择器）")
|
||
dump_page_state(page, "找不到登录按钮")
|
||
return False
|
||
|
||
log("INFO", "点击登录...")
|
||
login_btn.click()
|
||
|
||
return _wait_for_login(page)
|
||
|
||
|
||
# ── 手动登录（等待用户在浏览器操作） ─────────────────────────────────────────
|
||
def login_manual(page: ChromiumPage) -> bool:
|
||
"""打开登录页，等待用户手动完成登录操作。"""
|
||
log("INFO", f"打开登录页（手动模式）: {ERP_URL}")
|
||
page.get(ERP_URL)
|
||
|
||
# 确认登录页已加载
|
||
tenant_input = page.ele("@placeholder=租户代码", timeout=15)
|
||
if not tenant_input:
|
||
log("ERR", "未找到登录表单")
|
||
dump_page_state(page, "找不到表单")
|
||
return False
|
||
|
||
log("INFO", "═" * 50)
|
||
log("INFO", " 请在浏览器中手动输入账号并登录")
|
||
log("INFO", " 程序将在检测到登录成功后自动继续...")
|
||
log("INFO", "═" * 50)
|
||
|
||
return _wait_for_login(page, timeout=120) # 手动模式等待 2 分钟
|
||
|
||
|
||
# ── 等待登录结果（自动/手动共用） ────────────────────────────────────────────
|
||
def _wait_for_login(page: ChromiumPage, timeout: int = 15) -> bool:
|
||
"""轮询检测登录结果：表单消失 → 成功；error 提示 → 失败。"""
|
||
for elapsed in range(1, timeout + 1):
|
||
time.sleep(1)
|
||
has_form = bool(page.ele("@placeholder=密码", timeout=0))
|
||
err_el = page.ele(".el-message--error", timeout=0)
|
||
err_text = err_el.text.strip() if err_el else ""
|
||
|
||
if err_text:
|
||
log("ERR", f"登录失败: {err_text}")
|
||
dump_page_state(page)
|
||
return False
|
||
|
||
if not has_form:
|
||
log("OK", f"登录成功（{elapsed}s）→ {page.url}")
|
||
return True
|
||
|
||
if elapsed % 10 == 0: # 每 10 秒提示一次进度（手动模式有用）
|
||
log("INFO", f" 等待登录中... ({elapsed}s)")
|
||
|
||
log("WARN", f"超过 {timeout}s 未检测到登录成功")
|
||
dump_page_state(page, "登录超时")
|
||
return False
|
||
|
||
|
||
# ── 单独运行时的入口 ──────────────────────────────────────────────────────────
|
||
if __name__ == "__main__":
|
||
import sys
|
||
manual = "--manual" in sys.argv
|
||
page = get_page()
|
||
try:
|
||
ok = login_manual(page) if manual else login(page)
|
||
if ok:
|
||
log("OK", "登录完成，按 Enter 关闭浏览器")
|
||
input()
|
||
else:
|
||
log("ERR", "登录失败")
|
||
time.sleep(3)
|
||
except KeyboardInterrupt:
|
||
log("INFO", "用户中断")
|
||
finally:
|
||
page.quit()
|
||
log("INFO", "浏览器已关闭")
|