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

✨feat(project): add logging in project

Browse Source 
add logging in modules classier.py, parser.py, reporter.py. Update logging in cli.
This commit is contained in:
maurane glaude
2026-09-01 17:21:17 +02:00
parent 24a77363d0
commit a43afb9df7
4 changed files with 107 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/logwatcher/classifier.py
+18 -4
View File
@@ -1,5 +1,6 @@
import re
from logwatcher import logger
from logwatcher.models import LogEntry
# ============================================================
@@ -35,17 +36,19 @@ N2_PATTERNS = {
"PARTIAL_INTEGRATION": r"Erreur : (?P<bl_number>(?P<sap_code>\d{4})_(?P<order_number>\w+)_DESADV) : ISC ERROR CODE:(?P<isc_error_code>\d+) \(attention intégration partielle du BL ou d'une commande\)",
"SAME_LINE_EXCEPT_DISCOUNT": r"Erreur : (?P<group_number>GRP_\d+) / (?P<store_code>\d{5}/\d{3}) (?P<sap_code>\d{4})_(?P<order_number>\w+) ERREUR (?P<model_code>(?P<nosymag_model_code>\w+ (?P<color_code>\w+))(?P<brand_code>\w{3})) / SetCOMBCDE -> OrderNumber : \d+ / Une commande ne peut pas contenir plusieurs lignes avec le même modèle, la même collection, la même date de livraison et avoir des remises différentes",
"MULTIPLE_ROWS_IN_SINGLETON_SELECT": r"Erreur : (?P<group_number>GRP_\d+) / (?P<store_code>\d{5}/\d{3}) (?P<sap_code>\d{4})_(?P<order_number>\w+) ERREUR (?P<model_code>(?P<nosymag_model_code>\w+ (?P<color_code>\w+))(?P<brand_code>\w{3})) / SizeRange (?P<size_range>\d+\.\d+) / SetMrkFourn -> (?: \[[^\]]+\])+multiple rows in singleton select",
# low priority errors
"SIZE_NOT_FOUND": r"Erreur : (?P<group_number>GRP_\d+) / (?P<store_code>\d{5}/\d{3}) (?P<sap_code>\d{4})_(?P<order_number>\w+) WARNING (\d+ \w+) / SizeRange (?P<size_range>\d+\.\d+) - Columnx \d+ / Taille Inexistante",
"UNSYNCED_ORD_SD_FILES": r"Erreur : (?P<group_number>GRP_\d+) / (?P<store_code>\d{5}/\d{3}) (?P<sap_code>\d{4})_(?P<order_number>\w+) WARNING (?P<model_code>(?P<nosymag_model_code>\w+ (?P<color_code>\w+)(?P<brand_code>\w{3}))) / OrderNumber : \d+ / Le modèle n'existe pas dans le fichier SD correspondant",
"CODE_MODEL_SIZE_OUT_OF_BOUND": r"Erreur : (?P<group_number>GRP_\d+) / (?P<store_code>\d{5}/\d{3}) (?P<sap_code>\d{4})_(?P<order_number>\w+) WARNING [\w ]+ \w+ / SetArticle -> SizeRange \d+\.\d+ / (?: \[[^\]]+\])+-\w+\. Données trop volumineuses pour la variable (?: \[[^\]]+\]). Long. max. = \[\d+\], long. réelle = \[\d+\] Conseil : attribuez une valeur supérieure au paramètre TFDParam.Size",
"USER_NOT_CREATED_ON_NOSYMAG": r"Erreur : (?P<group_number>GRP_\d+) / (?P<store_code>\d{5}/\d{3}) (?P<sap_code>\d{4})_(?P<order_number>\w+) ERREUR / OrderNumber : \d+ / Utilisateur : configuration incorrecte \(valeur 0\) - vérifier dans IF_Config",
"INCORRECT_MDC_XML_FILES": r"Exception : Impossible de convertir le variant de type \(Null\) en type \(OleStr\)"
"INCORRECT_MDC_XML_FILES": r"Exception : Impossible de convertir le variant de type \(Null\) en type \(OleStr\)",
}
# compiled all patterns
_COMPILED_N2_PATTERNS = {error_name: re.compile(pattern) for error_name, pattern in N2_PATTERNS.items()}
_COMPILED_N2_PATTERNS = {
error_name: re.compile(pattern) for error_name, pattern in N2_PATTERNS.items()
}
def _match_n2_pattern(log_entry: LogEntry) -> str | None:
for error_code, pattern in _COMPILED_N2_PATTERNS.items():
@@ -53,19 +56,30 @@ def _match_n2_pattern(log_entry: LogEntry) -> str|None:
return error_code
return None
def classify_log_entries(log_entries: list[LogEntry]) -> tuple[list[LogEntry], list[LogEntry]]:
def classify_log_entries(
log_entries: list[LogEntry],
) -> tuple[list[LogEntry], list[LogEntry]]:
"""
Separate logs that require N2 intervention from those that don't.
"""
relevant_log_entries = []
irrelevant_log_entries = []
logger.info("\tclassification job started.")
for log_entry in log_entries:
error_name = _match_n2_pattern(log_entry)
if error_name:
logger.info(
f"\t\tError '{error_name}' associated with log entry '{log_entry.error_message}'. This is a n2 log entry"
)
log_entry.error_name = error_name
relevant_log_entries.append(log_entry)
else:
irrelevant_log_entries.append(log_entry)
logger.info(
f"\t\tn2 logs: {len(relevant_log_entries)}, other logs: {len(irrelevant_log_entries)}"
)
logger.info("\tclassification job completed.")
return relevant_log_entries, irrelevant_log_entries
src/logwatcher/cli.py
+6 -4
View File
@@ -97,25 +97,27 @@ def cli(
Returns:
None
"""
logger.info("logwatcher starts its work.")
logger.info("============ logwatcher started ============")
if input_files and input_dir:
raise typer.BadParameter(
"Only one of input_files or input_dir must be provided."
)
if input_files:
logger.info(f"input_files option selected.\n input_files: '{input_files}'")
logger.info(f"input_files option selected. input_files: {input_files}")
log_files = input_files
elif input_dir:
logger.info(f"input_dir option selected with value '{input_dir}'")
logger.info(f"input_dir option selected with value {input_dir}")
log_files = _get_files_from_dir(input_dir)
else:
raise typer.BadParameter("Either input_files or input_dir must be provided")
logger.info(f"reports will be located at '{output_dir.cwd()}'")
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 has finished its task.")
logger.info("============ logwatcher ended ============")
if __name__ == "__main__":
src/logwatcher/parser.py
+15 -10
View File
@@ -45,20 +45,24 @@ def parse_log_line(log_line: str) -> LogEntry | None:
"""
match = LOG_PATTERN.search(log_line)
if not match:
logger.debug(f"Unrecognize format. Ignoring line {log_line}.")
return None
data = match.groupdict()
return LogEntry(
server_ip=data["server_ip"],
mdc_server_name=data["mdc_server_name"],
start_time=datetime.strptime(data["start_time"], DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
start_time=datetime.strptime(data["start_time"], DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
store_name=data["store_name"],
error_time=datetime.strptime(data["error_time"], DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
error_time=datetime.strptime(data["error_time"], DATETIME_FORMAT).astimezone(
FRENCH_TIMEZONE
),
error_message=data["error_message"],
raw_line=log_line
raw_line=log_line,
)
def parse_log_file(log_file_path: Path) -> list[LogEntry]:
"""Transforms the content of a log file into a list of LogEntry.
Uses `parse_log_line` to parse each line of the choosen log file.
@@ -72,17 +76,18 @@ def parse_log_file(log_file_path: Path) -> list[LogEntry]:
"""
log_entry_list: list[LogEntry] = []
with open(log_file_path, 'r', encoding="windows-1252") as log_file:
with open(log_file_path, "r", encoding="windows-1252") as log_file:
logger.info(f"\tparsing log file '{log_file_path.name}' started.")
for index, log_line in enumerate(log_file):
try:
log_entry = parse_log_line(log_line)
except ValueError as error:
logger.error(f"Error while parsing the log file {log_file.name} at {log_line}: {error}")
logger.error(
f"\t\tline {index}: error while parsing the log file '{log_file.name}': {error}"
)
else:
if log_entry:
log_entry_list.append(log_entry)
logger.debug(f"Line {index} was successfully parsed.")
else:
logger.debug(f"Line {index} is empty. Skip.")
logger.info(f"\t\t{len(log_entry_list)} lines were parsed.")
logger.info("\tparsing completed.")
return log_entry_list
src/logwatcher/reporter.py
+42 -1
View File
@@ -2,6 +2,7 @@ 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.models import LogEntry
@@ -59,15 +60,30 @@ def write_log_report(
file.
Use the range date of relevant and irrelevant lists to get the period of time
the logs were generated.
Args:
relevant: List of n2 log entry
irrelevant: List of not n2 log entry
nb_files: Number of files scanned
output_dir: Location where all reports will be written
Returns:
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(relevant, irrelevant, start_date, end_date, nb_files)
for target in reports_dict:
logger.info(f"\t\tgenerating report '{target}'...")
output_dir.mkdir(parents=True, exist_ok=True)
logger.debug(f"\t\t\t{output_dir} exists.")
target_file = output_dir / str(target + ".log")
target_file.write_text(reports_dict[target], encoding="windows-1252")
logger.debug(f"\t\t\treport was written in {target_file.name}")
logger.info(f"\t\treport '{target}' was generated.")
logger.info("\twriting reports job completed.")
def build_reports(
@@ -77,7 +93,17 @@ def build_reports(
Build the three output reports: n2, other, and all.
"all" report contains both "n2" and "other" sections.
"n2" and "other" are two unique sections.
Args:
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
end_date: Date of the newest log entry in relevant + irrelevant list
nb_files: number of files scanned
Returns:
Dictionary of reports in string format
"""
logger.info("\t\tbuilding reports job started.")
relevant_report = _render_target_report(
log_entries=relevant, target_template=N2_SUPPORT_TEMPLATE
)
@@ -85,6 +111,8 @@ def build_reports(
irrelevant_report = _render_target_report(
log_entries=irrelevant, target_template=OTHER_TEMPLATE
)
logger.info("\t\t\tn2 and other reports were generated.")
logger.info("\t\tbuilding reports job completed.")
return {
"n2": BASE_TEMPLATE.substitute(
@@ -112,11 +140,19 @@ def build_reports(
# build templates
def _render_target_report(log_entries: list[LogEntry], target_template) -> str:
def _render_target_report(
log_entries: list[LogEntry], target_template: Template
) -> str:
"""
Render report using either N2_SUPPORT_TEMPLATE or OTHER_TEMPLATE.
Call _render_entries() and include rendered log entries in the
new report.
Args:
log_entries: List of log entries
target_template: Template to use for N2_SUPPORT_TEMPLATE
Returns:
content of report in string format
"""
error_list = _render_entries(log_entries)
return target_template.substitute(nb_errors=len(log_entries), error_list=error_list)
@@ -126,6 +162,11 @@ 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
Args:
log_entries: List of log entries
Returns:
error report in string format
"""
error_list = []
for index, log_entry in enumerate(log_entries, start=1):