"""Apache Common/Combined presets + the generic parser built from format_compiler (Chapter 07).""" from __future__ import annotations from app.services.log_parser import ParsedEntry from app.services.log_parser.format_compiler import ( CompiledFormat, compile_format, parse_apache_timestamp, parse_request_line, validate_ip, ) COMMON_LOG_FORMAT = '%h %l %u %t "%r" %>s %b' COMBINED_LOG_FORMAT = '%h %l %u %t "%r" %>s %b "%{Referer}i" "%{User-agent}i"' class FormatCompiledParser: """LogParser implementation driven by any CompiledFormat (Ch04/07 protocol).""" def __init__(self, compiled: CompiledFormat): self._compiled = compiled def parse_line(self, line: str) -> ParsedEntry | None: match = self._compiled.pattern.match(line.rstrip("\n")) if match is None: return None # malformed line — skip, never raise (Ch07) fields = match.groupdict() ip = validate_ip(fields["ip"]) if ip is None: return None req = parse_request_line(fields["request"]) if req is None: return None method, path = req try: timestamp = parse_apache_timestamp(fields["timestamp"]) status_code = int(fields["status"]) bytes_sent = 0 if fields["bytes"] == "-" else int(fields["bytes"]) except (ValueError, KeyError): return None referrer = fields.get("referrer") user_agent = fields.get("user_agent") return ParsedEntry( timestamp=timestamp, ip=ip, method=method, path=path, status_code=status_code, bytes_sent=bytes_sent, referrer=None if referrer in (None, "-") else referrer, user_agent=None if user_agent in (None, "-") else user_agent, ) def common_parser() -> FormatCompiledParser: return FormatCompiledParser(compile_format(COMMON_LOG_FORMAT)) def combined_parser() -> FormatCompiledParser: return FormatCompiledParser(compile_format(COMBINED_LOG_FORMAT))