Log-Analyse

[build-system]
requires = ["setuptools >= 77.0.3"]
build-backend = "setuptools.build_meta"

[project]
name = "loganalyser"
version = "1.0.0"
description = "Count log entries per level, source or hour, and write CSV or JSON."
requires-python = ">= 3.14"
dependencies = []

# TODO: a [project.scripts] table, so that pip install -e . creates a
# loganalyser command that calls main() in loganalyser/cli.py.

[tool.setuptools]
packages = ["loganalyser"]

"""Count log entries per level, source or hour, and write CSV or JSON."""

__version__ = "1.0.0"

"""python -m loganalyser runs the command line."""

import sys

from .cli import main

sys.exit(main())

"""The exceptions of the log analyser; the command line turns each into exit status 1."""


class LogAnalyserError(Exception):
    """Base class: every error this package raises on purpose."""


class InputError(LogAnalyserError):
    """A log file is missing, is a folder, or cannot be read."""


class ParseError(LogAnalyserError):
    """A line does not have the expected log format."""

    def __init__(self, line_no: int, line: str, reason: str) -> None:
        # TODO: call super().__init__ with a message such as
        # "line 3: not a log line: 'the line'", and keep line_no, line and reason
        # as attributes.
        raise NotImplementedError

"""The domain: log levels and parsed log entries."""

from dataclasses import dataclass
from datetime import datetime
from enum import IntEnum


class Level(IntEnum):
    """Log levels, ordered by severity so they compare: Level.ERROR > Level.INFO."""

    DEBUG = 10
    INFO = 20
    WARNING = 30
    ERROR = 40
    CRITICAL = 50

    @classmethod
    def parse(cls, text: str) -> "Level":
        """The level named by text, in any case; WARN and FATAL are accepted too."""
        # TODO: look the name up with cls[name]; turn a KeyError into ValueError.
        raise NotImplementedError


@dataclass(frozen=True, slots=True)
class Entry:
    """One parsed log line."""

    timestamp: datetime
    level: Level
    source: str
    message: str
    line_no: int = 0

"""Turning lines of text into Entry objects, lazily."""

import logging
import re
from collections.abc import Iterable, Iterator
from datetime import datetime
from pathlib import Path

from .errors import InputError, ParseError
from .model import Entry, Level

logger = logging.getLogger(__name__)

# 2026-09-28 14:03:12 INFO  auth: user alice logged in
# TODO: named groups timestamp, level, source and message.
LINE_RE = re.compile(r"TODO")
TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"


def parse_line(line: str, line_no: int = 0) -> Entry:
    """Parse one log line. Raise ParseError if it is not in the expected format."""
    # TODO: match LINE_RE, then datetime.strptime and Level.parse; turn their
    # ValueError into a ParseError chained with `from`.
    raise NotImplementedError


def parse_lines(lines: Iterable[str], *, source: str = "<input>", strict: bool = False) -> Iterator[Entry]:
    """Yield an Entry per log line. Blank lines are ignored.

    A bad line raises ParseError when strict, and is logged as a warning and
    skipped otherwise.
    """
    # TODO: a generator; number the lines from 1 with enumerate.
    raise NotImplementedError


def read_entries(paths: Iterable[Path], *, strict: bool = False) -> Iterator[Entry]:
    """Yield the entries of each file in turn, reading one line at a time."""
    # TODO: raise InputError for a path that is not a file; open each with
    # `with path.open(encoding="utf-8")` and `yield from parse_lines(...)`.
    raise NotImplementedError

"""Filtering, counting and writing the results as CSV or JSON."""

import csv
import json
from collections import Counter
from collections.abc import Iterable, Iterator
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import TextIO

from .model import Entry, Level


class GroupBy(Enum):
    """What to count entries by."""

    LEVEL = "level"
    SOURCE = "source"
    HOUR = "hour"

    def key(self, entry: Entry) -> str:
        """The group an entry falls into, as text: "ERROR", "db" or "2026-09-28 14:00"."""
        # TODO: match self.
        raise NotImplementedError


def within(entries: Iterable[Entry], since: datetime | None = None, until: datetime | None = None) -> Iterator[Entry]:
    """The entries at or after since and before until (either may be None)."""
    # TODO: a generator.
    raise NotImplementedError


def at_least(entries: Iterable[Entry], level: Level) -> Iterator[Entry]:
    """The entries whose level is level or more severe."""
    # TODO: a generator expression.
    raise NotImplementedError


