refactor(parser): change parser() by parse_log_line() function

src/logwatcher/app.py

@@ -1,10 +1,8 @@
 from logwatcher import logger
-from logwatcher.parser import parser
 
 
 def app():
     logger.info("logwatcher starts its work.")
-    parser()
     logger.info('logwatcher has finished its task.')
 
 if __name__ == "__main__":

src/logwatcher/parser.py

@@ -1,52 +1,29 @@
-import re
+from datetime import datetime
 from pathlib import Path
 
+from logwatcher import DATETIME_FORMAT, FRENCH_TIMEZONE, LOG_PATTERN, logger
+from logwatcher.models import LogEntry
+
 # misc log file
 misc_file_path = Path.cwd() / "output/misc.log"
 
 # effective logs
 effective_file_path = Path.cwd() / "output/effective.log"
 
-def _get_logs() -> str:
-    # read file
-    file_path = Path.cwd() / "tests\\fixtures\\CR_20260727110725.txt"
 
-    if not file_path.exists():
+def parse_log_line(log_line: str) -> LogEntry | None:
-        raise FileNotFoundError
+    match = LOG_PATTERN.search(log_line)
+    if not match:
+        logger.debug(f"Unrecognize format. Ignoring line {log_line}.")
+        return None
     
-    with open(file_path, "r") as file:
+    data = match.groupdict()
-        content = file.read()
+    return LogEntry(
-    return content
+        server_ip=data["server_ip"],
-
+        mdc_server_name=data["mdc_server_name"],
-
+        start_time=datetime.strptime(data["start_time"], DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
-def parser():
+        store_name=data["store_name"],
-    content = _get_logs()
+        error_time=datetime.strptime(data["error_time"], DATETIME_FORMAT).astimezone(FRENCH_TIMEZONE),
-
+        error_message=data["error_message"],
-    content_list = content.splitlines()
+        raw_line=log_line
-
+    )
-    pattern = r'^.*SetTailleTrav -> SizeRange.*$|^.*Code magasin non trouvé.*$|^.*DoTarifAchat.*$|^.*GetvalidVTA 2.1%.*$|^.*OrderNumber.*$'
-    
-    effective_content_list = re.findall(
-        pattern,
-        content,
-        flags=re.MULTILINE
-        )
-
-    misc_content_list = []
-
-    # misc_content_list = [line for line in content_list if line not in effective_content_list]
-
-    for line in content_list:
-        if line not in effective_content_list:
-            misc_content_list.append(line)
-        else:
-            print(line)
-
-    save_content("\n".join(effective_content_list), effective_file_path)
-    save_content("\n".join(misc_content_list), misc_file_path)
-
-
-def save_content(content_list: list, path: Path):
-    content = "".join(content_list)
-    with open(path, 'w') as file:
-        file.write(content)