|
| 1 | +#!/usr/bin/env python3 |
| 2 | +# -*- coding: utf-8 -*- |
| 3 | +"""TODO & Aufgaben Scanner Engine für CodeBox. |
| 4 | +
|
| 5 | +Stellt Datenstrukturen, reguläre Mustererkennung für standardisierte Aufgaben-Tags |
| 6 | +(TODO, FIXME, BUG, HACK, XXX, NOTE), Einzelfile-Scanner und einen asynchronen |
| 7 | +Hintergrund-Worker (QThread) für Multi-Root-Arbeitsbereiche bereit. |
| 8 | +""" |
| 9 | + |
| 10 | +from __future__ import annotations |
| 11 | + |
| 12 | +import os |
| 13 | +import re |
| 14 | +from dataclasses import dataclass |
| 15 | +from pathlib import Path |
| 16 | +from typing import Dict, Iterable, List, Optional, Set, Tuple |
| 17 | + |
| 18 | +from PySide6.QtCore import QObject, QThread, Signal |
| 19 | + |
| 20 | +from core.file_search import is_binary_file |
| 21 | +from core.workspace import DEFAULT_WORKSPACE_SKIP_DIRS |
| 22 | + |
| 23 | +DEFAULT_TODO_TAGS = ("TODO", "FIXME", "BUG", "HACK", "XXX", "NOTE") |
| 24 | + |
| 25 | +# Regulärer Ausdruck für Standard-Kommentare und Aufgaben-Tags |
| 26 | +# Erkennt Kommentare (#, //, /*, <!--, --, %, @, ;, *, - [ ], - [x]) |
| 27 | +# oder Zeilenanfang/Whitespace mit explizitem Tag gefolgt von Doppelpunkt. |
| 28 | +TODO_PATTERN = re.compile( |
| 29 | + r'(?:(?:#|//|/\*+|<!--|--|%|@|;\s*|\*\s*|-\s*\[[ xX]\]\s*)\s*(TODO|FIXME|BUG|HACK|XXX|NOTE)\b(?:\(([^)]+)\))?\s*:?\s*(.*))' |
| 30 | + r'|(?:^\s*(TODO|FIXME|BUG|HACK|XXX|NOTE)\b(?:\(([^)]+)\))?\s*:\s*(.*))', |
| 31 | + re.IGNORECASE, |
| 32 | +) |
| 33 | + |
| 34 | + |
| 35 | +@dataclass |
| 36 | +class TodoItem: |
| 37 | + """Repräsentiert eine gefundene Aufgabe im Quellcode.""" |
| 38 | + |
| 39 | + file_path: Path |
| 40 | + rel_path: str |
| 41 | + folder_name: str |
| 42 | + line_number: int # 1-basiert |
| 43 | + column: int # 1-basiert |
| 44 | + tag: str # "TODO", "FIXME", "BUG", "HACK", "XXX", "NOTE" |
| 45 | + author: Optional[str] # z. B. "lukas" in TODO(lukas): |
| 46 | + text: str # Beschreibungstext nach dem Tag |
| 47 | + line_text: str # Gesamte Quellcodezeile für Kontext und Vorschau |
| 48 | + |
| 49 | + |
| 50 | +def scan_file_for_todos( |
| 51 | + file_path: Path, |
| 52 | + rel_path: str = "", |
| 53 | + folder_name: str = "", |
| 54 | + tags: Optional[Iterable[str]] = None, |
| 55 | + max_file_size_kb: int = 4096, |
| 56 | +) -> List[TodoItem]: |
| 57 | + """Durchsucht eine einzelne Datei nach TODO/FIXME-Aufgabenkommentaren.""" |
| 58 | + try: |
| 59 | + size = file_path.stat().st_size |
| 60 | + if size > max_file_size_kb * 1024: |
| 61 | + return [] |
| 62 | + except (OSError, PermissionError): |
| 63 | + return [] |
| 64 | + |
| 65 | + if is_binary_file(file_path): |
| 66 | + return [] |
| 67 | + |
| 68 | + allowed_tags: Optional[Set[str]] = {t.upper() for t in tags} if tags else None |
| 69 | + |
| 70 | + items: List[TodoItem] = [] |
| 71 | + encodings = ["utf-8", "cp1252", "latin-1"] |
| 72 | + |
| 73 | + lines: Optional[List[str]] = None |
| 74 | + for enc in encodings: |
| 75 | + try: |
| 76 | + with open(file_path, "r", encoding=enc) as f: |
| 77 | + lines = f.readlines() |
| 78 | + break |
| 79 | + except (UnicodeDecodeError, OSError): |
| 80 | + continue |
| 81 | + |
| 82 | + if lines is None: |
| 83 | + return [] |
| 84 | + |
| 85 | + for line_idx, raw_line in enumerate(lines, start=1): |
| 86 | + clean_line = raw_line.rstrip("\r\n") |
| 87 | + m = TODO_PATTERN.search(clean_line) |
| 88 | + if not m: |
| 89 | + continue |
| 90 | + |
| 91 | + if m.group(1): |
| 92 | + tag = m.group(1).upper() |
| 93 | + author = m.group(2).strip() if m.group(2) else None |
| 94 | + raw_text = m.group(3) or "" |
| 95 | + col = m.start(1) + 1 |
| 96 | + else: |
| 97 | + tag = m.group(4).upper() |
| 98 | + author = m.group(5).strip() if m.group(5) else None |
| 99 | + raw_text = m.group(6) or "" |
| 100 | + col = m.start(4) + 1 |
| 101 | + |
| 102 | + if allowed_tags and tag not in allowed_tags: |
| 103 | + continue |
| 104 | + |
| 105 | + # Schließende Kommentarzeichen (*/ oder -->) am Zeilenende entfernen |
| 106 | + text = re.sub(r'(\*/|-->)\s*$', '', raw_text).strip() |
| 107 | + |
| 108 | + items.append( |
| 109 | + TodoItem( |
| 110 | + file_path=file_path, |
| 111 | + rel_path=rel_path or file_path.name, |
| 112 | + folder_name=folder_name, |
| 113 | + line_number=line_idx, |
| 114 | + column=col, |
| 115 | + tag=tag, |
| 116 | + author=author, |
| 117 | + text=text, |
| 118 | + line_text=clean_line, |
| 119 | + ) |
| 120 | + ) |
| 121 | + |
| 122 | + return items |
| 123 | + |
| 124 | + |
| 125 | +class TodoScanWorker(QThread): |
| 126 | + """Hintergrund-Thread für den asynchronen Scan aller Workspace-Ordner.""" |
| 127 | + |
| 128 | + itemFound = Signal(object) # TodoItem |
| 129 | + fileCompleted = Signal(object, object) # Path, List[TodoItem] |
| 130 | + scanProgress = Signal(int, int) # scanned_count, total_count |
| 131 | + scanFinished = Signal(object) # List[TodoItem] |
| 132 | + |
| 133 | + def __init__( |
| 134 | + self, |
| 135 | + folders: List[Path], |
| 136 | + tags: Optional[Iterable[str]] = None, |
| 137 | + exclude_dirs: Optional[Set[str]] = None, |
| 138 | + max_file_size_kb: int = 4096, |
| 139 | + parent: Optional[QObject] = None, |
| 140 | + ): |
| 141 | + super().__init__(parent) |
| 142 | + self.folders = [Path(f) for f in folders if Path(f).is_dir()] |
| 143 | + self.tags = tags |
| 144 | + self.exclude_dirs = exclude_dirs or set(DEFAULT_WORKSPACE_SKIP_DIRS) |
| 145 | + self.max_file_size_kb = max_file_size_kb |
| 146 | + self._is_stopped = False |
| 147 | + |
| 148 | + def stop(self): |
| 149 | + """Bricht den laufenden Scan sicher ab.""" |
| 150 | + self._is_stopped = True |
| 151 | + |
| 152 | + def run(self): |
| 153 | + """Führt den Scan im Hintergrund aus.""" |
| 154 | + candidate_files: List[Tuple[Path, str, str]] = [] |
| 155 | + |
| 156 | + # 1. Dateien einsammeln (mit Ausschluss von VCS- und Build-Ordnern) |
| 157 | + for folder in self.folders: |
| 158 | + if self._is_stopped: |
| 159 | + return |
| 160 | + folder_name = folder.name |
| 161 | + for root, dirs, files in os.walk(folder): |
| 162 | + if self._is_stopped: |
| 163 | + return |
| 164 | + # Verzeichnisse in-place filtern |
| 165 | + dirs[:] = [ |
| 166 | + d for d in dirs |
| 167 | + if d not in self.exclude_dirs and not d.startswith(".") |
| 168 | + ] |
| 169 | + for file in files: |
| 170 | + if file.startswith("."): |
| 171 | + continue |
| 172 | + file_path = Path(root) / file |
| 173 | + try: |
| 174 | + rel = str(file_path.relative_to(folder)) |
| 175 | + except ValueError: |
| 176 | + rel = file_path.name |
| 177 | + candidate_files.append((file_path, rel, folder_name)) |
| 178 | + |
| 179 | + total_files = len(candidate_files) |
| 180 | + all_items: List[TodoItem] = [] |
| 181 | + |
| 182 | + # 2. Dateien scannen |
| 183 | + for idx, (file_path, rel_path, folder_name) in enumerate(candidate_files, start=1): |
| 184 | + if self._is_stopped: |
| 185 | + return |
| 186 | + |
| 187 | + if idx % 20 == 0 or idx == total_files: |
| 188 | + self.scanProgress.emit(idx, total_files) |
| 189 | + |
| 190 | + file_items = scan_file_for_todos( |
| 191 | + file_path=file_path, |
| 192 | + rel_path=rel_path, |
| 193 | + folder_name=folder_name, |
| 194 | + tags=self.tags, |
| 195 | + max_file_size_kb=self.max_file_size_kb, |
| 196 | + ) |
| 197 | + |
| 198 | + if file_items: |
| 199 | + all_items.extend(file_items) |
| 200 | + self.fileCompleted.emit(file_path, file_items) |
| 201 | + for item in file_items: |
| 202 | + self.itemFound.emit(item) |
| 203 | + |
| 204 | + if not self._is_stopped: |
| 205 | + self.scanFinished.emit(all_items) |
| 206 | + |
| 207 | + |
| 208 | +class TodoScannerManager(QObject): |
| 209 | + """Verwaltet den Aufgaben-Cache und koordiniert asynchrone Workspace-Scans.""" |
| 210 | + |
| 211 | + scanStarted = Signal() |
| 212 | + scanProgress = Signal(int, int) # scanned, total |
| 213 | + scanFinished = Signal(int, int) # total_items, total_files |
| 214 | + todosUpdated = Signal(object) # List[TodoItem] |
| 215 | + |
| 216 | + def __init__(self, parent: Optional[QObject] = None): |
| 217 | + super().__init__(parent) |
| 218 | + self._cache: Dict[Path, List[TodoItem]] = {} |
| 219 | + self._folders: List[Path] = [] |
| 220 | + self._tags: Optional[Iterable[str]] = DEFAULT_TODO_TAGS |
| 221 | + self._worker: Optional[TodoScanWorker] = None |
| 222 | + |
| 223 | + @property |
| 224 | + def is_scanning(self) -> bool: |
| 225 | + return self._worker is not None and self._worker.isRunning() |
| 226 | + |
| 227 | + def set_folders(self, folders: List[Path | str]): |
| 228 | + self._folders = [Path(f) for f in folders if Path(f).is_dir()] |
| 229 | + |
| 230 | + def start_scan(self, folders: Optional[List[Path | str]] = None): |
| 231 | + """Startet einen vollständigen asynchronen Hintergrundscan.""" |
| 232 | + if folders is not None: |
| 233 | + self.set_folders(folders) |
| 234 | + |
| 235 | + self.stop_scan() |
| 236 | + self._cache.clear() |
| 237 | + self.scanStarted.emit() |
| 238 | + |
| 239 | + if not self._folders: |
| 240 | + self.todosUpdated.emit([]) |
| 241 | + self.scanFinished.emit(0, 0) |
| 242 | + return |
| 243 | + |
| 244 | + self._worker = TodoScanWorker( |
| 245 | + folders=self._folders, |
| 246 | + tags=self._tags, |
| 247 | + parent=self, |
| 248 | + ) |
| 249 | + self._worker.fileCompleted.connect(self._on_worker_file_completed) |
| 250 | + self._worker.scanProgress.connect(self.scanProgress.emit) |
| 251 | + self._worker.scanFinished.connect(self._on_worker_finished) |
| 252 | + self._worker.start() |
| 253 | + |
| 254 | + def stop_scan(self): |
| 255 | + """Stoppt einen aktuell laufenden Scan.""" |
| 256 | + if self._worker and self._worker.isRunning(): |
| 257 | + self._worker.stop() |
| 258 | + self._worker.wait(2000) |
| 259 | + self._worker = None |
| 260 | + |
| 261 | + def _on_worker_file_completed(self, file_path: Path, items: List[TodoItem]): |
| 262 | + self._cache[file_path] = items |
| 263 | + |
| 264 | + def _on_worker_finished(self, all_items: List[TodoItem]): |
| 265 | + self._worker = None |
| 266 | + sorted_items = self.get_all_todos() |
| 267 | + self.todosUpdated.emit(sorted_items) |
| 268 | + file_count = len(self._cache) |
| 269 | + self.scanFinished.emit(len(sorted_items), file_count) |
| 270 | + |
| 271 | + def rescan_file(self, file_path: Path | str) -> List[TodoItem]: |
| 272 | + """Scannt eine einzelne Datei synchron neu (z.B. direkt nach dem Speichern).""" |
| 273 | + file_path = Path(file_path) |
| 274 | + if not file_path.is_file(): |
| 275 | + if file_path in self._cache: |
| 276 | + del self._cache[file_path] |
| 277 | + self.todosUpdated.emit(self.get_all_todos()) |
| 278 | + return [] |
| 279 | + |
| 280 | + rel_path = file_path.name |
| 281 | + folder_name = "" |
| 282 | + for folder in self._folders: |
| 283 | + try: |
| 284 | + rel_path = str(file_path.relative_to(folder)) |
| 285 | + folder_name = folder.name |
| 286 | + break |
| 287 | + except ValueError: |
| 288 | + pass |
| 289 | + |
| 290 | + new_items = scan_file_for_todos( |
| 291 | + file_path=file_path, |
| 292 | + rel_path=rel_path, |
| 293 | + folder_name=folder_name, |
| 294 | + tags=self._tags, |
| 295 | + ) |
| 296 | + |
| 297 | + if new_items: |
| 298 | + self._cache[file_path] = new_items |
| 299 | + elif file_path in self._cache: |
| 300 | + del self._cache[file_path] |
| 301 | + |
| 302 | + self.todosUpdated.emit(self.get_all_todos()) |
| 303 | + return new_items |
| 304 | + |
| 305 | + def remove_file(self, file_path: Path | str): |
| 306 | + """Entfernt eine gelöschte Datei aus dem Aufgaben-Cache.""" |
| 307 | + file_path = Path(file_path) |
| 308 | + if file_path in self._cache: |
| 309 | + del self._cache[file_path] |
| 310 | + self.todosUpdated.emit(self.get_all_todos()) |
| 311 | + |
| 312 | + def clear(self): |
| 313 | + """Leert den gesamten Cache.""" |
| 314 | + self.stop_scan() |
| 315 | + self._cache.clear() |
| 316 | + self.todosUpdated.emit([]) |
| 317 | + |
| 318 | + def get_all_todos(self) -> List[TodoItem]: |
| 319 | + """Gibt alle gecachten Aufgaben sortiert nach Pfad und Zeilennummer zurück.""" |
| 320 | + items: List[TodoItem] = [] |
| 321 | + for file_path in sorted(self._cache.keys(), key=lambda p: str(p).lower()): |
| 322 | + file_todos = sorted(self._cache[file_path], key=lambda item: item.line_number) |
| 323 | + items.extend(file_todos) |
| 324 | + return items |
| 325 | + |
| 326 | + def get_todos_by_file(self) -> Dict[Path, List[TodoItem]]: |
| 327 | + """Gruppiert gecachte Aufgaben nach Datei.""" |
| 328 | + result: Dict[Path, List[TodoItem]] = {} |
| 329 | + for file_path in sorted(self._cache.keys(), key=lambda p: str(p).lower()): |
| 330 | + result[file_path] = sorted(self._cache[file_path], key=lambda item: item.line_number) |
| 331 | + return result |
| 332 | + |
| 333 | + def get_todos_by_tag(self) -> Dict[str, List[TodoItem]]: |
| 334 | + """Gruppiert gecachte Aufgaben nach Tag (TODO, FIXME etc.).""" |
| 335 | + result: Dict[str, List[TodoItem]] = {} |
| 336 | + for item in self.get_all_todos(): |
| 337 | + result.setdefault(item.tag, []).append(item) |
| 338 | + return result |
| 339 | + |
| 340 | + def get_counts(self) -> Tuple[int, int]: |
| 341 | + """Gibt (Gesamtzahl Aufgaben, Anzahl betroffener Dateien) zurück.""" |
| 342 | + total_items = sum(len(v) for v in self._cache.values()) |
| 343 | + return total_items, len(self._cache) |
0 commit comments