"""Squid custom logformat parser.

Squid's ``access.log`` is normally written in *native* format, but operators
can override it with a ``logformat`` directive, e.g.::

    logformat squid %ts.%03tu %6tr %>a %Ss/%03>Hs %<st %rm %ru %un %Sh/%<a %mt

This module turns such a format string into a regular expression and applies
it to individual log lines, returning a ``LogEntry``-compatible dict whose
field names line up with :mod:`log_parser` so the rest of the application
(dashboard, alerts, anomaly detection) does not need to change.

The two important design choices:

1. **Field name parity.** ``parse_with_format`` returns the same keys as
   ``log_parser.parse_line`` so ``get_parsed_logs`` can return a single
   uniform ``LogEntry`` regardless of which path produced it.
2. **Compose-then-split.** The combined tokens such as ``%Ss/%03>Hs`` or
   ``%Sh/%<a`` are matched as a single field (one whitespace-delimited
   token) and then split on ``/`` internally. This lets us handle the
   native-style "slash-joined" log line shape no matter which ``logformat``
   the operator uses.
"""
from __future__ import annotations

import re
from datetime import datetime, timezone
from urllib.parse import urlsplit

# ---------------------------------------------------------------------------
# Token map
# ---------------------------------------------------------------------------
#
# Each entry is ``(token_text, canonical_group_name, inner_pattern)``.
# ``canonical_group_name`` is a valid Python regex identifier that maps onto
# the field name used by :mod:`log_parser`.  The order of ``_TOKEN_DEFS``
# matters: longer / more specific tokens (with width specifiers like
# ``%03tu``) MUST be tried before shorter variants of the same prefix so
# we don't accidentally swallow them as ``%tu``.
#
# Tokens covered:
#   %ts   - Unix timestamp seconds (with optional fractional part)
#   %03tu / %tu - Sub-second fraction (milliseconds)
#   %6tr / %tr - Response time (microseconds with %6, otherwise raw)
#   %>a   - Client IP / source address
#   %<a   - Server IP / peer address
#   %Ss   - Squid result code (TCP_HIT etc)
#   %03>Hs / %>Hs - HTTP status code (with / without width spec)
#   %<st / %st - Reply size in bytes
#   %rm   - Request method (GET/POST/CONNECT ...)
#   %ru   - Request URL
#   %un   - Auth user / ident ("-" when anonymous)
#   %Sh   - Hierarchy code (HIER_DIRECT etc)
#   %mt   - MIME content type (greedy at the end)

# (token_text, canonical_group_name, regex_inner)
# ``regex_inner`` is the regex WITHOUT the ``(?P<name>...)`` wrapper so we
# can reuse it for compound tokens (where multiple tokens share the same
# captured string).
_TOKEN_DEFS: list[tuple[str, str, str]] = [
    # Timestamp: integer seconds.  The optional fractional part is
    # handled by %03tu / %tu if the operator's format splits it out
    # (e.g. "%ts.%03tu"); otherwise the native parser captures both
    # halves via its \d+(?:\.\d+)? pattern.
    ("%ts",      "time",          r"\d+"),
    ("%03tu",    "msec",          r"\d{1,3}"),
    ("%tu",      "msec",          r"\d+"),
    # Response time (microseconds with %6, otherwise raw)
    ("%6tr",     "elapsed",       r"\d+"),
    ("%tr",      "elapsed",       r"\d+"),
    # Client / peer address
    ("%>a",      "client",        r"\S+"),
    ("%<a",      "peer",          r"\S+"),
    # Result code (squid code / http status combined)
    ("%Ss",      "result",        r"\S+"),
    # HTTP status code field (becomes http_code_field; combined with result
    # in parse_with_format to yield http_code)
    ("%03>Hs",   "http_code_field", r"\d+"),
    ("%>Hs",     "http_code_field", r"\d+"),
    # Sizes
    ("%<st",     "size",          r"\d+"),
    ("%st",      "size",          r"\d+"),
    # Request
    ("%rm",      "method",        r"\S+"),
    ("%ru",      "url",           r"\S+"),
    ("%un",      "ident",         r"\S+"),
    # Hierarchy
    ("%Sh",      "hier_code",     r"\S+"),
    # Content type (greedy because it's the last column)
    ("%mt",      "content_type",  r".*"),
]

