Coverage for lintro / utils / console / logger.py: 99%
195 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""Thread-safe console logger for formatted output display.
3This module provides the ThreadSafeConsoleLogger class for console output
4with thread-safe message tracking for parallel execution.
5"""
7from __future__ import annotations
9import re
10import threading
11from collections.abc import Sequence
12from pathlib import Path
13from typing import Any
15import click
16from loguru import logger
18from lintro.enums.action import Action, normalize_action
19from lintro.enums.severity_level import SeverityLevel
20from lintro.enums.tool_name import ToolName
21from lintro.utils.ai_metadata import get_ai_count as _get_ai_count
22from lintro.utils.console.constants import (
23 BORDER_LENGTH,
24 RE_CANNOT_AUTOFIX,
25 RE_REMAINING_OR_CANNOT,
26 get_tool_emoji,
27)
28from lintro.utils.display_helpers import (
29 print_ascii_art,
30 print_final_status,
31 print_final_status_format,
32)
35class ThreadSafeConsoleLogger:
36 """Thread-safe logger for console output formatting and display.
38 This class handles both console output and message tracking with proper
39 thread synchronization for parallel tool execution.
40 """
42 def __init__(self, run_dir: Path | None = None) -> None:
43 """Initialize the ThreadSafeConsoleLogger.
45 Args:
46 run_dir: Optional run directory path for output location display.
47 """
48 self.run_dir = run_dir
49 self._messages: list[str] = []
50 self._lock = threading.Lock()
52 def console_output(self, text: str, color: str | None = None) -> None:
53 """Display text on console and track for console.log.
55 Thread-safe: Uses lock when appending to message list.
57 Args:
58 text: Text to display.
59 color: Optional color for the text.
60 """
61 if color:
62 click.echo(click.style(text, fg=color))
63 else:
64 click.echo(text)
66 # Track for console.log (thread-safe)
67 with self._lock:
68 self._messages.append(text)
70 def info(self, message: str, **kwargs: Any) -> None:
71 """Log an info message to the console.
73 Args:
74 message: The message to log.
75 **kwargs: Additional keyword arguments for logger formatting.
76 """
77 self.console_output(message)
78 logger.info(message, **kwargs)
80 def info_blue(self, message: str, **kwargs: Any) -> None:
81 """Log an info message to the console in blue color.
83 Args:
84 message: The message to log.
85 **kwargs: Additional keyword arguments for formatting.
86 """
87 self.console_output(message, color="cyan")
88 logger.info(message, **kwargs)
90 def debug(self, message: str) -> None:
91 """Log a debug message (only shown when debug logging is enabled).
93 Args:
94 message: Message to log.
95 """
96 logger.debug(message)
98 def warning(self, message: str, **kwargs: Any) -> None:
99 """Log a warning message to the console.
101 Args:
102 message: The message to log.
103 **kwargs: Additional keyword arguments for logger formatting.
104 """
105 warning_text = f"WARNING: {message}"
106 self.console_output(warning_text, color="yellow")
107 logger.warning(message, **kwargs)
109 def error(self, message: str, **kwargs: Any) -> None:
110 """Log an error message to the console.
112 Args:
113 message: The message to log.
114 **kwargs: Additional keyword arguments for logger formatting.
115 """
116 error_text = f"ERROR: {message}"
117 click.echo(click.style(error_text, fg="red", bold=True))
118 with self._lock:
119 self._messages.append(error_text)
120 logger.error(message, **kwargs)
122 def success(self, message: str, **kwargs: Any) -> None:
123 """Log a success message to the console.
125 Args:
126 message: The message to log.
127 **kwargs: Additional keyword arguments for logger formatting.
128 """
129 self.console_output(text=f"✅ {message}", color="green")
130 logger.info(f"SUCCESS: {message}", **kwargs)
132 def save_console_log(self, run_dir: str | Path | None = None) -> None:
133 """Save tracked console messages to console.log.
135 Thread-safe: Uses lock when reading message list.
137 Args:
138 run_dir: Directory to save the console log. If None, uses self.run_dir.
139 """
140 target_dir = Path(run_dir) if run_dir else self.run_dir
141 if not target_dir:
142 return
144 console_log_path = target_dir / "console.log"
145 try:
146 with self._lock:
147 messages = list(self._messages)
149 with open(console_log_path, "w", encoding="utf-8") as f:
150 for message in messages:
151 f.write(f"{message}\n")
152 logger.debug(f"Saved console output to {console_log_path}")
153 except OSError as e:
154 logger.error(f"Failed to save console log to {console_log_path}: {e}")
156 def print_execution_summary(
157 self,
158 action: Action,
159 tool_results: Sequence[object],
160 ) -> None:
161 """Print the execution summary for all tools.
163 Args:
164 action: The action being performed.
165 tool_results: The list of tool results.
166 """
167 # Add separation before Execution Summary
168 self.console_output(text="")
170 # Execution summary section
171 summary_header: str = click.style("📋 EXECUTION SUMMARY", fg="cyan", bold=True)
172 border_line: str = click.style("=" * 50, fg="cyan")
174 self.console_output(text=summary_header)
175 self.console_output(text=border_line)
177 # Build summary table
178 self._print_summary_table(action=action, tool_results=tool_results)
180 # Totals line and ASCII art
181 from lintro.utils.summary_tables import count_affected_files
183 affected_files = count_affected_files(tool_results)
185 if action == Action.FIX:
186 # For format commands, track both fixed and remaining issues
187 total_fixed: int = 0
188 total_remaining: int = 0
189 total_ai_applied: int = 0
190 total_ai_verified: int = 0
191 for result in tool_results:
192 fixed_std = getattr(result, "fixed_issues_count", None)
193 remaining_std = getattr(result, "remaining_issues_count", None)
194 success = getattr(result, "success", True)
195 total_ai_applied += _get_ai_count(result, "applied_count")
196 total_ai_verified += _get_ai_count(result, "verified_count")
198 if fixed_std is not None:
199 total_fixed += fixed_std
200 else:
201 total_fixed += getattr(result, "issues_count", 0)
203 if remaining_std is not None:
204 if isinstance(remaining_std, int):
205 total_remaining += remaining_std
206 elif not success:
207 pass
208 else:
209 output = getattr(result, "output", "")
210 if output and (
211 "remaining" in output.lower()
212 or "cannot be auto-fixed" in output.lower()
213 ):
214 remaining_match = RE_CANNOT_AUTOFIX.search(output)
215 if not remaining_match:
216 remaining_match = RE_REMAINING_OR_CANNOT.search(
217 output.lower(),
218 )
219 if remaining_match:
220 total_remaining += int(remaining_match.group(1))
222 self._print_totals_table(
223 action=action,
224 total_fixed=total_fixed,
225 total_remaining=total_remaining,
226 affected_files=affected_files,
227 total_ai_applied=total_ai_applied,
228 total_ai_verified=total_ai_verified,
229 )
230 self._print_ascii_art(total_issues=total_remaining)
231 logger.debug(
232 f"{action} completed with native={total_fixed}, "
233 f"ai_verified={total_ai_verified}, "
234 f"{total_remaining} remaining",
235 )
236 else:
237 total_issues: int = sum(
238 (getattr(result, "issues_count", 0) for result in tool_results),
239 )
240 any_failed: bool = any(
241 not getattr(result, "success", True) for result in tool_results
242 )
243 total_for_art: int = (
244 total_issues if not any_failed else max(1, total_issues)
245 )
247 # Tally severity breakdown across all issues
248 sev_errors = 0
249 sev_warnings = 0
250 sev_info = 0
251 for result in tool_results:
252 issues = getattr(result, "issues", None)
253 if issues:
254 for issue in issues:
255 get_sev = getattr(issue, "get_severity", None)
256 if get_sev:
257 level = get_sev()
258 if level == SeverityLevel.ERROR:
259 sev_errors += 1
260 elif level == SeverityLevel.WARNING:
261 sev_warnings += 1
262 elif level == SeverityLevel.INFO:
263 sev_info += 1
265 self._print_totals_table(
266 action=action,
267 total_issues=total_issues,
268 affected_files=affected_files,
269 severity_errors=sev_errors,
270 severity_warnings=sev_warnings,
271 severity_info=sev_info,
272 )
273 self._print_ascii_art(total_issues=total_for_art)
274 logger.debug(
275 f"{action} completed with {total_issues} total issues"
276 + (" and failures" if any_failed else ""),
277 )
279 def _print_summary_table(
280 self,
281 action: Action | str,
282 tool_results: Sequence[object],
283 ) -> None:
284 """Print the summary table for the run.
286 Args:
287 action: The action being performed.
288 tool_results: The list of tool results.
289 """
290 from lintro.utils.summary_tables import print_summary_table
292 action_enum = normalize_action(action)
293 print_summary_table(
294 console_output_func=self.console_output,
295 action=action_enum,
296 tool_results=tool_results,
297 )
299 def _print_totals_table(
300 self,
301 action: Action,
302 total_issues: int = 0,
303 total_fixed: int = 0,
304 total_remaining: int = 0,
305 affected_files: int = 0,
306 severity_errors: int = 0,
307 severity_warnings: int = 0,
308 severity_info: int = 0,
309 total_ai_applied: int = 0,
310 total_ai_verified: int = 0,
311 ) -> None:
312 """Print the totals summary table for the run.
314 Args:
315 action: The action being performed.
316 total_issues: Total number of issues found (CHECK/TEST mode).
317 total_fixed: Total number of issues fixed (FIX mode).
318 total_remaining: Total number of remaining issues (FIX mode).
319 affected_files: Number of unique files with issues.
320 severity_errors: Number of issues at ERROR severity.
321 severity_warnings: Number of issues at WARNING severity.
322 severity_info: Number of issues at INFO severity.
323 total_ai_applied: Total number of AI-applied fixes (FIX mode).
324 total_ai_verified: Total number of AI-resolved fixes (FIX mode).
325 """
326 from lintro.utils.summary_tables import print_totals_table
328 print_totals_table(
329 console_output_func=self.console_output,
330 action=action,
331 total_issues=total_issues,
332 total_fixed=total_fixed,
333 total_remaining=total_remaining,
334 affected_files=affected_files,
335 severity_errors=severity_errors,
336 severity_warnings=severity_warnings,
337 severity_info=severity_info,
338 total_ai_applied=total_ai_applied,
339 total_ai_verified=total_ai_verified,
340 )
342 def _print_final_status(
343 self,
344 action: Action | str,
345 total_issues: int,
346 ) -> None:
347 """Print the final status for the run.
349 Args:
350 action: The action being performed.
351 total_issues: The total number of issues found.
352 """
353 action_enum = normalize_action(action)
354 print_final_status(
355 console_output_func=self.console_output,
356 action=action_enum,
357 total_issues=total_issues,
358 )
360 def _print_final_status_format(
361 self,
362 total_fixed: int,
363 total_remaining: int,
364 ) -> None:
365 """Print the final status for format operations.
367 Args:
368 total_fixed: The total number of issues fixed.
369 total_remaining: The total number of remaining issues.
370 """
371 print_final_status_format(
372 console_output_func=self.console_output,
373 total_fixed=total_fixed,
374 total_remaining=total_remaining,
375 )
377 def _print_ascii_art(
378 self,
379 total_issues: int,
380 ) -> None:
381 """Print ASCII art based on the number of issues.
383 Args:
384 total_issues: The total number of issues found.
385 """
386 print_ascii_art(
387 console_output_func=self.console_output,
388 issue_count=total_issues,
389 )
391 def print_lintro_header(self) -> None:
392 """Print the main Lintro header with output directory information."""
393 if self.run_dir:
394 header_msg: str = (
395 f"[LINTRO] All output formats will be auto-generated in {self.run_dir}"
396 )
397 self.console_output(text=header_msg)
398 self.console_output(text="")
400 def print_tool_header(
401 self,
402 tool_name: str,
403 action: str,
404 ) -> None:
405 """Print a formatted header for a tool execution.
407 Args:
408 tool_name: Name of the tool.
409 action: The action being performed ("check" or "fmt").
410 """
411 emoji: str = get_tool_emoji(tool_name)
412 border: str = "=" * BORDER_LENGTH
413 header_text: str = (
414 f"✨ Running {tool_name} ({action}) "
415 f"{emoji} {emoji} {emoji} {emoji} {emoji}"
416 )
418 self.console_output(text=border)
419 self.console_output(text=header_text)
420 self.console_output(text=border)
421 self.console_output(text="")
423 def print_tool_result(
424 self,
425 tool_name: str,
426 output: str,
427 issues_count: int,
428 raw_output_for_meta: str | None = None,
429 action: Action | None = None,
430 success: bool = True,
431 ) -> None:
432 """Print the result of a tool execution.
434 Args:
435 tool_name: Name of the tool.
436 output: Tool output to display.
437 issues_count: Number of issues found.
438 raw_output_for_meta: Raw output for metadata parsing.
439 action: Action being performed.
440 success: Whether the tool execution was successful.
441 """
442 if output:
443 self.console_output(text=output)
444 self.console_output(text="")
446 if raw_output_for_meta and action == Action.CHECK:
447 self._print_metadata_messages(raw_output_for_meta)
449 if action == Action.TEST and tool_name == ToolName.PYTEST.value:
450 self._print_pytest_results(output, success)
452 def _print_metadata_messages(self, raw_output: str) -> None:
453 """Print metadata messages parsed from raw tool output.
455 Args:
456 raw_output: Raw tool output to parse for metadata.
457 """
458 output_lower = raw_output.lower()
460 fixable_match = re.search(r"(\d+)\s*fixable", output_lower)
461 if fixable_match:
462 fixable_count = int(fixable_match.group(1))
463 if fixable_count > 0:
464 self.console_output(
465 text=f"Info: Found {fixable_count} auto-fixable issue(s)",
466 )
467 else:
468 self.console_output(text="Info: No issues found")
469 return
471 if "cannot be auto-fixed" in output_lower:
472 self.console_output(text="Info: Found issues that cannot be auto-fixed")
473 return
475 if "would reformat" in output_lower:
476 self.console_output(text="Info: Files would be reformatted")
477 return
479 if "fixed" in output_lower:
480 self.console_output(text="Info: Issues were fixed")
481 return
483 self.console_output(text="Info: No issues found")
485 def _print_pytest_results(self, output: str, success: bool) -> None:
486 """Print formatted pytest results.
488 Args:
489 output: Pytest output.
490 success: Whether tests passed.
491 """
492 self.console_output(text="")
493 self.console_output(text="📋 Test Results", color="cyan")
494 self.console_output(text="=" * 50, color="cyan")
496 if success:
497 self.console_output(text="✅ All tests passed", color="green")
498 else:
499 self.console_output(text="❌ Some tests failed", color="red")
501 if output:
502 self.console_output(text="")
503 self.console_output(text=output)
505 def print_post_checks_header(
506 self,
507 ) -> None:
508 """Print a distinct header separating the post-checks phase."""
509 border_char: str = "━"
510 border: str = border_char * BORDER_LENGTH
511 title_styled: str = click.style(
512 text="🚦 POST-CHECKS",
513 fg="magenta",
514 bold=True,
515 )
516 subtitle_styled: str = click.style(
517 text=("Running optional follow-up checks after primary tools"),
518 fg="magenta",
519 )
520 border_styled: str = click.style(text=border, fg="magenta")
522 self.console_output(text=border_styled)
523 self.console_output(text=title_styled)
524 self.console_output(text=subtitle_styled)
525 self.console_output(text=border_styled)
526 self.console_output(text="")