Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.
from typing import Optional
from .time_iso_to_unix import iso_to_unix ← from time.iso-to-unix ^1.0.0 · built alongside by fune
from .monitor_parse_access_log_types import AccessLogEntry
_MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
def _at(t: str, i: int) -> str:
return t[i] if 0 <= i < len(t) else ""
def _digits(text: str, start: int, count: int) -> Optional[int]:
n = 0
for i in range(start, start + count):
c = _at(text, i)
# Not str.isdigit(): that accepts other scripts' digits and superscripts.
if not ("0" <= c <= "9") or c == "":
return None
n = n * 10 + ord(c) - 48
return n
def _dash(field: str) -> Optional[str]:
return None if field == "-" else field
def _parse_time(t: str, i: int) -> Optional[int]:
"""The time field, e.g. [10/Oct/2000:13:55:36 -0700], starting at `i` to Unix seconds, or None."""
for off, ch in ((0, "["), (3, "/"), (7, "/"), (12, ":"), (15, ":"), (18, ":"), (21, " "), (27, "]")):
if _at(t, i + off) != ch:
return None
day = _digits(t, i + 1, 2)
name = t[i + 4:i + 7]
month = _MONTHS.index(name) + 1 if name in _MONTHS else 0
year = _digits(t, i + 8, 4)
hour = _digits(t, i + 13, 2)
minute = _digits(t, i + 16, 2)
second = _digits(t, i + 19, 2)
sign = _at(t, i + 22)
oh = _digits(t, i + 23, 2)
om = _digits(t, i + 25, 2)
if day is None or year is None or hour is None or minute is None or second is None or oh is None or om is None:
return None
if month == 0 or year < 1 or hour > 23 or minute > 59 or second > 59 or oh > 23 or om > 59:
return None
if sign != "+" and sign != "-":
return None
leap = (year % 4 == 0 and year % 100 != 0) or year % 400 == 0
month_days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31][month - 1]
if day < 1 or day > month_days:
return None
# Every field is checked above, so iso_to_unix cannot raise here.
return iso_to_unix("%s-%02d-%sT%s%s%s:%s" % (t[i + 8:i + 12], month, t[i + 1:i + 3], t[i + 13:i + 21], sign, t[i + 23:i + 25], t[i + 25:i + 27]))
def _closing_quote(t: str, i: int) -> int:
"""The index of the quote closing a field that opens at `i`, honouring \\" escapes; -1 if none."""
if _at(t, i) != '"':
return -1
j = i + 1
while j < len(t):
if t[j] == "\\":
j += 1
elif t[j] == '"':
return j
j += 1
return -1
def _is_method(m: str) -> bool:
return len(m) > 0 and all("A" <= c <= "Z" for c in m)
def parse_access_log(line: str) -> Optional[AccessLogEntry]:
"""One Common or Combined Log Format line, or None when the line is in
neither format: logs hold junk, and a reader should skip it rather than stop."""
t = line
while t.endswith("\n") or t.endswith("\r"):
t = t[:-1]
host_end = t.find(" ")
if host_end < 1:
return None
ident_end = t.find(" ", host_end + 1)
if ident_end < host_end + 2:
return None
user_end = t.find(" [", ident_end + 1)
if user_end < ident_end + 2:
return None
at = _parse_time(t, user_end + 1)
if at is None:
return None
i = user_end + 29
if _at(t, i) != " ":
return None
req_end = _closing_quote(t, i + 1)
if req_end < 0:
return None
request = t[i + 2:req_end]
i = req_end + 1
# A three-digit status, then the byte count: both are required.
if _at(t, i) != " " or _at(t, i + 4) != " ":
return None
status = _digits(t, i + 1, 3)
if status is None or status < 100 or status > 599:
return None
i += 4
bytes_end = t.find(" ", i + 1)
if bytes_end < 0:
bytes_end = len(t)
bytes_text = t[i + 1:bytes_end]
size: Optional[int] = None
if bytes_text != "-":
if len(bytes_text) < 1 or len(bytes_text) > 15:
return None
size = _digits(bytes_text, 0, len(bytes_text))
if size is None:
return None
referer: Optional[str] = None
user_agent: Optional[str] = None
if bytes_end < len(t):
ref_end = _closing_quote(t, bytes_end + 1)
if ref_end < 0 or _at(t, ref_end + 1) != " ":
return None
ua_end = _closing_quote(t, ref_end + 2)
if ua_end < 0 or (ua_end + 1 < len(t) and t[ua_end + 1] != " "):
return None
referer = _dash(t[bytes_end + 2:ref_end])
user_agent = _dash(t[ref_end + 3:ua_end])
method: Optional[str] = None
path: Optional[str] = None
protocol: Optional[str] = None
parts = request.split(" ")
if len(parts) == 3 and _is_method(parts[0]) and parts[1] != "" and parts[2].startswith("HTTP/"):
method, path, protocol = parts
elif len(parts) == 2 and _is_method(parts[0]) and parts[1] != "":
method, path = parts
return AccessLogEntry(
remote_host=t[:host_end],
ident=_dash(t[host_end + 1:ident_end]),
user=_dash(t[ident_end + 1:user_end]),
at=at,
method=method,
path=path,
protocol=protocol,
status=status,
bytes=size,
referer=referer,
user_agent=user_agent,
)