62 lines
3.7 KiB
Python
62 lines
3.7 KiB
Python
"""Explicit first-send validation inside the main application, on a background worker."""
|
|
import json
|
|
from pathlib import Path
|
|
import threading
|
|
import time
|
|
from PySide6.QtWidgets import QInputDialog, QMessageBox
|
|
from runtime_paths import application_data_dir
|
|
|
|
|
|
def start_validation(host):
|
|
listening = getattr(host, '_thread', None)
|
|
if getattr(host, '_protocol_probe_running', False) or (listening is not None and listening.is_alive()):
|
|
host.append_log('请先停止监听并等待环境检测结束，再验证测试会话。', 'warn')
|
|
return
|
|
report = dict(getattr(host, '_protocol_report', {}) or {})
|
|
if report.get('status') != 'native_validation_required' or not report.get('loginVerified') or not report.get('databaseReady'):
|
|
host.append_log('请先检测本机协议；只有账号和消息库核对通过后才能验证发送。', 'warn')
|
|
host._refresh_protocol_probe()
|
|
return
|
|
name, accepted = QInputDialog.getText(host, '首次发送验证', '输入当前账号已有的专用测试单聊完整名称：')
|
|
name = str(name or '').strip()
|
|
if not accepted or not name:
|
|
return
|
|
response = QMessageBox.question(host, '核对测试消息',
|
|
f'当前账号：{report["nativeAccount"]}\n测试会话：{name}\n\n发送内容：协议通道测试，请忽略\n\n只发送这一条；取得原账号服务器回执后启用本账号的协议回复。',
|
|
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, QMessageBox.StandardButton.No)
|
|
if response != QMessageBox.StandardButton.Yes:
|
|
return
|
|
# Modal dialogs run a nested Qt event loop; a timer or another action may have changed state.
|
|
listening = getattr(host, '_thread', None)
|
|
if getattr(host, '_protocol_probe_running', False) or (listening is not None and listening.is_alive()):
|
|
host.append_log('运行状态已变化，本次验证已取消。', 'warn')
|
|
return
|
|
host._protocol_probe_running = True
|
|
host._protocol_probe = {**report, 'status': 'checking', 'label': '正在验证测试发送', 'canSend': False,
|
|
'message': '正在发送指定测试消息并核对原账号回执，请勿重复操作…'}
|
|
host._push_console_state(force=True)
|
|
target = application_data_dir() / 'protocol-validation-reports' / f'main-{time.time_ns()}.json'
|
|
|
|
def worker():
|
|
try:
|
|
from wecom_native_verification import run_send_test
|
|
from wxwork_protocol_probe import probe_installed_client
|
|
status = run_send_test(['--protocol-send-test', name, '--report', str(target),
|
|
'--expected-client', report['clientVersion'], '--expected-account', report['nativeAccount'],
|
|
'--expected-sha256', report['clientSha256']])
|
|
result = json.loads(target.read_text(encoding='utf-8'))
|
|
fresh = probe_installed_client(initialize_database=False)
|
|
fresh['lastSendValidation'] = {'status': result.get('status'), 'reportPath': str(target)}
|
|
if status:
|
|
fresh['message'] = '首次发送未完成验证：' + str(result.get('error', '请核对结果')) + '。' + fresh.get('message', '')
|
|
else:
|
|
fresh['message'] = '指定测试会话已取得并核对原账号服务器回执。' + fresh.get('message', '')
|
|
except Exception as exc:
|
|
fresh = {**report, 'checkedAt': time.time(), 'canSend': False,
|
|
'message': '首次发送验证未完成：' + str(exc)[:400]}
|
|
try:
|
|
host.protocolProbeLoaded.emit(fresh)
|
|
except RuntimeError:
|
|
pass
|
|
threading.Thread(target=worker, name='protocol-first-send-validation', daemon=True).start()
|