Add --top N to --summary

README.md

@@ -171,6 +171,33 @@ For the three example log lines above, this prints:
 
 The object has the keys `levels`, `top_errors` and `skipped_lines`.
 
+Top N: add `--top N` to `--summary` to change how many error messages are listed
+(default 5, at most 100).
+
+For example, with this `errors.log`:
+
+```text
+2026-10-10 12:00:00 ERROR Disk full
+2026-10-10 12:00:05 ERROR Timeout
+2026-10-10 12:00:09 ERROR Disk full
+2026-10-10 12:01:00 CRITICAL Out of memory
+```
+
+`logslice errors.log --summary --top 2` prints:
+
+```text
+Level counts:
+DEBUG: 0
+INFO: 0
+WARNING: 0
+ERROR: 3
+CRITICAL: 1
+Top 2 error messages:
+  2  Disk full
+  1  Timeout
+Skipped lines: 0
+```
+
 Input is processed line by line as UTF-8, replacing invalid bytes. LF and CRLF
 line endings are supported; output uses UTF-8. Summary mode keeps per-level and
 per-distinct-error-message counters rather than loading all records into memory.
@@ -191,8 +218,9 @@ have been written when a later file fails.
 | `--grep PATTERN` | Keep only records whose message matches the Python regular expression `PATTERN` (searched in the message only). |
 | `-i`, `--ignore-case` | Make `--grep` case-insensitive. |
 | `-v`, `--invert-match` | With `--grep`, keep only records whose message does not match; an error without `--grep`. |
-| `--summary` | Print level counts, the top 5 ERROR/CRITICAL messages and the skipped line count instead of records. |
-| `--summary-format {text,json}` | Output format of `--summary`, default `text`; `json` prints the level counts, the top 5 error messages and the skipped line count as a JSON object. |
+| `--summary` | Print level counts, up to the top 5 ERROR/CRITICAL messages by default, and the skipped line count instead of records. |
+| `--summary-format {text,json}` | Output format of `--summary`, default `text`; `json` prints the level counts, up to the top 5 error messages by default, and the skipped line count as a JSON object. |
+| `--top N` | With `--summary`, list the N most frequent ERROR/CRITICAL messages instead of 5; `N` is 1 to 100. An error without `--summary`. |
 | `-h`, `--help` | Show help. |
 
 ## Development

src/logslice/cli.py

@@ -6,7 +6,7 @@
 from contextlib import nullcontext
 from typing import TextIO
 
-from .filters import matches, parse_datetime, parse_level
+from .filters import matches, parse_datetime, parse_level, parse_top
 from .parser import Level, parse_line
 from .summary import Summary
 
@@ -59,6 +59,10 @@ def build_parser() -> argparse.ArgumentParser:
         "--summary-format", choices=["text", "json"], default="text",
         help="output format of --summary: text or json (default: text)",
     )
+    parser.add_argument(
+        "--top", type=parse_top, default=None, metavar="N",
+        help="number of ERROR/CRITICAL messages with --summary (1-100; default: 5)",
+    )
     return parser
 
 
@@ -77,6 +81,9 @@ def main(argv: list[str] | None = None) -> int:
     if args.summary_format == "json" and not args.summary:
         print("logslice: --summary-format requires --summary", file=sys.stderr)
         return 2
+    if args.top is not None and not args.summary:
+        print("logslice: --top requires --summary", file=sys.stderr)
+        return 2
     if args.invert_match and args.grep is None:
         print("logslice: --invert-match requires --grep", file=sys.stderr)
         return 2
@@ -120,5 +127,10 @@ def main(argv: list[str] | None = None) -> int:
             print(f"logslice: cannot read {path!r}: {error}", file=sys.stderr)
             return 2
     if args.summary:
-        print(summary.render_json() if args.summary_format == "json" else summary.render())
+        top = 5 if args.top is None else args.top
+        print(
+            summary.render_json(top)
+            if args.summary_format == "json"
+            else summary.render(top)
+        )
     return 0

src/logslice/filters.py

@@ -40,6 +40,20 @@ def parse_level(value: str) -> Level:
         ) from None
 
 
