afac18e843
- W26: NestJS API + Next.js Web + Prisma + SQLite scaffold - W26: Lead 模型 + Crawler + 询价发布助手 (线索池 + 一键发布) - W29.x: MCP 集成 (Integration 模型 + connectMcp + publish_rfq 5 工具端到端跑通) - W30: 3 字段补充 (notes / quotationType / quotationDemand) - W31 P1: 自动填 enterpriseId / userId (Prisma companyUserId 字段 + get_app_identity tool + lead.service.ts publishArgs) - W31: detail 详情页 + admin UI 集成管理 (OpenAppListView + AuditLogListView) 关联: ~/Documents/02-平台开发/产品开发/F002-W31-交接文档.md GIT-AUDIT-REPORT-2026-06-24.md
447 lines
15 KiB
Python
Executable File
447 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
direv 管理后台 - 自动审核（完全自动版）
|
||
登录检测改用页面元素判断，不需要人工按键确认
|
||
|
||
用法：
|
||
python3 auto_audit.py # 审核所有待审核项
|
||
python3 auto_audit.py --dry-run # 仅列出，不操作
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import sys
|
||
from datetime import datetime
|
||
from playwright.async_api import async_playwright
|
||
|
||
ADMIN_URL = "https://www.direv.com.cn/vue/#/"
|
||
ADMIN_LOGIN = "https://www.direv.com.cn/vue/#/login"
|
||
AUTH_FILE = "/tmp/direv-auth.json"
|
||
|
||
|
||
def log(msg: str):
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)
|
||
|
||
|
||
async def try_auth_from_file(page):
|
||
"""尝试从 auth 文件恢复登录态（注入 localStorage），返回 True 如果成功"""
|
||
if not os.path.exists(AUTH_FILE):
|
||
return False
|
||
try:
|
||
with open(AUTH_FILE) as f:
|
||
auth_data = json.load(f)
|
||
# 先访问站点，然后注入 localStorage
|
||
await page.goto(ADMIN_URL, wait_until="domcontentloaded")
|
||
await page.wait_for_timeout(2000)
|
||
for origin_entry in auth_data.get("origins", []):
|
||
if origin_entry.get("origin") == "https://www.direv.com.cn":
|
||
for item in origin_entry.get("localStorage", []):
|
||
key = item["name"]
|
||
val = item["value"]
|
||
await page.evaluate(f"localStorage.setItem('{key}', '{val}')")
|
||
log("✅ 已注入 localStorage（token + userInfo）")
|
||
break
|
||
# 重新导航以应用 localStorage
|
||
await page.goto(ADMIN_URL, wait_until="domcontentloaded")
|
||
await page.wait_for_timeout(4000)
|
||
# 检查是否在管理后台
|
||
has_admin = await page.query_selector(
|
||
".el-menu, [class*=sidebar], [class*=nav-bar], "
|
||
"span:has-text('数据统计'), span:has-text('询价管理')"
|
||
)
|
||
if has_admin:
|
||
log("✅ auth 文件登录成功！")
|
||
return True
|
||
log("⚠️ auth 文件注入后未能进入管理后台")
|
||
return False
|
||
except Exception as e:
|
||
log(f"⚠️ auth 文件恢复失败: {e}")
|
||
return False
|
||
|
||
|
||
async def wait_for_login(page, timeout=120):
|
||
"""
|
||
打开登录页，等用户登录。通过检测页面元素判断登录成功。
|
||
超时 120 秒。
|
||
"""
|
||
await page.goto(ADMIN_LOGIN, wait_until="domcontentloaded")
|
||
await page.wait_for_timeout(2000)
|
||
log("🔑 已打开 direv 管理后台登录页，浏览器窗口可见，请登录...")
|
||
|
||
# 轮询等待：URL 不再包含 login，或侧边栏出现
|
||
start = asyncio.get_event_loop().time()
|
||
while True:
|
||
await page.wait_for_timeout(3000)
|
||
url = page.url
|
||
has_login_url = "login" in url.lower()
|
||
|
||
# 检查是否有登录表单
|
||
has_login_form = await page.query_selector(
|
||
"input[placeholder*=手机], input[placeholder*=验证码], "
|
||
"button:has-text('登录'), button:has-text('登 录'), "
|
||
".login-form, [class*=login-form]"
|
||
)
|
||
|
||
# 检查是否出现了管理后台特有的元素
|
||
has_admin_elements = await page.query_selector(
|
||
".el-menu, [class*=sidebar], [class*=nav-bar], "
|
||
"span:has-text('数据统计'), span:has-text('询价管理'), "
|
||
"span:has-text('基础设置')"
|
||
)
|
||
|
||
elapsed = asyncio.get_event_loop().time() - start
|
||
|
||
if has_admin_elements:
|
||
log("✅ 检测到管理后台界面 - 登录成功！")
|
||
return True
|
||
|
||
if not has_login_form and not has_login_url:
|
||
log("✅ URL 已跳转离开登录页 - 登录成功！")
|
||
await page.wait_for_timeout(3000)
|
||
return True
|
||
|
||
if elapsed > timeout:
|
||
log("❌ 登录超时")
|
||
return False
|
||
|
||
# 进度提示
|
||
if int(elapsed) % 30 == 0 and elapsed > 0:
|
||
log(f" ⏳ 等待登录中... ({int(elapsed)}秒)")
|
||
|
||
|
||
async def navigate_to_inquiry_list(page):
|
||
"""导航到询价列表页"""
|
||
# 确保在管理后台首页
|
||
await page.goto(ADMIN_URL, wait_until="domcontentloaded")
|
||
await page.wait_for_timeout(3000)
|
||
|
||
# 找侧边栏中的「询价管理」点击展开
|
||
clicked = False
|
||
for el in await page.query_selector_all(".el-submenu__title, li.el-submenu > div, [class*=submenu] > div"):
|
||
try:
|
||
t = ((await el.text_content()) or "").strip()
|
||
if "询价管理" in t:
|
||
await el.click()
|
||
clicked = True
|
||
await page.wait_for_timeout(2500)
|
||
log(" → 展开询价管理")
|
||
break
|
||
except:
|
||
pass
|
||
|
||
if not clicked:
|
||
log("❌ 未找到「询价管理」菜单")
|
||
return False
|
||
|
||
# 点击「询价列表」— 通常在展开后的内联菜单中
|
||
await page.wait_for_timeout(1500)
|
||
|
||
# Element UI 展开的子菜单项
|
||
selectors = [
|
||
".el-menu--inline .el-menu-item",
|
||
"li.is-active.el-menu-item",
|
||
".el-menu-item",
|
||
"li[role=menuitem]",
|
||
]
|
||
|
||
for sel in selectors:
|
||
for el in await page.query_selector_all(sel):
|
||
try:
|
||
t = ((await el.text_content()) or "").strip()
|
||
if "询价列表" in t:
|
||
await el.click()
|
||
await page.wait_for_timeout(4000)
|
||
log(" → 进入询价列表")
|
||
return True
|
||
except:
|
||
pass
|
||
|
||
# 兜底：在所有元素中查找
|
||
for el in await page.query_selector_all("*"):
|
||
try:
|
||
t = ((await el.text_content()) or "").strip()
|
||
if t == "询价列表":
|
||
await el.click()
|
||
await page.wait_for_timeout(4000)
|
||
log(" → 进入询价列表")
|
||
return True
|
||
except:
|
||
pass
|
||
|
||
log("❌ 未找到「询价列表」")
|
||
return False
|
||
|
||
|
||
async def set_page_size(page, size=100):
|
||
"""调整分页器每页显示条数
|
||
|
||
页面结构（底部 pagination bar）：
|
||
左侧：Total 169（共169条）
|
||
右侧：小下拉框 10/page → 可选 100/page
|
||
对应 Element UI 结构：.el-pagination > .el-pagination__sizes > .el-select
|
||
"""
|
||
await page.wait_for_timeout(2000)
|
||
size_texts = [f"{size}条/页", f"{size} / 页", f"{size}/page", str(size)]
|
||
|
||
try:
|
||
size_selector = None
|
||
|
||
# 策略1：Element UI 标准选择器
|
||
size_selector = await page.query_selector(
|
||
".el-pagination__sizes .el-select .el-select__wrapper"
|
||
) or await page.query_selector(
|
||
".el-pagination__sizes .el-select"
|
||
) or await page.query_selector(
|
||
".el-pagination .el-select"
|
||
)
|
||
|
||
# 策略2：按文本内容找包含 "条/页" 或 "/page" 的 select/input
|
||
if not size_selector:
|
||
for el in await page.query_selector_all(".el-pagination *"):
|
||
try:
|
||
t = ((await el.text_content()) or "").strip()
|
||
if any(kw in t for kw in ("条/页", "/page", "条/页")):
|
||
# 往上找可点击的 select 容器
|
||
parent = await el.evaluate("""(el) => {
|
||
let p = el;
|
||
while (p) {
|
||
if (p.classList.contains('el-select') ||
|
||
p.classList.contains('el-select__wrapper') ||
|
||
p.tagName === 'INPUT') return p;
|
||
p = p.parentElement;
|
||
}
|
||
return el;
|
||
}""")
|
||
size_selector = el
|
||
break
|
||
except:
|
||
pass
|
||
|
||
if not size_selector:
|
||
log(" ⚠️ 未找到每页条数选择器")
|
||
return False
|
||
|
||
# 点击打开下拉
|
||
await size_selector.click()
|
||
await page.wait_for_timeout(1000)
|
||
|
||
# 策略3：在下拉列表中找目标条数
|
||
target_found = False
|
||
for el in await page.query_selector_all(
|
||
".el-select-dropdown__item, .el-select-dropdown li, "
|
||
".el-popper li, [class*=dropdown] li, .el-scrollbar__view li, "
|
||
".el-select-dropdown__wrap li, .el-select-dropdown__list li"
|
||
):
|
||
t = ((await el.text_content()) or "").strip()
|
||
for target in size_texts:
|
||
if t == target or t.startswith(target):
|
||
await el.click()
|
||
target_found = True
|
||
await page.wait_for_timeout(2000)
|
||
log(f" → 已调整每页显示 {size} 条")
|
||
return True
|
||
|
||
if not target_found:
|
||
log(f" ⚠️ 下拉中未找到 {size} 选项，保持默认")
|
||
await page.keyboard.press("Escape")
|
||
|
||
return target_found
|
||
except Exception as e:
|
||
log(f" ⚠️ 调整每页条数失败: {e}")
|
||
return False
|
||
|
||
|
||
async def get_table_data(page):
|
||
"""获取询价列表所有行数据"""
|
||
await page.wait_for_timeout(3000)
|
||
rows = await page.query_selector_all(
|
||
".el-table__body-wrapper tbody tr, .el-table__row, tbody tr"
|
||
)
|
||
|
||
data = []
|
||
for i, row in enumerate(rows):
|
||
cells = await row.query_selector_all("td, th")
|
||
texts = []
|
||
for cell in cells:
|
||
t = ((await cell.text_content()) or "").strip()
|
||
texts.append(t)
|
||
if texts:
|
||
data.append({"index": i, "row": row, "cells": texts})
|
||
return data
|
||
|
||
|
||
def find_targets(table_data):
|
||
"""筛选：发布人=用户6727740, 审核状态=待审核, 询价状态=进行中，按询价单号去重"""
|
||
seen_orders = set()
|
||
targets = []
|
||
for item in table_data:
|
||
s = " ".join(item["cells"])
|
||
if "用户6727740" in s and "待审核" in s and "进行中" in s:
|
||
# 提取询价单号去重 (X开头的编号)
|
||
order_no = None
|
||
for c in item["cells"]:
|
||
if c.startswith("X") and len(c) > 10:
|
||
order_no = c
|
||
break
|
||
if order_no and order_no in seen_orders:
|
||
continue
|
||
if order_no:
|
||
seen_orders.add(order_no)
|
||
targets.append(item)
|
||
return targets
|
||
|
||
|
||
async def audit_row(page, item, idx, total):
|
||
"""审核单条记录"""
|
||
row = item["row"]
|
||
cells = item["cells"]
|
||
title = cells[4] if len(cells) > 4 else "?"
|
||
log(f"\n[{idx+1}/{total}] {title[:50]}")
|
||
|
||
# 找到操作列中「审核」红色文字链接
|
||
# 操作列的审核是红色文字链接，非标准 button
|
||
audit_btn = None
|
||
for el in await row.query_selector_all("td:last-child *"):
|
||
try:
|
||
if ((await el.text_content()) or "").strip() == "审核":
|
||
audit_btn = el
|
||
break
|
||
except:
|
||
pass
|
||
|
||
# 兜底：在整行中找
|
||
if not audit_btn:
|
||
for el in await row.query_selector_all("span, a, button, td *"):
|
||
try:
|
||
if ((await el.text_content()) or "").strip() == "审核":
|
||
audit_btn = el
|
||
break
|
||
except:
|
||
pass
|
||
|
||
if not audit_btn:
|
||
log(" ⚠️ 未找到审核链接")
|
||
return False
|
||
|
||
# 用 evaluate 强制点击（红色文字链接可能不响应普通 click）
|
||
try:
|
||
await audit_btn.evaluate("el => el.click()")
|
||
except:
|
||
await audit_btn.click(force=True)
|
||
await page.wait_for_timeout(2500)
|
||
|
||
# 弹窗标题「更换管理员」，选「通过」radio，点「提交」
|
||
dialog = await page.query_selector(".el-dialog__wrapper:not([style*='display: none']), .el-dialog:not([style*='display: none'])")
|
||
if not dialog:
|
||
# 找最后出现的对话框
|
||
dialogs = await page.query_selector_all(".el-dialog, .el-dialog__wrapper, [role=dialog]")
|
||
dialog = dialogs[-1] if dialogs else None
|
||
|
||
if not dialog:
|
||
log(" ⚠️ 未找到审核弹窗")
|
||
return False
|
||
|
||
# 在弹窗中选「通过」（Element UI radio，通常 label 包裹 radio input）
|
||
for el in await dialog.query_selector_all(".el-radio, .el-radio__label, label"):
|
||
try:
|
||
t = ((await el.text_content()) or "").strip()
|
||
if t == "通过":
|
||
await el.click()
|
||
await page.wait_for_timeout(500)
|
||
break
|
||
except:
|
||
pass
|
||
|
||
# 点提交按钮
|
||
for btn in await dialog.query_selector_all("button, .el-button"):
|
||
try:
|
||
t = ((await btn.text_content()) or "").strip()
|
||
if t in ("提交", "提 交"):
|
||
await btn.click()
|
||
log(" ✅ 审核通过")
|
||
await page.wait_for_timeout(3500)
|
||
return True
|
||
except:
|
||
pass
|
||
|
||
log(" ⚠️ 未找到提交按钮")
|
||
await page.screenshot(path="/tmp/audit_dialog_err.png")
|
||
return False
|
||
|
||
|
||
async def main():
|
||
dry_run = "--dry-run" in sys.argv
|
||
|
||
async with async_playwright() as p:
|
||
browser = await p.chromium.launch(headless=False)
|
||
ctx = await browser.new_context(viewport={"width": 1440, "height": 900})
|
||
page = await ctx.new_page()
|
||
page.set_default_timeout(15000)
|
||
|
||
# 1. 登录：先尝试 auth 文件，失败再走手动登录
|
||
ok = await try_auth_from_file(page)
|
||
if not ok:
|
||
ok = await wait_for_login(page, timeout=180)
|
||
if not ok:
|
||
log("❌ 登录失败或超时")
|
||
await browser.close()
|
||
return
|
||
|
||
# 2. 导航到询价列表
|
||
ok = await navigate_to_inquiry_list(page)
|
||
if not ok:
|
||
await browser.close()
|
||
return
|
||
|
||
# 2.5. 调整每页显示 100 条
|
||
await set_page_size(page, 100)
|
||
|
||
# 3. 获取表格数据
|
||
table_data = await get_table_data(page)
|
||
log(f"\n📊 询价列表共 {len(table_data)} 行\n")
|
||
|
||
for item in table_data:
|
||
short = " | ".join(c[:16] for c in item["cells"][:10])
|
||
log(f" [{item['index']}] {short[:130]}")
|
||
|
||
# 4. 筛选
|
||
targets = find_targets(table_data)
|
||
|
||
if not targets:
|
||
log("\n✅ 无待审核项！")
|
||
await browser.close()
|
||
return
|
||
|
||
log(f"\n🎯 找到 {len(targets)} 条待审核（发布人=用户6727740 + 待审核 + 进行中）")
|
||
|
||
if dry_run:
|
||
for t in targets:
|
||
title = t["cells"][4] if len(t["cells"]) > 4 else "?"
|
||
log(f" → {title[:50]}")
|
||
log("\n⬆️ --dry-run 模式")
|
||
await browser.close()
|
||
return
|
||
|
||
# 5. 逐条审核
|
||
success = 0
|
||
for i, item in enumerate(targets):
|
||
try:
|
||
ok = await audit_row(page, item, i, len(targets))
|
||
if ok:
|
||
success += 1
|
||
except Exception as e:
|
||
log(f" ❌ {e}")
|
||
try:
|
||
await page.screenshot(path=f"/tmp/audit_err_{i}.png")
|
||
except:
|
||
pass
|
||
|
||
log(f"\n{'='*30}")
|
||
log(f"✅ 审核通过: {success} | ❌ 失败: {len(targets) - success}")
|
||
await browser.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|