chore: import original project baseline

Import the pre-repair source tree as the history baseline.
Runtime data (data/), virtualenvs, bytecode caches and logs are
gitignored so local secrets and user state stay out of the repo.
This commit is contained in:
2026-09-17 16:40:01 +08:00
commit a7412824e0
124 changed files with 26747 additions and 0 deletions
View File
+944
View File
@@ -0,0 +1,944 @@
# -*- coding: utf-8 -*-
"""右侧任务面板:本次会话的 bash 任务监控(运行中 / 已完成)
统一口径(与用户确认):
· 两栏 = 「运行中」/「已完成」(按状态分);两栏【同时可见、不互斥】,
中间是可拖动分隔线(260px 宽若真做左右并排,每栏仅 ~130px,
展开区放不下「参数 + 输出」两块列表,故用上下两栏)。
若要严格左右并排:把 _SPLIT_ORIENTATION 改成 Horizontal 即可。
· 一次 bash = 一层(BashLayer),自上而下按执行先后排列
· 单击层头 = 展开/收起;展开后分「参数」「输出」两块
· 运行中:输出为 tool_bash 的实时流(缓冲上限 200KB,超出丢弃最旧并标注)
· 已完成:输出为「进入上下文的原文」(DB messages.timeline[].result,不截断)
· 范围永远是【当前会话】;切换会话跟随刷新
· 面板【不自动收缩、也不自动展开】,开关就在面板自身:
收起态 = 栏正中按钮;展开态 = 标题行右上角按钮(永远只有一个可动按钮)
· 🆕 左边缘可拖拽调宽(最小 200px);松手立即记录到 config.json["bash_panel_width"]
下次按展开按钮自动恢复到记录的宽度
· 「已出上下文」= 该层在最近一次压缩切点之前(其输出已不在 API 上下文里)
"""
import json
import os
import re
from PyQt6 import QtCore, QtGui, QtWidgets
LIVE_BUF_CAP = 200 * 1024 # 每条实时输出缓冲上限
LAYER_LIMIT = 30 # 已完成栏默认只渲染最近 N 层
_SPLIT_ORIENTATION = QtCore.Qt.Orientation.Vertical # 改 Horizontal = 左右并排
_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
_DUR_RE = re.compile(r"\((\d+(?:\.\d+)?)s\)\s*$")
# ---------------------------------------------------------------- 宽度
PANEL_W_DEFAULT = 260 # 首次运行的默认展开宽度
PANEL_W_MIN = 200 # 🆕 拖拽最小宽度(再窄「参数/输出」两块就挤坏了)
PANEL_W_MAX = 560 # 绝对上限(实际还会受「不超过主窗口 50%」约束)
_CFG_KEY = "bash_panel_width"
_CFG_PATH = os.path.join(_ROOT, "data", "config.json")
def _cfg_path() -> str:
"""配置文件路径(测试可用 HAOCODE_CONFIG_FILE 指向临时文件 → 绝不污染真配置)"""
return os.environ.get("HAOCODE_CONFIG_FILE") or _CFG_PATH
def load_panel_width(default: int = PANEL_W_DEFAULT) -> int:
"""读回上次拖拽记录的宽度;缺失/异常 → 默认值(并强制夹在 [MIN, MAX])"""
try:
with open(_cfg_path(), "r", encoding="utf-8") as f:
w = int(json.load(f).get(_CFG_KEY, default))
except Exception:
return default
return max(PANEL_W_MIN, min(PANEL_W_MAX, w))
def save_panel_width(w: int) -> bool:
"""只改这一个键、其余配置原样保留;临时文件 + replace 原子写"""
p = _cfg_path()
try:
cfg = {}
if os.path.exists(p):
with open(p, "r", encoding="utf-8") as f:
cfg = json.load(f) or {}
cfg[_CFG_KEY] = int(w)
tmp = p + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(cfg, f, ensure_ascii=False, indent=2)
os.replace(tmp, p)
return True
except Exception as e:
print(f"[UI]: 保存任务面板宽度失败: {e}")
return False
def _svg(name: str) -> str:
return os.path.join(_ROOT, "svg", name)
def _fmt_dur(sec) -> str:
try:
sec = float(sec)
except Exception:
return ""
return f"{sec:.1f}s" if sec < 60 else f"{int(sec // 60)}m{int(sec % 60)}s"
def _parse_dur(text):
"""从上下文文本结尾的 (Ns) 里取耗时;取不到返回 None"""
m = _DUR_RE.search((text or "").strip())
if not m:
return None
try:
return float(m.group(1))
except Exception:
return None
def _cmd_of(args) -> str:
"""args 可能是 JSON 字符串或 dict → 取出 bash 的 command"""
if isinstance(args, dict):
return args.get("command", "") or ""
try:
d = json.loads(args or "{}")
if isinstance(d, dict):
return d.get("command", "") or ""
except Exception:
pass
return str(args or "")
def _iter_bash_entries(timeline):
"""timeline 可能是 JSON 字符串(DB)或 list[dict](内存)→ 产出 name=='bash' 的工具条目"""
data = timeline
if isinstance(timeline, str):
if not timeline.strip():
return
try:
data = json.loads(timeline)
except Exception:
return
if not isinstance(data, list):
return
for e in data:
if isinstance(e, dict) and e.get("t") == "tool" and e.get("name") == "bash":
yield e
class _ClickFrame(QtWidgets.QFrame):
"""整块可单击(单击即触发,不用双击)"""
clicked = QtCore.pyqtSignal()
def mouseReleaseEvent(self, e):
if (e.button() == QtCore.Qt.MouseButton.LeftButton
and self.rect().contains(e.position().toPoint())):
self.clicked.emit()
super().mouseReleaseEvent(e)
# ======================================================================
# 一层 = 一次 bash 执行
# ======================================================================
class BashLayer(QtWidgets.QFrame):
def __init__(self, call_id: str, command: str = "", parent=None):
super().__init__(parent)
self.setObjectName("bl_layer")
self.call_id = call_id
self.command = command or ""
self.ok = None # None=运行中 / True / False=已完成
self.in_context = True
self.expanded = False
self.elapsed = 0
self.timeout = 0
self._live = ""
self._live_truncated = False
self._final = None
self._build()
self._apply_status()
# ---------------- UI ----------------
def _build(self):
v = QtWidgets.QVBoxLayout(self)
v.setContentsMargins(0, 0, 0, 0)
v.setSpacing(0)
self.head = _ClickFrame()
self.head.setObjectName("bl_head")
self.head.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
self.head.setToolTip(self.command or "")
hl = QtWidgets.QHBoxLayout(self.head)
hl.setContentsMargins(8, 6, 8, 6)
hl.setSpacing(6)
self.dot = QtWidgets.QLabel("")
self.dot.setObjectName("bl_dot")
self.name = QtWidgets.QLabel("bash")
self.name.setObjectName("bl_name")
self.meta = QtWidgets.QLabel("")
self.meta.setObjectName("bl_meta")
self.tag = QtWidgets.QLabel("")
self.tag.setObjectName("bl_tag")
self.cmd = QtWidgets.QLabel(self._preview())
self.cmd.setObjectName("bl_cmd")
self.cmd.setSizePolicy(QtWidgets.QSizePolicy.Policy.Ignored,
QtWidgets.QSizePolicy.Policy.Preferred)
self.chev = QtWidgets.QLabel("")
self.chev.setObjectName("bl_chev")
hl.addWidget(self.dot, 0)
hl.addWidget(self.name, 0)
hl.addWidget(self.meta, 0)
hl.addWidget(self.tag, 0)
hl.addWidget(self.cmd, 1)
hl.addWidget(self.chev, 0)
v.addWidget(self.head)
self.body = QtWidgets.QWidget()
self.body.setObjectName("bl_body")
bl = QtWidgets.QVBoxLayout(self.body)
bl.setContentsMargins(8, 0, 8, 8)
bl.setSpacing(4)
self.lbl_arg = QtWidgets.QLabel("参数")
self.lbl_arg.setObjectName("bl_sect_label")
self.arg_box = self._code_box(80, wrap=True)
bl.addWidget(self.lbl_arg)
bl.addWidget(self.arg_box)
self.lbl_out = QtWidgets.QLabel("输出")
self.lbl_out.setObjectName("bl_sect_label")
self.out_box = self._code_box(230, wrap=False)
bl.addWidget(self.lbl_out)
bl.addWidget(self.out_box)
self.body.hide()
v.addWidget(self.body)
self.head.clicked.connect(self.toggle)
def _code_box(self, max_h, wrap):
box = QtWidgets.QPlainTextEdit()
box.setObjectName("bl_code")
box.setReadOnly(True)
box.setMaximumHeight(max_h)
box.setLineWrapMode(
QtWidgets.QPlainTextEdit.LineWrapMode.WidgetWidth if wrap
else QtWidgets.QPlainTextEdit.LineWrapMode.NoWrap)
box.setFrameShape(QtWidgets.QFrame.Shape.NoFrame)
box.setVerticalScrollBarPolicy(QtCore.Qt.ScrollBarPolicy.ScrollBarAsNeeded)
box.setHorizontalScrollBarPolicy(
QtCore.Qt.ScrollBarPolicy.ScrollBarAsNeeded if not wrap
else QtCore.Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
return box
def _preview(self) -> str:
s = " ".join((self.command or "").split())
return s[:58] + ("" if len(s) > 58 else "")
# ---------------- 状态 ----------------
def _apply_status(self):
if self.ok is None:
self.dot.setProperty("state", "run")
self.meta.setText(f"{self.elapsed}/{self.timeout}s" if self.timeout
else (f"{self.elapsed}s" if self.elapsed else ""))
else:
self.dot.setProperty("state", "ok" if self.ok else "bad")
self.meta.setText(_fmt_dur(self.elapsed) if self.elapsed else "")
if self.ok is not None and not self.in_context:
self.tag.setText("已出上下文")
self.tag.setProperty("kind", "out")
elif self.ok is None:
self.tag.setText("运行中")
self.tag.setProperty("kind", "run")
elif self.ok:
self.tag.setText("已完成")
self.tag.setProperty("kind", "ok")
else:
self.tag.setText("失败")
self.tag.setProperty("kind", "bad")
for w in (self.dot, self.tag):
w.style().unpolish(w)
w.style().polish(w)
def set_running(self, elapsed, timeout):
self.ok = None
self.elapsed = int(elapsed or 0)
self.timeout = int(timeout or 0)
self._apply_status()
if self.expanded:
self._render_output()
def append_live(self, text: str):
if not text:
return
self._live += text
if len(self._live) > LIVE_BUF_CAP:
self._live = self._live[-LIVE_BUF_CAP:]
self._live_truncated = True
if self.expanded:
self._render_output()
def set_finished(self, ok, text):
self.ok = bool(ok)
self._final = text or ""
d = _parse_dur(self._final)
if d is not None:
self.elapsed = d
self._apply_status()
if self.expanded:
self._render_output()
def set_in_context(self, in_ctx: bool):
self.in_context = bool(in_ctx)
self._apply_status()
# ---------------- 展开 ----------------
def toggle(self):
self.expanded = not self.expanded
self.body.setVisible(self.expanded)
self.chev.setText("" if self.expanded else "")
if self.expanded:
self.arg_box.setPlainText(self.command or "(无参数)")
self._render_output()
def _render_output(self):
if self.ok is None:
body = self._live
if self._live_truncated:
body = "…(实时缓冲已截断,仅保留最后 200KB)\n" + body
self.lbl_out.setText("输出(实时)")
self.out_box.setPlainText(body if body else "(等待输出…)")
sb = self.out_box.verticalScrollBar()
sb.setValue(sb.maximum())
else:
self.lbl_out.setText("输出(进入上下文)")
self.out_box.setPlainText(self._final or "(无输出)")
# ======================================================================
# 一栏:标题(可折叠)+ 层列表
# ======================================================================
class _Section(QtWidgets.QWidget):
def __init__(self, title, empty_hint, on_fold=None, parent=None):
super().__init__(parent)
self.empty_hint = empty_hint
self.folded = False
self._on_fold = on_fold # 回调:交面板做高度动画(抽屉式)
self._effect = None # 内容区淡入淡出用(按需创建,静止时移除)
self._head_hint = 0 # 头部行高缓存(恒定值)
v = QtWidgets.QVBoxLayout(self)
v.setContentsMargins(0, 0, 0, 0)
v.setSpacing(0)
self.head = _ClickFrame()
self.head.setObjectName("bl_sect_head")
self.head.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
hl = QtWidgets.QHBoxLayout(self.head)
hl.setContentsMargins(10, 6, 8, 6)
hl.setSpacing(6)
self.chev = QtWidgets.QLabel("")
self.chev.setObjectName("bl_sect_chev")
self.title = QtWidgets.QLabel(title)
self.title.setObjectName("bl_sect_title")
self.count = QtWidgets.QLabel("0")
self.count.setObjectName("bl_sect_count")
hl.addWidget(self.chev, 0)
hl.addWidget(self.title, 0)
hl.addStretch(1)
hl.addWidget(self.count, 0)
v.addWidget(self.head)
self.scroll = QtWidgets.QScrollArea()
self.scroll.setObjectName("bl_scroll")
self.scroll.setWidgetResizable(True)
self.scroll.setFrameShape(QtWidgets.QFrame.Shape.NoFrame)
self.scroll.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
# 🐛 修复:QScrollArea 的 viewport 默认按 palette base(白)自画背景,
# 会把父级 #f7f8fa 盖成白色 → 强制透明
self.scroll.viewport().setStyleSheet("background: transparent;")
self.scroll.viewport().setAutoFillBackground(False)
self.host = QtWidgets.QWidget()
self.host.setObjectName("bl_host")
self.host.setAutoFillBackground(False)
self.lay = QtWidgets.QVBoxLayout(self.host)
self.lay.setContentsMargins(6, 2, 6, 6)
self.lay.setSpacing(5)
self.scroll.setWidget(self.host)
# 🆕 「下面填充的 bash 层」整块放进 body:折叠/展开动画只针对它,
# 头部行(self.head)高度永远不变
self.body = QtWidgets.QWidget()
self.body.setObjectName("bl_body")
self.body.setAutoFillBackground(False)
bv = QtWidgets.QVBoxLayout(self.body)
bv.setContentsMargins(0, 0, 0, 0)
bv.setSpacing(0)
bv.addWidget(self.scroll, 1)
self.hint = QtWidgets.QLabel("")
self.hint.setObjectName("bl_hint")
self.hint.setWordWrap(True)
bv.addWidget(self.hint)
v.addWidget(self.body, 1)
self.head.clicked.connect(self._toggle_fold)
self._set_hint(self.empty_hint)
# 🐛 头部行高【硬固定】:收起时 body 被隐、布局里只剩 head,而 QFrame 默认
# 竖直策略可伸长 → head 会被拉满整栏(实测 793px 巨条 = 「按钮变大很高」)
self._head_hint = max(16, self.head.sizeHint().height())
self.head.setFixedHeight(self._head_hint)
# ---------------- 头部行高(收起/展开恒定的那一行)----------------
def head_height(self) -> int:
# 🐛 必须缓存 sizeHint:若用实时 self.head.height(),会形成正反馈——
# 头部被压矮 → minimumSizeHint 变小 → splitter 允许更矮 → 头部再被压矮…
if self._head_hint <= 0:
self._head_hint = max(16, self.head.sizeHint().height())
return self._head_hint
def minimumSizeHint(self):
# 🐛 关键修复:默认最小高 ≈ 113px(头部 + QScrollArea 最小高 + 提示语),
# QSplitter 完全没法把本栏收缩 → 收起后留一大块空白(用户报的 bug)。
# 改为「只剩头部行」,内容区允许被压到 0(超出部分天然被父级裁切)。
return QtCore.QSize(0, self.head_height())
# ---------------- 内容区透明度(动画帧驱动)----------------
def set_body_opacity(self, op: float):
op = max(0.0, min(1.0, float(op)))
if op >= 0.999:
self._drop_effect()
return
if self._effect is None:
self._effect = QtWidgets.QGraphicsOpacityEffect(self.body)
self.body.setGraphicsEffect(self._effect)
self._effect.setOpacity(op)
def _drop_effect(self):
"""去掉离屏渲染开销(静止时不挂 effect)"""
if self.body.graphicsEffect() is not None:
self.body.setGraphicsEffect(None)
self._effect = None
def show_body(self):
self.body.setVisible(True)
self.scroll.setVisible(True)
self.hint.setVisible(bool(self.hint.text()))
def hide_body(self):
self.body.setVisible(False)
def set_folded(self, folded: bool):
"""应用折叠状态(高度由面板的 splitter 动画负责)"""
self.folded = bool(folded)
self.chev.setText("" if self.folded else "")
if self.folded:
self._drop_effect()
else:
self.show_body()
self.set_body_opacity(1.0)
def _toggle_fold(self):
if self._on_fold is not None:
self._on_fold(self) # 交面板:状态切换 + 高度抽屉动画
return
self.set_folded(not self.folded)
self.scroll.setVisible(not self.folded)
self.hint.setVisible(not self.folded and bool(self.hint.text()))
def _set_hint(self, text):
self.hint.setText(text or "")
self.hint.setVisible(not self.folded and bool(text))
def set_layers(self, layers, total=None):
"""重排层(复用同一批 widget → 实时缓冲不丢)"""
while self.lay.count():
it = self.lay.takeAt(0)
w = it.widget()
if w is not None:
w.setParent(None)
for w in layers:
self.lay.addWidget(w)
w.show()
self.lay.addStretch(1)
self.count.setText(str(len(layers)))
total = len(layers) if total is None else total
if total > len(layers):
self._set_hint(f"仅显示最近 {len(layers)} 层(本会话共 {total} 层)")
else:
self._set_hint("" if layers else self.empty_hint)
# ======================================================================
# 右侧任务面板(外壳,镜像左侧边栏)
# ======================================================================
class _ResizeHandle(QtWidgets.QWidget):
"""面板左边缘的拖拽手柄(4px 隐形条,hover 才亮;光标 SizeHor"""
W = 4
def __init__(self, panel):
super().__init__(panel)
self._panel = panel
self.setObjectName("bl_resize_handle")
# QWidget 子类必须显式开启才会绘制样式表背景(hover 高亮靠它)
self.setAttribute(QtCore.Qt.WidgetAttribute.WA_StyledBackground, True)
self.setCursor(QtCore.Qt.CursorShape.SizeHorCursor)
self.setStyleSheet("#bl_resize_handle:hover { background: #dfe3ea; }")
self.setToolTip("拖拽调整任务面板宽度(最小 %dpx" % PANEL_W_MIN)
self._drag = False
def mousePressEvent(self, e):
if e.button() == QtCore.Qt.MouseButton.LeftButton:
self._drag = True
self._panel._drag_begin(e.globalPosition().x())
e.accept()
return
super().mousePressEvent(e)
def mouseMoveEvent(self, e):
if self._drag:
self._panel._drag_to_global_x(e.globalPosition().x())
e.accept()
return
super().mouseMoveEvent(e)
def mouseReleaseEvent(self, e):
if self._drag:
self._drag = False
self._panel._commit_drag_width()
e.accept()
return
super().mouseReleaseEvent(e)
class BashPanel(QtWidgets.QWidget):
W_EXPAND = PANEL_W_DEFAULT
W_COLLAPSE = 52
W_MIN = PANEL_W_MIN
W_MAX = PANEL_W_MAX
FOLD_MS = 200 # 栏折叠/展开时长(与 ModelSelectPopup.DRAWER_DURATION 一致)
FOLD_STEP = 16 # 帧间隔(与 DRAWER_STEP 一致,~60fps
MIN_BODY_H = 60 # 展开后「下面填充的层」至少保留的高度
def __init__(self, parent=None):
super().__init__(parent)
self.setObjectName("right_sidebar")
# 🐛 关键修复:QWidget 的自定义子类默认【不绘制样式表背景/边框】
# #sidebar 是原生 QWidget 实例所以正常,BashPanel 是子类所以失效——
# 表现为右侧栏底色变白、左侧分隔线丢失、空间感消失)
self.setAttribute(QtCore.Qt.WidgetAttribute.WA_StyledBackground, True)
# 🆕 展开宽度:优先用上次拖拽记录的值(config.json: bash_panel_width
self.W_EXPAND = load_panel_width()
self._end_w = self.W_EXPAND
self._drag_right = None
self._w = self.W_COLLAPSE
self.setFixedWidth(self._w) # 默认收起
self.collapsed = True
self._target = True
self._anim = None
self._session_id = None
self._db = None
self._layers = {} # call_id -> BashLayer
self._order = [] # 执行先后
self._running = set()
self._done = set()
self._saved_h = [120, 520] # 每栏收起前的高度(展开时恢复)
self._fold = None # 折叠动画状态
self._fold_timer = QtCore.QTimer(self)
self._fold_timer.setInterval(self.FOLD_STEP)
self._fold_timer.timeout.connect(self._fold_tick)
self._build()
# 🆕 左边缘拖拽手柄(浮在最左 4px,收起/展开两种状态都在)
self._handle = _ResizeHandle(self)
self._handle.setFixedWidth(_ResizeHandle.W)
self._handle.raise_()
# ---------------- 外壳 ----------------
def _build(self):
v = QtWidgets.QVBoxLayout(self)
v.setContentsMargins(0, 0, 0, 0)
v.setSpacing(0)
self.stack = QtWidgets.QStackedWidget()
v.addWidget(self.stack)
# ---- 展开页(顶部留白 20px + 36px 标题行,与左侧栏完全对齐)----
self.expand_page = QtWidgets.QWidget()
ev = QtWidgets.QVBoxLayout(self.expand_page)
ev.setContentsMargins(0, 20, 0, 12)
ev.setSpacing(0)
self.header = QtWidgets.QWidget()
self.header.setObjectName("bl_panel_head")
self.header.setFixedHeight(36)
hl = QtWidgets.QHBoxLayout(self.header)
hl.setContentsMargins(10, 0, 10, 0)
hl.setSpacing(6)
self.lbl_title = QtWidgets.QLabel("任务面板")
self.lbl_title.setObjectName("bl_panel_title")
self.lbl_total = QtWidgets.QLabel("")
self.lbl_total.setObjectName("bl_panel_total")
# 🐛 修复双按钮:面板内不再自带收起按钮(唯一开关 = 顶部工具栏按钮),
# 与左侧栏的区别:左侧栏的开关在栏内,右侧栏的开关在顶部(原「导出」位置)
hl.addWidget(self.lbl_title, 0)
hl.addWidget(self.lbl_total, 0)
hl.addStretch(1)
# 🆕 展开态开关:标题行右端(与左侧栏 collapse_btn 完全同规格)
self.fold_btn = QtWidgets.QPushButton("")
self.fold_btn.setObjectName("collapse_btn")
self.fold_btn.setIcon(QtGui.QIcon(_svg("panel_right.svg")))
self.fold_btn.setIconSize(QtCore.QSize(16, 16))
self.fold_btn.setFixedSize(28, 28)
self.fold_btn.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
self.fold_btn.setToolTip("收起任务面板")
hl.addWidget(self.fold_btn, 0)
ev.addWidget(self.header)
ev.addSpacing(10)
self.splitter = QtWidgets.QSplitter(_SPLIT_ORIENTATION)
self.splitter.setObjectName("bl_splitter")
self.splitter.setChildrenCollapsible(False)
self.splitter.setHandleWidth(5)
self.sec_running = _Section("运行中", "暂无正在运行的 bash", on_fold=self._toggle_section)
self.sec_done = _Section("已完成", "本会话还没有已完成的 bash", on_fold=self._toggle_section)
self.splitter.addWidget(self.sec_running)
self.splitter.addWidget(self.sec_done)
# 🆕 底部留白占位:最后一栏收起时由它吸收余量(否则栏头会被顶到面板底部,
# 或被拉伸成巨条)
self.spacer = QtWidgets.QWidget()
self.spacer.setObjectName("bl_spacer")
self.spacer.setAttribute(QtCore.Qt.WidgetAttribute.WA_StyledBackground, True)
self.spacer.setStyleSheet("#bl_spacer { background: transparent; }")
self.splitter.addWidget(self.spacer)
self.splitter.setCollapsible(2, True) # 占位层允许被压到 0
self.splitter.setSizes(self._saved_h + [0])
_h2 = self.splitter.handle(2)
if _h2 is not None: # 占位层上方的分隔条:不可拖、不可见
_h2.setEnabled(False)
_h2.setStyleSheet("background: transparent;")
ev.addWidget(self.splitter, 1)
self.stack.addWidget(self.expand_page)
# ---- 收起页:栏正中一个展开按钮(与左侧栏 collapse_expand_btn 完全同规格)----
self.collapse_page = QtWidgets.QWidget()
cl = QtWidgets.QVBoxLayout(self.collapse_page)
cl.setContentsMargins(0, 0, 0, 0)
cl.addStretch(1)
self.expand_btn = QtWidgets.QPushButton("")
self.expand_btn.setObjectName("collapse_btn")
self.expand_btn.setIcon(QtGui.QIcon(_svg("panel_right.svg")))
self.expand_btn.setIconSize(QtCore.QSize(18, 18))
self.expand_btn.setFixedSize(34, 34)
self.expand_btn.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
self.expand_btn.setToolTip("展开任务面板")
cl.addWidget(self.expand_btn, 0, QtCore.Qt.AlignmentFlag.AlignHCenter)
cl.addStretch(1)
self.stack.addWidget(self.collapse_page)
# 两个按钮分居 QStackedWidget 两页 → 任何时刻可见开关恒为 1 个
self.stack.setCurrentWidget(self.collapse_page)
self.fold_btn.clicked.connect(self.toggle)
self.expand_btn.clicked.connect(self.toggle)
# ---------------- 两栏折叠/展开:抽屉式高度动画(头部行高恒定)----------------
def _toggle_section(self, sec):
"""点栏头 → 只把「下面填充的 bash 层」的高度收到 0,头部行高度不变"""
self._finish_fold() # 连点:先把上一段动画落定
idx = 0 if sec is self.sec_running else 1
sizes = self.splitter.sizes()
avail = sum(sizes)
head_h = sec.head_height()
other = self.sec_done if idx == 0 else self.sec_running
if not sec.folded and not other.folded:
# 即将收起 + 另一栏展开 → 记住当前高度(用于下次展开恢复);
# 另一栏已收起时不记(此时本栏高度是「余量」而非用户意图高度)
self._saved_h[idx] = max(sizes[idx], head_h + self.MIN_BODY_H)
sec.set_folded(not sec.folded)
target = self._layout_targets(actor=sec)
if not sec.folded: # 展开 → 内容从透明开始淡入
sec.show_body()
sec.set_body_opacity(0.0)
self._sync_handle()
if avail <= 0 or self.splitter.width() <= 0:
self.splitter.setSizes(target)
self._end_fold_visual(sec)
return
self._fold = {"sec": sec, "from": list(sizes), "to": target, "t": 0.0}
self._fold_timer.start()
def _layout_targets(self, actor=None):
"""统一布局策略(层从顶部堆叠):
· 收起栏 = 只剩头部行
· 本次被点开的那栏 → 用记录的高度
· 余量只由「展开着的已完成」吸收(填满到底部);
若「已完成」已收起 → 余量进底部留白(栏头紧跟上方内容)
"""
s = self.splitter.sizes()
avail = sum(s)
hr = self.sec_running.head_height()
hd = self.sec_done.head_height()
h0, h1 = s[0], s[1]
run_f, done_f = self.sec_running.folded, self.sec_done.folded
if run_f:
h0 = hr
if done_f:
h1 = hd
if (not run_f) and actor is self.sec_running:
h0 = self._saved_h[0]
if (not done_f) and actor is self.sec_done:
h1 = self._saved_h[1]
if not done_f: # 已完成展开 → 吃掉全部余量
floor0 = hr if run_f else hr + self.MIN_BODY_H
h1 = max(hd + self.MIN_BODY_H, avail - max(h0, floor0))
h0 = max(hr if run_f else hr + self.MIN_BODY_H, h0)
h1 = max(hd if done_f else hd + self.MIN_BODY_H, h1)
return [h0, h1, max(0, avail - h0 - h1)]
def _fold_tick(self):
"""抽屉式每帧:ease-out 三次方插值高度 + 内容淡入淡出(同模型选择窗口)"""
d = self._fold
if d is None:
self._fold_timer.stop()
return
d["t"] = min(1.0, d["t"] + self.FOLD_STEP / self.FOLD_MS)
e = 1.0 - (1.0 - d["t"]) ** 3
a, b = d["from"], d["to"]
self.splitter.setSizes([int(round(x + (y - x) * e)) for x, y in zip(a, b)])
sec = d["sec"]
sec.set_body_opacity((1.0 - e) if sec.folded else e)
if d["t"] >= 1.0:
self._finish_fold()
def _finish_fold(self):
self._fold_timer.stop()
d = self._fold
if d is None:
return
self._fold = None
self.splitter.setSizes(d["to"])
self._end_fold_visual(d["sec"])
def _end_fold_visual(self, sec):
if sec.folded:
sec.set_body_opacity(0.0)
sec.hide_body() # 隐藏后连离屏渲染也省了
else:
sec.show_body()
sec.set_body_opacity(1.0) # 置 1 → 自动移除 effect
def _sync_handle(self):
"""两栏都展开 → 中间分隔条可拖;任一栏收起 → 禁用(避免「刚收起又被拖大」)"""
h = self.splitter.handle(1)
if h is not None:
ok = not (self.sec_running.folded or self.sec_done.folded)
h.setEnabled(ok)
h.setCursor(QtCore.Qt.CursorShape.SplitVCursor if ok
else QtCore.Qt.CursorShape.ArrowCursor)
def _repin_fold(self):
"""窗口/面板尺寸变化后,按统一策略重新钉住(收起栏 = 只剩头部行)"""
if self._fold is not None or self.collapsed:
return
if not (self.sec_running.folded or self.sec_done.folded):
return
want = self._layout_targets()
if want != self.splitter.sizes():
self.splitter.setSizes(want)
def get_panelWidth(self):
return self._w
def set_panelWidth(self, w):
self._w = int(w)
self.setFixedWidth(int(w))
panelWidth = QtCore.pyqtProperty(int, fget=get_panelWidth, fset=set_panelWidth)
def toggle(self):
self._set_collapsed(not self._target)
def _set_collapsed(self, collapsed: bool):
self._stop_anim()
self._target = collapsed
self.stack.setCurrentWidget(
self.collapse_page if collapsed else self.expand_page)
start_w = self.width()
# 展开时若记录的宽度超出当前窗口可承受范围 → 夹一下
self._end_w = self.W_COLLAPSE if collapsed else self._clamp_w(self.W_EXPAND)
end_w = self._end_w
self._anim = QtCore.QPropertyAnimation(self, b"panelWidth")
self._anim.setDuration(260)
self._anim.setStartValue(float(start_w))
self._anim.setEndValue(float(end_w))
self._anim.setEasingCurve(QtCore.QEasingCurve.Type.InOutCubic)
self._anim.finished.connect(self._on_anim_finished)
self._anim.start()
def _on_anim_finished(self):
self.setFixedWidth(self.W_COLLAPSE if self._target else self._end_w)
self.collapsed = self._target
self._stop_anim()
# ---------------- 宽度拖拽(右边界固定,左边界跟随鼠标)----------------
def _stop_anim(self):
if self._anim is not None:
try:
self._anim.stop()
self._anim.deleteLater()
except Exception:
pass
self._anim = None
def _clamp_w(self, w) -> int:
"""夹在 [W_MIN, min(W_MAX, 主窗口 50%)](最小宽度就在这里生效)"""
hi = self.W_MAX
par = self.parentWidget()
if par is not None and par.width() > 0:
hi = min(hi, max(self.W_MIN, int(par.width() * 0.5)))
return max(self.W_MIN, min(hi, int(w)))
def resizeEvent(self, e):
super().resizeEvent(e)
h = getattr(self, "_handle", None)
if h is not None:
h.setGeometry(0, 0, _ResizeHandle.W, self.height())
h.raise_()
self._repin_fold()
def _drag_begin(self, gx):
self._stop_anim()
# 动画中途按下 → 以当前实际宽度为准确定状态
self.collapsed = self.width() <= self.W_COLLAPSE + 8
self._target = self.collapsed
self.stack.setCurrentWidget(
self.collapse_page if self.collapsed else self.expand_page)
self._drag_right = self.mapToGlobal(QtCore.QPoint(self.width(), 0)).x()
def _drag_to_global_x(self, gx):
if self._drag_right is None:
return
w = self._clamp_w(self._drag_right - int(gx))
if self.collapsed:
if w <= self.W_COLLAPSE + 8:
return # 还在收起条内拖 → 不响应
self.collapsed = False # 从收起态直接拖开 → 立即进展开态
self._target = False
self.stack.setCurrentWidget(self.expand_page)
self._w = w
self.setFixedWidth(w)
def _commit_drag_width(self):
self._drag_right = None
if self.collapsed:
return
self.W_EXPAND = self._w
if save_panel_width(self._w):
print(f"[UI]: 任务面板宽度 -> {self._w}px(已记录,下次展开自动使用)")
# ---------------- 数据 ----------------
def set_session(self, session_id, db, active_stream=None):
"""切换会话 → 全量重建。已完成来自 DB 链;未落库的在跑任务来自 active_stream。"""
self._session_id = session_id
self._db = db
self._layers.clear()
self._order.clear()
self._running.clear()
self._done.clear()
# ① DB:本会话链上的历史 bash(标注是否还在上下文内)
try:
chain = db.get_message_chain(session_id) if db else []
except Exception:
chain = []
mark_idx = -1
for i, m in enumerate(chain):
if m.get("role") == "compaction":
mark_idx = i
for i, m in enumerate(chain):
msg_id = m.get("id") or ""
for e in _iter_bash_entries(m.get("timeline")):
cid = e.get("id") or f"dbmsg_{msg_id}_{i}"
if cid in self._layers:
continue
lay = BashLayer(cid, _cmd_of(e.get("args")))
lay.set_finished(bool(e.get("ok")), e.get("result") or "")
lay.set_in_context(i > mark_idx)
self._layers[cid] = lay
self._order.append(cid)
self._done.add(cid)
# ② 内存:正在进行的这一轮(包括已开始但未落库、以及刚结束未落库的)
for e in _iter_bash_entries((active_stream or {}).get("timeline")):
cid = e.get("id")
if not cid or cid in self._layers:
continue
lay = BashLayer(cid, _cmd_of(e.get("args")))
if e.get("ok") is None:
lay.set_running(0, 0)
if e.get("result"):
lay.append_live(str(e.get("result"))) # 尽力回放(上游上限 4000 字符)
self._running.add(cid)
else:
lay.set_finished(bool(e.get("ok")), e.get("result") or "")
self._done.add(cid)
self._layers[cid] = lay
self._order.append(cid)
self._refresh()
def _refresh(self):
run_ids = [c for c in self._order if c in self._running]
done_ids = [c for c in self._order if c in self._done]
self.sec_running.set_layers([self._layers[c] for c in run_ids])
shown = done_ids[-LAYER_LIMIT:]
self.sec_done.set_layers([self._layers[c] for c in shown], total=len(done_ids))
self.lbl_total.setText(f"{len(self._order)}" if self._order else "")
# ---------------- 事件 ----------------
def on_started(self, call_id, name, args):
if name != "bash" or not call_id or call_id in self._layers:
return
lay = BashLayer(call_id, _cmd_of(args))
lay.set_running(0, 0)
self._layers[call_id] = lay
self._order.append(call_id)
self._running.add(call_id)
self._refresh()
def on_output(self, call_id, text):
lay = self._layers.get(call_id)
if lay is not None and lay.ok is None:
lay.append_live(text)
def on_timed(self, call_id, elapsed, timeout):
lay = self._layers.get(call_id)
if lay is not None and lay.ok is None:
lay.set_running(elapsed, timeout)
def on_finished(self, call_id, name, ok, text):
lay = self._layers.get(call_id)
if lay is None:
if name != "bash" or not call_id:
return
lay = BashLayer(call_id, "")
self._layers[call_id] = lay
self._order.append(call_id)
self._running.discard(call_id)
self._done.add(call_id)
lay.set_finished(bool(ok), text or "")
lay.set_in_context(True)
self._refresh()
def clear_all(self):
self._layers.clear()
self._order.clear()
self._running.clear()
self._done.clear()
self._refresh()
# ---------------- 测试/调试辅助 ----------------
def layer_ids(self, which="all"):
if which == "running":
return [c for c in self._order if c in self._running]
if which == "done":
return [c for c in self._order if c in self._done]
return list(self._order)
+188
View File
@@ -0,0 +1,188 @@
# chat_bridge.py
import json
from PyQt6.QtCore import QObject, pyqtSlot, pyqtSignal
class ChatBridge(QObject):
# 信号定义
regenerate_clicked = pyqtSignal(str) # 携带需要重新回答的助手消息ID
branch_switch_clicked = pyqtSignal(str, int) # 分支切换信号 (msg_id, direction: -1 为上一条, 1 为下一条)
delete_message_requested = pyqtSignal(str) # 删除消息信号 (msg_id)
attachment_clicked = pyqtSignal(str)
scroll_changed = pyqtSignal(float, float, float) # 页面 scrollY / scrollHeight / innerHeight
def __init__(self, page, channel, js_runner=None):
"""
:param page: QWebEnginePage 实例(QtWebEngine 路径)
:param channel: QWebChannel 实例(QtWebEngine 路径)
:param js_runner: 可选 JS 执行函数(WebView2 路径:script -> None
"""
super().__init__()
self.page = page
self._js_runner = js_runner
if channel is not None:
# 将自身注册到 channel 中,前端可通过 bridge 对象调用
channel.registerObject("bridge", self)
print("ChatBridge被初始化")
def run_js(self, script: str):
if self._js_runner is not None:
self._js_runner(script)
elif self.page is not None:
self.page.runJavaScript(script)
else:
print("[ChatBridge] run_js: 无可用执行通道")
# ---------- 消息构造 (Python -> JS) ----------
def create_message(self, msg_id: str, role: str, text: str = "", sender_name: str = "", branch_info: dict = None):
"""
在前端创建一条消息容器
:param branch_info: 格式如 {"current": 0, "total": 2},若为 None 则不显示分支切换器
"""
safe_text = json.dumps(text)
safe_name = json.dumps(sender_name)
safe_branch = json.dumps(branch_info) if branch_info else "null"
self.run_js(f"createMessage('{msg_id}', '{role}', {safe_text}, {safe_name}, {safe_branch});")
def create_long_message(self, msg_id: str, role: str, text: str, sender_name: str = ""):
safe_text = json.dumps(text)
safe_name = json.dumps(sender_name)
size_kb = round(len(text.encode('utf-8')) / 1024, 2)
self.run_js(f"createLongMessage('{msg_id}', '{role}', {safe_text}, {safe_name}, {size_kb});")
def create_user_message_with_attachments(self, msg_id: str, text: str, attachments: list):
safe_text = json.dumps(text)
safe_attachments = json.dumps(attachments)
self.run_js(f"createUserMessageWithAttachments('{msg_id}', {safe_text}, {safe_attachments});")
# ---------- 流式输出 (Python -> JS) ----------
def append_token(self, msg_id: str, token: str):
safe_token = json.dumps(token)
self.run_js(f"appendToken('{msg_id}', {safe_token});")
def append_reasoning(self, msg_id: str, token: str):
safe_token = json.dumps(token)
self.run_js(f"appendReasoning('{msg_id}', {safe_token});")
def finish_message(self, msg_id: str):
self.run_js(f"finishMessage('{msg_id}');")
# ---------- 工具执行事件 (Python -> JS) —— pi tool_execution_* ----------
def tool_execution_started(self, msg_id: str, call_id: str, name: str, args: str):
"""工具开始执行:前端在时间线当前位置插入 chip(带 call_id 对号入座)"""
safe_call = json.dumps(call_id)
safe_name = json.dumps(name)
safe_args = json.dumps(args)
self.run_js(f"toolExecutionStarted('{msg_id}', {safe_call}, {safe_name}, {safe_args});")
def tool_execution_updated(self, msg_id: str, call_id: str, text: str):
"""工具执行中的增量输出(bash stdout 等)"""
safe_call = json.dumps(call_id)
safe_text = json.dumps(text)
self.run_js(f"toolExecutionUpdated('{msg_id}', {safe_call}, {safe_text});")
def tool_execution_timed(self, msg_id: str, call_id: str,
elapsed: int, timeout: int):
"""🆕 bash 运行中每秒读秒:气泡摘要行刷新 N/Ts"""
safe_call = json.dumps(call_id)
self.run_js(f"toolExecutionTimed('{msg_id}', {safe_call}, "
f"{int(elapsed)}, {int(timeout)});")
def tool_execution_finished(self, msg_id: str, call_id: str, name: str, ok: bool, text: str):
"""工具执行结束:chip 按 call_id 标记成功/失败并显示结果摘要"""
safe_call = json.dumps(call_id)
safe_name = json.dumps(name)
safe_text = json.dumps(text)
self.run_js(f"toolExecutionFinished('{msg_id}', {safe_call}, {safe_name}, {str(ok).lower()}, {safe_text});")
# ---------- 时间线 (Python -> JS) ----------
def restore_streaming_timeline(self, msg_id: str, timeline_json: str):
"""切回进行中的会话:按时间线 JSON 恢复 思考/文本/工具 块(续流)"""
safe_json = json.dumps(timeline_json)
self.run_js(f"restoreStreamingTimeline('{msg_id}', {safe_json});")
def render_timeline_history(self, msg_id: str, timeline_json: str):
"""历史消息:按时间线 JSON 静态渲染 思考/文本/工具 块"""
safe_json = json.dumps(timeline_json)
self.run_js(f"renderTimelineHistory('{msg_id}', {safe_json});")
# ---------- 压缩可视化 (Python -> JS) ----------
def compaction_started(self, msg_id: str, path: str):
"""压缩开始 → 当前消息时间线内显示「上下文压缩」思考气泡(与深度思考同款)"""
safe_id = json.dumps(msg_id or "")
safe_path = json.dumps(path or "")
self.run_js(f"compactionStarted({safe_id}, {safe_path});")
def compaction_finished(self, msg_id: str, payload: dict):
"""压缩完成 → 同一气泡原地更新(前→后 token + 摘要全文)"""
safe_id = json.dumps(msg_id or "")
safe_payload = json.dumps(payload or {}, ensure_ascii=False)
self.run_js(f"compactionFinished({safe_id}, {safe_payload});")
# ---------- 系统提示 (Python -> JS) ----------
def show_note(self, text: str):
"""在聊天流中插入一条居中系统提示(如:已自动压缩上下文)"""
safe_text = json.dumps(text)
self.run_js(f"showSystemNote({safe_text});")
# ---------- 错误与清理 (Python -> JS) ----------
def show_error(self, msg_id: str, error_text: str):
safe_text = json.dumps(error_text)
self.run_js(f"showError('{msg_id}', {safe_text});")
def clear_chat(self):
self.run_js("clearChat();")
def delete_message(self, msg_id: str):
"""通知前端从 DOM 中移除特定消息"""
self.run_js(f"deleteMessage('{msg_id}');")
# ---------- 历史记录渲染 ----------
def render_history_message(self, msg_id: str, role: str, content: str, reasoning: str = "", branch_info: dict = None):
self.create_message(msg_id, role, content, branch_info=branch_info)
if reasoning:
safe_reasoning = json.dumps(reasoning)
self.run_js(f"insertThinkBlock('{msg_id}', {safe_reasoning});")
self.finish_message(msg_id)
def show_welcome(self):
self.run_js("showWelcome();")
# ---------- 会话加载界面 (Python -> JS) ----------
def show_loading(self):
"""显示统一加载界面(图标 + 从左到右扫描条)"""
self.run_js("showLoadingOverlay();")
def hide_loading(self):
"""隐藏加载界面(带渐变退场)"""
self.run_js("hideLoadingOverlay();")
# ---------- 交互 Slot (JS -> Python) ----------
@pyqtSlot(str)
def onRegenerateClicked(self, msg_id):
"""前端点击“重新生成”按钮时触发"""
print(f"Regenerate requested for: {msg_id}")
self.regenerate_clicked.emit(msg_id)
@pyqtSlot(str, int)
def onBranchSwitch(self, msg_id, direction):
"""前端点击分支切换箭头时触发"""
print(f"Branch switch requested for: {msg_id}, direction: {direction}")
self.branch_switch_clicked.emit(msg_id, direction)
@pyqtSlot(str)
def onDeleteMessageClicked(self, msg_id):
"""接收 JS 发来的确认删除指令(🌟 已去重,原先定义了两次)"""
print(f"Delete requested for: {msg_id}")
self.delete_message_requested.emit(msg_id)
@pyqtSlot(str)
def onAttachmentClicked(self, meta_json_str):
"""接收 JS 发来的附件点击事件"""
self.attachment_clicked.emit(meta_json_str)
@pyqtSlot(float, float, float)
def onScrollChanged(self, y, content_h, client_h):
"""接收前端上报的页面滚动信息,用于自定义滚动条位置/滑块长度镜像"""
self.scroll_changed.emit(y, content_h, client_h)
+66
View File
@@ -0,0 +1,66 @@
from PyQt6.QtWebEngineCore import QWebEnginePage, QWebEngineScript
from PyQt6.QtCore import QUrl
from PyQt6.QtGui import QDesktopServices
class CustomWebPage(QWebEnginePage):
"""
自定义网页类,集中约束内置 QtWebEngine 的不合规浏览器行为:
1. 拦截所有外部链接跳转,改用系统默认浏览器打开;
2. 禁用 Ctrl/Meta + 滚轮 及 Ctrl + +/-/0 快捷键的页面缩放
(浏览器式缩放对桌面聊天工具无意义且易误触)。
"""
def __init__(self, parent=None):
super().__init__(parent)
self._inject_zoom_lock()
def _inject_zoom_lock(self):
"""
🌟 缩放锁定:通过网页脚本在文档创建阶段(DocumentCreation)注入监听,
使用隔离世界(ApplicationWorld)注入,不污染页面自身逻辑。
原理:Chromium 的 Ctrl+滚轮缩放会先经过页面 wheel 事件的
preventDefault() 判定 —— 页面阻止即取消缩放,这是地图类应用
禁用缩放的官方做法;Ctrl+=/-/0 快捷键同理用 keydown 拦截。
"""
script = QWebEngineScript()
script.setName("haocode_zoom_lock")
script.setInjectionPoint(QWebEngineScript.InjectionPoint.DocumentCreation)
script.setWorldId(QWebEngineScript.ScriptWorldId.ApplicationWorld)
script.setRunsOnSubFrames(True)
script.setSourceCode(r"""
(function () {
if (window.__haocodeZoomLocked) return; // 防重复注入
window.__haocodeZoomLocked = true;
// 1) Ctrl/Meta + 滚轮 → 阻止缩放
document.addEventListener('wheel', function (e) {
if (e.ctrlKey || e.metaKey) {
e.preventDefault();
e.stopPropagation();
}
}, { passive: false }); // 必须非被动,preventDefault 才有效
// 2) Ctrl + = / + / - / 0 → 阻止缩放快捷键
document.addEventListener('keydown', function (e) {
if ((e.ctrlKey || e.metaKey) && ['+', '-', '=', '0'].indexOf(e.key) >= 0) {
e.preventDefault();
}
}, false);
})();
""")
self.scripts().insert(script)
def acceptNavigationRequest(self, url: QUrl, nav_type, is_main_frame):
# 如果是本地文件(我们自己的 index.html),允许加载
if url.scheme() == "file":
return True
# 如果是外部链接(http/https),用系统浏览器打开
if url.scheme() in ["http", "https"]:
print(f"[System]: 在系统浏览器中打开 -> {url.toString()}")
QDesktopServices.openUrl(url)
return False # 阻止在应用内跳转
# 其他情况(如 javascript:void(0)),允许
return True
+214
View File
@@ -0,0 +1,214 @@
# -*- coding: utf-8 -*-
"""独立调试器窗口 —— 与主窗口/项目树完全解耦(顶层窗口,parent=None
功能:
Tab1「调试会话」 实时 tail data/debug_session.log,按来源着色:
[USER]蓝 / [AGENT]绿 / [APP]灰 / [SYS]紫
Tab2「应用日志」 实时 tail diag.log + compaction_diag.log + stream_diag.log
底部输入框 用户输入观察到的情况,回车 → 记 [USER]
按钮 暂停显示 / 清空会话日志 / 打开日志文件
控制: 代理写 data/debug_window.cmd (show/hide),主窗口 2s 轮询后调用本窗口。
本文件不 import main_window,可独立离屏测试。
"""
import os
import re
from PyQt6.QtCore import Qt, QTimer
from PyQt6.QtGui import QTextCharFormat, QTextCursor, QColor
from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QPlainTextEdit,
QLineEdit, QPushButton, QCheckBox, QLabel,
QTabWidget, QMessageBox)
from core.debug_log import DEBUG_LOG_PATH, DEBUG_CMD_PATH, debug_log
_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
_APP_LOGS = [
("DIAG", os.path.join(_ROOT, "diag.log")),
("COMPACT", os.path.join(_ROOT, "compaction_diag.log")),
("STREAM", os.path.join(_ROOT, "stream_diag.log")),
]
_TAG_COLORS = {"USER": "#2563eb", "AGENT": "#16a34a",
"APP": "#6b7280", "SYS": "#9333ea"}
_LINE_RE = re.compile(r"^\[([^\]]+)\]\s+\[([A-Z]+)\]\s?(.*)$")
class _TailReader:
"""单文件增量读取器(文件被截断时自动重置偏移)"""
def __init__(self, path: str):
self.path = path
self.offset = 0
def read_new(self) -> str:
try:
if not os.path.exists(self.path):
return ""
size = os.path.getsize(self.path)
if size < self.offset: # 被清空/轮转
self.offset = 0
if size == self.offset:
return ""
with open(self.path, "r", encoding="utf-8", errors="replace") as f:
f.seek(self.offset)
data = f.read()
self.offset = size
return data
except Exception:
return ""
class DebugWindow(QWidget):
def __init__(self):
super().__init__(None) # 顶层独立窗口
self.setWindowTitle("Haocode 调试器")
self.resize(780, 540)
self.setWindowFlags(Qt.WindowType.Window)
# 默认停靠主屏右上角,避免被主窗口挡住
try:
from PyQt6.QtGui import QGuiApplication
_geo = QGuiApplication.primaryScreen().availableGeometry()
self.move(_geo.right() - self.width() - 24, _geo.top() + 24)
except Exception:
pass
self._paused = False
self._announced = False
root = QVBoxLayout(self)
root.setContentsMargins(8, 8, 8, 8)
# ---- Tab 容器 ----
self._tabs = QTabWidget()
self._view_session = QPlainTextEdit()
self._view_session.setReadOnly(True)
self._view_session.setMaximumBlockCount(3000)
self._view_session.setLineWrapMode(
QPlainTextEdit.LineWrapMode.NoWrap)
self._view_app = QPlainTextEdit()
self._view_app.setReadOnly(True)
self._view_app.setMaximumBlockCount(3000)
self._view_app.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
self._tabs.addTab(self._view_session, "调试会话")
self._tabs.addTab(self._view_app, "应用日志")
root.addWidget(self._tabs, 1)
# ---- 状态行 ----
status = QLabel(f"会话日志: {DEBUG_LOG_PATH}\n"
f"控制文件: {DEBUG_CMD_PATH} (show/hide)")
status.setStyleSheet("color:#888; font-size:11px;")
status.setWordWrap(True)
root.addWidget(status)
# ---- 按钮行 ----
btn_row = QHBoxLayout()
self._chk_pause = QCheckBox("暂停显示(记录继续)")
self._chk_pause.toggled.connect(self._on_pause_toggled)
self._chk_top = QCheckBox("置顶")
self._chk_top.toggled.connect(self._on_top_toggled)
btn = QPushButton("清空会话日志")
btn.clicked.connect(self._on_clear)
btn2 = QPushButton("打开日志文件")
btn2.clicked.connect(self._on_open_file)
btn_row.addWidget(self._chk_pause)
btn_row.addWidget(self._chk_top)
btn_row.addStretch(1)
btn_row.addWidget(btn)
btn_row.addWidget(btn2)
root.addLayout(btn_row)
# ---- 用户输入行 ----
in_row = QHBoxLayout()
hint = QLabel("观察到的情况(回车记录为 [USER]):")
hint.setStyleSheet("color:#555; font-size:12px;")
self._input = QLineEdit()
self._input.setPlaceholderText("例如: 上下文标签显示 40.5k,刚发送了「接着输出」")
self._input.returnPressed.connect(self._on_submit)
in_row.addWidget(hint)
in_row.addWidget(self._input, 1)
root.addLayout(in_row)
# ---- 文件增量读取器 + 轮询 ----
self._reader_session = _TailReader(DEBUG_LOG_PATH)
self._readers_app = {tag: _TailReader(p) for tag, p in _APP_LOGS}
self._timer = QTimer(self)
self._timer.setInterval(500)
self._timer.timeout.connect(self._tick)
self._timer.start()
# ==================== 轮询 ====================
def _tick(self):
try:
if not self._announced:
self._announced = True
debug_log("调试窗口开启", "SYS")
if not self._paused:
data = self._reader_session.read_new()
if data:
self._append_tagged(self._view_session, data)
for tag, _p in _APP_LOGS:
d = self._readers_app[tag].read_new()
if d:
self._view_app.appendPlainText(
f"── [{tag}] {os.path.basename(_p)} ──")
self._view_app.appendPlainText(d.rstrip("\n"))
except Exception:
pass
def _append_tagged(self, view: QPlainTextEdit, data: str):
cur = view.textCursor()
cur.movePosition(QTextCursor.MoveOperation.End)
for line in data.splitlines():
if not line:
continue
m = _LINE_RE.match(line)
color = _TAG_COLORS.get(m.group(2)) if m else None
fmt = QTextCharFormat()
if color:
fmt.setForeground(QColor(color))
cur.insertText(line + "\n", fmt)
view.setTextCursor(cur)
view.ensureCursorVisible()
# ==================== 交互 ====================
def _on_submit(self):
text = self._input.text().strip()
if not text:
return
debug_log(text, "USER")
self._input.clear()
def _on_pause_toggled(self, checked: bool):
self._paused = checked
def _on_top_toggled(self, checked: bool):
f = self.windowFlags()
if checked:
f |= Qt.WindowType.WindowStaysOnTopHint
else:
f &= ~Qt.WindowType.WindowStaysOnTopHint
self.setWindowFlags(f)
self.show() # setWindowFlags 会隐藏窗口,需重新 show
def _on_clear(self):
if QMessageBox.question(
self, "清空会话日志",
"将清空 debug_session.logUSER/AGENT/APP 记录全部丢失),确定?") \
!= QMessageBox.StandardButton.Yes:
return
try:
open(DEBUG_LOG_PATH, "w", encoding="utf-8").close()
self._reader_session.offset = 0
self._view_session.clear()
debug_log("会话日志已清空", "SYS")
except Exception:
pass
def _on_open_file(self):
try:
import subprocess
if os.name == "nt":
os.startfile(DEBUG_LOG_PATH) # noqa
else:
subprocess.Popen(["xdg-open", DEBUG_LOG_PATH])
except Exception:
pass
File diff suppressed because it is too large Load Diff
+7
View File
@@ -0,0 +1,7 @@
# -*- coding: utf-8 -*-
"""
系统级工具集合 (system_tools)
- screen_capture: 屏幕截图覆盖层
- global_hotkey: 全局热键监听线程
- file_reader: 文本/代码文件读取(编码探测 + 二进制探测)
"""
+71
View File
@@ -0,0 +1,71 @@
# -*- coding: utf-8 -*-
"""
文本/代码文件读取工具 (file_reader)
编码探测 + 二进制探测 + 大小守卫,供附件系统(拖拽/粘贴/上传文本与代码文件)复用。
用法:read_text_file(path) -> (content, encoding, size_kb, lines),失败抛 ValueError。
说明:纯函数模块,不依赖 Qt,便于单元测试。
"""
import os
# 明确拒绝的二进制/文档格式(Word/Excel/PPT/PDF/压缩包/可执行/媒体等)
BINARY_EXTS = frozenset({
# Office 文档(明确排除 Word/Excel
'.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx',
'.odt', '.ods', '.odp', '.rtf',
# 电子书(.pdf 已交由 tools/builtin_tools/pdf_reader.py 专门解析,不在此拦截)
'.epub', '.mobi',
# 压缩包 / 磁盘镜像
'.zip', '.rar', '.7z', '.tar', '.gz', '.bz2', '.xz', '.iso', '.dmg',
# 可执行 / 编译产物
'.exe', '.dll', '.so', '.dylib', '.msi', '.bin', '.apk', '.jar',
'.class', '.pyc', '.pyd', '.o', '.a',
# 音视频
'.mp3', '.wav', '.flac', '.m4a', '.aac', '.ogg',
'.mp4', '.avi', '.mov', '.mkv', '.webm', '.wmv',
# 字体
'.ttf', '.otf', '.woff', '.woff2', '.eot',
# 数据库
'.db', '.sqlite', '.sqlite3', '.mdb',
# 设计稿
'.psd', '.ai', '.sketch', '.fig',
})
# 单文件大小上限(字节):超过则拒绝,避免内存与上下文爆炸
MAX_ATTACH_FILE_BYTES = 2 * 1024 * 1024 # 2 MB
# 编码降级链:UTF-8(含BOM) → GB18030(⊇GBK/GB2312) → latin-1(永不失败)
_ENCODINGS = ('utf-8-sig', 'gb18030', 'latin-1')
def read_text_file(path, max_bytes=MAX_ATTACH_FILE_BYTES):
"""读取一个文本/代码文件。
返回 (content, encoding, size_kb, lines)。
文件过大或判定为二进制时抛出 ValueError(附带可读原因)。
"""
size = os.path.getsize(path)
if size > max_bytes:
raise ValueError(
f"文件过大({size / 1024 / 1024:.1f} MB > {max_bytes / 1024 / 1024:.0f} MB"
)
with open(path, 'rb') as f:
raw = f.read()
# 二进制探测:前 8KB 含 NUL 字节即判定为二进制(git 同款启发式)
if b'\x00' in raw[:8192]:
raise ValueError("二进制文件,无法作为文本读取")
text, used_enc = None, 'utf-8'
for enc in _ENCODINGS:
try:
text = raw.decode(enc)
used_enc = enc
break
except UnicodeDecodeError:
continue
if text is None: # 理论上 latin-1 兜底永不失败
raise ValueError("无法识别文件编码")
lines = text.count('\n') + 1
return text, used_enc, round(size / 1024, 2), lines
+105
View File
@@ -0,0 +1,105 @@
# -*- coding: utf-8 -*-
"""
全局热键监听器 (Windows)
独立线程 + Win32 RegisterHotKey 消息循环,实现主窗口失焦也能触发的系统级全局快捷键。
用法:
hotkey = GlobalHotkeyThread() # 默认 Alt+S
hotkey.triggered.connect(on_triggered) # 跨线程信号,槽函数在主线程执行
hotkey.start() # 启动线程
hotkey.stop() # 注销热键并退出线程
"""
import ctypes
import sys
import threading
from ctypes import wintypes
from PyQt6 import QtCore
# Win32 常量
WM_HOTKEY = 0x0312
WM_QUIT = 0x0012
MOD_ALT = 0x0001
MOD_CONTROL = 0x0002
MOD_SHIFT = 0x0004
MOD_NOREPEAT = 0x4000 # 按住不重复触发 (Windows 7+)
VK_S = 0x53
_is_windows = sys.platform == "win32"
if _is_windows:
_user32 = ctypes.windll.user32
_kernel32 = ctypes.windll.kernel32
# 显式声明函数签名,避免 ctypes 默认 int 截断指针/句柄
_user32.RegisterHotKey.argtypes = [wintypes.HWND, ctypes.c_int, wintypes.UINT, wintypes.UINT]
_user32.RegisterHotKey.restype = wintypes.BOOL
_user32.UnregisterHotKey.argtypes = [wintypes.HWND, ctypes.c_int]
_user32.UnregisterHotKey.restype = wintypes.BOOL
_user32.GetMessageW.argtypes = [
ctypes.POINTER(wintypes.MSG), wintypes.HWND, wintypes.UINT, wintypes.UINT
]
_user32.GetMessageW.restype = wintypes.BOOL # >0 正常 / 0 WM_QUIT / -1 出错
_user32.PostThreadMessageW.argtypes = [
wintypes.DWORD, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM
]
_user32.PostThreadMessageW.restype = wintypes.BOOL
_kernel32.GetCurrentThreadId.restype = wintypes.DWORD
class GlobalHotkeyThread(QtCore.QThread):
"""在独立线程注册系统级全局热键并运行 Win32 消息循环。
收到 WM_HOTKEY 后通过 Qt 信号 triggered 通知主线程,
信号槽机制保证槽函数在主线程执行,操作 Qt 控件安全。
"""
triggered = QtCore.pyqtSignal()
def __init__(self, hotkey_id=9001, mod=MOD_ALT, vk=VK_S, parent=None):
super().__init__(parent)
self._hotkey_id = hotkey_id
self._mod = mod
self._vk = vk
self._thread_id = 0
self._ready = threading.Event() # run() 记录线程 ID 后置位
self._registered = False
def run(self):
"""线程入口:注册热键 -> 消息循环 -> 退出时注销"""
if not _is_windows:
self._ready.set()
return
self._thread_id = _kernel32.GetCurrentThreadId()
ok = _user32.RegisterHotKey(
None, self._hotkey_id, self._mod | MOD_NOREPEAT, self._vk
)
self._registered = bool(ok)
self._ready.set()
if not ok:
print(
f"[GlobalHotkey] 注册热键失败 (id={self._hotkey_id}, mod={self._mod:#x}, "
f"vk={self._vk:#x}),可能已被其他程序占用"
)
return
# Win32 消息循环:hwnd=None 取本线程所有消息
msg = wintypes.MSG()
while _user32.GetMessageW(ctypes.byref(msg), None, 0, 0) > 0:
if msg.message == WM_HOTKEY and msg.wParam == self._hotkey_id:
self.triggered.emit()
# 收到 WM_QUIT 退出循环 -> 注销热键
_user32.UnregisterHotKey(None, self._hotkey_id)
self._registered = False
def stop(self):
"""请求线程退出:向线程消息队列投递 WM_QUIT,然后等待结束"""
if not _is_windows:
return
# 等 run() 至少记录好线程 ID(注册成功或失败都行)
self._ready.wait(timeout=2.0)
if self._thread_id:
_user32.PostThreadMessageW(self._thread_id, WM_QUIT, 0, 0)
self.wait(2000)
+193
View File
@@ -0,0 +1,193 @@
# -*- coding: utf-8 -*-
"""
屏幕截图覆盖层 (ScreenCaptureOverlay)
全屏半透明遮罩 + 鼠标拖拽选区 + 确认/取消工具条 + 截图完成发射 QImage 信号
用法:调用 start() 启动截图,监听 screenshot_captured 信号获取结果
"""
from PyQt6 import QtCore, QtGui, QtWidgets
import os
class ScreenCaptureOverlay(QtWidgets.QWidget):
"""全屏截图覆盖层:半透明遮罩 + 鼠标拖拽选区 + 确认/取消按钮"""
screenshot_captured = QtCore.pyqtSignal(QtGui.QImage)
# 按钮尺寸
BTN_W = 36
BTN_H = 30
BTN_GAP = 4
def __init__(self, parent=None):
super().__init__(parent)
self.setWindowFlags(
QtCore.Qt.WindowType.FramelessWindowHint
| QtCore.Qt.WindowType.WindowStaysOnTopHint
| QtCore.Qt.WindowType.Tool
)
self.setCursor(QtCore.Qt.CursorShape.CrossCursor)
self._full_pixmap = None
self._start_pos = QtCore.QPoint()
self._end_pos = QtCore.QPoint()
self._is_drawing = False
self._has_selection = False
self._current_rect = QtCore.QRect()
# SVG 图标路径(项目根目录下 svg/ 文件夹)
root_dir = os.path.abspath(
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "..")
)
# 确认 / 取消按钮(使用 SVG 图标,灰白简约风背景)
self._btn_confirm = QtWidgets.QPushButton(self)
self._btn_cancel = QtWidgets.QPushButton(self)
self._btn_confirm.setIcon(QtGui.QIcon(os.path.join(root_dir, "svg", "check.svg")))
self._btn_cancel.setIcon(QtGui.QIcon(os.path.join(root_dir, "svg", "cross.svg")))
btn_style = (
"QPushButton { background-color: #f9f9f9; border: 1px solid #ddd; border-radius: 4px; }"
"QPushButton:hover { background-color: #eee; border-color: #ccc; }"
)
for btn in (self._btn_confirm, self._btn_cancel):
btn.setFixedSize(self.BTN_W, self.BTN_H)
btn.setIconSize(QtCore.QSize(20, 20))
btn.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
btn.setStyleSheet(btn_style)
btn.hide()
self._btn_confirm.clicked.connect(self._on_confirm)
self._btn_cancel.clicked.connect(self._on_cancel)
def start(self):
"""开始截图:抓取屏幕全图并显示覆盖层"""
screen = QtWidgets.QApplication.primaryScreen()
if not screen:
return
self._full_pixmap = screen.grabWindow(0)
self.setGeometry(screen.geometry())
self.show()
self.activateWindow()
self.raise_()
def paintEvent(self, event):
if not self._full_pixmap:
return
painter = QtGui.QPainter(self)
# 1. 绘制屏幕截图作为背景
painter.drawPixmap(0, 0, self._full_pixmap)
# 2. 半透明遮罩
painter.fillRect(self.rect(), QtGui.QColor(0, 0, 0, 100))
# 绘制中 或 已有选区 时,绘制选区高亮
if self._is_drawing or self._has_selection:
if self._is_drawing:
rect = QtCore.QRect(self._start_pos, self._end_pos).normalized()
else:
rect = self._current_rect
if rect.width() > 0 and rect.height() > 0:
# 3. 选区内重绘原图(去掉遮罩,形成高亮效果)
painter.drawPixmap(rect, self._full_pixmap, rect)
# 4. 蓝色边框
pen = QtGui.QPen(QtGui.QColor(0, 120, 215), 2)
painter.setPen(pen)
painter.setBrush(QtCore.Qt.BrushStyle.NoBrush)
painter.drawRect(rect)
# 5. 尺寸标注
size_text = f"{rect.width()} x {rect.height()}"
font = painter.font()
font.setPointSize(9)
painter.setFont(font)
fm = painter.fontMetrics()
text_w = fm.horizontalAdvance(size_text)
text_h = fm.height()
text_x = rect.x()
text_y = rect.y() - text_h - 2
if text_y < 0:
text_y = rect.bottom() + 2
painter.fillRect(text_x, text_y, text_w + 10, text_h, QtGui.QColor(0, 120, 215))
painter.setPen(QtGui.QColor(255, 255, 255))
painter.drawText(text_x + 5, text_y + fm.ascent(), size_text)
painter.end()
def mousePressEvent(self, event):
if event.button() == QtCore.Qt.MouseButton.LeftButton:
# 开始新选区,隐藏按钮
self._hide_buttons()
self._has_selection = False
self._start_pos = event.position().toPoint()
self._end_pos = self._start_pos
self._is_drawing = True
self.update()
elif event.button() == QtCore.Qt.MouseButton.RightButton:
self.close()
def mouseMoveEvent(self, event):
if self._is_drawing:
self._end_pos = event.position().toPoint()
self.update()
def mouseReleaseEvent(self, event):
if event.button() == QtCore.Qt.MouseButton.LeftButton and self._is_drawing:
self._is_drawing = False
rect = QtCore.QRect(self._start_pos, self._end_pos).normalized()
if rect.width() > 5 and rect.height() > 5:
# 保留选区,显示确认/取消按钮
self._current_rect = rect
self._has_selection = True
self._position_buttons(rect)
self._btn_confirm.show()
self._btn_cancel.show()
self.update()
else:
self.close()
def keyPressEvent(self, event):
if event.key() == QtCore.Qt.Key.Key_Escape:
self.close()
elif event.key() in (QtCore.Qt.Key.Key_Return, QtCore.Qt.Key.Key_Enter):
if self._has_selection:
self._on_confirm()
def _position_buttons(self, rect):
"""将确认/取消按钮定位到选区右下角"""
total_w = self.BTN_W * 2 + self.BTN_GAP
# 默认放在选区右下角外侧
x = rect.right() - total_w
y = rect.bottom() + 4
# 边界检测:超出屏幕底部时翻到选区内侧
if y + self.BTN_H > self.height():
y = rect.bottom() - self.BTN_H - 4
if x < 0:
x = 0
self._btn_confirm.move(x, y)
self._btn_cancel.move(x + self.BTN_W + self.BTN_GAP, y)
def _hide_buttons(self):
self._btn_confirm.hide()
self._btn_cancel.hide()
def _on_confirm(self):
"""确认截图:裁剪并发射信号"""
if self._full_pixmap and self._current_rect.width() > 5 and self._current_rect.height() > 5:
dpr = self._full_pixmap.devicePixelRatio()
phys_rect = QtCore.QRect(
int(self._current_rect.x() * dpr),
int(self._current_rect.y() * dpr),
int(self._current_rect.width() * dpr),
int(self._current_rect.height() * dpr),
)
captured = self._full_pixmap.toImage().copy(phys_rect)
self.screenshot_captured.emit(captured)
self.close()
def _on_cancel(self):
"""取消截图:直接关闭"""
self.close()
def closeEvent(self, event):
self._is_drawing = False
self._has_selection = False
self._hide_buttons()
self._full_pixmap = None
super().closeEvent(event)
+217
View File
@@ -0,0 +1,217 @@
# -*- coding: utf-8 -*-
"""
WebView2View —— 外观兼容 QWebEngineView 的 QWidget(仅实现 main_window 用到的 API 面):
setUrl / loadFinished / page().runJavaScript / page().setBackgroundColor / grab
内部由 core.webview2.Wv2Session 驱动:
- 30ms timer + resizeEvent 双路 SetBoundsAndZoomFactorslot 矩形 → 父窗客户区物理像素)
- WebMessageReceived → ChatBridge 的 JS→Python slot
- run_js → ExecuteScriptAsync(与 QWebChannel 路径生成的 JS 文本同构)
"""
import json
import ctypes
import ctypes.wintypes as wintypes
from PyQt6 import QtCore, QtGui, QtWidgets
from PyQt6.QtCore import Qt, pyqtSignal, QTimer, QUrl
# JS → Python 消息名 → ChatBridge slot 名(与 QWebChannel 注册的对象方法一一对应)
_BRIDGE_METHODS = (
"onRegenerateClicked",
"onBranchSwitch",
"onDeleteMessageClicked",
"onAttachmentClicked",
"onScrollChanged",
)
class _PageShim:
"""QWebEnginePage 的最小替身"""
def __init__(self, view):
self._view = view
def runJavaScript(self, script, callback=None):
if callback is None:
self._view.run_js(script)
else:
self._view.execute_js_async(script, callback)
def setBackgroundColor(self, color):
pass # WebView2 底色由页面 CSS 决定(页面本身就是白底)
class WebView2View(QtWidgets.QWidget):
loadFinished = pyqtSignal(bool)
def __init__(self, session, parent=None):
super().__init__(parent)
self._session = session
self._page = _PageShim(self)
self._bridge = None
self._session.on_load_finished = self._on_page_loaded
self.setAttribute(Qt.WidgetAttribute.WA_OpaquePaintEvent, True)
# 子窗口发现(异步创建)
self._child_timer = QTimer(self)
self._child_timer.setInterval(100)
self._child_timer.timeout.connect(self._try_find_child)
self._child_timer.start()
# bounds 同步兜底(resizeEvent 主路 + 30ms 兜底,与 P1 实测一致)
self._sync_timer = QTimer(self)
self._sync_timer.setInterval(30)
self._sync_timer.timeout.connect(self.sync_bounds)
self._sync_timer.start()
# 几何去重:值没变就不发跨进程 COM(移动窗口时避免主线程被 WebView2 阻塞)
self._last_sent_bounds = None
# ---------- 页面加载完成 ----------
def _on_page_loaded(self):
# 立即启动 JS 管线(子窗口全程可见,无节流/竞态)。
# 首帧前的"透底色"问题由 Qt 侧白色 paintEvent 解决(透明子窗表面露出白底)。
self.loadFinished.emit(True)
def paintEvent(self, ev):
# 子窗口隐藏期间(启动首帧前)槽位必须自绘白底,
# 否则 WA_OpaquePaintEvent + 无 paintEvent = 未绘制表面 = 纯黑
p = QtGui.QPainter(self)
p.fillRect(self.rect(), QtGui.QColor("#ffffff"))
p.end()
# ---------- QWebEngineView 兼容 API ----------
def page(self):
return self._page
def setUrl(self, url):
# 接受 QUrl 或 str
s = url.toString() if isinstance(url, QUrl) else str(url)
self._session.navigate(s)
def grab(self):
"""子窗口区域截图(调试工具用)"""
h = self._session.child_hwnd
if not h:
return QtGui.QPixmap(self.size())
r = wintypes.RECT()
if not ctypes.windll.user32.GetWindowRect(wintypes.HWND(h), ctypes.byref(r)):
return QtGui.QPixmap(self.size())
w = max(1, r.right - r.left)
hgt = max(1, r.bottom - r.top)
screen_dc = ctypes.windll.user32.GetDC(0)
mem_dc = ctypes.windll.gdi32.CreateCompatibleDC(screen_dc)
bmp = ctypes.windll.gdi32.CreateCompatibleBitmap(screen_dc, w, hgt)
ctypes.windll.gdi32.SelectObject(mem_dc, bmp)
ctypes.windll.gdi32.BitBlt(mem_dc, 0, 0, w, hgt, screen_dc, r.left, r.top, 0x00CC0020)
# BITMAPINFO
class _BMI(ctypes.Structure):
_fields_ = [("biSize", wintypes.DWORD), ("biWidth", ctypes.c_long),
("biHeight", ctypes.c_long), ("biPlanes", wintypes.WORD),
("biBitCount", wintypes.WORD), ("biCompression", wintypes.DWORD),
("biSizeImage", wintypes.DWORD), ("biXPelsPerMeter", ctypes.c_long),
("biYPelsPerMeter", ctypes.c_long), ("biClrUsed", wintypes.DWORD),
("biClrImportant", wintypes.DWORD)]
bmi = _BMI()
bmi.biSize = ctypes.sizeof(_BMI)
bmi.biWidth = w
bmi.biHeight = -hgt # 顶向下
bmi.biPlanes = 1
bmi.biBitCount = 32
bmi.biCompression = 0
buf = ctypes.create_string_buffer(w * hgt * 4)
ctypes.windll.gdi32.GetDIBits(mem_dc, bmp, 0, hgt, buf, ctypes.byref(bmi), 0)
ctypes.windll.gdi32.DeleteObject(bmp)
ctypes.windll.gdi32.DeleteDC(mem_dc)
ctypes.windll.user32.ReleaseDC(0, screen_dc)
img = QtGui.QImage(buf, w, hgt, w * 4, QtGui.QImage.Format.Format_RGBA8888).copy()
return QtGui.QPixmap.fromImage(img)
# ---------- 桥接 ----------
def attach_bridge(self, bridge):
self._bridge = bridge
self._session.on_message = self._dispatch_message
def _dispatch_message(self, data):
if not isinstance(data, dict) or not self._bridge:
return
name = data.get("m")
args = data.get("a", [])
if name in _BRIDGE_METHODS:
fn = getattr(self._bridge, name, None)
if fn:
try:
fn(*args)
except Exception as ex:
print(f"[WV2] bridge {name} error:", ex)
# ---------- JS → Python 用的执行通道 ----------
def run_js(self, script):
self._session.execute_js(script)
def execute_js_async(self, script, cb):
self._session.execute_js_async(script, cb)
# ---------- 几何同步 ----------
def _dpr(self):
try:
d = self.dpr()
return d if d > 0 else 1.0
except Exception:
return 1.0
def sync_bounds(self):
if not self.isVisible() or not self._session.child_hwnd:
return
top = self.window()
p = self.mapTo(top, QtCore.QPoint(0, 0))
d = self._dpr()
L, T = p.x() * d, p.y() * d
W, H = max(1, self.width() * d), max(1, self.height() * d)
# 去重:值没变就不发跨进程 COM(移动窗口时避免主线程被 WebView2 阻塞 → 整窗黑屏)
# 注:旧版 resize hold(放大时钳制旧尺寸→露白边)已移除 ——
# 拖动黑边的真凶是假异步 JS 泵(已修真异步),高频 SetBounds 实测 Chromium 完全跟得上;
# 用户实测:立即发送 = 缩小/放大都实时渲染、无阻塞。
if self._last_sent_bounds == (L, T, W, H):
return
self._session.set_bounds(L, T, W, H)
self._last_sent_bounds = (L, T, W, H)
def resizeEvent(self, ev):
super().resizeEvent(ev)
self.sync_bounds()
def showEvent(self, ev):
super().showEvent(ev)
# 确保渲染层点亮(SDK 初始状态下控制器可能不可见)
self._session.set_visible(True)
self.sync_bounds()
def _force_repaint(self):
"""WebView2 子窗口创建会打断首帧合成 → 强制整窗重绘(治启动时控件渲染丢失)"""
top = self.window()
if top is None:
return
top.update()
QtWidgets.QApplication.processEvents()
def _try_find_child(self):
if self._session.child_hwnd:
self._child_timer.stop()
self._session.set_visible(True)
self.sync_bounds()
self._force_repaint()
QtCore.QTimer.singleShot(300, self._force_repaint)
return
if self._session.find_child_once():
self._child_timer.stop()
self._session.set_visible(True)
self.sync_bounds()
self._force_repaint()
QtCore.QTimer.singleShot(300, self._force_repaint)
# ---------- 生命周期 ----------
def closeEvent(self, ev):
try:
self._child_timer.stop()
self._sync_timer.stop()
self._session.close()
except Exception:
pass
super().closeEvent(ev)