# Lookup helpers - built once at import time
_TOKEN_TO_GROUP: dict[str, str] = {tok: grp for tok, grp, _ in _TOKEN_DEFS}
_GROUP_TO_INNER: dict[str, str] = {grp: inner for _, grp, inner in _TOKEN_DEFS}

# Order to try matching tokens at any given '%' position: longest first.
_TOKEN_ORDER: list[str] = sorted(_TOKEN_TO_GROUP.keys(), key=lambda x: -len(x))


# Built-in logformat examples presented in the UI dropdown.  "native" is
# the default Squid format and is handled by log_parser directly - it is
# listed here so the dropdown has a consistent shape and the user can
# explicitly choose "go back to native" from the picker.
PRESET_FORMATS: dict[str, str] = {
    "native": "",  # empty = fall back to log_parser (Squid native)
    "squid": "%ts.%03tu %6tr %>a %Ss/%03>Hs %<st %rm %ru %un %Sh/%<a %mt",
    "common": '%>a %[ui] [%{%d/%b/%Y:%H:%M:%S +0000}] "%rm %ru HTTP/%rv" %>Hs %<st "%{Referer}>h" "%{User-Agent}>h"',
    "combined": '%>a %[ui] [%{%d/%b/%Y:%H:%M:%S +0000}] "%rm %ru HTTP/%rv" %>Hs %<st "%{Referer}>h" "%{User-Agent}>h" %Sh/%<a %mt',
    "referer": '%ts.%03tu %>a "%{Referer}>h" "%rm %ru" %>Hs %<st',
    "useragent": '%ts.%03tu %>a "%rm %ru" %>Hs %<st "%{User-Agent}>h"',
}


# ---------------------------------------------------------------------------
# Compilation
# ---------------------------------------------------------------------------

class CompiledFormat:
    """Bundles a compiled regex with the list of named groups it produces.

    The caller can introspect ``fields`` to decide what to display in the
    UI without re-running the parser.
    """

    __slots__ = ("pattern", "fields", "source")

    def __init__(self, pattern: re.Pattern, fields: list[str], source: str):
        self.pattern = pattern
        self.fields = fields
        self.source = source

    def __repr__(self) -> str:  # pragma: no cover - debug only
        return f"CompiledFormat(fields={self.fields!r})"


def _sanitize_group_name(name: str) -> str:
    """Turn an arbitrary string into a valid Python regex group identifier."""
    # Group names in Python's re module must match [A-Za-z_][A-Za-z0-9_]*
    # and must not be a pure integer.  Strip invalid chars and prefix
    # if the result would otherwise start with a digit.
    out = re.sub(r"[^A-Za-z0-9_]", "_", name)
    if not out or out[0].isdigit():
        out = "g_" + out
    return out


def _inner_without_slash(inner: str) -> str:
    """Return a variant of ``inner`` that does not match ``/``.

    Used for the first half of a compound token so the regex engine
    knows where to place the literal ``/`` separator.  We rewrite the
    common character classes to exclude ``/``:

    - ``\\S+`` -> ``[^/\\s]+``
    - ``\\S*`` -> ``[^/\\s]*``
    - ``.*``   -> ``[^/]*``  (only used for trailing content_type)
    """
    out = inner
    out = out.replace(r"\S+", r"[^/\s]+")
    out = out.replace(r"\S*", r"[^/\s]*")
    out = out.replace(r".*",   r"[^/]*")
    return out


