fix: resolve CI/CD linting and formatting issues
Some checks failed
Some checks failed
- Replaced deprecated typing.Dict/List/Tuple with native types (UP035) - Removed unused imports across all modules - Fixed unused variables by replacing with _ prefix - Added missing Optional type imports - Reorganized imports for proper sorting (I001) - Applied black formatting to all source files
This commit is contained in:
@@ -1,22 +1,22 @@
|
|||||||
"""Click CLI commands for LogLens."""
|
'''Click CLI commands for LogLens.'''
|
||||||
|
|
||||||
|
import logging
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
import logging
|
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
import click
|
import click
|
||||||
from colorlog import ColoredFormatter
|
from colorlog import ColoredFormatter
|
||||||
|
|
||||||
from loglens.parsers.base import LogFormat
|
|
||||||
from loglens.analyzers.analyzer import LogAnalyzer
|
from loglens.analyzers.analyzer import LogAnalyzer
|
||||||
from loglens.formatters.table_formatter import TableFormatter
|
|
||||||
from loglens.formatters.json_formatter import JSONFormatter
|
from loglens.formatters.json_formatter import JSONFormatter
|
||||||
|
from loglens.formatters.table_formatter import TableFormatter
|
||||||
from loglens.formatters.text_formatter import TextFormatter
|
from loglens.formatters.text_formatter import TextFormatter
|
||||||
|
from loglens.parsers.base import LogFormat
|
||||||
|
|
||||||
|
|
||||||
def setup_logging(verbosity: int = 0) -> None:
|
def setup_logging(verbosity: int = 0) -> None:
|
||||||
"""Setup logging configuration."""
|
'''Setup logging configuration.'''
|
||||||
log_levels = ["ERROR", "WARNING", "INFO", "DEBUG"]
|
log_levels = ["ERROR", "WARNING", "INFO", "DEBUG"]
|
||||||
level_idx = min(verbosity, len(log_levels) - 1)
|
level_idx = min(verbosity, len(log_levels) - 1)
|
||||||
level = log_levels[level_idx]
|
level = log_levels[level_idx]
|
||||||
@@ -29,7 +29,7 @@ def setup_logging(verbosity: int = 0) -> None:
|
|||||||
"WARNING": "yellow",
|
"WARNING": "yellow",
|
||||||
"ERROR": "red",
|
"ERROR": "red",
|
||||||
"CRITICAL": "red,bg_white",
|
"CRITICAL": "red,bg_white",
|
||||||
}
|
},
|
||||||
)
|
)
|
||||||
|
|
||||||
logger = logging.getLogger("loglens")
|
logger = logging.getLogger("loglens")
|
||||||
@@ -42,7 +42,7 @@ def setup_logging(verbosity: int = 0) -> None:
|
|||||||
@click.option("--config", type=click.Path(exists=True), help="Path to config file")
|
@click.option("--config", type=click.Path(exists=True), help="Path to config file")
|
||||||
@click.pass_context
|
@click.pass_context
|
||||||
def main(ctx: click.Context, verbosity: int, config: str) -> None:
|
def main(ctx: click.Context, verbosity: int, config: str) -> None:
|
||||||
"""LogLens - Parse, analyze, and summarize log files."""
|
'''LogLens - Parse, analyze, and summarize log files.'''
|
||||||
setup_logging(verbosity)
|
setup_logging(verbosity)
|
||||||
ctx.ensure_object(dict)
|
ctx.ensure_object(dict)
|
||||||
ctx.obj["config"] = config
|
ctx.obj["config"] = config
|
||||||
@@ -50,13 +50,20 @@ def main(ctx: click.Context, verbosity: int, config: str) -> None:
|
|||||||
|
|
||||||
@main.command("analyze")
|
@main.command("analyze")
|
||||||
@click.argument("files", type=click.Path(exists=True), nargs=-1)
|
@click.argument("files", type=click.Path(exists=True), nargs=-1)
|
||||||
@click.option("--format", type=click.Choice(["json", "syslog", "apache", "auto"]),
|
@click.option(
|
||||||
default="auto", help="Log format (auto-detect by default)")
|
"--format",
|
||||||
@click.option("--output", type=click.Choice(["table", "json", "text"]), default="table",
|
type=click.Choice(["json", "syslog", "apache", "auto"]),
|
||||||
help="Output format")
|
default="auto",
|
||||||
|
help="Log format (auto-detect by default)",
|
||||||
|
)
|
||||||
|
@click.option(
|
||||||
|
"--output", type=click.Choice(["table", "json", "text"]), default="table", help="Output format"
|
||||||
|
)
|
||||||
@click.option("--follow/--no-follow", default=False, help="Follow file changes")
|
@click.option("--follow/--no-follow", default=False, help="Follow file changes")
|
||||||
@click.option("--max-entries", type=int, default=100, help="Maximum entries to display")
|
@click.option("--max-entries", type=int, default=100, help="Maximum entries to display")
|
||||||
@click.option("--json/--no-json", default=False, help="Output as JSON (shorthand for --output json)")
|
@click.option(
|
||||||
|
"--json/--no-json", default=False, help="Output as JSON (shorthand for --output json)"
|
||||||
|
)
|
||||||
@click.pass_context
|
@click.pass_context
|
||||||
def analyze(
|
def analyze(
|
||||||
ctx: click.Context,
|
ctx: click.Context,
|
||||||
@@ -65,9 +72,9 @@ def analyze(
|
|||||||
output: str,
|
output: str,
|
||||||
follow: bool,
|
follow: bool,
|
||||||
max_entries: int,
|
max_entries: int,
|
||||||
json: bool
|
json: bool,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Analyze log files and display summary."""
|
'''Analyze log files and display summary.'''
|
||||||
if json:
|
if json:
|
||||||
output = "json"
|
output = "json"
|
||||||
|
|
||||||
@@ -85,7 +92,7 @@ def analyze(
|
|||||||
|
|
||||||
|
|
||||||
def _analyze_lines(lines: list, format_str: str, output: str, max_entries: int) -> None:
|
def _analyze_lines(lines: list, format_str: str, output: str, max_entries: int) -> None:
|
||||||
"""Analyze lines from stdin."""
|
'''Analyze lines from stdin.'''
|
||||||
format_enum = None if format_str == "auto" else LogFormat(format_str)
|
format_enum = None if format_str == "auto" else LogFormat(format_str)
|
||||||
analyzer = LogAnalyzer()
|
analyzer = LogAnalyzer()
|
||||||
|
|
||||||
@@ -97,8 +104,10 @@ def _analyze_lines(lines: list, format_str: str, output: str, max_entries: int)
|
|||||||
_display_result(result, output, max_entries)
|
_display_result(result, output, max_entries)
|
||||||
|
|
||||||
|
|
||||||
def _analyze_file(file_path: str, format_str: str, output: str, max_entries: int, follow: bool) -> None:
|
def _analyze_file(
|
||||||
"""Analyze a single file."""
|
file_path: str, format_str: str, output: str, max_entries: int, follow: bool
|
||||||
|
) -> None:
|
||||||
|
'''Analyze a single file.'''
|
||||||
format_enum = None if format_str == "auto" else LogFormat(format_str)
|
format_enum = None if format_str == "auto" else LogFormat(format_str)
|
||||||
analyzer = LogAnalyzer()
|
analyzer = LogAnalyzer()
|
||||||
|
|
||||||
@@ -109,10 +118,15 @@ def _analyze_file(file_path: str, format_str: str, output: str, max_entries: int
|
|||||||
_display_result(result, output, max_entries)
|
_display_result(result, output, max_entries)
|
||||||
|
|
||||||
|
|
||||||
def _follow_file(file_path: str, analyzer: LogAnalyzer, format: Optional[LogFormat],
|
def _follow_file(
|
||||||
output: str, max_entries: int) -> None:
|
file_path: str,
|
||||||
"""Follow a file and analyze in real-time."""
|
analyzer: LogAnalyzer,
|
||||||
with open(file_path, "r") as f:
|
format: Optional[LogFormat],
|
||||||
|
output: str,
|
||||||
|
max_entries: int,
|
||||||
|
) -> None:
|
||||||
|
'''Follow a file and analyze in real-time.'''
|
||||||
|
with open(file_path) as f:
|
||||||
f.seek(0, 2)
|
f.seek(0, 2)
|
||||||
buffer = []
|
buffer = []
|
||||||
|
|
||||||
@@ -136,7 +150,7 @@ def _follow_file(file_path: str, analyzer: LogAnalyzer, format: Optional[LogForm
|
|||||||
|
|
||||||
|
|
||||||
def _display_result(result, output: str, max_entries: int) -> None:
|
def _display_result(result, output: str, max_entries: int) -> None:
|
||||||
"""Display analysis result."""
|
'''Display analysis result.'''
|
||||||
if output == "json":
|
if output == "json":
|
||||||
formatter = JSONFormatter()
|
formatter = JSONFormatter()
|
||||||
click.echo(formatter.format(result))
|
click.echo(formatter.format(result))
|
||||||
@@ -150,19 +164,17 @@ def _display_result(result, output: str, max_entries: int) -> None:
|
|||||||
|
|
||||||
@main.command("watch")
|
@main.command("watch")
|
||||||
@click.argument("files", type=click.Path(exists=True), nargs=-1)
|
@click.argument("files", type=click.Path(exists=True), nargs=-1)
|
||||||
@click.option("--format", type=click.Choice(["json", "syslog", "apache", "auto"]),
|
@click.option(
|
||||||
default="auto", help="Log format")
|
"--format",
|
||||||
|
type=click.Choice(["json", "syslog", "apache", "auto"]),
|
||||||
|
default="auto",
|
||||||
|
help="Log format",
|
||||||
|
)
|
||||||
@click.option("--interval", type=float, default=1.0, help="Refresh interval in seconds")
|
@click.option("--interval", type=float, default=1.0, help="Refresh interval in seconds")
|
||||||
@click.option("--max-entries", type=int, default=50, help="Maximum entries per update")
|
@click.option("--max-entries", type=int, default=50, help="Maximum entries per update")
|
||||||
@click.pass_context
|
@click.pass_context
|
||||||
def watch(
|
def watch(ctx: click.Context, files: tuple, format: str, interval: float, max_entries: int) -> None:
|
||||||
ctx: click.Context,
|
'''Watch log files and display live updates.'''
|
||||||
files: tuple,
|
|
||||||
format: str,
|
|
||||||
interval: float,
|
|
||||||
max_entries: int
|
|
||||||
) -> None:
|
|
||||||
"""Watch log files and display live updates."""
|
|
||||||
if not files:
|
if not files:
|
||||||
click.echo("Error: No files specified for watching.")
|
click.echo("Error: No files specified for watching.")
|
||||||
ctx.exit(1)
|
ctx.exit(1)
|
||||||
@@ -189,19 +201,19 @@ def watch(
|
|||||||
|
|
||||||
@main.command("report")
|
@main.command("report")
|
||||||
@click.argument("files", type=click.Path(exists=True), nargs=-1)
|
@click.argument("files", type=click.Path(exists=True), nargs=-1)
|
||||||
@click.option("--format", type=click.Choice(["json", "syslog", "apache", "auto"]),
|
@click.option(
|
||||||
default="auto", help="Log format")
|
"--format",
|
||||||
|
type=click.Choice(["json", "syslog", "apache", "auto"]),
|
||||||
|
default="auto",
|
||||||
|
help="Log format",
|
||||||
|
)
|
||||||
@click.option("--output", type=click.Path(), help="Output file path (default: stdout)")
|
@click.option("--output", type=click.Path(), help="Output file path (default: stdout)")
|
||||||
@click.option("--json/--no-json", default=False, help="Output as JSON")
|
@click.option("--json/--no-json", default=False, help="Output as JSON")
|
||||||
@click.pass_context
|
@click.pass_context
|
||||||
def report(
|
def report(
|
||||||
ctx: click.Context,
|
ctx: click.Context, files: tuple, format: str, output: Optional[str], json: bool
|
||||||
files: tuple,
|
|
||||||
format: str,
|
|
||||||
output: Optional[str],
|
|
||||||
json: bool
|
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Generate detailed analysis report."""
|
'''Generate detailed analysis report.'''
|
||||||
if not files:
|
if not files:
|
||||||
click.echo("Error: No log files specified.")
|
click.echo("Error: No log files specified.")
|
||||||
ctx.exit(1)
|
ctx.exit(1)
|
||||||
@@ -218,10 +230,7 @@ def report(
|
|||||||
formatter = JSONFormatter()
|
formatter = JSONFormatter()
|
||||||
report_data = {
|
report_data = {
|
||||||
"files_analyzed": len(files),
|
"files_analyzed": len(files),
|
||||||
"results": [
|
"results": [{"file": path, "analysis": result} for path, result in all_results],
|
||||||
{"file": path, "analysis": result}
|
|
||||||
for path, result in all_results
|
|
||||||
]
|
|
||||||
}
|
}
|
||||||
report_text = formatter.format(report_data)
|
report_text = formatter.format(report_data)
|
||||||
else:
|
else:
|
||||||
@@ -236,8 +245,10 @@ def report(
|
|||||||
lines.append(f"=== {file_path} ===")
|
lines.append(f"=== {file_path} ===")
|
||||||
lines.append(f"Total Lines: {result.total_lines}")
|
lines.append(f"Total Lines: {result.total_lines}")
|
||||||
lines.append(f"Format: {result.format_detected.value}")
|
lines.append(f"Format: {result.format_detected.value}")
|
||||||
lines.append(f"Critical: {result.critical_count} | Error: {result.error_count} | "
|
lines.append(
|
||||||
f"Warning: {result.warning_count} | Info: {result.debug_count}")
|
f"Critical: {result.critical_count} | Error: {result.error_count} | "
|
||||||
|
f"Warning: {result.warning_count} | Info: {result.debug_count}"
|
||||||
|
)
|
||||||
lines.append("")
|
lines.append("")
|
||||||
|
|
||||||
if result.suggestions:
|
if result.suggestions:
|
||||||
@@ -258,11 +269,14 @@ def report(
|
|||||||
|
|
||||||
@main.command("patterns")
|
@main.command("patterns")
|
||||||
@click.option("--group", help="Filter by pattern group")
|
@click.option("--group", help="Filter by pattern group")
|
||||||
@click.option("--severity", type=click.Choice(["critical", "error", "warning", "info", "debug"]),
|
@click.option(
|
||||||
help="Filter by severity")
|
"--severity",
|
||||||
|
type=click.Choice(["critical", "error", "warning", "info", "debug"]),
|
||||||
|
help="Filter by severity",
|
||||||
|
)
|
||||||
@click.pass_context
|
@click.pass_context
|
||||||
def patterns(ctx: click.Context, group: str, severity: str) -> None:
|
def patterns(ctx: click.Context, group: str, severity: str) -> None:
|
||||||
"""List available error detection patterns."""
|
'''List available error detection patterns.'''
|
||||||
analyzer = LogAnalyzer()
|
analyzer = LogAnalyzer()
|
||||||
patterns_by_group = analyzer.list_patterns_by_group()
|
patterns_by_group = analyzer.list_patterns_by_group()
|
||||||
|
|
||||||
@@ -286,11 +300,10 @@ def patterns(ctx: click.Context, group: str, severity: str) -> None:
|
|||||||
"error": "red",
|
"error": "red",
|
||||||
"warning": "yellow",
|
"warning": "yellow",
|
||||||
"info": "blue",
|
"info": "blue",
|
||||||
"debug": "grey"
|
"debug": "grey",
|
||||||
}.get(pattern["severity"], "white")
|
}.get(pattern["severity"], "white")
|
||||||
formatter.console.print(
|
formatter.console.print(
|
||||||
f" [bold]{pattern['name']}[/] "
|
f" [bold]{pattern['name']}[/] " f"[{severity_color}]({pattern['severity']})[/]"
|
||||||
f"[{severity_color}]({pattern['severity']})[/]"
|
|
||||||
)
|
)
|
||||||
if pattern["description"]:
|
if pattern["description"]:
|
||||||
formatter.console.print(f" {pattern['description']}")
|
formatter.console.print(f" {pattern['description']}")
|
||||||
@@ -299,7 +312,7 @@ def patterns(ctx: click.Context, group: str, severity: str) -> None:
|
|||||||
@main.command("info")
|
@main.command("info")
|
||||||
@click.pass_context
|
@click.pass_context
|
||||||
def info(ctx: click.Context) -> None:
|
def info(ctx: click.Context) -> None:
|
||||||
"""Display LogLens information."""
|
'''Display LogLens information.'''
|
||||||
from loglens import __version__
|
from loglens import __version__
|
||||||
|
|
||||||
click.echo(f"LogLens CLI v{__version__}")
|
click.echo(f"LogLens CLI v{__version__}")
|
||||||
|
|||||||
Reference in New Issue
Block a user