diff --git a/src/logwatcher/__init__.py b/src/logwatcher/__init__.py index beccdb8..7055442 100644 --- a/src/logwatcher/__init__.py +++ b/src/logwatcher/__init__.py @@ -3,8 +3,6 @@ from importlib.metadata import version from dotenv import load_dotenv from logwatcher.config import * -from logwatcher.logging_config import setup_logging __version__ = version("logwatcher") -logger = setup_logging() load_dotenv() diff --git a/src/logwatcher/classifier.py b/src/logwatcher/classifier.py index 6bd33be..72731d9 100644 --- a/src/logwatcher/classifier.py +++ b/src/logwatcher/classifier.py @@ -1,8 +1,10 @@ +import logging import re -from logwatcher import logger from logwatcher.models import LogEntry +logger = logging.getLogger(__name__) + # ============================================================ # Atomic patterns — reusable building blocks (no named groups) # ============================================================ diff --git a/src/logwatcher/cli.py b/src/logwatcher/cli.py index 8898c1b..2565e20 100644 --- a/src/logwatcher/cli.py +++ b/src/logwatcher/cli.py @@ -1,16 +1,26 @@ +import logging from pathlib import Path from typing import Annotated import typer -from logwatcher import __version__, logger +from logwatcher import __version__ from logwatcher.classifier import classify_log_entries -from logwatcher.mail_reader import extract_logs_from_mails +from logwatcher.config import SourceType +from logwatcher.logging_config import setup_logging +from logwatcher.mail_reader import ( + connect_to_mailbox, + extract_logs_from_mails, + fetch_log_messages, +) from logwatcher.models import LogEntry -from logwatcher.parser import parse_log_file +from logwatcher.parser import parse_log_file, parse_log_line from logwatcher.reporter import write_log_report -app = typer.Typer() +app = typer.Typer(help="Logwatcher - Analysis and sorting of LAME MDC logs.") + + +logger = logging.getLogger("logwatcher") def version_callback(value: bool): @@ -59,15 +69,44 @@ def _files_to_log_entries(input_files: list[Path]) -> list[LogEntry]: return log_entries -@app.command() -def from_mail(): - logger.info("============ logwatcher started ============") - logs = extract_logs_from_mails() - logger.info("============ logwatcher ended ============") +def _mail_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 analyze_logs(): + pass @app.command() -def cli( +def from_mail( + output_dir: Annotated[ + Path, + typer.Option( + "--output-dir", help="Output directories where reports will be stored." + ), + ] = Path("output"), +): + logger.info("============ logwatcher mail started ============") + account = connect_to_mailbox() + messages = fetch_log_messages(account) + logs = extract_logs_from_mails(messages) + log_entries = _mail_to_log_entries(logs) + + nb_files = messages.count() + relevant, irrelevant = classify_log_entries(log_entries) + write_log_report( + SourceType.MAIL, relevant, irrelevant, nb_files, output_dir=output_dir + ) + logger.info("============ logwatcher mail ended ============") + + +@app.command() +def from_files( output_dir: Annotated[ Path, typer.Option( @@ -81,19 +120,8 @@ def cli( Path | None, typer.Option("--input-dir", help="Directories containing log files."), ] = None, - version: Annotated[ - bool, - typer.Option( - "--version", - help="Get the version of the package", - callback=version_callback, - is_eager=True, - ), - ] = False, -) -> None: +): """ - Entry point for logwatcher. - Convert log files into two reports, one of n2 related logs, one without any n2 related logs. These reports are sent by mail to n2 support technicians. These log entries are used to generate incidents on Ivanti using its API. @@ -107,7 +135,7 @@ def cli( Returns: None """ - logger.info("============ logwatcher started ============") + logger.info("============ logwatcher file started ============") if input_files and input_dir: raise typer.BadParameter( "Only one of input_files or input_dir must be provided." @@ -127,8 +155,40 @@ def cli( log_entries = _files_to_log_entries(log_files) relevant, irrelevant = classify_log_entries(log_entries) - write_log_report(relevant, irrelevant, len(log_files), output_dir) - logger.info("============ logwatcher ended ============") + write_log_report(SourceType.FILE, relevant, irrelevant, len(log_files), output_dir) + logger.info("============ logwatcher file ended ============") + + +@app.callback() +def cli( + version: Annotated[ + bool, + typer.Option( + "--version", + help="Get the version of the package", + callback=version_callback, + is_eager=True, + ), + ] = False, + verbose: Annotated[ + bool, + typer.Option( + "--verbose", + "-v", + help="Get more information in standard output while the script is running", + ), + ] = False, +) -> None: + """ + Entrypoint for logwatcher. + + Args: + verbose: if provided, script returns info and debug logs too. Otherwise, it returns only warnings and errors. + version: if provided, script returns the version of package + + Returns: + """ + setup_logging(verbose=verbose) if __name__ == "__main__": diff --git a/src/logwatcher/config.py b/src/logwatcher/config.py index 1370b10..c2aa03c 100644 --- a/src/logwatcher/config.py +++ b/src/logwatcher/config.py @@ -1,4 +1,5 @@ from datetime import timedelta, timezone +from enum import Enum from pathlib import Path ### GENERAL DATA @@ -13,4 +14,10 @@ FIXTURE_PATH = TEST_PATH / "fixtures" # date and time DATETIME_FORMAT = "%d/%m/%Y %H:%M:%S" -FRENCH_TIMEZONE = timezone(offset=timedelta(hours=2)) # UTC+2 = CEST \ No newline at end of file +FRENCH_TIMEZONE = timezone(offset=timedelta(hours=2)) # UTC+2 = CEST + + +# file type +class SourceType(Enum): + FILE = "fichier" + MAIL = "mail" diff --git a/src/logwatcher/logging_config.py b/src/logwatcher/logging_config.py index 62876d3..a5e4662 100644 --- a/src/logwatcher/logging_config.py +++ b/src/logwatcher/logging_config.py @@ -45,7 +45,7 @@ def _convert_string_to_log_level(log_level: str) -> int: def setup_logging( formatter: logging.Formatter | None = None, file_path: Path = Path("output/logs/logwatcher.log"), - verbose: str = "warning", + verbose: bool = False, ) -> logging.Logger: """ Create a new instance of Logger named 'logger' customized for the logwatcher package. @@ -54,7 +54,10 @@ def setup_logging( """ _formatter = formatter if formatter else _setup_formatter() - standard_output = _setup_handler(_formatter, logging.WARNING) + if verbose: + standard_output = _setup_handler(_formatter, logging.DEBUG) + else: + standard_output = _setup_handler(_formatter, logging.WARNING) file_output = _setup_handler(_formatter, logging.DEBUG, file_path) root = logging.getLogger() diff --git a/src/logwatcher/mail_reader.py b/src/logwatcher/mail_reader.py index d951545..42b68bf 100644 --- a/src/logwatcher/mail_reader.py +++ b/src/logwatcher/mail_reader.py @@ -1,3 +1,4 @@ +import logging import os from exchangelib import ( @@ -14,7 +15,8 @@ from exchangelib import ( from exchangelib.errors import UnauthorizedError from exchangelib.queryset import QuerySet -from logwatcher import logger +logger = logging.getLogger(__name__) + MAILBOX_DIR = "Logs" LOG_IN_ATTACHMENT_PATTERN = "Le compte-rendu contient plus de 100 lignes." @@ -155,7 +157,7 @@ def extract_log_lines(content: str) -> list[str]: return [line.strip("\r\n") for line in content.splitlines() if line.strip()] -def extract_logs_from_mails() -> list[str]: +def extract_logs_from_mails(messages: QuerySet) -> list[str]: """ Connect to the mailbox, retrieves emails that are N2 log-related, clean them and return a list of logs in string format. diff --git a/src/logwatcher/parser.py b/src/logwatcher/parser.py index a3450a8..97a0bc5 100644 --- a/src/logwatcher/parser.py +++ b/src/logwatcher/parser.py @@ -1,11 +1,13 @@ +import logging import re from datetime import datetime from pathlib import Path -from logwatcher import DATETIME_FORMAT, FRENCH_TIMEZONE, logger +from logwatcher import DATETIME_FORMAT, FRENCH_TIMEZONE from logwatcher.models import LogEntry -# DEFINITIONS OF LOG PATTERNS +logger = logging.getLogger(__name__) + SERVER_IP_PATTERN = r"([1-9]\d{1,2}\.)(0{1}|\d{1,3}\.){2}(0{1}|\d{1,3})" MDC_SERVER_NAME_PATTERN = r"MDC_\d+" DATE_TIME_PATTERN = r"((0[1-9])|([1-2][0-9])|(3[0-1]))/((0[1-9])|(1[0-2]))/\d{4} (([0-1][0-9])|(2[0-3])):[0-5][0-9]:[0-5][0-9]" diff --git a/src/logwatcher/reporter.py b/src/logwatcher/reporter.py index 0bc71a0..e998455 100644 --- a/src/logwatcher/reporter.py +++ b/src/logwatcher/reporter.py @@ -1,17 +1,19 @@ +import logging from datetime import datetime from pathlib import Path from string import Template -from logwatcher import logger -from logwatcher.config import DATETIME_FORMAT, RESULT_PATH +from logwatcher.config import DATETIME_FORMAT, RESULT_PATH, SourceType from logwatcher.models import LogEntry +logger = logging.getLogger(__name__) + ### report templates BASE_TEMPLATE = Template(""" RAPPORT D'ANALYSE DE LOGS ========================= Période\t : $start_date -> $end_date -Fichier(s) lu(s)\t : $nb_files +$source_type(s) lu(s)\t : $nb_files Nombre total d'erreur(s)\t: $nb_errors $sub_content """) @@ -50,6 +52,7 @@ def _get_period(entries: list[LogEntry]) -> tuple[str, str]: def write_log_report( + source_type: SourceType, relevant: list[LogEntry], irrelevant: list[LogEntry], nb_files: int, @@ -62,6 +65,7 @@ def write_log_report( the logs were generated. Args: + source_type: relevant: List of n2 log entry irrelevant: List of not n2 log entry nb_files: Number of files scanned @@ -73,7 +77,9 @@ def write_log_report( start_date, end_date = _get_period(relevant + irrelevant) logger.info(f"\t\tperiod found: {start_date} -> {end_date}") - reports_dict = build_reports(relevant, irrelevant, start_date, end_date, nb_files) + reports_dict = build_reports( + source_type, relevant, irrelevant, start_date, end_date, nb_files + ) for target in reports_dict: logger.info(f"\t\tgenerating report '{target}'...") @@ -87,7 +93,12 @@ def write_log_report( def build_reports( - relevant: list[LogEntry], irrelevant: list[LogEntry], start_date, end_date, nb_files + source_type: SourceType, + relevant: list[LogEntry], + irrelevant: list[LogEntry], + start_date, + end_date, + nb_files, ) -> dict[str, str]: """ Build the three output reports: n2, other, and all. @@ -95,6 +106,7 @@ def build_reports( "n2" and "other" are two unique sections. Args: + source_type: relevant: List of n2 log entry irrelevant: List of not n2 log entry start_date: Date of the oldest log entry in relevant + irrelevant list @@ -114,8 +126,11 @@ def build_reports( logger.info("\t\t\tn2 and other reports were generated.") logger.info("\t\tbuilding reports job completed.") + source_value = source_type.value.capitalize() + return { "n2": BASE_TEMPLATE.substitute( + source_type=source_value, start_date=start_date, end_date=end_date, nb_files=nb_files, @@ -123,6 +138,7 @@ def build_reports( sub_content=relevant_report, ), "other": BASE_TEMPLATE.substitute( + source_type=source_value, start_date=start_date, end_date=end_date, nb_files=nb_files, @@ -130,6 +146,7 @@ def build_reports( sub_content=irrelevant_report, ), "all": BASE_TEMPLATE.substitute( + source_type=source_value, start_date=start_date, end_date=end_date, nb_files=nb_files, diff --git a/tests/test_cli.py b/tests/test_cli.py index c3724e9..9c02e27 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -40,7 +40,7 @@ def get_argument_from_files(log_dir: Path, output_path: Path) -> list[str]: str(file) for file in log_dir.iterdir() if file.suffix in ("", ".txt", ".log") ] - arguments = ["--output-dir", str(output_path)] + arguments = ["from-files", "--output-dir", str(output_path)] for file in files: arguments += ["--input-files", file] return arguments @@ -55,11 +55,10 @@ def test_cli_help(): """ results = runner.invoke(app, ["--help"]) assert results.exit_code == 0 - assert "Usage: cli [OPTIONS]" in results.output - assert "--input-dir" in results.output - assert "--input-files" in results.output - assert "--output-dir" in results.output + assert "--verbose" in results.output assert "--version" in results.output + assert "from-mail" in results.output + assert "from-files" in results.output def test_version_cli(): @@ -162,6 +161,7 @@ def test_cli_input_files_and_input_dir(output_path: Path): results = runner.invoke( app, [ + "from-files", "--input-files", "a.log", "--input-dir", @@ -183,7 +183,7 @@ def test_cli_neither_input_files_nor_input_dir(): Returns: None """ - results = runner.invoke(app) + results = runner.invoke(app, ["from-files"]) assert results.exit_code == 2 assert "Either input_files or input_dir must be provided" in results.output @@ -205,7 +205,7 @@ def test_cli_bad_input_file(tmp_path: Path, output_path: Path, invalid_log_dir: directory = str(tmp_path) results = runner.invoke( app, - ["--input-files", directory, "--output-dir", str(output_path)], + ["from-files", "--input-files", directory, "--output-dir", str(output_path)], ) assert results.exit_code == 1 assert isinstance(results.exception, ValueError) @@ -231,7 +231,7 @@ def test_cli_empty_input_files(output_path: Path, invalid_log_dir: Path): empty_file = str(next(invalid_log_dir.glob("empty_file.txt"))) results = runner.invoke( app, - ["--input-files", empty_file, "--output-dir", str(output_path)], + ["from-files", "--input-files", empty_file, "--output-dir", str(output_path)], ) assert results.exit_code == 0 # check files @@ -288,6 +288,7 @@ def test_cli_input_files_or_input_dir( result = runner.invoke( app, [ + "from-files", "--input-dir", str(original_log_dir), "--output-dir", @@ -310,7 +311,14 @@ def test_cli_input_files_or_input_dir( def test_cli_input_files_one_file(valid_log_dir: Path, output_path: Path): mixed_logs_file = valid_log_dir / "mixed_logs.txt" result = runner.invoke( - app, ["--input-files", str(mixed_logs_file), "--output-dir", str(output_path)] + app, + [ + "from-files", + "--input-files", + str(mixed_logs_file), + "--output-dir", + str(output_path), + ], ) assert result.exit_code == 0 @@ -345,6 +353,7 @@ def test_cli_empty_input_directory(tmp_path: Path, output_path: Path): result = runner.invoke( app, [ + "from-files", "--input-dir", str(input_dir), "--output-dir", diff --git a/tests/test_logging.py b/tests/test_logging.py index 45cad44..e756d55 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -6,14 +6,15 @@ logwatcher is launched. - the StreamHandler must watch at an error level - the FileHandler must watch at a debug level -- +- """ + +import logging import re from logging import DEBUG, ERROR, INFO, WARNING, FileHandler, StreamHandler, getLogger import pytest -from logwatcher import logger from logwatcher.logging_config import _setup_formatter, _setup_handler, setup_logging @@ -22,9 +23,10 @@ def test_setup_handler(tmp_log_file): handler = _setup_handler(formatter=formatter, level=DEBUG) assert isinstance(handler, StreamHandler) - handler = _setup_handler(formatter=formatter, level=INFO,filename=tmp_log_file) + handler = _setup_handler(formatter=formatter, level=INFO, filename=tmp_log_file) assert isinstance(handler, FileHandler) + def test_count_of_handler(tmp_log_file): logger = getLogger() count_file_handler = 0 @@ -37,11 +39,13 @@ def test_count_of_handler(tmp_log_file): count_stream_handler += 1 else: pass - + assert count_file_handler >= 1 assert count_stream_handler >= 1 + def test_type_of_handlers(): + logger = logging.getLogger("logwatcher") for handler in logger.handlers: if isinstance(handler, FileHandler): assert handler.level == DEBUG @@ -49,53 +53,56 @@ def test_type_of_handlers(): assert handler.level == ERROR else: pytest.fail("handlers must be either a file or a stream handler") - assert handler.formatter._fmt == '[%(asctime)s] - %(levelname)s: %(message)s' - assert handler.formatter.datefmt == '%d/%m/%Y %H:%M:%S' + assert handler.formatter._fmt == "[%(asctime)s] - %(levelname)s: %(message)s" + assert handler.formatter.datefmt == "%d/%m/%Y %H:%M:%S" + def test_modified_format(): """ - when starting, the logger is first intialized, thus two handlers are at position - 1 and 2 in logger.handlers list. + when starting, the logger is first intialized, thus two handlers are at position + 1 and 2 in logger.handlers list. """ logger = setup_logging() for handler in logger.handlers[2:]: - assert handler.formatter._fmt == '[%(asctime)s] - %(levelname)s: %(message)s' + assert handler.formatter._fmt == "[%(asctime)s] - %(levelname)s: %(message)s" - formatter_different_format = _setup_formatter(format='%(asctime)s - %(levelname)s: %(message)s') + formatter_different_format = _setup_formatter( + format="%(asctime)s - %(levelname)s: %(message)s" + ) logger = setup_logging(formatter=formatter_different_format) for handler in logger.handlers[4:]: - assert handler.formatter._fmt == '%(asctime)s - %(levelname)s: %(message)s' + assert handler.formatter._fmt == "%(asctime)s - %(levelname)s: %(message)s" - formatter_different_datefmt = _setup_formatter(datefmt='%a %b %Y at %Hh%M') + formatter_different_datefmt = _setup_formatter(datefmt="%a %b %Y at %Hh%M") logger = setup_logging(formatter=formatter_different_datefmt) for handler in logger.handlers[6:]: - assert handler.formatter._fmt == '[%(asctime)s] - %(levelname)s: %(message)s' - assert handler.formatter.datefmt == '%a %b %Y at %Hh%M' + assert handler.formatter._fmt == "[%(asctime)s] - %(levelname)s: %(message)s" + assert handler.formatter.datefmt == "%a %b %Y at %Hh%M" def test_logging_in_correct_path(tmp_log_file): + logger = logging.getLogger("logwatcher") # must be both in file and stdout logger.critical("hi- BYE") logger.error("hi... ok, bye") logger.warning("hi, are you alright?") # only in file - logger.info("hi there") - logger.debug("hi, how are you? it's been a while!") + logger.info("hi there") + logger.debug("hi, how are you? it's been a while!") with open(tmp_log_file) as log_file: content = log_file.readlines() - regex = r'\[\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2}\] - ' + regex = r"\[\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2}\] - " assert len(content) == 5 for line in content: assert re.search(regex, line) # test critical to warning - assert re.search(r'CRITICAL: hi- BYE', content[0]) - assert re.search(r'ERROR: hi... ok, bye', content[1]) - assert re.search(r'WARNING: hi, are you alright?', content[2]) - assert re.search(r'INFO: hi there', content[3]) + assert re.search(r"CRITICAL: hi- BYE", content[0]) + assert re.search(r"ERROR: hi... ok, bye", content[1]) + assert re.search(r"WARNING: hi, are you alright?", content[2]) + assert re.search(r"INFO: hi there", content[3]) assert logger.getEffectiveLevel() < WARNING assert re.search(r"DEBUG: hi, how are you\? it's been a while!", content[4]) - \ No newline at end of file diff --git a/tests/test_reporter.py b/tests/test_reporter.py index 5e97aad..10296eb 100644 --- a/tests/test_reporter.py +++ b/tests/test_reporter.py @@ -4,7 +4,7 @@ from pathlib import Path import pytest from logwatcher.classifier import classify_log_entries -from logwatcher.config import DATETIME_FORMAT, FIXTURE_PATH +from logwatcher.config import DATETIME_FORMAT, FIXTURE_PATH, SourceType from logwatcher.models import LogEntry from logwatcher.parser import parse_log_file from logwatcher.reporter import ( @@ -160,7 +160,12 @@ def test_build_reports( """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]) reports = build_reports( - get_all_log_entries[0], get_all_log_entries[1], start_date, end_date, nb_files + SourceType.FILE, + get_all_log_entries[0], + get_all_log_entries[1], + start_date, + end_date, + nb_files, ) assert "all" in reports and "n2" in reports and "other" in reports @@ -201,7 +206,7 @@ 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(relevant, irrelevant, nb_files, results_path) + write_log_report(SourceType.FILE, relevant, irrelevant, nb_files, results_path) # check if files were created all_file, n2_file, other_file = ( (results_path / "all.log"), @@ -214,7 +219,9 @@ def _test_write_log_report( # check content start_date, end_date = _get_period(relevant + irrelevant) - reports = build_reports(relevant, irrelevant, start_date, end_date, nb_files) + reports = build_reports( + SourceType.FILE, relevant, irrelevant, start_date, end_date, nb_files + ) nb_errors = { "all": len(relevant + irrelevant), "n2": len(relevant), @@ -271,7 +278,11 @@ def test_write_log_report_all_logs( """The full fixture writes three files whose content matches build_reports exactly.""" results_path = tmp_path / "results" write_log_report( - get_all_log_entries[0], get_all_log_entries[1], nb_files, results_path + SourceType.FILE, + get_all_log_entries[0], + get_all_log_entries[1], + nb_files, + results_path, ) # check if files were created @@ -287,7 +298,12 @@ def test_write_log_report_all_logs( # check content start_date, end_date = _get_period(get_all_log_entries[0] + get_all_log_entries[1]) reports = build_reports( - get_all_log_entries[0], get_all_log_entries[1], start_date, end_date, nb_files + SourceType.FILE, + get_all_log_entries[0], + get_all_log_entries[1], + start_date, + end_date, + nb_files, ) assert reports["all"] == all_file.read_text(encoding="windows-1252") assert reports["n2"] == n2_file.read_text(encoding="windows-1252")