PKx~ENxw~}}nginx_error_log/__init__.py"""Nginx error log parser.""" from ._utils import ParseError, Level, parse_level, parse_timestamp from ._parser import LogEntry from ._helpers import parse_lines, parse_lines_merge_multiple parse_line = LogEntry.parse_line __version__ = "0.1.0" __all__ = ( "ParseError", "Level", "LogEntry", "parse_line", "parse_lines", "parse_lines_merge_multiple", ) PK~ENLLnginx_error_log/_helpers.py"""Parser helpers.""" from typing import Iterable from ._utils import ParseError from ._parser import LogEntry def parse_lines(lines: Iterable[str]) -> Iterable[LogEntry]: """Parse an iterable of error_log lines. Raises: `ParseError` if a line could not be parsed. """ for line in lines: yield LogEntry.parse_line(line) def parse_lines_merge_multiple(lines: Iterable[str]) -> Iterable[LogEntry]: """Parse an iterable of error_log debug lines. If one of the logs are split over multiple lines, this will attempt to merge these into a single `LogEntry`. Only use this on `debug` level log files! Raises: `ParseError` if the first line could not be parsed. """ prev = None for line in lines: try: current = LogEntry.parse_line(line) except ParseError: if not prev: raise object.__setattr__(prev, "message", prev.message + "\n" + line.strip()) continue if prev: yield prev prev = current if prev: yield prev PK[~EN:nginx_error_log/_parser.py"""Log entry parser.""" import dataclasses import datetime from typing import Optional, Type, TypeVar from ._utils import ParseError, Level, parse_level, parse_timestamp, LOG_REGEX T = TypeVar("T", bound="LogEntry") @dataclasses.dataclass(frozen=True) class LogEntry: timestamp: datetime.datetime level: Level message: str pid: int tid: int cid: Optional[int] @classmethod def parse_line(cls: Type[T], line: str) -> T: """Parse a single error_log line. Raises: `ParseError` if the line could not be parsed. """ m = LOG_REGEX.match(line) if not m: raise ParseError("Could not parse {!r}.".format(line)) return cls( timestamp=parse_timestamp(m.group("date_string")), level=parse_level(m.group("level")), message=m.group("message").strip(), pid=int(m.group("pid")), tid=int(m.group("tid")), cid=int(m.group("cid")) if m.group("cid") is not None else None, ) PK}EN*uunginx_error_log/_utils.py"""Various utilities.""" import datetime import enum import logging import re # Format: # YYYY/MM/DD HH:MM:SS [LEVEL] PID#TID: *CID MESSAGE # See https://stackoverflow.com/a/26125951 LOG_REGEX = re.compile( r"^(?P[\d/: ]{19}) " r"\[(?P[a-z]+)\] " r"(?P\d+)#(?P\d+): " r"(\*(?P\d+) )?" r"(?P.*)", re.DOTALL, ) DATETIME_FORMAT = "%Y/%m/%d %H:%M:%S" class ParseError(Exception): pass class Level(enum.IntEnum): DEBUG = logging.DEBUG # 10 INFO = logging.INFO # 20 NOTICE = 25 WARNING = logging.WARNING # 30 ERROR = logging.ERROR # 40 CRITICAL = logging.CRITICAL # 50 ALERT = 60 EMERGENCY = 70 level_mapping = { "debug": Level.DEBUG, "info": Level.INFO, "notice": Level.NOTICE, "warn": Level.WARNING, "error": Level.ERROR, "crit": Level.CRITICAL, "alert": Level.ALERT, "emerg": Level.EMERGENCY, } def parse_level(data: str) -> Level: if data not in level_mapping: raise ParseError("Could not parse level {!r}.".format(data)) return level_mapping[data] def parse_timestamp(date_string: str) -> datetime.datetime: try: return datetime.datetime.strptime(date_string, DATETIME_FORMAT) except ValueError as e: raise ParseError( "Could not parse {!r} to a datetime.".format(date_string) ) from e PKl\N<..+nginx_error_log-0.1.0.dist-info/LICENSE.txtMIT License Copyright (c) 2019 Mads Marquart Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!H>*RQ%nginx_error_log-0.1.0.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,rzd&Y)r$[)T&UrPK!Hs (nginx_error_log-0.1.0.dist-info/METADATAVo8~_mu"5Ii{ڜ*]VZtE>S1|ul_e͌gƹK3ji7häHE:2Vò(w8qqDQm@G7Y@hTskI8gv^NTqA3S"PKA4뎂^H[CWY>:K;͐ϒj۬PP҈΍~L 51. T3e1ް#aQ+ O\é1lo.UX2Ԗ$ i 8| udLvL:ҹzͮɥCvɵRZVHl|Gq9BE A5xԔyg"p-sM.~=]A^kpGZt q#'#XGu IjV޽גlg0e܆/x͎̀[sheoS[ ܾSEv f;8=[79d2]1ErykT bizuxC.Cҥ3bMNPPK!Hi俓&nginx_error_log-0.1.0.dist-info/RECORDɒ@}= P HL% Æ@eHdH3SJ7F͢_O;px%#λs8@Juv|eI}ҧ?[qQg$bV"P|,< kcl~뇶&pGJp $d g0<D&%~#M+Qo֮Jbk82<%{9!yUG`g0ϫuREQSTݍBdsBBY8Tmi}W#jsˑ7u׫6C[ ;Z{w8Ģ`"IROcffhf@I9/q 1BUWCǖc枵ou_iuYT^Dt}޺*#țGC:|PKx~ENxw~}}nginx_error_log/__init__.pyPK~ENLLnginx_error_log/_helpers.pyPK[~EN:;nginx_error_log/_parser.pyPK}EN*uu nginx_error_log/_utils.pyPKl\N<..+6nginx_error_log-0.1.0.dist-info/LICENSE.txtPK!H>*RQ%nginx_error_log-0.1.0.dist-info/WHEELPK!Hs (Bnginx_error_log-0.1.0.dist-info/METADATAPK!Hi俓&Pnginx_error_log-0.1.0.dist-info/RECORDPKw'