128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""用 DB 里用户真实消息内容回放流式渲染，验证新路径是否正确。
|
|
每个 chunk 走独立 runJavaScript（模拟 Python 逐 token 推送的真实路径）。"""
|
|
import os, sys, json, sqlite3
|
|
# 不加 --disable-gpu：复现真实应用的 GPU 渲染环境
|
|
os.environ["QTWEBENGINE_CHROMIUM_FLAGS"] = os.environ.get("QTWEBENGINE_CHROMIUM_FLAGS", "")
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from PyQt6.QtWidgets import QApplication
|
|
import PyQt6.QtWebEngineWidgets # noqa
|
|
from PyQt6.QtCore import QTimer
|
|
from ui.views.main_window import MainWindow
|
|
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
PASS = 0
|
|
FAIL = 0
|
|
|
|
def check(name, cond, detail=""):
|
|
global PASS, FAIL
|
|
if cond:
|
|
PASS += 1
|
|
print(f" PASS {name}")
|
|
else:
|
|
FAIL += 1
|
|
print(f" FAIL {name} {detail}")
|
|
|
|
def esc(s):
|
|
return s.replace("\\", "\\\\").replace("'", "\\'").replace("\n", "\\n")
|
|
|
|
# 取用户真实消息
|
|
conn = sqlite3.connect(os.path.join(ROOT, "data", "chat_history.db"))
|
|
conn.row_factory = sqlite3.Row
|
|
row = conn.execute(
|
|
"SELECT content, timeline FROM messages WHERE id LIKE 'msg-99235c06%' "
|
|
"ORDER BY created_at DESC LIMIT 1").fetchone()
|
|
assert row, "DB 中找不到测试消息"
|
|
timeline = json.loads(row["timeline"] or "[]")
|
|
print(f"消息: content={len(row['content'])}c, timeline={len(timeline)} 条")
|
|
|
|
MID = "repro-real"
|
|
# 构造回放脚本序列：think/text 按 20 字符切块，每批 20 个 chunk 一次 JS 调用
|
|
CHUNK = 20
|
|
BATCH = 20
|
|
steps = [] # JS 片段列表
|
|
for e in timeline:
|
|
t = e.get("t")
|
|
txt = e.get("text", "")
|
|
if t in ("think", "text") and txt:
|
|
for i in range(0, len(txt), CHUNK):
|
|
fn = "appendReasoning" if t == "think" else "appendToken"
|
|
steps.append(f"{fn}('{MID}', '{esc(txt[i:i+CHUNK])}');")
|
|
steps_batched = [chr(10).join(steps[i:i+BATCH]) for i in range(0, len(steps), BATCH)]
|
|
|
|
app = QApplication(sys.argv)
|
|
window = MainWindow()
|
|
window.resize(1280, 800)
|
|
window.show()
|
|
|
|
state = {"idx": 0, "probed": False}
|
|
|
|
def next_batch():
|
|
"""一次送一批（20 个 chunk 拼接在一个 JS 任务里）"""
|
|
if state["idx"] >= len(steps_batched):
|
|
QTimer.singleShot(300, mid_probe)
|
|
return
|
|
bi = state["idx"]
|
|
js = steps_batched[bi]
|
|
total = len(steps_batched)
|
|
|
|
def done(r, bi=bi):
|
|
state["idx"] = bi + 1
|
|
# 过半时探针一次
|
|
if not state["probed"] and bi + 1 >= total // 2:
|
|
state["probed"] = True
|
|
window.browser.page().runJavaScript(
|
|
f"probeStream('{MID}')", on_mid_probe)
|
|
else:
|
|
QTimer.singleShot(0, next_batch)
|
|
window.browser.page().runJavaScript(js, done)
|
|
|
|
def on_mid_probe(res):
|
|
d = json.loads(str(res))
|
|
print(" [过半探针]", str(d)[:400])
|
|
check("中途: 缓冲已累积", any(s["b"] > 0 for s in d.get("segs", [])), str(d))
|
|
check("中途: DOM 已写入部分正文",
|
|
any(s["d"] > 0 for s in d.get("segs", [])), str(d))
|
|
check("中途: 思考 DOM 已写入",
|
|
any(t["d"] > 0 for t in d.get("thinks", [])), str(d))
|
|
finish_seq()
|
|
|
|
def mid_probe():
|
|
window.browser.page().runJavaScript(f"probeStream('{MID}')", on_mid_probe)
|
|
|
|
def finish_seq():
|
|
window.browser.page().runJavaScript(
|
|
f"finishMessage('{MID}'); probeStream('{MID}')", on_final)
|
|
|
|
def on_final(res):
|
|
d = json.loads(str(res))
|
|
print(" [最终探针]", str(d)[:400])
|
|
tot_b = sum(s["b"] for s in d.get("segs", []))
|
|
tot_d = sum(s["d"] for s in d.get("segs", []))
|
|
check("最终: 正文缓冲完整", tot_b > 500, f"b={tot_b}")
|
|
check("最终: 正文 DOM 完整", tot_d > 500, f"d={tot_d}")
|
|
check("最终: 思考 DOM 完整",
|
|
sum(t["d"] for t in d.get("thinks", [])) > 500,
|
|
str([t['d'] for t in d.get('thinks', [])]))
|
|
# DOM 实际包含关键子串
|
|
window.browser.page().runJavaScript(
|
|
"(function(){ var w = document.getElementById('" + MID + "');"
|
|
" return w ? w.textContent.length : -1; })()", on_text)
|
|
|
|
def on_text(res):
|
|
n = int(str(res) or 0)
|
|
check("DOM 总文本量正常", n > 1000, f"total={n}")
|
|
print(f"===== {'ALL PASS' if FAIL == 0 else 'HAS FAILURES'}: {PASS}/{PASS+FAIL} =====")
|
|
app.quit()
|
|
|
|
def start():
|
|
window.browser.page().runJavaScript(
|
|
f"createMessage('{MID}', 'assistant', '', 'Real');",
|
|
lambda r: QTimer.singleShot(200, next_batch))
|
|
|
|
QTimer.singleShot(1500, start)
|
|
app.exec()
|
|
sys.exit(1 if FAIL else 0)
|