def compile_logformat(fmt: str) -> CompiledFormat:
    """Compile a Squid ``logformat`` string into a regex.

    The algorithm:

    1. Walk the format string, looking for ``%`` and trying to match the
       longest known token at each position.
    2. Replace each matched token with its regex fragment (which carries a
       named capture group).
    3. Escape any literal text between tokens, but treat **whitespace as a
       separator** rather than a literal character - we expect log lines
       to be whitespace-separated and want the regex to remain robust
       even if the operator accidentally uses tabs vs. spaces.
    4. Return a :class:`CompiledFormat` carrying the compiled pattern, the
       ordered list of named fields, and the original source string.

    Notes / limitations:
    - Squid has hundreds of format codes; we only know the subset listed in
      :data:`_TOKEN_DEFS`. Unknown ``%xxx`` sequences are passed through
      as ``\\S+`` so the line still parses.
    - Compound tokens like ``%Ss/%03>Hs`` are detected at compile time
      and rendered as a single compound field (``hierarchy_blob``) that
      is split on ``/`` in :func:`parse_with_format`.  The compound
      shares the first half's group name so existing field-name lookups
      still work.
    """
    fmt = (fmt or "").strip()
    if not fmt:
        raise ValueError("empty logformat string")

    # Detect whether the operator splits the timestamp via ``%ts.%03tu``
    # - if so, we constrain ``%ts`` to integer seconds so the literal
    # dot between the two halves is unambiguous.  Otherwise we accept
    # the bare ``seconds.msec`` shape from the native format too.
    has_split_ts = bool(re.search(r"%ts\s*\.\s*%(03)?tu", fmt))

    # Pass 1: tokenise into a stream of (kind, value) pairs.
    # kind in {"tok","lit","unk","slash"} - "slash" is a marker we emit
    # when a literal begins with "/" so we can later detect compound
    # sequences like "%A/%B".
    tokens: list[tuple[str, str]] = []
    i = 0
    n = len(fmt)
    while i < n:
        if fmt[i] == "%":
            # Try longest-known-token first
            matched_tok: str | None = None
            for tok in _TOKEN_ORDER:
                if fmt.startswith(tok, i):
                    matched_tok = tok
                    break
            if matched_tok is not None:
                grp = _TOKEN_TO_GROUP[matched_tok]
                # ``%ts`` gets a context-sensitive inner pattern: if the
                # format splits it (``%ts.%03tu``), only integer seconds
                # are valid; otherwise we accept either integer or
                # ``seconds.msec`` to also cover the native format.
                if matched_tok == "%ts" and not has_split_ts:
                    grp = "time_loose"
                tokens.append(("tok", grp))
                i += len(matched_tok)
                continue
            # Try special %>xx / %<xx forms that aren't in the literal
            # table but we still want to capture (e.g. %>ru, %>ha).
            if i + 1 < n and fmt[i + 1] in (">", "<"):
                end = i + 2
                while end < n and (fmt[end].isalpha() or fmt[end] == "_"):
                    end += 1
                candidate = fmt[i + 1:end]  # e.g. "a", "ru", "Hs"
                name = _sanitize_group_name("ext_" + candidate.lower())
                # Emit as a known field if we happen to map it
                mapped = _GROUP_TO_INNER.get(name)
                # Generic capture; cannot recover semantics so just \S+
                tokens.append(("tok", name))
                i = end
                continue
            # Unknown %-code: consume the % and the following
            # identifier characters; emit a generic "unknown" field.
            end = i + 1
            while end < n and (fmt[end].isalpha() or fmt[end].isdigit()
                               or fmt[end] in "<>_{}[]"):
                end += 1
            if end == i + 1:
                # bare '%' with nothing after it: emit literally
                tokens.append(("lit", "%"))
                i += 1
            else:
                name = _sanitize_group_name("u_" + fmt[i + 1:end].lower())
                tokens.append(("unk", name))
                i = end
        else:
            # Literal run - collect everything until the next % or end
            j = i
            while j < n and fmt[j] != "%":
                j += 1
            lit = fmt[i:j]
            # If a token was just emitted and the next literal begins
            # with "/", mark this as a compound separator.
            if lit.startswith("/") and tokens and tokens[-1][0] == "tok":
                tokens.append(("slash", ""))
                lit = lit[1:]
            tokens.append(("lit", lit))
            i = j

    # Pass 2: detect compound tokens (tok - slash - [lit] - tok) and
    # merge them into a single field that captures the slash-joined
    # string.  The result is rendered as one capture group whose name is
    # the first half's name; the post-processor in parse_with_format
    # splits it.
    merged: list[tuple[str, str]] = []
    j = 0
    while j < len(tokens):
        kind, value = tokens[j]
        # Look ahead for: tok - slash - [optional empty literal] - tok
        is_compound = False
        if kind == "tok" and j + 1 < len(tokens) and tokens[j + 1][0] == "slash":
            # Skip any empty literal between slash and second half.
            k = j + 2
            while k < len(tokens) and tokens[k][0] == "lit" and tokens[k][1] == "":
                k += 1
            if k < len(tokens) and tokens[k][0] in ("tok", "unk"):
                v1 = tokens[j][1]
                v2 = tokens[k][1]
                inner1 = _GROUP_TO_INNER.get(v1, r"\S+")
                inner2 = _GROUP_TO_INNER.get(v2, r"\S+")
                merged.append(("compound", f"{v1}|{inner1}|{inner2}"))
                j = k + 1
                is_compound = True
        if is_compound:
            continue
        if kind == "slash":
            # Stray slash: shouldn't happen given the tokeniser but
            # be defensive.
            j += 1
            continue
        merged.append((kind, value))
        j += 1

    # Pass 3: translate into regex
    parts: list[str] = []
    fields: list[str] = []
    seen_groups: set[str] = set()
    for kind, value in merged:
        if kind == "lit":
            # Whitespace runs -> \s+, anything else -> re.escape
            k = 0
            while k < len(value):
                ch = value[k]
                if ch.isspace():
                    m = k
                    while m < len(value) and value[m].isspace():
                        m += 1
                    parts.append(r"\s+")
                    k = m
                else:
                    parts.append(re.escape(ch))
                    k += 1
        elif kind == "tok":
            inner = _GROUP_TO_INNER.get(value, r"\S+")
            grp_name = value
            if grp_name in seen_groups:
                # Duplicate group name (shouldn't happen for our token
                # table but be defensive).  Use a suffixed alias.
                grp_name = grp_name + "_dup"
            seen_groups.add(grp_name)
            parts.append(rf"(?P<{grp_name}>{inner})")
            if grp_name not in fields:
                fields.append(grp_name)
        elif kind == "unk":
            parts.append(rf"(?P<{value}>\S+)")
            if value not in fields:
                fields.append(value)
        elif kind == "compound":
            # value is "v1|inner1|inner2" - encode the slash-joined
            # capture under v1's group name.
            v1, inner1, inner2 = value.split("|", 2)
            grp_name = v1
            if grp_name in seen_groups:
                grp_name = grp_name + "_c"
            seen_groups.add(grp_name)
            # The combined pattern captures the full slash-joined value.
            # The *first* half must be non-greedy and constrained to
            # *not* contain "/" so the "/" separator falls in the right
            # place.  E.g. ``\S+/`` would otherwise over-eat the
            # ``HIER_DIRECT/13.107.5.93`` style string and return
            # ``HIER_DIRECT/13.107.5.9`` (3 captured).
            #
            # We rewrite inner1 to disallow "/" and make it non-greedy
            # so the engine backtracks to the first valid separator.
            non_greedy_inner1 = _inner_without_slash(inner1)
            parts.append(rf"(?P<{grp_name}>{non_greedy_inner1}/{inner2})")
            if grp_name not in fields:
                fields.append(grp_name)

    pattern = re.compile("^" + "".join(parts) + r"\s*$")
    return CompiledFormat(pattern=pattern, fields=fields, source=fmt)


# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------

def parse_with_format(line: str, compiled) -> dict | None:
    """Parse a single log line using a compiled logformat.

    ``compiled`` may be either a :class:`CompiledFormat` or the bare
    ``(pattern, fields)`` tuple returned by older call sites.  Returns a
    dict shaped exactly like :func:`log_parser.parse_line` so the result
    can flow through the same downstream code.
    """
    if not line or not line.strip():
        return None
    if isinstance(compiled, tuple):
        pattern = compiled[0]
    else:
        pattern = compiled.pattern

    m = pattern.match(line.strip())
    if not m:
        return None
    g = m.groupdict()

    # --- timestamp ------------------------------------------------------
    # If the format splits the timestamp into integer seconds and a
    # separate msec field (%ts.%03tu), recombine them so consumers
    # always see the canonical ``time`` string in ``seconds.msec`` form
    # - matching log_parser's output and avoiding surprising downstream
    # code that expects ``float(time)`` to give a sub-second timestamp.
    time_str = g.get("time") or "0"
    msec = g.get("msec")
    if msec and "." not in time_str:
        # pad msec to 3 digits to match Squid's native formatting
        time_str = f"{time_str}.{int(msec):03d}"
    try:
        ts = float(time_str)
        dt = datetime.fromtimestamp(ts, tz=timezone.utc)
    except (ValueError, OSError):
        dt = None

    # --- elapsed --------------------------------------------------------
    # log_parser treats this field as milliseconds directly.  Squid's
    # ``%6tr`` width specifier is just left-padding - the value itself
    # is reported in milliseconds.  Mirror that behaviour so downstream
    # code can compare LogEntry dicts interchangeably.
    elapsed_raw = g.get("elapsed") or "0"
    try:
        elapsed_ms = int(elapsed_raw)
    except ValueError:
        elapsed_ms = 0

    # --- result / http code --------------------------------------------
    # Compound tokens like %Ss/%03>Hs end up captured under the *first*
    # half's group name (e.g. "result").  Split on "/" to recover the
    # two halves.  The http_code_field group is used when the result is
    # not slash-joined but the http code is a standalone token.
    raw_result = g.get("result", "") or ""
    http_code_field = g.get("http_code_field", "") or ""
    if "/" in raw_result:
        squid_code, _, http_code = raw_result.partition("/")
        if not http_code:
            http_code = http_code_field
    else:
        squid_code = raw_result
        http_code = http_code_field

    # --- hierarchy / peer ----------------------------------------------
    # The compound %Sh/%<a lives in the "hier_code" group as
    # "HIER_DIRECT/13.107.5.93".  Split on "/" to recover both halves.
    # If the user only specified %Sh (no compound), the value is the
    # bare code and peer stays empty.
    raw_hier = g.get("hier_code", "") or ""
    if "/" in raw_hier:
        hier_code, _, peer = raw_hier.partition("/")
    else:
        hier_code, peer = raw_hier, ""

    # If the user explicitly captured a peer separately (e.g. via %>a
    # followed by %<a) prefer that over the compound value.
    if g.get("peer") and not peer:
        peer = g["peer"]

    # --- URL / host -----------------------------------------------------
    url = g.get("url", "") or ""
    host = _extract_host(url)

    # --- ident ----------------------------------------------------------
    ident = g.get("ident", "") or ""
    if ident == "-":
        ident = ""

    # --- category -------------------------------------------------------
    # Imported lazily to avoid a circular import at module load time.
    from log_parser import RESULT_CATEGORIES
    category = RESULT_CATEGORIES.get(squid_code, "Other")

    # --- size / content_type -------------------------------------------
    try:
        size = int(g.get("size") or 0)
    except ValueError:
        size = 0
    content_type = (g.get("content_type") or "").strip()

    return {
        # Mirror log_parser.LogEntry keys exactly
        "time": time_str,
        "timestamp": dt,
        "elapsed_ms": elapsed_ms,
        "client": g.get("client", "") or "",
        "result": raw_result,
        "result_code": squid_code,
        "http_code": http_code,
        "category": category,
        "size": size,
        "method": g.get("method", "") or "",
        "url": url,
        "host": host,
        "ident": ident,
        "hierarchy": f"{hier_code}/{peer}" if (hier_code or peer) else "",
        "hier_code": hier_code,
        "peer": peer,
        "content_type": content_type,
    }


# ---------------------------------------------------------------------------
# Host extraction (mirrors log_parser._extract_host so we behave the same)
# ---------------------------------------------------------------------------

def _extract_host(url: str) -> str:
    """Extract hostname from a Squid log URL.

    Mirrors :func:`log_parser._extract_host`.  Squid logs URLs in two
    forms: ``http(s)://host/path?query`` for normal GET/POST, and
    ``host:port`` for CONNECT tunnels.  ``urlsplit`` chokes on the second
    form so we detect it explicitly.
    """
    if not url or url == "-":
        return ""
    if "://" not in url:
        if ":" in url:
            host, _, _port = url.rpartition(":")
            return host.strip()
        return url.strip()
    try:
        h = urlsplit(url).hostname
        return h or ""
    except Exception:
        return ""


# ---------------------------------------------------------------------------
# Convenience: parse multi-line text
# ---------------------------------------------------------------------------

def parse_lines_with_format(text: str, compiled) -> list[dict]:
    """Parse a multi-line log blob.  Unparseable lines are silently skipped."""
    out = []
    for line in text.splitlines():
        e = parse_with_format(line, compiled)
        if e:
            out.append(e)
    return out