feat(reporter): ✨ add reporter main functions

we can now generate reports based on log files. Three reports are processed: n2 report, other report and general report combining the two previous reports. All reports would be available in the output directory.
This commit is contained in:
2026-08-26 18:00:55 +02:00
parent 23afe3f185
commit b83ee451a3
4 changed files with 153 additions and 6 deletions
+9 -6
View File
@@ -45,12 +45,13 @@ N2_PATTERNS = {
}
# compiled all patterns
_COMPILED_N2_PATTERNS = [re.compile(p) for p in N2_PATTERNS.values()]
_COMPILED_N2_PATTERNS = {error_name: re.compile(pattern) for error_name, pattern in N2_PATTERNS.items()}
def _match_n2_pattern(log_entry: LogEntry) -> bool:
return any(
pattern.search(log_entry.error_message) for pattern in _COMPILED_N2_PATTERNS
)
def _match_n2_pattern(log_entry: LogEntry) -> str|None:
for error_code, pattern in _COMPILED_N2_PATTERNS.items():
if pattern.search(log_entry.error_message):
return error_code
return None
def classify_log_entries(log_entries: list[LogEntry]) -> tuple[list[LogEntry], list[LogEntry]]:
"""
@@ -60,7 +61,9 @@ def classify_log_entries(log_entries: list[LogEntry]) -> tuple[list[LogEntry], l
irrelevant_log_entries = []
for log_entry in log_entries:
if _match_n2_pattern(log_entry):
error_name = _match_n2_pattern(log_entry)
if error_name:
log_entry.error_name = error_name
relevant_log_entries.append(log_entry)
else:
irrelevant_log_entries.append(log_entry)
+1
View File
@@ -16,6 +16,7 @@ class LogEntry:
error_time: datetime
error_message: str
raw_line: str # full initial log
error_name: str = ""
def __str__(self):
return self.error_message
+143
View File
@@ -0,0 +1,143 @@
import json
from datetime import datetime
from string import Template
from logwatcher.config import DATETIME_FORMAT, RESULT_PATH
from logwatcher.models import LogEntry
### report templates
BASE_TEMPLATE = Template("""
RAPPORT D'ANALYSE DE LOGS
=========================
Période\t : $start_date → $end_date
Fichiers lu\t : $nb_files
Nombre total d'erreurs\t: $nb_errors
$sub_content
""")
N2_SUPPORT_TEMPLATE = Template("""
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
ERREURS SUPPORT N2
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Nombre d'erreurs\t : $nb_errors
$error_list
""")
OTHER_TEMPLATE = Template("""
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
AUTRES ERREURS
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Nombre d'erreurs\t : $nb_errors
$error_list
""")
ERROR_TEMPLATE = Template("""
[$index] $error_name
\tMagasin: $store_name
\tHeure: $error_time
\tRaison: $error_message
""")
def _get_period(entries: list[LogEntry]) -> tuple[str, str]:
if not entries:
return "", ""
times: list[datetime] = [e.error_time for e in entries]
return min(times).strftime(DATETIME_FORMAT), max(times).strftime(DATETIME_FORMAT)
def write_log_report(relevant: list[LogEntry], irrelevant: list[LogEntry], nb_files: int):
"""
Write each relevant, irrelevant and general reports in their respective output
file.
Use the range date of relevant and irrelevant lists to get the period of time
the logs were generated.
"""
end_date, start_date = _get_period(relevant + irrelevant)
reports_dict = build_reports(relevant, irrelevant, start_date, end_date, nb_files)
for target, content in reports_dict.items():
target_file = RESULT_PATH / str(target + ".log")
target_file.write_text(content, encoding="windows-1252")
def generate_json_report(relevant: list[LogEntry]) -> str: # TODO: put it in a serializer module
"""
Serialize from a Python LogEntry object to a JSON object.
Used in the Ivanti API calls
"""
return json.dumps(relevant)
def build_reports(relevant: list[LogEntry], irrelevant: list[LogEntry], start_date, end_date, nb_files) -> dict[str, str]:
"""
Build the three output reports: n2, other, and all.
"all" report contains both "n2" and "other" sections.
"n2" and "other" are two unique sections.
"""
relevant_report = _render_target_report(
log_entries=relevant,
target_template=N2_SUPPORT_TEMPLATE
)
irrelevant_report = _render_target_report(
log_entries=irrelevant,
target_template=OTHER_TEMPLATE
)
return {
"n2": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(relevant),
sub_content=relevant_report,
),
"other": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(irrelevant),
sub_content=irrelevant_report,
),
"all": BASE_TEMPLATE.substitute(
start_date=start_date,
end_date=end_date,
nb_files=nb_files,
nb_errors=len(relevant + irrelevant),
sub_content=relevant_report + "\n" + irrelevant_report,
)
}
def _render_entries(log_entries: list[LogEntry]) -> str:
"""
Render a list of log entries using the `ERROR_TEMPLATE`.
Used in generated report files in the `output` directory
"""
error_list = []
for index, log_entry in enumerate(log_entries, start=1):
error_list.append(
ERROR_TEMPLATE.substitute(
index=index,
error_name=log_entry.error_name,
store_name=log_entry.store_name,
error_time=log_entry.get_error_time(),
error_message=log_entry.error_message
)
)
return "".join(error_list)
# build templates
def _render_target_report(log_entries: list[LogEntry], target_template) -> str:
"""
Render report using either N2_SUPPORT_TEMPLATE or OTHER_TEMPLATE.
Call _render_entries() and include rendered log entries in the
new report.
"""
error_list = _render_entries(log_entries)
return target_template.substitute(
nb_errors=len(log_entries),
error_list=error_list
)
View File