Coverage for src / ezpl / cli / commands / logs.py: 37.70%
276 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-13 19:35 +0000
1# ///////////////////////////////////////////////////////////////
2# EZPL - CLI Logs Commands
3# Project: ezpl
4# ///////////////////////////////////////////////////////////////
6"""
7CLI commands for log file operations.
9This module provides commands for viewing, searching, analyzing,
10and managing log files.
11"""
13from __future__ import annotations
15# ///////////////////////////////////////////////////////////////
16# IMPORTS
17# ///////////////////////////////////////////////////////////////
18# Standard library imports
19import json
20import time
21from datetime import datetime, timedelta
22from pathlib import Path
24import click
26# Third-party imports
27from rich.table import Table
29# Local imports
30from ...config import ConfigurationManager
31from .._console import console
32from ..utils.log_parser import LogParser
33from ..utils.log_stats import LogStatistics
35# ///////////////////////////////////////////////////////////////
36# HELPER FUNCTIONS
37# ///////////////////////////////////////////////////////////////
40def _get_log_file(file: Path | None) -> Path:
41 """
42 Get log file path from parameter or configuration.
44 Args:
45 file: Optional file path from command line
47 Returns:
48 Path to log file
50 Raises:
51 click.ClickException: If file doesn't exist
52 """
53 if file: 53 ↛ 56line 53 didn't jump to line 56 because the condition on line 53 was always true
54 log_file = Path(file)
55 else:
56 config_manager = ConfigurationManager()
57 log_file = config_manager.get_log_file()
59 if not log_file.exists(): 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 raise click.ClickException(f"Log file not found: {log_file}")
62 return log_file
65def _get_log_dir(dir: Path | None) -> Path:
66 """
67 Get log directory from parameter or configuration.
69 Args:
70 dir: Optional directory path from command line
72 Returns:
73 Path to log directory
74 """
75 if dir: 75 ↛ 78line 75 didn't jump to line 78 because the condition on line 75 was always true
76 return Path(dir)
77 else:
78 config_manager = ConfigurationManager()
79 log_dir = config_manager.get("log-dir")
80 if isinstance(log_dir, str):
81 return Path(log_dir)
82 return Path(log_dir) if log_dir else Path.home() / ".ezpl" / "logs"
85def _parse_size(size_str: str) -> int:
86 """
87 Parse size string to bytes.
89 Args:
90 size_str: Size string (e.g., "100MB", "1GB", "500KB")
92 Returns:
93 Size in bytes
94 """
95 size_str = size_str.upper().strip()
96 multipliers = {"KB": 1024, "MB": 1024**2, "GB": 1024**3, "TB": 1024**4}
98 for unit, multiplier in multipliers.items():
99 if size_str.endswith(unit):
100 try:
101 value = float(size_str[: -len(unit)])
102 return int(value * multiplier)
103 except ValueError as e:
104 raise click.ClickException(f"Invalid size format: {size_str}") from e
106 # Try to parse as bytes
107 try:
108 return int(size_str)
109 except ValueError as e:
110 raise click.ClickException(f"Invalid size format: {size_str}") from e
113# ///////////////////////////////////////////////////////////////
114# COMMAND GROUP
115# ///////////////////////////////////////////////////////////////
118@click.group(name="logs", help="📊 Manage and view log files")
119def logs_group() -> None:
120 """
121 Log file management commands.
123 View, search, analyze, and manage Ezpl log files.
124 """
127# ///////////////////////////////////////////////////////////////
128# COMMANDS
129# ///////////////////////////////////////////////////////////////
132@logs_group.command(name="view", help="View log file contents")
133@click.option(
134 "--file",
135 "-f",
136 type=click.Path(exists=True, path_type=Path),
137 help="Path to log file (default: from config)",
138)
139@click.option("--lines", "-n", type=int, default=50, help="Number of lines to display")
140@click.option(
141 "--level",
142 "-l",
143 type=str,
144 help="Filter by log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)",
145)
146@click.option(
147 "--follow",
148 "-F",
149 is_flag=True,
150 help="Follow log file (like tail -f)",
151)
152def view_command(
153 file: Path | None, lines: int, level: str | None, follow: bool
154) -> None:
155 """
156 View log file contents.
158 Display log entries from the specified file with optional filtering.
159 """
160 try:
161 log_file = _get_log_file(file)
162 parser = LogParser(log_file)
164 if follow: 164 ↛ 166line 164 didn't jump to line 166 because the condition on line 164 was never true
165 # Follow mode (tail -f)
166 console.print(f"[cyan]Following {log_file}...[/cyan]")
167 console.print("[dim]Press Ctrl+C to stop[/dim]\n")
169 try:
170 with open(log_file, encoding="utf-8") as f:
171 # Go to end of file
172 f.seek(0, 2)
174 while True:
175 line = f.readline()
176 if line:
177 entry = parser.parse_line(line, 0)
178 if entry: # noqa: SIM102
179 if not level or entry.level.upper() == level.upper():
180 console.print(entry.raw_line)
181 else:
182 time.sleep(0.1)
183 except KeyboardInterrupt:
184 console.print("\n[yellow]Stopped following[/yellow]")
185 else:
186 # Regular view
187 entries = parser.parse_lines(max_lines=lines)
189 if level: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 entries = [e for e in entries if e.level.upper() == level.upper()]
192 if not entries: 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true
193 console.print("[yellow]No log entries found[/yellow]")
194 return
196 # Display entries
197 for entry in entries:
198 console.print(entry.raw_line)
200 except click.ClickException:
201 raise
202 except (OSError, ValueError, json.JSONDecodeError) as e:
203 raise click.ClickException(str(e)) from e
206@logs_group.command(name="search", help="Search log entries")
207@click.option(
208 "--file",
209 "-f",
210 type=click.Path(exists=True, path_type=Path),
211 help="Path to log file (default: from config)",
212)
213@click.option(
214 "--pattern",
215 "-p",
216 type=str,
217 required=True,
218 help="Search pattern (regex supported)",
219)
220@click.option(
221 "--level",
222 "-l",
223 type=str,
224 help="Filter by log level",
225)
226@click.option(
227 "--case-sensitive",
228 "-c",
229 is_flag=True,
230 help="Case-sensitive search",
231)
232def search_command(
233 file: Path | None,
234 pattern: str,
235 level: str | None,
236 case_sensitive: bool,
237) -> None:
238 """
239 Search for log entries matching a pattern.
241 Search through log files using regex patterns with optional level filtering.
242 """
243 try:
244 log_file = _get_log_file(file)
245 parser = LogParser(log_file)
247 # Search entries
248 results = list(parser.search(pattern, case_sensitive=case_sensitive))
250 # Filter by level if specified
251 if level: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 results = [e for e in results if e.level.upper() == level.upper()]
254 if not results: 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true
255 console.print(
256 f"[yellow]No entries found matching pattern: {pattern}[/yellow]"
257 )
258 return
260 # Display results
261 console.print(f"[green]Found {len(results)} matching entries:[/green]\n")
262 for entry in results:
263 console.print(entry.raw_line)
265 except click.ClickException:
266 raise
267 except (OSError, ValueError, json.JSONDecodeError) as e:
268 raise click.ClickException(str(e)) from e
271@logs_group.command(name="stats", help="Display log statistics")
272@click.option(
273 "--file",
274 "-f",
275 type=click.Path(exists=True, path_type=Path),
276 help="Path to log file (default: from config)",
277)
278@click.option(
279 "--format",
280 "-F",
281 type=click.Choice(["table", "json"], case_sensitive=False),
282 default="table",
283 help="Output format",
284)
285def stats_command(file: Path | None, format: str) -> None:
286 """
287 Display statistics about log files.
289 Show counts by level, file size, date ranges, and temporal distribution.
290 """
291 try:
292 log_file = _get_log_file(file)
293 stats = LogStatistics(log_file)
294 all_stats = stats.get_all_stats()
296 if format == "json": 296 ↛ 298line 296 didn't jump to line 298 because the condition on line 296 was never true
297 # JSON output
298 console.print(json.dumps(all_stats, indent=2, default=str))
299 else:
300 # Table output
301 # File info
302 file_info = all_stats["file_info"]
303 info_table = Table(
304 title="File Information", show_header=True, header_style="bold blue"
305 )
306 info_table.add_column("Property", style="cyan")
307 info_table.add_column("Value", style="white")
309 info_table.add_row("Path", file_info["file_path"])
310 info_table.add_row(
311 "Size", f"{file_info['size_mb']} MB ({file_info['size_bytes']} bytes)"
312 )
313 info_table.add_row("Entries", str(file_info["line_count"]))
315 if file_info["date_range"]: 315 ↛ 334line 315 didn't jump to line 334 because the condition on line 315 was always true
316 date_range = file_info["date_range"]
317 info_table.add_row(
318 "First Entry",
319 (
320 date_range["first"].strftime("%Y-%m-%d %H:%M:%S")
321 if date_range["first"]
322 else "N/A"
323 ),
324 )
325 info_table.add_row(
326 "Last Entry",
327 (
328 date_range["last"].strftime("%Y-%m-%d %H:%M:%S")
329 if date_range["last"]
330 else "N/A"
331 ),
332 )
334 console.print(info_table)
336 # Level counts
337 level_counts = all_stats["level_counts"]
338 if level_counts: 338 ↛ exitline 338 didn't return from function 'stats_command' because the condition on line 338 was always true
339 level_table = Table(
340 title="Level Distribution",
341 show_header=True,
342 header_style="bold blue",
343 )
344 level_table.add_column("Level", style="cyan")
345 level_table.add_column("Count", style="green")
347 for level, count in sorted(
348 level_counts.items(), key=lambda x: x[1], reverse=True
349 ):
350 level_table.add_row(level, str(count))
352 console.print("\n")
353 console.print(level_table)
355 except click.ClickException:
356 raise
357 except (OSError, ValueError, json.JSONDecodeError) as e:
358 raise click.ClickException(str(e)) from e
361@logs_group.command(name="tail", help="Display last lines of log file")
362@click.option(
363 "--file",
364 "-f",
365 type=click.Path(exists=True, path_type=Path),
366 help="Path to log file (default: from config)",
367)
368@click.option("--lines", "-n", type=int, default=20, help="Number of lines to display")
369@click.option(
370 "--follow",
371 "-F",
372 is_flag=True,
373 help="Follow log file (like tail -f)",
374)
375def tail_command(file: Path | None, lines: int, follow: bool) -> None:
376 """
377 Display the last lines of a log file.
379 Similar to Unix 'tail' command with optional follow mode.
380 """
381 try:
382 log_file = _get_log_file(file)
383 parser = LogParser(log_file)
385 if follow:
386 # Use view command's follow mode
387 view_command(file, lines=lines, level=None, follow=True)
388 else:
389 # Get last N lines
390 entries = parser.get_last_lines(lines)
392 if not entries:
393 console.print("[yellow]No log entries found[/yellow]")
394 return
396 # Display entries
397 for entry in entries:
398 console.print(entry.raw_line)
400 except click.ClickException:
401 raise
402 except (OSError, ValueError, json.JSONDecodeError) as e:
403 raise click.ClickException(str(e)) from e
406@logs_group.command(name="list", help="List log files")
407@click.option(
408 "--dir",
409 "-d",
410 type=click.Path(exists=True, path_type=Path),
411 help="Directory to search (default: from config)",
412)
413def list_command(dir: Path | None) -> None:
414 """
415 List available log files.
417 Display all log files in the configured log directory.
418 """
419 try:
420 log_dir = _get_log_dir(dir)
422 if not log_dir.exists(): 422 ↛ 423line 422 didn't jump to line 423 because the condition on line 422 was never true
423 console.print(f"[yellow]Log directory does not exist: {log_dir}[/yellow]")
424 return
426 # Find all log files
427 log_files = sorted(
428 log_dir.glob("*.log"), key=lambda p: p.stat().st_mtime, reverse=True
429 )
431 if not log_files: 431 ↛ 432line 431 didn't jump to line 432 because the condition on line 431 was never true
432 console.print(f"[yellow]No log files found in {log_dir}[/yellow]")
433 return
435 # Display as table
436 table = Table(
437 title=f"Log Files in {log_dir}", show_header=True, header_style="bold blue"
438 )
439 table.add_column("File", style="cyan")
440 table.add_column("Size", style="green")
441 table.add_column("Modified", style="white")
443 for log_file in log_files:
444 try:
445 size = log_file.stat().st_size
446 size_mb = size / (1024 * 1024)
447 modified = datetime.fromtimestamp(log_file.stat().st_mtime)
448 table.add_row(
449 log_file.name,
450 f"{size_mb:.2f} MB",
451 modified.strftime("%Y-%m-%d %H:%M:%S"),
452 )
453 except OSError as e:
454 console.print(f"[bold red]Error:[/bold red] {e}")
456 console.print(table)
458 except click.ClickException:
459 raise
460 except (OSError, ValueError, json.JSONDecodeError) as e:
461 raise click.ClickException(str(e)) from e
464@logs_group.command(name="clean", help="Clean old log files")
465@click.option(
466 "--file",
467 "-f",
468 type=click.Path(exists=True, path_type=Path),
469 help="Specific file to clean",
470)
471@click.option(
472 "--days",
473 "-d",
474 type=int,
475 help="Delete files older than N days",
476)
477@click.option(
478 "--size",
479 "-s",
480 type=str,
481 help="Delete files larger than SIZE (e.g., '100MB')",
482)
483@click.option(
484 "--confirm",
485 "-y",
486 is_flag=True,
487 help="Skip confirmation prompt",
488)
489def clean_command(
490 file: Path | None, days: int | None, size: str | None, confirm: bool
491) -> None:
492 """
493 Clean old or large log files.
495 Remove log files based on age or size criteria.
496 """
497 try:
498 if file:
499 # Clean specific file
500 files_to_clean = [Path(file)]
501 else:
502 # Clean from directory
503 log_dir = _get_log_dir(None)
504 if not log_dir.exists():
505 console.print(
506 f"[yellow]Log directory does not exist: {log_dir}[/yellow]"
507 )
508 return
510 files_to_clean = list(log_dir.glob("*.log"))
512 if not files_to_clean:
513 console.print("[yellow]No log files to clean[/yellow]")
514 return
516 # Filter by criteria
517 files_to_delete = []
519 for log_file in files_to_clean:
520 should_delete = False
522 if days:
523 # Check age
524 file_age = datetime.now() - datetime.fromtimestamp(
525 log_file.stat().st_mtime
526 )
527 if file_age > timedelta(days=days):
528 should_delete = True
530 if size:
531 # Check size
532 file_size = log_file.stat().st_size
533 size_bytes = _parse_size(size)
534 if file_size > size_bytes:
535 should_delete = True
537 if should_delete:
538 files_to_delete.append(log_file)
540 if not files_to_delete:
541 console.print("[green]No files match the cleanup criteria[/green]")
542 return
544 # Confirm deletion
545 if not confirm:
546 console.print(
547 f"[yellow]Files to be deleted ({len(files_to_delete)}):[/yellow]"
548 )
549 for f in files_to_delete:
550 console.print(f" - {f}")
552 if not click.confirm("\nAre you sure you want to delete these files?"):
553 console.print("[yellow]Cleanup cancelled[/yellow]")
554 return
556 # Delete files
557 deleted_count = 0
558 for log_file in files_to_delete:
559 try:
560 log_file.unlink()
561 deleted_count += 1
562 console.print(f"[green]✓[/green] Deleted: {log_file}")
563 except OSError as e:
564 console.print(f"[red]✗[/red] Failed to delete {log_file}: {e}")
566 console.print(f"\n[green]Deleted {deleted_count} file(s)[/green]")
568 except click.ClickException:
569 raise
570 except (OSError, ValueError, json.JSONDecodeError) as e:
571 raise click.ClickException(str(e)) from e
574@logs_group.command(name="export", help="Export log file")
575@click.option(
576 "--file",
577 "-f",
578 type=click.Path(exists=True, path_type=Path),
579 help="Path to log file (default: from config)",
580)
581@click.option(
582 "--format",
583 "-F",
584 type=click.Choice(["json", "csv", "txt"], case_sensitive=False),
585 default="json",
586 help="Export format",
587)
588@click.option(
589 "--output",
590 "-o",
591 type=click.Path(path_type=Path),
592 help="Output file path (default: stdout)",
593)
594def export_command(file: Path | None, format: str, output: Path | None) -> None:
595 """
596 Export log file to different formats.
598 Convert log files to JSON, CSV, or plain text format.
599 """
600 try:
601 log_file = _get_log_file(file)
602 parser = LogParser(log_file)
603 entries = list(parser.parse())
605 if not entries:
606 console.print("[yellow]No log entries to export[/yellow]")
607 return
609 # Export based on format
610 if format == "json":
611 data = [entry.to_dict() for entry in entries]
612 content = json.dumps(data, indent=2, default=str)
613 elif format == "csv":
614 import csv
615 from io import StringIO
617 output_buffer = StringIO()
618 writer = csv.DictWriter(
619 output_buffer,
620 fieldnames=[
621 "timestamp",
622 "level",
623 "module",
624 "function",
625 "line",
626 "message",
627 ],
628 )
629 writer.writeheader()
630 for entry in entries:
631 writer.writerow(entry.to_dict())
632 content = output_buffer.getvalue()
633 else: # txt
634 content = "\n".join(entry.raw_line for entry in entries)
636 # Write output
637 if output:
638 output_path = Path(output)
639 output_path.parent.mkdir(parents=True, exist_ok=True)
640 with open(output_path, "w", encoding="utf-8") as f:
641 f.write(content)
642 console.print(f"[green]✓[/green] Exported to {output_path}")
643 else:
644 console.print(content)
646 except click.ClickException:
647 raise
648 except (OSError, ValueError, json.JSONDecodeError) as e:
649 raise click.ClickException(str(e)) from e