diff --git a/crash_watcher.py b/crash_watcher.py new file mode 100644 index 0000000..858a37b --- /dev/null +++ b/crash_watcher.py @@ -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}") diff --git a/main_gui.py b/main_gui.py index b0587d8..b14382e 100644 --- a/main_gui.py +++ b/main_gui.py @@ -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, @@ -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) @@ -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") @@ -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) @@ -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 diff --git a/tests/test_crash_watcher.py b/tests/test_crash_watcher.py new file mode 100644 index 0000000..ec114f1 --- /dev/null +++ b/tests/test_crash_watcher.py @@ -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