@dataclass
class Report:
    """Counts per group, plus the size and time span of what was counted."""

    by: GroupBy
    counts: list[tuple[str, int]] = field(default_factory=list)
    entries: int = 0
    first: datetime | None = None
    last: datetime | None = None

    def to_dict(self) -> dict[str, object]:
        """A JSON-ready dict: by, entries, first, last and counts."""
        # TODO: first and last as "2026-09-28 09:00:01"; counts as
        # [{"level": "INFO", "count": 2}, ...], keyed by self.by.value.
        raise NotImplementedError


def summarise(entries: Iterable[Entry], by: GroupBy) -> Report:
    """Count the entries per group in one pass over them.

    Order the counts: levels by severity, sources most common first (ties by
    name), hours in time order.
    """
    # TODO: a collections.Counter, then sort.
    raise NotImplementedError


def write_csv(report: Report, out: TextIO) -> None:
    """One header row (the group name and count), then a row per group."""
    # TODO: csv.writer(out, lineterminator="\n").
    raise NotImplementedError


def write_json(report: Report, out: TextIO) -> None:
    """The report as an indented JSON object."""
    # TODO: json.dump(report.to_dict(), ...).
    raise NotImplementedError

"""The command line: loganalyser [options] FILE [FILE ...]."""

import argparse
import logging
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from typing import TextIO

from . import __version__
from .errors import LogAnalyserError, ParseError
from .model import Level
from .parser import read_entries
from .report import GroupBy, at_least, summarise, within, write_csv, write_json

logger = logging.getLogger("loganalyser")

EXIT_OK = 0
EXIT_ERROR = 1  # bad input: a missing file, or a bad line with --strict
# argparse itself exits with status 2 on a usage error.


def when(text: str) -> datetime:
    """An argparse type: 2026-09-28, 2026-09-28T14:00 or "2026-09-28 14:00"."""
    # TODO: datetime.fromisoformat; raise argparse.ArgumentTypeError if it fails.
    raise NotImplementedError


def level(text: str) -> Level:
    """An argparse type: a level name such as warning or ERROR."""
    # TODO: Level.parse; raise argparse.ArgumentTypeError if it fails.
    raise NotImplementedError


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="loganalyser",
        description="Count log entries per level, source or hour, and write the counts as CSV or JSON.",
    )
    parser.add_argument("files", nargs="+", type=Path, metavar="FILE", help="log files to read, in order")
    # TODO: --by, --format, -o/--output, --since, --until, --level, --strict,
    # -v/--verbose (a count) and --version.
    return parser


def configure_logging(verbosity: int) -> None:
    """Progress and warnings go to standard error, so they never mix with the output."""
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
    logger.handlers[:] = [handler]
    logger.setLevel(logging.WARNING if verbosity == 0 else logging.INFO if verbosity == 1 else logging.DEBUG)
    logger.propagate = False


@contextmanager
def open_output(path: Path | None) -> Iterator[TextIO]:
    """Standard output, or the file at path, closed afterwards."""
    # TODO: yield sys.stdout for None; otherwise open the file with
    # newline="" (the csv module wants it) inside a with statement.
    raise NotImplementedError
    yield sys.stdout


def main(argv: list[str] | None = None) -> int:
    """Run the tool with argv (default: sys.argv[1:]); return the exit status."""
    # TODO: parse, configure logging, read -> filter -> summarise -> write,
    # and turn each LogAnalyserError into a logged error and EXIT_ERROR.
    raise NotImplementedError


if __name__ == "__main__":
    sys.exit(main())

2026-09-28 08:59:58 INFO  web: server started on port 8080
2026-09-28 09:00:01 INFO  auth: user alice logged in
2026-09-28 09:00:07 DEBUG db: pool size 5
2026-09-28 09:02:13 WARN  web: slow response for /reports (2.4 s)
2026-09-28 09:15:42 ERROR db: connection lost, retrying
2026-09-28 09:15:43 INFO  db: reconnected
this line was cut off by a crash
2026-09-28 10:01:09 INFO  auth: user bob logged in
2026-09-28 10:05:30 ERROR web: 500 on /export (KeyError: 'month')
2026-09-28 10:05:31 WARNING auth: 3 failed logins for carol

2026-09-28 11:20:00 CRITICAL db: disk full
2026-09-28 11:20:05 INFO  web: server stopping