+def parse_top(value: str) -> int:
+    """Parse a top-error limit from one to 100."""
+    if re.fullmatch(r"[0-9]{1,3}", value) is None:
+        raise argparse.ArgumentTypeError(
+            f"invalid top count {value!r}: expected an integer from 1 through 100"
+        )
+    count = int(value)
+    if not 1 <= count <= 100:
+        raise argparse.ArgumentTypeError(
+            f"invalid top count {value!r}: expected an integer from 1 through 100"
+        )
+    return count
+
+
 def matches(
     record: Record,
     *,

src/logslice/summary.py

@@ -21,26 +21,28 @@ def add(self, record: Record) -> None:
         if record.level >= Level.ERROR:
             self.error_messages[record.message] += 1
 
-    def render(self) -> str:
-        """Return all levels, up to five error messages, and skipped count."""
+    def render(self, top: int = 5) -> str:
+        """Return all levels, up to top error messages, and skipped count."""
         lines = ["Level counts:"]
         lines.extend(f"{level.name}: {self.level_counts[level]}" for level in Level)
-        lines.append("Top 5 error messages:")
-        top = self.error_messages.most_common(5)
-        if top:
-            lines.extend(f"  {count}  {message}" for message, count in top)
+        lines.append(f"Top {top} error messages:")
+        top_messages = self.error_messages.most_common(top)
+        if top_messages:
+            lines.extend(
+                f"  {count}  {message}" for message, count in top_messages
+            )
         else:
             lines.append("  (none)")
         lines.append(f"Skipped lines: {self.skipped_lines}")
         return "\n".join(lines)
 
-    def render_json(self) -> str:
-        """Return level counts, up to five error messages, and skipped count as JSON."""
+    def render_json(self, top: int = 5) -> str:
+        """Return level counts, up to top error messages, and skipped count as JSON."""
         payload = {
             "levels": {level.name: self.level_counts[level] for level in Level},
             "top_errors": [
                 {"count": count, "message": message}
-                for message, count in self.error_messages.most_common(5)
+                for message, count in self.error_messages.most_common(top)
             ],
             "skipped_lines": self.skipped_lines,
         }

tests/test_cli.py

@@ -49,7 +49,7 @@ def test_help_documents_inputs_options_and_accepted_forms():
         "--since", "--until", "--level", "--summary", "stdin", "'-'",
         "YYYY-MM-DD", "HH:MM:SS", "fraction", "DEBUG", "INFO", "WARNING",
         "WARN", "ERROR", "CRITICAL", "case-insensitive",
-        "--grep", "--ignore-case", "--summary-format",
+        "--grep", "--ignore-case", "--summary-format", "--top",
     ):
         assert text in result.stdout
 

tests/test_summary.py

@@ -104,6 +104,13 @@ def test_top_five_descending_counts_first_occurrence_ties_and_skipped_count():
     )
 
 
+def test_render_default_matches_explicit_top_five():
+    summary = Summary()
+    summary.add(make_record(Level.ERROR, "failure"))
+
+    assert summary.render() == summary.render(5)
+
+
 def test_error_and_critical_messages_share_counts_but_lower_levels_do_not():
     summary = Summary()
     for level in Level:

tests/test_top.py

