logtail.py

#!/usr/bin/env python3
"""
logtail -- Tail and filter log files. Multi-file, color, JSON-aware. Zero deps.

Like tail -f + grep + jq combined. Works on Windows, macOS, Linux.
Handles multiple files, highlights patterns, pretty-prints JSON lines.

Usage:
    py logtail.py app.log                              # Tail a file
    py logtail.py app.log -f                           # Follow (like tail -f)
    py logtail.py app.log error.log -f                 # Tail multiple files
    py logtail.py app.log -g "ERROR|WARN"              # Filter by regex
    py logtail.py app.log -l error                     # Filter by log level
    py logtail.py app.log -f --json                    # Pretty-print JSON lines
    py logtail.py app.log -f -H "timeout|exception"    # Highlight keywords
    py logtail.py app.log -n 50                        # Last 50 lines
    py logtail.py app.log -f -x "DEBUG|TRACE"          # Exclude patterns
    py logtail.py *.log -f -l error -H "OOM"           # Combine filters
"""

import argparse
import fnmatch
import glob
import json
import os
import re
import sys
import time

# Colors
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
MAGENTA = "\033[35m"
BG_RED = "\033[41m"
BG_YELLOW = "\033[43;30m"

FILE_COLORS = [CYAN, MAGENTA, YELLOW, GREEN, "\033[34m", "\033[91m", "\033[92m", "\033[93m"]

LEVEL_COLORS = {
    "error": RED,
    "err": RED,
    "fatal": BG_RED,
    "critical": BG_RED,
    "warn": YELLOW,
    "warning": YELLOW,
    "info": GREEN,
    "debug": DIM,
    "trace": DIM,
}

LEVEL_PATTERNS = re.compile(
    r"\b(FATAL|CRITICAL|ERROR|ERR|WARN|WARNING|INFO|DEBUG|TRACE)\b", re.IGNORECASE
)


def color_supported() -> bool:
    if os.environ.get("NO_COLOR"):
        return False
    if sys.platform == "win32":
        return bool(os.environ.get("TERM") or os.environ.get("WT_SESSION"))
    return hasattr(sys.stdout, "isatty") and sys.stdout.isatty()


USE_COLOR = color_supported()


def c(code: str, text: str) -> str:
    return f"{code}{text}{RESET}" if USE_COLOR else text


def colorize_level(line: str) -> str:
    """Color-code log levels in a line."""
    if not USE_COLOR:
        return line
    def repl(m):
        level = m.group(0)
        color = LEVEL_COLORS.get(level.lower(), "")
        return f"{color}{level}{RESET}" if color else level
    return LEVEL_PATTERNS.sub(repl, line)


def highlight(line: str, pattern: re.Pattern) -> str:
    """Highlight matching text in a line."""
    if not USE_COLOR or not pattern:
        return line
    return pattern.sub(lambda m: f"{BG_YELLOW}{m.group()}{RESET}", line)


def format_json_line(line: str) -> str:
    """Try to pretty-print a JSON line."""
    stripped = line.strip()
    if not stripped.startswith("{"):
        return None
    try:
        data = json.loads(stripped)
        # Extract common fields
        parts = []
        ts = data.pop("timestamp", data.pop("time", data.pop("ts", data.pop("@timestamp", None))))
        level = data.pop("level", data.pop("severity", data.pop("lvl", None)))
        msg = data.pop("message", data.pop("msg", data.pop("text", None)))

        if ts:
            parts.append(c(DIM, str(ts)))
        if level:
            lc = LEVEL_COLORS.get(str(level).lower(), "")
            parts.append(c(lc, str(level).upper()) if lc else str(level).upper())
        if msg:
            parts.append(str(msg))
        if data:
            extra = " ".join(f"{k}={v}" for k, v in data.items())
            parts.append(c(DIM, extra))

        return " ".join(parts) if parts else None
    except (json.JSONDecodeError, TypeError, AttributeError):
        return None


def read_last_n_lines(filepath: str, n: int, encoding: str = "utf-8") -> list[str]:
    """Read last N lines from a file efficiently."""
    try:
        with open(filepath, "rb") as f:
            f.seek(0, 2)  # End of file
            size = f.tell()
            if size == 0:
                return []

            # Read backwards in chunks
            chunk_size = min(8192, size)
            lines = []
            pos = size

            while pos > 0 and len(lines) <= n:
                read_size = min(chunk_size, pos)
                pos -= read_size
                f.seek(pos)
                chunk = f.read(read_size)
                lines = chunk.split(b"\n") + lines[1:] if lines else chunk.split(b"\n")

            result = []
            for raw in lines[-n:]:
                try:
                    result.append(raw.decode(encoding, errors="replace").rstrip("\r"))
                except Exception:
                    result.append(raw.decode("latin-1", errors="replace").rstrip("\r"))
            return result
    except Exception:
        return []