import contextlib
import csv
import inspect
import io
import json
import tempfile
import tomllib
from datetime import datetime
from pathlib import Path

from loganalyser.cli import main
from loganalyser.errors import InputError, LogAnalyserError, ParseError
from loganalyser.model import Entry, Level
from loganalyser.parser import parse_line, parse_lines, read_entries
from loganalyser.report import GroupBy, at_least, summarise, within, write_csv, write_json

LOG = """2026-09-28 09:00:01 INFO  auth: user alice logged in
2026-09-28 09:15:42 ERROR db: connection lost
not a log line
2026-09-28 10:05:30 WARN  web: slow response

2026-09-28 10:59:59 ERROR web: 500 on /export
2026-09-28 11:00:00 INFO  auth: user bob logged in
"""


def entries_of(text=LOG):
    """The entries of text, parsed leniently (the warning for line 3 is dropped)."""
    with contextlib.redirect_stderr(io.StringIO()):
        return list(parse_lines(io.StringIO(text)))


def run(argv):
    """Call main(argv) and return (status, stdout, stderr)."""
    out, err = io.StringIO(), io.StringIO()
    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
        status = main(argv)
    return status, out.getvalue(), err.getvalue()


def write_log(folder, text=LOG, name="app.log"):
    path = Path(folder) / name
    path.write_text(text, encoding="utf-8")
    return str(path)


def test_level_parse():
    """Level.parse liest Namen in beliebiger Schreibung, akzeptiert WARN, und Stufen vergleichen sich nach Schwere"""
    got = Level.parse("error"), Level.parse("WARN"), Level.parse("Info")
    assert got == (Level.ERROR, Level.WARNING, Level.INFO), f"Level.parse ergab {got!r}"
    assert Level.ERROR > Level.WARNING > Level.INFO > Level.DEBUG, "Stufen sollten sich nach Schwere vergleichen: ERROR > WARNING > INFO > DEBUG"
    try:
        Level.parse("loud")
    except ValueError:
        return
    raise AssertionError("Level.parse('loud') sollte einen ValueError auslösen")


def test_parse_line():
    """parse_line macht aus einer Logzeile ein Entry"""
    got = parse_line("2026-09-28 09:15:42 ERROR db: connection lost, retrying\n", 7)
    want = Entry(datetime(2026, 9, 28, 9, 15, 42), Level.ERROR, "db", "connection lost, retrying", 7)
    assert got == want, f"parse_line ergab {got!r}, erwartet: {want!r}"


def test_parse_line_rejects():
    """parse_line löst ParseError aus, einen LogAnalyserError mit der Zeilennummer"""
    for line in ["not a log line", "2026-02-30 09:00:00 INFO web: no such day", "2026-09-28 09:00:00 LOUD web: odd level"]:
        try:
            got = parse_line(line, 3)
        except ParseError as err:
            assert isinstance(err, LogAnalyserError), "ParseError sollte eine Unterklasse von LogAnalyserError sein"
            assert err.line_no == 3, f"bei {line!r} hat der Fehler line_no {err.line_no!r}, erwartet: 3"
            continue
        raise AssertionError(f"parse_line({line!r}) gab {got!r} zurück, erwartet: ein ParseError")


def test_read_entries_skips_bad_lines():
    """read_entries ist ein Generator, der eine fehlerhafte Zeile mit Warnung überspringt oder mit strict abbricht"""
    with tempfile.TemporaryDirectory() as folder:
        path = Path(write_log(folder))
        gen = read_entries([path])
        assert inspect.isgenerator(gen), f"read_entries gab {type(gen).__name__} zurück, erwartet: ein Generator"
        with contextlib.redirect_stderr(io.StringIO()):
            got = [e.line_no for e in gen]
        assert got == [1, 2, 4, 6, 7], f"read_entries lieferte Einträge aus den Zeilen {got!r}, erwartet: [1, 2, 4, 6, 7]"
        try:
            list(read_entries([path], strict=True))
        except ParseError as err:
            assert err.line_no == 3, f"mit strict hat der ParseError line_no {err.line_no}, erwartet: 3"
        else:
            raise AssertionError("mit strict=True sollte read_entries bei Zeile 3 einen ParseError auslösen")
        try:
            list(read_entries([Path(folder) / "missing.log"]))
        except InputError:
            pass
        else:
            raise AssertionError("bei einer fehlenden Datei sollte read_entries einen InputError auslösen")


