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

✨feat(project): add from-mails command

Browse Source 
add `from-mails` command that connect to the mailbox using configurations in .env file, parse, classify and report all not analyzed files.
This commit is contained in:
maurane
2026-09-08 18:32:49 +02:00
parent 348ed05d91
commit 150eb5145d
11 changed files with 200 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/logwatcher/__init__.py
-2
View File
@@ -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()
src/logwatcher/classifier.py
+3 -1
View File
@@ -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)
# ============================================================
src/logwatcher/cli.py
+85 -25
View File
@@ -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__":
src/logwatcher/config.py
+7
View File
@@ -1,4 +1,5 @@
from datetime import timedelta, timezone
from enum import Enum
from pathlib import Path
### GENERAL DATA
@@ -14,3 +15,9 @@ 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
# file type
class SourceType(Enum):
FILE = "fichier"
MAIL = "mail"
src/logwatcher/logging_config.py
+4 -1
View File
@@ -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,6 +54,9 @@ def setup_logging(
"""
_formatter = formatter if formatter else _setup_formatter()
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)
src/logwatcher/mail_reader.py
+4 -2
View File
@@ -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.
src/logwatcher/parser.py
+4 -2
View File
@@ -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]"
src/logwatcher/reporter.py
+22 -5
View File
@@ -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,
tests/test_cli.py
+18 -9
View File
@@ -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",
tests/test_logging.py
+22 -15
View File
@@ -8,12 +8,13 @@ logwatcher is launched.
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
@@ -25,6 +26,7 @@ def test_setup_handler(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
@@ -41,7 +43,9 @@ def test_count_of_handler(tmp_log_file):
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,8 +53,9 @@ 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():
"""
@@ -59,21 +64,24 @@ def test_modified_format():
"""
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")
@@ -85,17 +93,16 @@ def test_logging_in_correct_path(tmp_log_file):
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])
tests/test_reporter.py
+22 -6
View File
@@ -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")