From b1409dfea5493cb96ef27f5feb76a0d196fe2533 Mon Sep 17 00:00:00 2001 From: maurane Date: Mon, 14 Sep 2026 18:36:20 +0200 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8Frefactor(reporter):=20factori?= =?UTF-8?q?ze=20reporter.py=20for=20future=20notifier?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit let cli call get_period instead of functions in `reporter.py`, add notifier.py --- src/logwatcher/cli.py | 41 +++++++++++++++++++--- src/logwatcher/notifier.py | 5 +++ src/logwatcher/reporter.py | 71 +++++++++++++++++++++++++++++++------- tests/test_reporter.py | 65 +++++++++++++++++++++++----------- 4 files changed, 146 insertions(+), 36 deletions(-) create mode 100644 src/logwatcher/notifier.py diff --git a/src/logwatcher/cli.py b/src/logwatcher/cli.py index 3d83ac9..16679a3 100644 --- a/src/logwatcher/cli.py +++ b/src/logwatcher/cli.py @@ -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 ============") diff --git a/src/logwatcher/notifier.py b/src/logwatcher/notifier.py new file mode 100644 index 0000000..602664d --- /dev/null +++ b/src/logwatcher/notifier.py @@ -0,0 +1,5 @@ +from exchangelib import Account, FileAttachment, Message + + +def send_n2_report(summary: str): + pass diff --git a/src/logwatcher/reporter.py b/src/logwatcher/reporter.py index 3763be2..04742a6 100644 --- a/src/logwatcher/reporter.py +++ b/src/logwatcher/reporter.py @@ -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: diff --git a/tests/test_reporter.py b/tests/test_reporter.py index e2764ba..be17eee 100644 --- a/tests/test_reporter.py +++ b/tests/test_reporter.py @@ -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")