class FileTailer:
    """Tail a single file with follow support."""

    def __init__(self, filepath: str, encoding: str = "utf-8"):
        self.filepath = filepath
        self.encoding = encoding
        self.name = os.path.basename(filepath)
        self._fh = None
        self._pos = 0

    def open(self, from_end: bool = True):
        try:
            self._fh = open(self.filepath, "r", encoding=self.encoding, errors="replace")
            if from_end:
                self._fh.seek(0, 2)
                self._pos = self._fh.tell()
        except FileNotFoundError:
            print(f"Error: file not found: {self.filepath}", file=sys.stderr)
        except Exception as e:
            print(f"Error opening {self.filepath}: {e}", file=sys.stderr)

    def read_new_lines(self) -> list[str]:
        """Read any new lines since last check."""
        if not self._fh:
            return []
        try:
            # Check for file truncation
            current_size = os.path.getsize(self.filepath)
            if current_size < self._pos:
                self._fh.seek(0)
                self._pos = 0

            self._fh.seek(self._pos)
            lines = []
            for line in self._fh:
                lines.append(line.rstrip("\n\r"))
            self._pos = self._fh.tell()
            return lines
        except Exception:
            return []

    def close(self):
        if self._fh:
            self._fh.close()


def should_include(line: str, grep_pat: re.Pattern | None,
                   exclude_pat: re.Pattern | None,
                   level_filter: set | None) -> bool:
    """Check if a line passes all filters."""
    if grep_pat and not grep_pat.search(line):
        return False
    if exclude_pat and exclude_pat.search(line):
        return False
    if level_filter:
        m = LEVEL_PATTERNS.search(line)
        if m:
            if m.group(0).lower() not in level_filter:
                return False
        else:
            return False  # No level found, filter it out
    return True


def format_line(line: str, prefix: str, json_mode: bool,
                highlight_pat: re.Pattern | None) -> str:
    """Format a line for output."""
    if json_mode:
        formatted = format_json_line(line)
        if formatted:
            line = formatted

    line = colorize_level(line)

    if highlight_pat:
        line = highlight(line, highlight_pat)

    if prefix:
        return f"{prefix} {line}"
    return line


def expand_file_args(file_args: list[str]) -> list[str]:
    """Expand glob patterns in file arguments."""
    result = []
    for arg in file_args:
        if "*" in arg or "?" in arg:
            expanded = glob.glob(arg)
            result.extend(sorted(expanded))
        else:
            result.append(arg)
    return result


def main():
    parser = argparse.ArgumentParser(
        description="logtail -- tail and filter log files",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("files", nargs="+", help="Log file(s) to tail (supports globs)")
    parser.add_argument("-f", "--follow", action="store_true", help="Follow file (like tail -f)")
    parser.add_argument("-n", "--lines", type=int, default=10, help="Number of lines from end (default: 10)")
    parser.add_argument("-g", "--grep", help="Include only lines matching regex")
    parser.add_argument("-x", "--exclude", help="Exclude lines matching regex")
    parser.add_argument("-l", "--level", help="Filter by log level (error,warn,info,debug; comma-separated)")
    parser.add_argument("-H", "--highlight", help="Highlight regex pattern in output")
    parser.add_argument("--json", action="store_true", help="Pretty-print JSON log lines")
    parser.add_argument("--no-prefix", action="store_true", help="Don't show filename prefix")
    parser.add_argument("-e", "--encoding", default="utf-8", help="File encoding (default: utf-8)")

    args = parser.parse_args()

    # Expand globs
    files = expand_file_args(args.files)
    if not files:
        print("Error: no files matched", file=sys.stderr)
        sys.exit(1)

    multi = len(files) > 1

    # Compile patterns
    grep_pat = re.compile(args.grep, re.IGNORECASE) if args.grep else None
    exclude_pat = re.compile(args.exclude, re.IGNORECASE) if args.exclude else None
    highlight_pat = re.compile(f"({args.highlight})", re.IGNORECASE) if args.highlight else None
    level_filter = None
    if args.level:
        level_filter = set()
        for l in args.level.split(","):
            l = l.strip().lower()
            level_filter.add(l)
            # Add aliases
            if l == "error":
                level_filter.add("err")
            if l == "warn":
                level_filter.add("warning")
            if l == "fatal":
                level_filter.add("critical")

    # Show initial lines (tail -n)
    if not args.follow or args.lines > 0:
        for fi, filepath in enumerate(files):
            color = FILE_COLORS[fi % len(FILE_COLORS)]
            prefix = c(color, f"[{os.path.basename(filepath)}]") if multi and not args.no_prefix else ""

            lines = read_last_n_lines(filepath, args.lines, args.encoding)
            for line in lines:
                if should_include(line, grep_pat, exclude_pat, level_filter):
                    print(format_line(line, prefix, args.json, highlight_pat))

    # Follow mode
    if args.follow:
        tailers = []
        for fi, filepath in enumerate(files):
            t = FileTailer(filepath, args.encoding)
            t.open(from_end=True)
            tailers.append((fi, t))

        if multi:
            print(c(DIM, f"\n  Following {len(files)} file(s)... (Ctrl+C to stop)\n"))
        else:
            print(c(DIM, f"\n  Following {files[0]}... (Ctrl+C to stop)\n"))

        try:
            while True:
                any_new = False
                for fi, tailer in tailers:
                    color = FILE_COLORS[fi % len(FILE_COLORS)]
                    prefix = c(color, f"[{tailer.name}]") if multi and not args.no_prefix else ""

                    new_lines = tailer.read_new_lines()
                    for line in new_lines:
                        if should_include(line, grep_pat, exclude_pat, level_filter):
                            print(format_line(line, prefix, args.json, highlight_pat), flush=True)
                            any_new = True

                if not any_new:
                    time.sleep(0.2)

        except KeyboardInterrupt:
            print(c(DIM, "\n  Stopped."))
        finally:
            for _, t in tailers:
                t.close()


if __name__ == "__main__":
    main()