def test_filters():
    """within behält since <= Zeit < until, und at_least behält eine Stufe und schwerere"""
    entries = entries_of()
    got = [e.line_no for e in within(entries, datetime(2026, 9, 28, 9, 15, 42), datetime(2026, 9, 28, 11, 0))]
    assert got == [2, 4, 6], f"within 09:15:42 und 11:00 behielt die Zeilen {got!r}, erwartet: [2, 4, 6]"
    got = [e.line_no for e in at_least(entries, Level.WARNING)]
    assert got == [2, 4, 6], f"at_least WARNING behielt die Zeilen {got!r}, erwartet: [2, 4, 6]"


def test_summarise():
    """summarise zählt je Stufe nach Schwere, je Quelle die häufigste zuerst und je Stunde"""
    entries = entries_of()
    got = summarise(entries, GroupBy.LEVEL)
    assert got.counts == [("INFO", 2), ("WARNING", 1), ("ERROR", 2)], f"die Zählung nach Stufe ist {got.counts!r}"
    assert (got.entries, got.first, got.last) == (5, datetime(2026, 9, 28, 9, 0, 1), datetime(2026, 9, 28, 11, 0)), f"entries, first und last sind {(got.entries, got.first, got.last)!r}"
    got = summarise(entries, GroupBy.SOURCE).counts
    assert got == [("auth", 2), ("web", 2), ("db", 1)], f"die Zählung nach Quelle ist {got!r}, erwartet: [('auth', 2), ('web', 2), ('db', 1)]"
    got = summarise(entries, GroupBy.HOUR).counts
    want = [("2026-09-28 09:00", 2), ("2026-09-28 10:00", 2), ("2026-09-28 11:00", 1)]
    assert got == want, f"die Zählung nach Stunde ist {got!r}, erwartet: {want!r}"


def test_write_csv_and_json():
    """write_csv schreibt eine Kopfzeile und eine Zeile je Gruppe; write_json den ganzen Bericht"""
    report = summarise(entries_of(), GroupBy.SOURCE)
    out = io.StringIO()
    write_csv(report, out)
    rows = list(csv.reader(io.StringIO(out.getvalue())))
    assert rows == [["source", "count"], ["auth", "2"], ["web", "2"], ["db", "1"]], f"die CSV-Zeilen sind {rows!r}"
    out = io.StringIO()
    write_json(report, out)
    got = json.loads(out.getvalue())
    want = {
        "by": "source",
        "entries": 5,
        "first": "2026-09-28 09:00:01",
        "last": "2026-09-28 11:00:00",
        "counts": [{"source": "auth", "count": 2}, {"source": "web", "count": 2}, {"source": "db", "count": 1}],
    }
    assert got == want, f"das JSON ist {got!r}, erwartet: {want!r}"


def test_cli_csv_to_stdout():
    """main([DATEI]) gibt die Zählung nach Stufe als CSV aus und gibt 0 zurück"""
    with tempfile.TemporaryDirectory() as folder:
        status, out, _ = run([write_log(folder)])
    assert status == 0, f"main gab {status} zurück, erwartet: 0"
    assert out.splitlines() == ["level,count", "INFO,2", "WARNING,1", "ERROR,2"], f"main gab {out!r} aus"


def test_cli_json_file():
    """--format json -o DATEI schreibt den Bericht in die Datei und gibt nichts aus"""
    with tempfile.TemporaryDirectory() as folder:
        target = Path(folder) / "report.json"
        status, out, _ = run([write_log(folder), "--by", "hour", "--format", "json", "-o", str(target)])
        assert status == 0, f"main gab {status} zurück, erwartet: 0"
        assert out == "", f"mit -o gab main trotzdem {out!r} aus"
        got = json.loads(target.read_text(encoding="utf-8"))
    assert got["by"] == "hour" and got["entries"] == 5, f"die JSON-Datei enthält {got!r}"


def test_cli_filters():
    """--since, --until und --level grenzen ein, was gezählt wird"""
    with tempfile.TemporaryDirectory() as folder:
        path = write_log(folder)
        _, out, _ = run([path, "--since", "2026-09-28T09:10", "--until", "2026-09-28 11:00"])
        assert out.splitlines() == ["level,count", "WARNING,1", "ERROR,2"], f"mit --since und --until gab main {out!r} aus"
        _, out, _ = run([path, "--level", "error", "--by", "source"])
        assert out.splitlines() == ["source,count", "db,1", "web,1"], f"mit --level error --by source gab main {out!r} aus"


