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

♻️refactor(reporter): factorize reporter.py for future notifier

Browse Source 
let cli call get_period instead of functions in `reporter.py`, add notifier.py
This commit is contained in:
maurane
2026-09-14 18:36:20 +02:00
parent 91e76e6079
commit b1409dfea5
4 changed files with 146 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/logwatcher/cli.py
+37 -4
View File
@@ -1,4 +1,5 @@
import logging
from datetime import datetime
from pathlib import Path
from typing import Annotated
@@ -6,7 +7,7 @@ import typer
from logwatcher import __version__
from logwatcher.classifier import classify_log_entries
from logwatcher.config import SourceType
from logwatcher.config import DATETIME_FORMAT, FRENCH_TIMEZONE, SourceType
from logwatcher.logging_config import setup_logging
from logwatcher.mail_reader import (
connect_to_mailbox,
@@ -16,7 +17,7 @@ from logwatcher.mail_reader import (
)
from logwatcher.models import LogEntry
from logwatcher.parser import parse_file, parse_lines
from logwatcher.reporter import write_log_report
from logwatcher.reporter import build_mail_summary, get_period, write_log_report
app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.")
@@ -88,9 +89,30 @@ def from_mail(
nb_files = len(messages)
relevant, irrelevant = classify_log_entries(log_entries)
today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT)
start_date, end_date = get_period(relevant + irrelevant)
write_log_report(
SourceType.MAIL, relevant, irrelevant, nb_files, output_dir=output_dir
SourceType.MAIL,
relevant,
irrelevant,
nb_files,
today,
start_date,
end_date,
output_dir=output_dir,
)
mail_body = build_mail_summary(
nb_files=nb_files,
nb_errors_total=len(relevant + irrelevant),
nb_errors_n2=len(relevant),
today=today,
start_date=start_date,
end_date=end_date,
)
# send mail with n2.log as attachment
move_analyzed_mails(account, messages)
logger.info("============ logwatcher mail ended ============")
@@ -145,7 +167,18 @@ def from_files(
log_entries = _files_to_log_entries(log_files)
relevant, irrelevant = classify_log_entries(log_entries)
write_log_report(SourceType.FILE, relevant, irrelevant, len(log_files), output_dir)
today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT)
start_date, end_date = get_period(relevant + irrelevant)
write_log_report(
source_type=SourceType.FILE,
relevant=relevant,
irrelevant=irrelevant,
nb_files=len(log_files),
today=today,
start_date=start_date,
end_date=end_date,
output_dir=output_dir,
)
logger.info("============ logwatcher file ended ============")
src/logwatcher/notifier.py
+5
View File
@@ -0,0 +1,5 @@
from exchangelib import Account, FileAttachment, Message
def send_n2_report(summary: str):
pass
src/logwatcher/reporter.py
+59 -12
View File
@@ -3,7 +3,7 @@ from datetime import datetime
from pathlib import Path
from string import Template
from logwatcher.config import DATETIME_FORMAT, FRENCH_TIMEZONE, RESULT_PATH, SourceType
from logwatcher.config import DATETIME_FORMAT, RESULT_PATH, SourceType
from logwatcher.models import LogEntry
logger = logging.getLogger(__name__)
@@ -44,8 +44,19 @@ ERROR_TEMPLATE = Template("""
\tRaison: $error_message
""")
### mail template
MAIL_TEMPLATE = Template("""
Rapport analyse de log - $generation_date
def _get_period(entries: list[LogEntry]) -> tuple[str, str]:
Nb de fichiers analysés\t : $nb_files
Erreurs N2/Erreurs totales\t : $n2_errors/$total_errors
Période : $start_date -> $end_date
La liste des erreurs se trouvent en pièce jointe `n2.log`.
""")
def get_period(entries: list[LogEntry]) -> tuple[str, str]:
if not entries:
return "", ""
times: list[datetime] = [e.error_time for e in entries]
@@ -57,6 +68,9 @@ def write_log_report(
relevant: list[LogEntry],
irrelevant: list[LogEntry],
nb_files: int,
today: str,
start_date: str,
end_date: str,
output_dir: Path = RESULT_PATH,
) -> None:
"""
@@ -66,7 +80,10 @@ def write_log_report(
the logs were generated.
Args:
source_type:
end_date: # TODO
start_date: # TODO
today: # TODO
source_type: # TODO
relevant: List of n2 log entry
irrelevant: List of not n2 log entry
nb_files: Number of files scanned
@@ -75,11 +92,8 @@ def write_log_report(
None
"""
logger.info("\twriting reports job started.")
start_date, end_date = _get_period(relevant + irrelevant)
logger.info(f"\t\tperiod found: {start_date} -> {end_date}")
reports_dict = build_reports(
source_type, relevant, irrelevant, start_date, end_date, nb_files
source_type, relevant, irrelevant, today, start_date, end_date, nb_files
)
for target in reports_dict:
@@ -97,9 +111,10 @@ def build_reports(
source_type: SourceType,
relevant: list[LogEntry],
irrelevant: list[LogEntry],
start_date,
end_date,
nb_files,
today: str,
start_date: str,
end_date: str,
nb_files: int,
) -> dict[str, str]:
"""
Build the three output reports: n2, other, and all.
@@ -107,9 +122,10 @@ def build_reports(
"n2" and "other" are two unique sections.
Args:
source_type:
source_type: # TODO
relevant: List of n2 log entry
irrelevant: List of not n2 log entry
today: # TODO
start_date: Date of the oldest log entry in relevant + irrelevant list
end_date: Date of the newest log entry in relevant + irrelevant list
nb_files: number of files scanned
@@ -127,7 +143,6 @@ def build_reports(
logger.info("\t\t\tn2 and other reports were generated.")
logger.info("\t\tbuilding reports job completed.")
today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT)
source_value = source_type.value.capitalize()
return {
@@ -162,6 +177,38 @@ def build_reports(
# build templates
def build_mail_summary(
nb_files: int,
nb_errors_total: int,
nb_errors_n2: int,
today: str,
start_date: str,
end_date: str,
) -> str:
"""Build a short plain-text summary of a logwatcher run.
Args:
nb_files: Number of log files parsed.
nb_errors_total: Total number of errors across all reports.
nb_errors_n2: Number of N2-relevant errors.
today: # TODO
start_date: Earliest error timestamp.
end_date: Latest error timestamp.
Returns:
A short summary suitable for an email body.
"""
logger.info("generating mail body...")
return MAIL_TEMPLATE.substitute(
generation_date=today,
nb_files=nb_files,
n2_errors=nb_errors_n2,
total_errors=nb_errors_total,
start_date=start_date,
end_date=end_date,
)
def _render_target_report(
log_entries: list[LogEntry], target_template: Template
) -> str:
tests/test_reporter.py
+45 -20
View File
@@ -4,16 +4,16 @@ from pathlib import Path
import pytest
from logwatcher.classifier import classify_log_entries
from logwatcher.config import DATETIME_FORMAT, FIXTURE_PATH, SourceType
from logwatcher.config import DATETIME_FORMAT, FIXTURE_PATH, FRENCH_TIMEZONE, SourceType
from logwatcher.models import LogEntry
from logwatcher.parser import parse_file
from logwatcher.reporter import (
N2_SUPPORT_TEMPLATE,
OTHER_TEMPLATE,
_get_period,
_render_entries,
_render_target_report,
build_reports,
get_period,
write_log_report,
)
@@ -61,14 +61,14 @@ def get_only_irrelevant_log_entries_fixture(valid_log_dir: Path):
def test_get_period_empty_list():
"""An empty entry list yields an empty (start, end) period."""
assert _get_period([]) == ("", "")
assert get_period([]) == ("", "")
def test_get_period_format(
get_all_log_entries: tuple[list[LogEntry], list[LogEntry]],
):
"""Period bounds are formatted according to the configured datetime format."""
period = _get_period(get_all_log_entries[0] + get_all_log_entries[1])
period = get_period(get_all_log_entries[0] + get_all_log_entries[1])
for period_bound in period:
try:
assert datetime.strptime(period_bound, DATETIME_FORMAT) # noqa: DTZ007
@@ -158,11 +158,12 @@ def test_build_reports(
get_all_log_entries: tuple[list[LogEntry], list[LogEntry]], nb_files
):
"""The three reports are built with correct headers, counts, and section separation."""
start_date, end_date = _get_period(get_all_log_entries[0] + get_all_log_entries[1])
start_date, end_date = get_period(get_all_log_entries[0] + get_all_log_entries[1])
reports = build_reports(
SourceType.FILE,
get_all_log_entries[0],
get_all_log_entries[1],
datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT),
start_date,
end_date,
nb_files,
@@ -206,7 +207,18 @@ def _test_write_log_report(
"""Write the reports, then verify files exist and their content matches build_reports."""
results_path = output_dir / "results"
relevant, irrelevant = log_entries_tuple
write_log_report(SourceType.FILE, relevant, irrelevant, nb_files, results_path)
today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT)
start_date, end_date = get_period(relevant + irrelevant)
write_log_report(
SourceType.FILE,
relevant,
irrelevant,
nb_files,
today,
start_date,
end_date,
results_path,
)
# check if files were created
all_file, n2_file, other_file = (
(results_path / "all.log"),
@@ -218,9 +230,15 @@ def _test_write_log_report(
assert other_file.exists()
# check content
start_date, end_date = _get_period(relevant + irrelevant)
start_date, end_date = get_period(relevant + irrelevant)
reports = build_reports(
SourceType.FILE, relevant, irrelevant, start_date, end_date, nb_files
SourceType.FILE,
relevant,
irrelevant,
datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT),
start_date,
end_date,
nb_files,
)
nb_errors = {
"all": len(relevant + irrelevant),
@@ -277,12 +295,18 @@ def test_write_log_report_all_logs(
):
"""The full fixture writes three files whose content matches build_reports exactly."""
results_path = tmp_path / "results"
today = datetime.now(FRENCH_TIMEZONE).strftime(DATETIME_FORMAT)
start_date, end_date = get_period(get_all_log_entries[0] + get_all_log_entries[1])
write_log_report(
SourceType.FILE,
get_all_log_entries[0],
get_all_log_entries[1],
nb_files,
results_path,
source_type=SourceType.FILE,
relevant=get_all_log_entries[0],
irrelevant=get_all_log_entries[1],
nb_files=nb_files,
today=today,
start_date=start_date,
end_date=end_date,
output_dir=results_path,
)
# check if files were created
@@ -296,14 +320,15 @@ def test_write_log_report_all_logs(
assert other_file.exists()
# check content
start_date, end_date = _get_period(get_all_log_entries[0] + get_all_log_entries[1])
start_date, end_date = get_period(get_all_log_entries[0] + get_all_log_entries[1])
reports = build_reports(
SourceType.FILE,
get_all_log_entries[0],
get_all_log_entries[1],
start_date,
end_date,
nb_files,
source_type=SourceType.FILE,
relevant=get_all_log_entries[0],
irrelevant=get_all_log_entries[1],
today=today,
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
)
assert reports["all"] == all_file.read_text(encoding="windows-1252")
assert reports["n2"] == n2_file.read_text(encoding="windows-1252")