@@ -0,0 +1,221 @@
+"""Tests for configurable top-error summaries through the CLI."""
+
+import json
+
+import pytest
+
+from logslice.cli import main
+
+LOG_CONTENT = (
+    "2026-10-10 12:00:00 ERROR Disk full\n"
+    "2026-10-10 12:00:05 ERROR Timeout\n"
+    "2026-10-10 12:00:09 ERROR Disk full\n"
+    "2026-10-10 12:01:00 CRITICAL Out of memory\n"
+)
+
+
+def write_log(tmp_path, content=LOG_CONTENT):
+    path = tmp_path / "app.log"
+    path.write_text(content, encoding="utf-8")
+    return path
+
+
+def run_cli(capsys, *args):
+    exit_code = main([str(arg) for arg in args])
+    captured = capsys.readouterr()
+    return exit_code, captured.out, captured.err
+
+
+def run_argparse_cli(capsys, *args):
+    with pytest.raises(SystemExit) as error:
+        main([str(arg) for arg in args])
+    captured = capsys.readouterr()
+    return error.value.code, captured.out, captured.err
+
+
+def test_top_two_text_summary_matches_worked_example(tmp_path, capsys):
+    path = write_log(tmp_path)
+
+    exit_code, stdout, stderr = run_cli(capsys, path, "--summary", "--top", "2")
+
+    assert exit_code == 0
+    assert stderr == ""
+    assert stdout == (
+        "Level counts:\n"
+        "DEBUG: 0\n"
+        "INFO: 0\n"
+        "WARNING: 0\n"
+        "ERROR: 3\n"
+        "CRITICAL: 1\n"
+        "Top 2 error messages:\n"
+        "  2  Disk full\n"
+        "  1  Timeout\n"
+        "Skipped lines: 0\n"
+    )
+
+
+def test_top_limits_json_summary_without_changing_keys_or_counts(tmp_path, capsys):
+    path = write_log(tmp_path)
+
+    exit_code, stdout, stderr = run_cli(
+        capsys, path, "--summary", "--top", "1", "--summary-format", "json"
+    )
+
+    assert exit_code == 0
+    assert stderr == ""
+    summary = json.loads(stdout)
+    assert list(summary) == ["levels", "top_errors", "skipped_lines"]
+    assert summary["levels"]["ERROR"] == 3
+    assert summary["levels"]["CRITICAL"] == 1
+    assert summary["top_errors"] == [{"count": 2, "message": "Disk full"}]
+    assert summary["skipped_lines"] == 0
+
+
+@pytest.mark.parametrize(
+    ("top_args", "expected_header", "expected_entries"),
+    [
+        (("--top", "1"), "Top 1 error messages:\n", "  2  Disk full\n"),
+        (
+            ("--top=2",),
+            "Top 2 error messages:\n",
+            "  2  Disk full\n  1  Timeout\n",
+        ),
+        (
+            ("--top", "07"),
+            "Top 7 error messages:\n",
+            "  2  Disk full\n  1  Timeout\n  1  Out of memory\n",
+        ),
+        (
+            ("--top", "100"),
+            "Top 100 error messages:\n",
+            "  2  Disk full\n  1  Timeout\n  1  Out of memory\n",
+        ),
+    ],
+)
+def test_top_accepts_valid_values_and_normalizes_the_header(
+    tmp_path, capsys, top_args, expected_header, expected_entries
+):
+    path = write_log(tmp_path)
+
+    exit_code, stdout, stderr = run_cli(capsys, path, "--summary", *top_args)
+
+    assert exit_code == 0
+    assert stderr == ""
+    assert expected_header in stdout
+    assert expected_entries in stdout
+
+
+@pytest.mark.parametrize(
+    "value",
+    ["0", "101", "1000", "-3", "+5", " 5", "5 ", "1e1", "x", "٣", ""],
+)
+def test_invalid_top_values_are_argparse_usage_errors(capsys, value):
+    exit_code, stdout, stderr = run_argparse_cli(
+        capsys, "--summary", "--top", value
+    )
+
+    assert exit_code == 2
+    assert stdout == ""
+    assert stderr.rstrip().endswith(
+        "argument --top: invalid top count "
+        f"{value!r}: expected an integer from 1 through 100"
+    )
+    assert "Traceback" not in stderr
+
+
+def test_top_without_value_keeps_argparse_missing_value_error(capsys):
+    exit_code, stdout, stderr = run_argparse_cli(capsys, "--summary", "--top")
+
+    assert exit_code == 2
+    assert stdout == ""
+    assert "argument --top: expected one argument" in stderr
+
+
+def test_top_without_summary_fails_before_reading_input(tmp_path, capsys):
+    missing_path = tmp_path / "missing.log"
+
+    exit_code, stdout, stderr = run_cli(capsys, missing_path, "--top", "5")
+
+    assert exit_code == 2
+    assert stdout == ""
+    assert stderr == "logslice: --top requires --summary\n"
+
+
+def test_summary_format_validation_precedes_top_validation(capsys):
+    exit_code, stdout, stderr = run_cli(
+        capsys, "--top", "5", "--summary-format", "json"
+    )
+
+    assert exit_code == 2
+    assert stdout == ""
+    assert stderr == "logslice: --summary-format requires --summary\n"
+
+
+def test_top_validation_precedes_invert_match_validation(capsys):
+    exit_code, stdout, stderr = run_cli(capsys, "--top", "2", "-v")
+
+    assert exit_code == 2
+    assert stdout == ""
+    assert stderr == "logslice: --top requires --summary\n"
+
+
+def test_repeated_top_uses_last_value(tmp_path, capsys):
+    path = write_log(tmp_path)
+
+    exit_code, stdout, stderr = run_cli(
+        capsys, path, "--summary", "--top", "1", "--top", "2"
+    )
+
+    assert exit_code == 0
+    assert stderr == ""
+    assert "Top 2 error messages:\n" in stdout
+    assert "  1  Timeout\n" in stdout
+
+
+def test_top_with_fewer_distinct_errors_lists_only_available_messages(tmp_path, capsys):
+    path = write_log(
+        tmp_path,
+        "2026-10-10 12:00:00 ERROR Disk full\n"
+        "2026-10-10 12:00:01 ERROR Disk full\n",
+    )
+
+    exit_code, stdout, stderr = run_cli(capsys, path, "--summary", "--top", "100")
+
+    assert exit_code == 0
+    assert stderr == ""
+    assert "Top 100 error messages:\n  2  Disk full\n" in stdout
+    assert "  (none)" not in stdout
+
+
+def test_top_with_no_error_messages_keeps_none_marker(tmp_path, capsys):
+    path = write_log(
+        tmp_path, "2026-10-10 12:00:00 INFO Started\n"
+    )
+
+    exit_code, stdout, stderr = run_cli(capsys, path, "--summary", "--top", "3")
+
+    assert exit_code == 0
+    assert stderr == ""
+    assert "Top 3 error messages:\n  (none)\n" in stdout
+
+
+@pytest.mark.parametrize(
+    ("filters", "expected_entries"),
+    [
+        (("--level", "critical"), "  1  Out of memory\n"),
+        (("--grep", "Disk"), "  2  Disk full\n"),
+        (("--grep", "Disk", "-v"), "  1  Timeout\n  1  Out of memory\n"),
+    ],
+)
+def test_top_preserves_level_and_grep_filter_interactions(
+    tmp_path, capsys, filters, expected_entries
+):
+    path = write_log(tmp_path)
+
+    exit_code, stdout, stderr = run_cli(
+        capsys, path, "--summary", "--top", "3", *filters
+    )
+
+    assert exit_code == 0
+    assert stderr == ""
+    assert expected_entries in stdout