def test_cli_several_files():
    """Mehrere Dateien werden nacheinander gelesen und zusammen gezählt"""
    with tempfile.TemporaryDirectory() as folder:
        first = write_log(folder, "2026-09-28 09:00:00 INFO a: one\n", "a.log")
        second = write_log(folder, "2026-09-29 09:00:00 INFO b: two\n2026-09-29 09:00:01 ERROR b: three\n", "b.log")
        status, out, _ = run([first, second, "--by", "source"])
    assert (status, out.splitlines()) == (0, ["source,count", "b,2", "a,1"]), f"main gab {status} zurück und {out!r} aus"


def test_cli_missing_file():
    """Eine fehlende Datei ergibt Exit-Status 1 und eine Fehlermeldung auf der Standardfehlerausgabe"""
    with tempfile.TemporaryDirectory() as folder:
        missing = str(Path(folder) / "nope.log")
        status, out, err = run([missing])
    assert status == 1, f"main gab bei fehlender Datei {status} zurück, erwartet: 1"
    assert "nope.log" in err and out == "", f"erwartet: ein Fehler mit nope.log auf stderr und nichts auf stdout; stderr war {err!r}"


def test_cli_strict():
    """Eine fehlerhafte Zeile ergibt eine protokollierte Warnung, mit --strict Exit-Status 1"""
    with tempfile.TemporaryDirectory() as folder:
        path = write_log(folder)
        status, _, err = run([path])
        assert status == 0 and "3" in err and "WARNING" in err, f"ohne --strict: Status {status}, stderr {err!r}; erwartet: 0 und eine Warnung zu Zeile 3"
        status, out, err = run([path, "--strict"])
    assert status == 1 and out == "", f"mit --strict gab main {status} zurück und {out!r} aus, erwartet: 1 und nichts"
    assert "line 3" in err, f"mit --strict sollte der Fehler {err!r} line 3 nennen"


def test_cli_bad_arguments():
    """Bei falschen Argumenten beendet argparse das Programm mit Status 2"""
    for argv in [[], ["app.log", "--since", "yesterday"], ["app.log", "--format", "xml"], ["app.log", "--level", "loud"]]:
        try:
            with contextlib.redirect_stderr(io.StringIO()):
                main(argv)
        except SystemExit as exc:
            assert exc.code == 2, f"bei {argv!r} war der Exit-Status {exc.code!r}, erwartet: 2"
        else:
            raise AssertionError(f"main({argv!r}) kehrte zurück, statt mit Status 2 zu enden")


def test_cli_logs_progress():
    """-v protokolliert jede gelesene Datei und die Anzahl auf der Standardfehlerausgabe"""
    with tempfile.TemporaryDirectory() as folder:
        path = write_log(folder)
        status, out, err = run([path, "-v"])
    assert status == 0, f"main gab {status} zurück, erwartet: 0"
    assert "INFO" in err and "app.log" in err and "5" in err, f"mit -v war stderr {err!r}, erwartet: INFO-Zeilen mit app.log und der Anzahl 5"
    assert "INFO" not in out.replace("INFO,", ""), f"Logzeilen sind in die Standardausgabe geraten: {out!r}"


def test_sample_log():
    """samples/app.log ergibt die erwartete Zählung"""
    status, out, _ = run(["samples/app.log"])
    want = ["level,count", "DEBUG,1", "INFO,5", "WARNING,2", "ERROR,2", "CRITICAL,1"]
    assert (status, out.splitlines()) == (0, want), f"bei samples/app.log gab main {status} zurück und {out.splitlines()!r} aus, erwartet: {want!r}"


def test_pyproject_script():
    """pyproject.toml deklariert das Konsolenskript loganalyser"""
    with open("pyproject.toml", "rb") as f:
        project = tomllib.load(f).get("project", {})
    scripts = project.get("scripts", {})
    assert scripts.get("loganalyser") == "loganalyser.cli:main", f"[project.scripts] ist {scripts!r}, erwartet: loganalyser = \"loganalyser.cli:main\""
    assert project.get("requires-python"), "[project] sollte angeben, welches Python es braucht (requires-python)"