Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
101 changes: 101 additions & 0 deletions crash_watcher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
"""CrashWatcher module recording recent events and emitting crash reports."""

from __future__ import annotations

import atexit
import threading
import time
from collections import deque
from pathlib import Path
from typing import Deque

try:
import tkinter as tk
from tkinter import messagebox
from tkinter.scrolledtext import ScrolledText
except Exception: # pragma: no cover - tkinter may be unavailable in some envs
tk = None # type: ignore
messagebox = None # type: ignore
ScrolledText = None # type: ignore


class CrashWatcher(threading.Thread):
"""Background thread storing recent events in a circular buffer."""

def __init__(self, max_events: int = 100):
super().__init__(daemon=True)
self._events: Deque[str] = deque(maxlen=max_events)
self._lock = threading.Lock()

def record_event(self, msg: str) -> None:
ts = time.strftime("%Y-%m-%d %H:%M:%S")
with self._lock:
self._events.append(f"{ts} - {msg}")

def dump_events(self) -> str:
with self._lock:
return "\n".join(self._events)

def run(self) -> None: # pragma: no cover - thread performs no active work
while True:
time.sleep(1)


_watcher: CrashWatcher | None = None
clean_shutdown = False
_docs_dir: Path | None = None


def start(max_events: int = 100) -> None:
"""Start the global crash watcher thread and register exit handler."""
global _watcher
if _watcher is None:
_watcher = CrashWatcher(max_events=max_events)
_watcher.start()
atexit.register(_handle_exit)


def record_event(msg: str) -> None:
"""Record an event in the crash buffer."""
if _watcher is not None:
_watcher.record_event(msg)


def mark_clean_shutdown() -> None:
"""Set the global clean shutdown flag."""
global clean_shutdown
clean_shutdown = True


def set_library_path(root: str | Path) -> None:
"""Configure the library root for crash reports."""
global _docs_dir
root_path = Path(root)
_docs_dir = root_path / "Docs"
_docs_dir.mkdir(parents=True, exist_ok=True)


def _handle_exit() -> None:
if clean_shutdown or _watcher is None:
return

report_dir = _docs_dir or Path(__file__).resolve().parent / "docs"
report_path = report_dir / "crash_report.txt"
data = _watcher.dump_events()
report_dir.mkdir(parents=True, exist_ok=True)
report_path.write_text(data, encoding="utf-8")

if tk and ScrolledText:
try:
root = tk.Tk()
root.title("Crash Report")
text = ScrolledText(root, width=80, height=24)
text.pack(fill="both", expand=True)
text.insert("1.0", data)
text.configure(state="disabled")
tk.Button(root, text="Close", command=root.destroy).pack(pady=5)
root.mainloop()
except Exception:
pass
else: # pragma: no cover - in headless environments
print(f"Crash report written to {report_path}")
11 changes: 10 additions & 1 deletion main_gui.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@
from indexer_control import cancel_event, IndexCancelled
import library_sync
import playlist_generator
import crash_watcher

from controllers.library_controller import (
load_last_path,
Expand Down Expand Up @@ -551,7 +552,7 @@ def build_ui(self):
file_menu.add_command(label="Compare Libraries", command=self.compare_libraries)
file_menu.add_command(label="Show All Files", command=self._on_show_all)
file_menu.add_separator()
file_menu.add_command(label="Exit", command=self.quit)
file_menu.add_command(label="Exit", command=self._on_exit)
menubar.add_cascade(label="File", menu=file_menu)

settings_menu = tk.Menu(menubar, tearoff=False)
Expand Down Expand Up @@ -1068,6 +1069,7 @@ def select_library(self):

info = open_library(chosen)
self.library_path = info["path"]
crash_watcher.set_library_path(self.library_path)
self.library_name_var.set(info["name"])
self.library_path_var.set(info["path"])
self.mapping_path = os.path.join(self.library_path, ".genre_mapping.json")
Expand Down Expand Up @@ -2600,8 +2602,14 @@ def open_metadata_settings(self):
frame.pack(fill="both", expand=True, padx=10, pady=10)
self._metadata_win = win

def _on_exit(self) -> None:
"""Triggered by File→Exit to mark a clean shutdown."""
crash_watcher.mark_clean_shutdown()
self.quit()

def _on_close(self):
"""Handle application close event."""
crash_watcher.record_event("WM_DELETE_WINDOW")
self.preview_player.stop_preview()
if self._preview_thread and self._preview_thread.is_alive():
self._preview_thread.join(timeout=0.1)
Expand Down Expand Up @@ -2647,6 +2655,7 @@ def _log(self, msg):

LOG_PATH = "soundvault_crash.log"
install_crash_logger(log_path=LOG_PATH, level=logging.DEBUG)
crash_watcher.start()
app = SoundVaultImporterApp()
add_context_provider(lambda: {"library_path": app.library_path})
app.log_path = LOG_PATH
Expand Down
16 changes: 16 additions & 0 deletions tests/test_crash_watcher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
import crash_watcher

def test_crash_report_written(monkeypatch, tmp_path):
crash_watcher._watcher = None # ensure clean state
crash_watcher.clean_shutdown = False
crash_watcher.start()
crash_watcher.record_event("test event")
crash_watcher.set_library_path(tmp_path)
monkeypatch.setattr(crash_watcher, "tk", None)
monkeypatch.setattr(crash_watcher, "ScrolledText", None)
crash_watcher._handle_exit()
report = tmp_path / "Docs" / "crash_report.txt"
assert report.exists()
assert "test event" in report.read_text()
crash_watcher.clean_shutdown = True
crash_watcher._watcher = None