提交 85ef6050 authored 作者: 陈泽健's avatar 陈泽健

feat(远程自动化部署): X86统信5.70 业务功能验证与授权验证脚本入库

- verify_business.py 8步业务验证脚本(教训L/M 修复版)+ verify_business_result.json 8/8通过存档
- phase4_run/phase4_verify/post_deploy_verify 授权闭环与复验脚本
- _verify_login/_diag_verify_apis 登录与4大接口实时复核脚本
- create_permgroup/diag_* 业务验证排障脚本(SKILL 教训L 引用)
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 460aa4e5
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""复验四大业务接口(对外、预定、运维集控、讯飞)在路由修改后是否仍正常"""
import sys, io, requests
if sys.stdout.encoding != 'utf-8':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
HOST = '192.168.5.70'
tests = [
('对外接口', f'https://{HOST}/exapi/message/getMsgPageList', '无效token'),
('预定系统接口', f'https://{HOST}/meetingV3/api/systemConfiguration/globalConfig?companyNumber=CN-SZ-00-0201', 'accessToken为空'),
('运维集控接口', f'https://{HOST}/monitor/api2/api/servermonitor/', '用户不存在'),
('讯飞转录接口', f'https://{HOST}/voice/api/iflytek/roommaster?company_id=1&user_id=8&company_secret=57d00f9f-020f-5f1f-b788-55fae843bceb&getall=1', '"success":true'),
]
all_pass = True
for name, url, expect in tests:
try:
r = requests.get(url, verify=False, timeout=10)
hit = expect in r.text
print(f'[{hit and "✅" or "❌"}] {name}: HTTP {r.status_code}')
print(f' 响应: {r.text[:150]}')
if not hit:
all_pass = False
except Exception as e:
print(f'[❌] {name} 异常: {e}')
all_pass = False
print(f'\n接口复验结果: {"全部通过 ✅" if all_pass else "有接口异常 ❌"}')
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""验证修复后 /api/system/login 能否正常登录,返回 JWT"""
import sys, io
import paramiko
from pathlib import Path
import json
if sys.stdout.encoding != 'utf-8':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('192.168.5.70', port=22, username='root', password='Ubains@2026', timeout=15)
HASH = '84ef2b3b9e6eea45541b1771efcef4e01e6c3b585b0a38f26054bb8e53f16043'
VERIFY_CODE = 'csba'
# 1. 抓取 datacode
stdin, stdout, stderr = client.exec_command('curl -s -i "http://172.17.0.1:8999/system/getVerifyCode"')
res = stdout.read().decode('utf-8', errors='ignore')
datacode = next((line.split(':', 1)[1].strip() for line in res.splitlines() if 'datacode:' in line), None)
print('=== datacode ===')
print(datacode or '(无 datacode)')
# 2. 带 datacode 登录
if datacode:
cmd = f'curl -s -i -X POST "http://172.17.0.1:8999/system/login?account=superadmin&password={HASH}&verifyCode={VERIFY_CODE}" -H "datacode: {datacode}" -H "Content-Type: application/json"'
stdin, stdout, stderr = client.exec_command(cmd)
out = stdout.read().decode('utf-8', errors='ignore')
print('=== 带 datacode 登录响应 ===')
print(out.strip())
else:
print('跳过 datacode 登录')
client.close()
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""业务验证 - 创建测试权限组(固定选择器 + 批量全选 + 绑定)"""
import sys, io, json, time
from pathlib import Path
from playwright.sync_api import sync_playwright
if sys.stdout.encoding != 'utf-8':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
CODE_DIR = Path(__file__).parent
CONFIG = json.loads((CODE_DIR / 'deploy_config.json').read_text(encoding='utf-8'))
HOST = CONFIG['server']['host']
ADMIN_URL = f'https://{HOST}/#/LoginAdmin'
VB = CONFIG['license']['verify_code']
SHOT_DIR = Path.home() / 'deploy_logs' / 'screenshots'
SHOT_DIR.mkdir(parents=True, exist_ok=True)
def all_frames(page):
return [page] + [f for f in page.frames if f != page.main_frame]
def snap(page, name):
p = SHOT_DIR / f"biz_{name}.png"
try:
page.screenshot(path=str(p))
print(f"[截图] {p}", flush=True)
except Exception as e:
print(f"[截图失败] {e}", flush=True)
def main():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True, args=[
'--ignore-certificate-errors', '--disable-web-security', '--no-sandbox'])
ctx = browser.new_context(ignore_https_errors=True)
page = ctx.new_page()
page.goto(ADMIN_URL, timeout=60000, wait_until='domcontentloaded')
page.wait_for_timeout(5000)
vis = page.locator('input:visible')
vis.nth(0).fill('admin')
vis.nth(1).fill('Ubains@13579')
vis.nth(2).fill(VB)
try:
page.locator('input[type="submit"]:visible').first.click(timeout=5000)
except Exception:
page.locator('button:has-text("登")').first.click(timeout=5000)
page.wait_for_timeout(6000)
snap(page, "login_filled")
# 导航权限管理
for f2 in all_frames(page):
try:
t = f2.locator('.el-submenu__title:has-text("系统管理"), .el-submenu__title:has-text("系 统管理")')
if t.count() > 0:
t.first.click()
break
except Exception:
pass
page.wait_for_timeout(2000)
clicked = False
for f2 in all_frames(page):
try:
m = f2.locator('text=权限管理')
for i in range(m.count()):
if m.nth(i).is_visible():
m.nth(i).click()
clicked = True
break
if clicked:
break
except Exception:
pass
print(f"导航权限管理: {clicked}", flush=True)
page.wait_for_timeout(3000)
snap(page, "perm_nav")
# 进入新增权限组页
for f2 in all_frames(page):
try:
b = f2.locator('button:has-text("添加")')
if b.count() > 0:
b.first.click()
print("已点击添加", flush=True)
break
except Exception:
pass
page.wait_for_timeout(4000)
snap(page, "perm_add")
# 填写权限组名称
for i, f2 in enumerate(all_frames(page)):
try:
# 查找包含"权限组名称"的表单项
html = f2.evaluate('''() => {
const out = [];
const walker = document.createTreeWalker(document.body, NodeFilter.SHOW_ELEMENT);
let node;
while ((node = walker.nextNode())) {
const ownText = Array.from(node.childNodes)
.filter(n => n.nodeType === 3).map(n => n.textContent.trim()).join('');
if (ownText.includes('权限组名称')) {
let p = node.parentElement;
let chain = [];
for (let k = 0; k < 5 && p; k++) {
chain.push(p.tagName + '.' + p.className.split(' ')[0]);
p = p.parentElement;
}
out.push('NODE: ' + node.tagName + ' class=' + node.className.split(' ')[0] + ' chain=' + chain.join(' <- '));
const fi = node.closest('.el-form-item, .el-form-item__label');
if (fi) {
const inp = fi.querySelector('input');
if (inp) {
out.push('INPUT: ' + inp.tagName + ' placeholder=' + (inp.placeholder || inp.getAttribute('placeholder')));
}
}
}
}
return out;
}''')
if html:
print(f"--- frame[{i}] ---")
for line in html:
print(' ' + line)
except Exception:
pass
# 填写表单
page.locator('input').first.fill('测试权限组')
snap(page, "perm_filled")
print("已填权限组名称: 测试权限组", flush=True)
# 批量全选复选框
checkboxes = page.locator('input[type="checkbox"]')
for i in range(checkboxes.count()):
try:
checkboxes.nth(i).check()
except Exception:
pass
print(f"已勾选 {checkboxes.count()} 个复选框", flush=True)
snap(page, "perm_checkboxes")
# 点击保存
for f2 in all_frames(page):
try:
b = f2.locator('button:has-text("保存"), button:has-text("确定")')
if b.count() > 0:
b.first.click()
print("已点击保存", flush=True)
break
except Exception:
pass
page.wait_for_timeout(5000)
snap(page, "perm_saved")
# 返回列表页,检查是否创建成功
page.reload(wait_until='domcontentloaded')
page.wait_for_timeout(5000)
for i, f2 in enumerate(all_frames(page)):
try:
txt = f2.locator('body').inner_text()[:500].replace('\n', ' | ')
if '测试权限组' in txt:
print(f"✅ 创建成功: 列表中出现 测试权限组", flush=True)
break
except Exception:
pass
snap(page, "perm_list")
browser.close()
if __name__ == '__main__':
main()
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""诊断权限组绑定页:角色搜索下拉框展开后的选项 DOM"""
import sys, io, json
from pathlib import Path
from playwright.sync_api import sync_playwright
if sys.stdout.encoding != 'utf-8':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
CODE_DIR = Path(__file__).parent
CONFIG = json.loads((CODE_DIR / 'deploy_config.json').read_text(encoding='utf-8'))
HOST = CONFIG['server']['host']
ADMIN_URL = f'https://{HOST}/#/LoginAdmin'
VB = CONFIG['license']['verify_code']
SHOT_DIR = Path.home() / 'deploy_logs' / 'screenshots'
def all_frames(page):
return [page] + [f for f in page.frames if f != page.main_frame]
def snap(page, name):
p = SHOT_DIR / f"diag_{name}.png"
try:
page.screenshot(path=str(p))
print(f"[截图] {p}", flush=True)
except Exception as e:
print(f"[截图失败] {e}", flush=True)
def main():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True, args=[
'--ignore-certificate-errors', '--disable-web-security', '--no-sandbox'])
ctx = browser.new_context(ignore_https_errors=True)
page = ctx.new_page()
page.goto(ADMIN_URL, timeout=60000, wait_until='domcontentloaded')
page.wait_for_timeout(5000)
vis = page.locator('input:visible')
vis.nth(0).fill('admin')
vis.nth(1).fill('Ubains@13579')
vis.nth(2).fill(VB)
try:
page.locator('input[type="submit"]:visible').first.click(timeout=5000)
except Exception:
page.locator('button:has-text("登")').first.click(timeout=5000)
page.wait_for_timeout(6000)
page.reload(wait_until='domcontentloaded')
page.wait_for_timeout(6000)
# 系统管理→权限管理
for f2 in all_frames(page):
try:
t = f2.locator('.el-submenu__title:has-text("系统管理"), .el-submenu__title:has-text("系 统管理")')
if t.count() > 0:
t.first.click()
break
except Exception:
pass
page.wait_for_timeout(2000)
for f2 in all_frames(page):
try:
m = f2.locator('text=权限管理')
done = False
for i in range(m.count()):
if m.nth(i).is_visible():
m.nth(i).click()
done = True
break
if done:
break
except Exception:
pass
page.wait_for_timeout(3000)
# 点击测试权限组行的绑定
for f2 in all_frames(page):
try:
for sel in ['.el-table__fixed-right tr:has-text("测试权限组") .el-tag:has-text("绑定")',
'tr:has-text("测试权限组") .el-tag:has-text("绑定")']:
b = f2.locator(sel)
if b.count() > 0:
b.first.click()
print("已点击绑定", flush=True)
break
else:
continue
break
except Exception:
pass
page.wait_for_timeout(3000)
# 点击"搜索角色"下拉框(el-select 容器,输入框 readonly 点击无效)
clicked = False
for f2 in all_frames(page):
try:
sel = f2.locator('.el-select:has(input[placeholder="搜索角色"])')
if sel.count() > 0:
sel.first.click()
clicked = True
print("已点击搜索角色下拉框(el-select)", flush=True)
break
except Exception:
pass
print(f"点击角色下拉框: {clicked}", flush=True)
page.wait_for_timeout(2500)
snap(page, "bind_role_dropdown")
# 抓取所有可见下拉层/列表选项 DOM
for i, f2 in enumerate(all_frames(page)):
try:
info = f2.evaluate('''() => {
const out = {};
// el-select-dropdown
out.drops = Array.from(document.querySelectorAll('.el-select-dropdown')).map(d => ({
visible: getComputedStyle(d).display !== 'none' && d.offsetParent !== null,
items: Array.from(d.querySelectorAll('.el-select-dropdown__item')).map(it => it.innerText.trim())
}));
// 其他 popper
out.poppers = Array.from(document.querySelectorAll('.el-popover, .el-tooltip__popper, [x-placement]')).map(d => ({
cls: d.className.slice(0, 80), visible: d.offsetParent !== null,
text: d.innerText.replace(/\\n/g, ' | ').slice(0, 200)
})).filter(x => x.visible);
// 搜索角色附近容器 HTML
const inp = document.querySelector('input[placeholder="搜索角色"]');
if (inp) {
let box = inp.closest('.el-select') || inp.parentElement;
out.selectHTML = box ? box.outerHTML.slice(0, 1200) : null;
}
return out;
}''')
if info and (info.get('drops') or info.get('poppers') or info.get('selectHTML')):
print(f"--- frame[{i}] ---", flush=True)
print(json.dumps(info, ensure_ascii=False, indent=1), flush=True)
except Exception:
pass
browser.close()
if __name__ == '__main__':
main()
此差异已折叠。
此差异已折叠。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""诊断新增权限组页面的表单 DOM 结构"""
import sys, io, json
from pathlib import Path
from playwright.sync_api import sync_playwright
if sys.stdout.encoding != 'utf-8':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
CODE_DIR = Path(__file__).parent
CONFIG = json.loads((CODE_DIR / 'deploy_config.json').read_text(encoding='utf-8'))
HOST = CONFIG['server']['host']
ADMIN_URL = f'https://{HOST}/#/LoginAdmin'
VB = CONFIG['license']['verify_code']
def all_frames(page):
return [page] + [f for f in page.frames if f != page.main_frame]
def main():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True, args=[
'--ignore-certificate-errors', '--disable-web-security', '--no-sandbox'])
ctx = browser.new_context(ignore_https_errors=True)
page = ctx.new_page()
page.goto(ADMIN_URL, timeout=60000, wait_until='domcontentloaded')
page.wait_for_timeout(5000)
vis = page.locator('input:visible')
vis.nth(0).fill('admin')
vis.nth(1).fill('Ubains@13579')
vis.nth(2).fill(VB)
try:
page.locator('input[type="submit"]:visible').first.click(timeout=5000)
except Exception:
page.locator('button:has-text("登")').first.click(timeout=5000)
page.wait_for_timeout(6000)
page.reload(wait_until='domcontentloaded')
page.wait_for_timeout(6000)
# 导航 系统管理 → 权限管理
for f2 in all_frames(page):
try:
t = f2.locator('.el-submenu__title:has-text("系统管理"), .el-submenu__title:has-text("系 统管理")')
if t.count() > 0:
t.first.click()
break
except Exception:
pass
page.wait_for_timeout(2000)
clicked = False
for f2 in all_frames(page):
try:
m = f2.locator('text=权限管理')
for i in range(m.count()):
if m.nth(i).is_visible():
m.nth(i).click()
clicked = True
break
if clicked:
break
except Exception:
pass
print(f"导航权限管理: {clicked}", flush=True)
page.wait_for_timeout(3000)
# 点击添加
for f2 in all_frames(page):
try:
b = f2.locator('button:has-text("添加")')
if b.count() > 0:
b.first.click()
print("已点击添加", flush=True)
break
except Exception:
pass
page.wait_for_timeout(4000)
# 在所有 frame 中查找 label 权限组名称 附近的 HTML
for i, f2 in enumerate(all_frames(page)):
try:
html = f2.evaluate('''() => {
const out = [];
// 1) 所有 input 及其 placeholder/maxlength
document.querySelectorAll('input').forEach(inp => {
if (inp.offsetParent !== null) {
out.push('INPUT: type=' + inp.type + ' placeholder=' + inp.placeholder + ' maxlength=' + inp.maxLength);
}
});
// 2) 包含"权限组名称"文本的元素及其父级结构
const walker = document.createTreeWalker(document.body, NodeFilter.SHOW_ELEMENT);
let node;
while ((node = walker.nextNode())) {
const ownText = Array.from(node.childNodes)
.filter(n => n.nodeType === 3).map(n => n.textContent.trim()).join('');
if (ownText.includes('权限组名称')) {
let p = node.parentElement;
let chain = [];
for (let k = 0; k < 4 && p; k++) { chain.push(p.tagName + '.' + p.className); p = p.parentElement; }
out.push('LABEL_NODE: <' + node.tagName + '> class=' + node.className + ' parents=' + chain.join(' <- '));
const fi = node.closest('.el-form-item');
if (fi) {
const inp = fi.querySelector('input');
out.push('FORM_ITEM_INPUT: ' + (inp ? inp.placeholder : 'null') + ' formItemClass=' + fi.className);
} else {
out.push('NO .el-form-item ancestor');
}
}
}
return out;
}''')
if html:
print(f"--- frame[{i}] name={getattr(f2, 'name', 'page')} ---", flush=True)
for line in html:
print(f" {line}", flush=True)
except Exception as e:
pass
browser.close()
if __name__ == '__main__':
main()
\ No newline at end of file
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""诊断 admin 首次登录强制改密流程(分步截图+状态输出)"""
import sys, io, time, json
from pathlib import Path
from playwright.sync_api import sync_playwright
if sys.stdout.encoding != 'utf-8':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
CODE_DIR = Path(__file__).parent
CONFIG = json.loads((CODE_DIR / 'deploy_config.json').read_text(encoding='utf-8'))
HOST = CONFIG['server']['host']
ADMIN_URL = f'https://{HOST}/#/LoginAdmin'
VB = CONFIG['license']['verify_code']
SHOT_DIR = Path.home() / 'deploy_logs' / 'screenshots'
SHOT_DIR.mkdir(parents=True, exist_ok=True)
def snap(page, name):
p = SHOT_DIR / f"diag_{name}.png"
try:
page.screenshot(path=str(p))
print(f"[截图] {p}", flush=True)
except Exception as e:
print(f"[截图失败] {e}", flush=True)
def dump_all_frames(page, tag):
print(f"\n===== {tag} =====", flush=True)
print(f"url: {page.url}", flush=True)
for i, fr in enumerate([page] + [f for f in page.frames if f != page.main_frame]):
try:
txt = fr.locator('body').inner_text(timeout=2000)[:600].replace('\n', ' | ')
print(f"--- frame[{i}] name={fr.name} text: {txt}", flush=True)
except Exception as e:
print(f"--- frame[{i}] 读取失败: {e}", flush=True)
# 所有可见对话框
for i, fr in enumerate([page] + [f for f in page.frames if f != page.main_frame]):
try:
dlgs = fr.locator('.el-dialog:visible')
print(f"frame[{i}] 可见对话框数: {dlgs.count()}", flush=True)
for d in range(dlgs.count()):
dlg = dlgs.nth(d)
dtxt = dlg.inner_text()[:200].replace('\n', ' | ')
print(f" 对话框[{d}]: {dtxt}", flush=True)
pws = dlg.locator('input[type="password"]')
print(f" 密码框数: {pws.count()}", flush=True)
except Exception:
pass
# message 提示
try:
msgs = page.evaluate('Array.from(document.querySelectorAll(".el-message")).map(e=>e.innerText)')
print(f"el-message: {msgs}", flush=True)
except Exception:
pass
# messagebox
for i, fr in enumerate([page] + [f for f in page.frames if f != page.main_frame]):
try:
mbs = fr.locator('.el-message-box:visible')
for m in range(mbs.count()):
print(f"frame[{i}] messagebox[{m}]: {mbs.nth(m).inner_text()[:200]}", flush=True)
except Exception:
pass
def main():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True, args=[
'--ignore-certificate-errors', '--disable-web-security', '--no-sandbox'])
ctx = browser.new_context(ignore_https_errors=True)
page = ctx.new_page()
page.goto(ADMIN_URL, timeout=60000, wait_until='domcontentloaded')
page.wait_for_timeout(5000)
vis = page.locator('input:visible')
print(f"登录页可见输入框: {vis.count()}", flush=True)
if vis.count() >= 3:
vis.nth(0).fill('admin')
vis.nth(1).fill('Ubains@1357')
vis.nth(2).fill(VB)
snap(page, "pwd_login_filled")
# 提交登录
clicked = False
try:
sub = page.locator('input[type="submit"]:visible')
if sub.count() > 0:
sub.first.click(timeout=5000)
clicked = True
except Exception:
pass
if not clicked:
for sel in ['button:has-text("登 录")', 'button:has-text("登")']:
try:
loc = page.locator(sel)
for i in range(loc.count()):
if loc.nth(i).is_visible():
loc.nth(i).click(timeout=5000)
clicked = True
break
if clicked:
break
except Exception:
continue
print(f"登录按钮点击: {clicked}", flush=True)
page.wait_for_timeout(6000)
dump_all_frames(page, "登录提交后6秒")
snap(page, "pwd_after_login")
# 若仍处于登录页尝试识别错误消息并再点一次
on_login = 'Login' in page.url and page.locator('input[type="password"]:visible').count() > 0
print(f"仍在登录页? {on_login}", flush=True)
if not on_login:
page.wait_for_timeout(3000)
dump_all_frames(page, "登录后9秒")
snap(page, "pwd_after_login9s")
page.reload(wait_until='domcontentloaded')
page.wait_for_timeout(6000)
dump_all_frames(page, "刷新后6秒")
snap(page, "pwd_after_reload")
browser.close()
if __name__ == '__main__':
main()
\ No newline at end of file
......@@ -138,14 +138,15 @@ def main():
try:
# ========== 步骤2: 访问维护平台 ==========
log("\n[步骤2] 访问维护平台...")
page.goto(MAINTENANCE_URL, timeout=60000, wait_until='networkidle')
# networkidle 因后台长轮询请求永远无法达成,改用 domcontentloaded + 选择器等待
page.goto(MAINTENANCE_URL, timeout=60000, wait_until='domcontentloaded')
log(f"页面标题: {page.title()}")
save_screenshot(page, "01_maintenance_platform")
# ========== 步骤3: 登录 ==========
log("\n[步骤3] 登录超管账户...")
# 等待页面完全加载
page.wait_for_load_state('networkidle', timeout=60000)
page.wait_for_load_state('domcontentloaded', timeout=60000)
time.sleep(5) # 额外等待Vue渲染
# 尝试多种选择器找登录表单
......@@ -266,7 +267,7 @@ def main():
if file_input.count() > 0:
file_input.first.set_input_files(LICENSE_FILE)
log(f"✓ 授权文件已选择(直接设置): {LICENSE_FILE}")
page.wait_for_load_state('networkidle', timeout=60000)
page.wait_for_load_state('domcontentloaded', timeout=60000)
time.sleep(5)
log("✓ 授权文件上传成功")
save_screenshot(page, "04_upload_license")
......@@ -284,7 +285,7 @@ def main():
if file_input.count() > 0:
file_input.first.set_input_files(LICENSE_FILE)
log(f"✓ 授权文件已选择(JS可见后设置): {LICENSE_FILE}")
page.wait_for_load_state('networkidle', timeout=60000)
page.wait_for_load_state('domcontentloaded', timeout=60000)
time.sleep(5)
save_screenshot(page, "04_upload_license")
else:
......@@ -302,7 +303,7 @@ def main():
log("说明:需要进入'服务升级'页面,勾选服务并重启")
# 刷新页面确保菜单完整渲染
page.reload(wait_until='networkidle')
page.reload(wait_until='domcontentloaded')
time.sleep(3)
save_screenshot(page, "05_page_refreshed")
......@@ -317,7 +318,7 @@ def main():
service_upgrade = page.locator('text=服务升级')
if service_upgrade.count() > 0:
service_upgrade.click()
page.wait_for_load_state('networkidle', timeout=30000)
page.wait_for_load_state('domcontentloaded', timeout=30000)
log("已进入'服务升级'页面")
save_screenshot(page, "06_service_upgrade_page")
......
此差异已折叠。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
授权后等待10分钟 + 复验脚本(已知新密码已改过)
"""
import time
import json
from pathlib import Path
import urllib.request
import ssl
import sys
sys.stdout.reconfigure(encoding='utf-8')
CODE_DIR = Path("E:\\github\\ubains-module-test\\develop\\.claude\\skills\\X86-TX-XTYBS\\code")
CONFIG_FILE = CODE_DIR / 'deploy_config.json'
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
config = json.load(f)
BASE_URL = f"https://{config['server']['host']}"
RESTART_TS = '2026-09-08 17:09:59' # 实际重启时间
INTERFACES = [
{'name': '对外接口', 'url': f'{BASE_URL}/exapi/message/getMsgPageList', 'ok_keywords': ['无效token', 'Full authentication is required'], 'fail_keywords': ['nginx Error']},
{'name': '运维集控接口', 'url': f'{BASE_URL}/monitor/api2/api/servermonitor/', 'ok_keywords': ['用户不存在'], 'fail_keywords': ['nginx Error']},
{'name': '讯飞转录接口', 'url': f'{BASE_URL}/voice/api/iflytek/roommaster?company_id=1&user_id=8&company_secret=57d00f9f-020f-5f1f-b788-55fae843bceb&getall=1', 'ok_keywords': ['缺少关键参数', '"success":true'], 'fail_keywords': ['nginx Error']},
{'name': '预定系统接口', 'url': f'{BASE_URL}/meetingV3/api/systemConfiguration/globalConfig?companyNumber=CN-SZ-00-0201', 'ok_keywords': ['accessToken为空'], 'fail_keywords': ['内部服务器错误']},
]
MAX_RETRY = 5
RETRY_INTERVAL = 30
CONFIRM_WAIT = 30
SSL_CTX = ssl.create_default_context()
SSL_CTX.check_hostname = False
SSL_CTX.verify_mode = ssl.CERT_NONE
def log(msg, level='INFO'):
ts = time.strftime('%Y-%m-%d %H:%M:%S')
print(f"[{ts}] [{level}] {msg}")
def http_get(url, timeout=20):
try:
req = urllib.request.Request(url, headers={'User-Agent': 'deploy-verify/1.0'})
with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp:
return resp.status, resp.read().decode('utf-8', 'replace')
except Exception as e:
return 0, str(e)
def wait_for_startup():
import datetime
restart = datetime.datetime.strptime(RESTART_TS, '%Y-%m-%d %H:%M:%S')
target = restart + datetime.timedelta(minutes=12)
now = datetime.datetime.now()
remain = (target - now).total_seconds()
if remain > 0:
log(f"等待服务启动:距重启时刻+12分钟还有 {int(remain)} 秒...")
time.sleep(remain)
else:
log("已过等待期,直接开始复验")
def verify_one(iface):
name = iface['name']
for attempt in range(1, MAX_RETRY + 1):
code, body = http_get(iface['url'])
snippet = body[:150].replace('\n', ' ')
ok = any(k in body for k in iface['ok_keywords'])
log(f"[{name}] 第{attempt}次: HTTP {code} | {'✅ 命中成功标志' if ok else '❌ 未命中'} | {snippet}")
if ok:
time.sleep(CONFIRM_WAIT)
code2, body2 = http_get(iface['url'])
ok2 = any(k in body2 for k in iface['ok_keywords'])
if ok2:
return True, f"HTTP {code},二次确认通过"
if attempt < MAX_RETRY:
time.sleep(RETRY_INTERVAL)
return False, f"{MAX_RETRY}次重试均未命中成功标志(最后响应: {snippet})"
def main():
log("=" * 60)
log("阶段4授权后接口复验")
log("=" * 60)
wait_for_startup()
results = []
for iface in INTERFACES:
ok, note = verify_one(iface)
results.append((iface['name'], ok, note))
log(f"[结果] {iface['name']}: {'✅' if ok else '❌'} {note}")
log("\n" + "=" * 60)
log("复验汇总:")
ok_count = 0
for name, ok, note in results:
log(f" {'✅' if ok else '❌'} {name}: {note}")
if ok:
ok_count += 1
log(f" 通过 {ok_count}/{len(results)}")
out = CODE_DIR / 'phase4_verify_result.json'
with open(out, 'w', encoding='utf-8') as f:
json.dump({
'verify_time': time.strftime('%Y-%m-%d %H:%M:%S'),
'restart_ts': RESTART_TS,
'results': [{'name': n, 'ok': o, 'note': t} for n, o, t in results],
}, f, ensure_ascii=False, indent=2)
log(f"结果已保存: {out}")
return 0 if ok_count == len(results) else 1
if __name__ == '__main__':
sys.exit(main())
{
"verify_time": "2026-09-08 17:25:29",
"restart_ts": "2026-09-08 17:09:59",
"results": [
{
"name": "对外接口",
"ok": true,
"note": "HTTP 200,二次确认通过"
},
{
"name": "运维集控接口",
"ok": true,
"note": "HTTP 200,二次确认通过"
},
{
"name": "讯飞转录接口",
"ok": true,
"note": "HTTP 200,二次确认通过"
},
{
"name": "预定系统接口",
"ok": true,
"note": "HTTP 200,二次确认通过"
}
]
}
\ No newline at end of file
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
完成 root 改密后,进行完整部署流程验证:
1. docker ps 检查容器状态
2. 执行阶段3 验收检查(4大接口)
3. 运行 phase4_run.py 完成授权
4. 运行 phase4_verify.py 复验
5. 生成部署分析报告
"""
import paramiko
import os
import sys
import time
import json
from pathlib import Path
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
HOST = '192.168.5.70'
KEY_PATH = os.path.expanduser('~/.ssh/192.168.5.70/id_rsa')
PASSWORD = 'Ubains@2026'
CODE_DIR = Path("E:\\github\\ubains-module-test\\develop\\.claude\\skills\\X86-TX-XTYBS\\code")
def exec_cmd(client, cmd):
stdin, stdout, stderr = client.exec_command(cmd)
out = stdout.read().decode('utf-8', errors='ignore')
err = stderr.read().decode('utf-8', errors='ignore')
return out, err
def main():
key = paramiko.RSAKey.from_private_key_file(KEY_PATH)
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
print("[1] 用新密码认证 ...")
client.connect(HOST, port=22, username='root', password=PASSWORD, timeout=15)
print("[2] docker ps ...")
out, err = exec_cmd(client, 'docker ps --format "{{.Names}}\t{{.Status}}"')
print(out)
if 'Up' not in out:
print('>> 容器未全部正常,退出')
client.close()
return 1
print("[3] 执行阶段3 验收检查 ...")
out, err = exec_cmd(client, 'cd /data/offline_auto_unifiedPlatform && python -m verify_x86_api')
print(out)
print("[4] 执行阶段4 授权 ...")
out, err = exec_cmd(client, 'cd /data/offline_auto_unifiedPlatform && python -m phase4_run')
print(out)
print("[5] 执行阶段4 复验 ...")
out, err = exec_cmd(client, 'cd /data/offline_auto_unifiedPlatform && python -m phase4_verify')
print(out)
print("[6] 生成部署分析报告 ...")
out, err = exec_cmd(client, 'cd /data/offline_auto_unifiedPlatform && python -m generate_report')
print(out)
client.close()
return 0
if __name__ == '__main__':
sys.exit(main())
此差异已折叠。
{
"admin": {
"time": "2026-09-08 18:34:31",
"results": [
{
"step": "1.添加管理员",
"ok": true,
"note": "列表已出现精确 admin 记录,页面消息: []"
}
],
"all_ok": true
},
"pwd": {
"time": "2026-09-08 18:37:29",
"results": [
{
"step": "2.触发强制改密",
"ok": true,
"note": "检测到修改密码对话框"
},
{
"step": "2.改密后用新密码登录",
"ok": true,
"note": "新密码登录成功"
}
],
"all_ok": true
},
"permgroup": {
"time": "2026-09-08 18:39:01",
"results": [
{
"step": "3.填写权限组名称",
"ok": true,
"note": "已填 测试权限组"
},
{
"step": "3.创建权限组",
"ok": true,
"note": "列表出现 测试权限组"
},
{
"step": "3.点击绑定",
"ok": true,
"note": "已点击绑定"
},
{
"step": "3.点击角色下拉框",
"ok": true,
"note": "已点击角色下拉框"
},
{
"step": "3.选择角色",
"ok": true,
"note": "已选 公司管理员"
}
],
"all_ok": true
},
"user": {
"time": "2026-09-08 18:39:44",
"results": [
{
"step": "4.添加用户",
"ok": true,
"note": "列表出现 admin@test"
}
],
"all_ok": true
},
"dept": {
"time": "2026-09-08 18:40:31",
"results": [
{
"step": "5.新增部门",
"ok": true,
"note": "树形结构出现 默认部门名称"
}
],
"all_ok": true
},
"room": {
"time": "2026-09-08 18:41:24",
"results": [
{
"step": "6.填写会议室名称",
"ok": true,
"note": "已填 测试会议室"
},
{
"step": "6.选择预定授权",
"ok": true,
"note": "已选授权码"
},
{
"step": "6.新增会议室",
"ok": true,
"note": "列表出现 测试会议室"
}
],
"all_ok": true
},
"authcode": {
"time": "2026-09-08 18:42:21",
"results": [
{
"step": "7.批量启用授权码",
"ok": true,
"note": "授权码状态变为 已激活"
}
],
"all_ok": true
},
"meeting": {
"time": "2026-09-08 18:43:40",
"results": [
{
"step": "8.前台登录",
"ok": true,
"note": "前台已登录"
},
{
"step": "8.选择开会区域",
"ok": true,
"note": "已点击 选择开会区域"
},
{
"step": "8.勾选会议室",
"ok": true,
"note": "已勾选 测试会议室"
},
{
"step": "8.新建会议",
"ok": true,
"note": "弹出 会议创建成功 提示"
}
],
"all_ok": true
}
}
\ No newline at end of file
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论