maurane/logwatcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

♻️refactor(parser): refactor parsing methods

Browse Source
This commit is contained in:
maurane
2026-09-10 09:46:16 +02:00
parent d37733457f
commit cf83a06cfe
3 changed files with 91 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/logwatcher/cli.py
+14 -26
View File
@@ -14,7 +14,7 @@ from logwatcher.mail_reader import (
fetch_log_messages,
)
from logwatcher.models import LogEntry
from logwatcher.parser import parse_log_file, parse_log_line
from logwatcher.parser import parse_file, parse_lines
from logwatcher.reporter import write_log_report
app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.")
@@ -23,19 +23,6 @@ app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.")
logger = logging.getLogger("logwatcher")
def version_callback(value: bool):
"""
Print the version of package.
Stop the CLI after printing version.
Args:
value: True if '--version' is provided, False otherwise.
"""
if value:
print(f"logwatcher version: {__version__}")
raise typer.Exit()
def _is_log_file(path: Path) -> bool:
"""
Return True if file is a log file (.txt, .log, or extensionless).
@@ -60,26 +47,27 @@ def _get_files_from_dir(input_dir: Path) -> list[Path]:
def _files_to_log_entries(input_files: list[Path]) -> list[LogEntry]:
"""Parse each input file, raising if any path is not a file."""
log_entries = []
for file in input_files:
if not file.is_file():
logger.error(f"Unrecognized input file '{file}'.")
raise ValueError(f"'{file}' is not a file")
log_entries.extend(parse_log_file(file))
log_entries.extend(parse_file(file))
return log_entries
def _lines_to_log_entries(logs: list[str]) -> list[LogEntry]:
log_entries = []
for log in logs:
log_entry = parse_log_line(log)
if log_entry:
log_entries.append(log_entry)
return log_entries
def version_callback(value: bool):
"""
Print the version of package.
Stop the CLI after printing version.
def analyze_logs():
pass
Args:
value: True if '--version' is provided, False otherwise.
"""
if value:
print(f"logwatcher version: {__version__}")
raise typer.Exit()
@app.command()
@@ -95,7 +83,7 @@ def from_mail(
account = connect_to_mailbox()
messages = fetch_log_messages(account)
logs = extract_logs_from_mails(messages)
log_entries = _lines_to_log_entries(logs)
log_entries = parse_lines(logs)
nb_files = len(messages)
relevant, irrelevant = classify_log_entries(log_entries)
src/logwatcher/parser.py
+33 -20
View File
@@ -1,5 +1,6 @@
import logging
import re
from collections.abc import Iterable
from datetime import datetime
from pathlib import Path
@@ -35,8 +36,9 @@ misc_file_path = Path.cwd() / "output/misc.log"
effective_file_path = Path.cwd() / "output/effective.log"
def parse_log_line(log_line: str) -> LogEntry | None:
"""Parse a raw log line into a LogEntry object.
def parse_line(log_line: str) -> LogEntry | None:
"""
Parse a raw log line into a LogEntry object.
Args:
log_line: a line in a log file received by N2 technicians
@@ -65,10 +67,32 @@ def parse_log_line(log_line: str) -> LogEntry | None:
)
def parse_log_file(log_file_path: Path) -> list[LogEntry]:
"""Transforms the content of a log file into a list of LogEntry.
Uses `parse_log_line` to parse each line of the choosen log file.
Raise exception if a log line encounters an error.
def parse_lines(lines: Iterable[str]) -> list[LogEntry]:
"""
Parse an iterable of log lines into LogEntry objects.
Unrecognized lines are skipped. Logs an error for lines that raise a ValueError.
Args:
lines: Iterable of raw log lines.
Returns:
List of parsed LogEntry objects.
"""
log_entries: list[LogEntry] = []
for index, line in enumerate(lines):
try:
log_entry = parse_line(line)
except ValueError as error:
logger.error(f"\t\tline {index}: error while parsing the log: {error}")
else:
if log_entry:
log_entries.append(log_entry)
return log_entries
def parse_file(log_file_path: Path) -> list[LogEntry]:
"""
Transforms the content of a log file into a list of LogEntry.
Args:
log_file_path: Path of a log file
@@ -76,20 +100,9 @@ def parse_log_file(log_file_path: Path) -> list[LogEntry]:
Returns:
A list of LogEntry
"""
log_entry_list: list[LogEntry] = []
with open(log_file_path, "r", encoding="windows-1252") as log_file:
logger.info(f"\tparsing log file '{log_file_path.name}' started.")
for index, log_line in enumerate(log_file):
try:
log_entry = parse_log_line(log_line)
except ValueError as error:
logger.error(
f"\t\tline {index}: error while parsing the log file '{log_file.name}': {error}"
)
else:
if log_entry:
log_entry_list.append(log_entry)
logger.info(f"\t\t{len(log_entry_list)} lines were parsed.")
log_entries = parse_lines(log_file)
logger.info(f"\t\t{len(log_entries)} lines were parsed.")
logger.info("\tparsing completed.")
return log_entry_list
return log_entries
tests/test_parser.py
+44 -38
View File
@@ -3,48 +3,50 @@ from pathlib import Path
import pytest
from logwatcher.models import LogEntry
from logwatcher.parser import parse_log_file, parse_log_line
from logwatcher.parser import parse_file, parse_line
@pytest.fixture(name="empty_log_file")
def empty_log_file_fixture(invalid_log_dir: Path) -> Path:
"""
Path of an empty log file.
Ensures the parser does not raise an exception
and returns an empty list.
Path of an empty log file.
Ensures the parser does not raise an exception
and returns an empty list.
"""
return invalid_log_dir / "empty_file.txt"
@pytest.fixture(name="bad_log_file")
def bad_log_file_fixture(invalid_log_dir: Path) -> Path:
"""
Path of a file containing only incorrectly formatted lines.
Each line fails at a different point in LOG_PATTERN
(invalid IP, missing timestamp, missing DOSSIER EN COURS, etc.).
No LogEntry should be produced.
Path of a file containing only incorrectly formatted lines.
Each line fails at a different point in LOG_PATTERN
(invalid IP, missing timestamp, missing DOSSIER EN COURS, etc.).
No LogEntry should be produced.
"""
return invalid_log_dir / "bad_log_file.txt"
@pytest.fixture(name="specific_logs")
def specific_logs_fixture(valid_log_dir: Path) -> list[str]:
"""
Single log lines chosen to cover specific cases.
Each line is structurally valid and must be parsed
successfully. Used for unit tests of parse_log_line().
Single log lines chosen to cover specific cases.
Each line is structurally valid and must be parsed
successfully. Used for unit tests of parse_line().
"""
log_file = valid_log_dir / "specific_logs.txt"
with open(log_file, 'r') as file:
with open(log_file, "r") as file:
return file.readlines()
def test_parse_log_line(specific_logs: list[str]):
def test_parse_line(specific_logs: list[str]):
"""
Every structurally valid line produces a complete LogEntry.
Ensures no required field is empty or None after parsing
a line conforming to the LAME MDC format.
Every structurally valid line produces a complete LogEntry.
Ensures no required field is empty or None after parsing
a line conforming to the LAME MDC format.
"""
for log_line in specific_logs:
log_entry = parse_log_line(log_line)
log_entry = parse_line(log_line)
assert log_entry is not None
assert log_entry.server_ip is not None
assert log_entry.mdc_server_name is not None
@@ -52,49 +54,53 @@ def test_parse_log_line(specific_logs: list[str]):
assert log_entry.store_name is not None
assert log_entry.error_time is not None
assert log_entry.error_message is not None
assert log_entry.raw_line != ''
assert log_entry.raw_line != ""
def test_parse_empty_log_line():
"""
An empty or structurally invalid line returns None.
Covered cases:
- Empty string.
- Directory header (\"Répertoire scanné : \\\\...\").
An empty or structurally invalid line returns None.
Covered cases:
- Empty string.
- Directory header (\"Répertoire scanné : \\\\...\").
"""
log_entry = parse_log_line("")
log_entry = parse_line("")
assert not log_entry
log_entry = parse_log_line("Répertoire scanné : \\\\192.168.60.48\\e$\\MDC_9110\\Logs")
log_entry = parse_line("Répertoire scanné : \\\\192.168.60.48\\e$\\MDC_9110\\Logs")
assert not log_entry
def test_parse_valid_log_file(original_log_dir: Path):
"""
Each real log file produces at least one valid LogEntry.
Integration test: files provided by technicians contain
a mix of valid lines and lines to be ignored.
Ensures the parser extracts at least one entry per file.
Each real log file produces at least one valid LogEntry.
Integration test: files provided by technicians contain
a mix of valid lines and lines to be ignored.
Ensures the parser extracts at least one entry per file.
"""
for log_file in original_log_dir.glob("CR_*"):
log_list = parse_log_file(log_file)
log_list = parse_file(log_file)
assert len(log_list) > 0
for log_entry in log_list:
assert isinstance(log_entry, LogEntry)
def test_parse_empty_log_file(empty_log_file):
"""
An empty log file can be parsed and must returns an
empty LogEntry list.
An empty log file can be parsed and must returns an
empty LogEntry list.
"""
log_list = parse_log_file(empty_log_file)
log_list = parse_file(empty_log_file)
assert len(log_list) == 0
def test_parse_bad_log_file(bad_log_file):
"""
A file containing only invalid lines produces nothing.
Any line that does not match the LAME MDC structure
(IP, timestamps, DOSSIER EN COURS, error message)
must be silently ignored.
A file containing only invalid lines produces nothing.
Any line that does not match the LAME MDC structure
(IP, timestamps, DOSSIER EN COURS, error message)
must be silently ignored.
"""
log_list = parse_log_file(bad_log_file)
assert len(log_list) == 0
log_list = parse_file(bad_log_file)
assert len(log_list) == 0