sorrow404null
a7412824e0
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.
215 lines
8.0 KiB
Python
215 lines
8.0 KiB
Python
# -*- 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.log（USER/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
|