nexus/tools/log_analyzer.py

"""
Log Analyzer Tool - Phân tích log files: filter, count, top errors, exceptions.
===========================================
Pure stdlib (re + collections.Counter). Hỗ trợ log levels chuẩn
(DEBUG/INFO/WARN/ERROR/FATAL) và pattern matching linh hoạt.

Author: Hieu Louis (2026)
"""
from __future__ import annotations

import os
import re
from collections import Counter
from typing import Any, Dict, List, Optional

from .base import Tool, ToolResult, ToolContext, ToolCategory, ToolSafety


LEVELS = {"DEBUG", "INFO", "WARN", "WARNING", "ERROR", "FATAL", "CRITICAL", "TRACE"}
OPERATIONS = {"filter", "count", "top_errors", "extract_exceptions", "summary", "tail", "head"}


# Pattern bắt log level thường gặp / common log-line level pattern
_LEVEL_RE = re.compile(r"\b(DEBUG|INFO|WARN(?:ING)?|ERROR|FATAL|CRITICAL|TRACE)\b", re.IGNORECASE)
# Pattern bắt exception/stacktrace / exception stacktrace pattern
_EXCEPTION_RE = re.compile(
    r"^(?:Traceback|Caused by:|^\s+at\s+|^\s*File\s+|^\s*\.\.\.|"
    r"[A-Za-z_][A-Za-z0-9_.]*(?:Error|Exception|Fault|Failure|Warning):)",
    re.MULTILINE,
)
# Pattern bắt message lỗi / error message capture
_ERROR_MSG_RE = re.compile(
    r"((?:[A-Za-z_][\w.]*)(?:Error|Exception|Fault|Failure|Warning))(?::\s*(.*))?",
    re.IGNORECASE,
)


class LogAnalyzerTool(Tool):
    """Phân tích log files: filter/count/top_errors/extract_exceptions/summary."""

    category = ToolCategory.MONITOR
    safety = ToolSafety.SAFE

    @property
    def name(self) -> str:
        return "log_analyzer"

    @property
    def description(self) -> str:
        return "Analyze log files: filter by level, count, top errors, extract exceptions, summary."

    @property
    def parameters(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Đường dẫn log file"},
                "operation": {
                    "type": "string",
                    "enum": sorted(OPERATIONS),
                    "default": "summary",
                },
                "level": {
                    "type": "string",
                    "enum": sorted(LEVELS),
                    "description": "Log level để filter (cho operation='filter')",
                },
                "min_level": {
                    "type": "string",
                    "enum": sorted(LEVELS),
                    "description": "Lọc từ level này trở lên (mức nghiêm trọng tăng dần)",
                },
                "pattern": {"type": "string", "description": "Regex pattern để match line"},
                "last_n": {"type": "integer", "default": 100, "description": "Cho tail/head"},
                "top_k": {"type": "integer", "default": 10, "description": "Cho top_errors"},
            },
            "required": ["path", "operation"],
        }

    def validate_args(self, args: Dict[str, Any]) -> Optional[str]:
        if not args.get("path"):
            return "Missing required arg: path"
        op = args.get("operation", "summary")
        if op not in OPERATIONS:
            return f"Invalid operation='{op}'. Supported: {sorted(OPERATIONS)}"
        if op == "filter" and not args.get("level") and not args.get("min_level") and not args.get("pattern"):
            return "filter cần ít nhất 1 trong: level, min_level, pattern"
        return None

    # ---- Helpers --------------------------------------------------------

    def _level_priority(self, level: str) -> int:
        """Trả về thứ tự nghiêm trọng của level. / Severity rank of level."""
        norm = level.upper()
        if norm in ("TRACE", "DEBUG"):
            return 0
        if norm == "INFO":
            return 1
        if norm in ("WARN", "WARNING"):
            return 2
        if norm == "ERROR":
            return 3
        if norm in ("FATAL", "CRITICAL"):
            return 4
        return -1

    def _detect_level(self, line: str) -> Optional[str]:
        m = _LEVEL_RE.search(line)
        if not m:
            return None
        return m.group(1).upper().replace("WARNING", "WARN")

    def _read_lines(self, path: str) -> List[str]:
        """Đọc file log (xử lý encoding linh hoạt). / Read log file with encoding fallback."""
        # Thử utf-8, fallback latin-1 (không bao giờ lỗi) / utf-8 first, latin-1 fallback
        for enc in ("utf-8", "latin-1"):
            try:
                with open(path, "r", encoding=enc) as f:
                    return f.readlines()
            except UnicodeDecodeError:
                continue
        # Fallback cuối / last-resort
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            return f.readlines()

    # ---- Operations -----------------------------------------------------

    def _op_count(self, lines: List[str]) -> Dict[str, Any]:
        counts: Counter = Counter()
        no_level = 0
        for line in lines:
            lvl = self._detect_level(line)
            if lvl:
                counts[lvl] += 1
            else:
                no_level += 1
        return {
            "counts": dict(counts),
            "no_level": no_level,
            "total_lines": len(lines),
            "total_with_level": sum(counts.values()),
        }

    def _op_filter(
        self,
        lines: List[str],
        level: Optional[str],
        min_level: Optional[str],
        pattern: Optional[str],
    ) -> Dict[str, Any]:
        regex = re.compile(pattern) if pattern else None
        threshold = self._level_priority(min_level) if min_level else None
        target = level.upper().replace("WARNING", "WARN") if level else None
        matched: List[str] = []
        for line in lines:
            lvl = self._detect_level(line)
            if target and lvl != target:
                continue
            if threshold is not None:
                p = self._level_priority(lvl) if lvl else -1
                if p < threshold:
                    continue
            if regex and not regex.search(line):
                continue
            matched.append(line.rstrip("\n"))
        return {
            "matched_count": len(matched),
            "sample_lines": matched[:50],  # Giới hạn output / limit output
            "filter": {"level": target, "min_level": min_level, "pattern": pattern},
        }

    def _op_top_errors(self, lines: List[str], top_k: int) -> Dict[str, Any]:
        """Đếm top exception types và message patterns."""
        exception_counts: Counter = Counter()
        error_line_counts: Counter = Counter()
        for line in lines:
            m = _ERROR_MSG_RE.search(line)
            if m:
                exc_type = m.group(1)
                exception_counts[exc_type] += 1
                # Normalize message: bỏ số / digits, rút gọn
                msg = (m.group(2) or "").strip()
                # Bỏ số, path, ID / strip digits, paths, IDs
                norm = re.sub(r"\d+", "N", msg)
                norm = re.sub(r"/[\w/\.]+", "/path", norm)
                norm = re.sub(r"\s+", " ", norm)[:120]
                if norm:
                    error_line_counts[f"{exc_type}: {norm}"] += 1
        return {
            "top_exception_types": exception_counts.most_common(top_k),
            "top_error_messages": error_line_counts.most_common(top_k),
        }

    def _op_extract_exceptions(self, lines: List[str]) -> Dict[str, Any]:
        """Trích các block stacktrace (từ 'Traceback' đến khi gặp dòng trống/log-level)."""
        blocks: List[Dict[str, Any]] = []
        current: List[str] = []
        for line in lines:
            if line.startswith("Traceback") or line.startswith("Caused by:"):
                if current:
                    blocks.append({"lines": current, "preview": "".join(current[:3]).strip()})
                current = [line.rstrip("\n")]
            elif current and (
                line.startswith("  ")
                or line.startswith("\t")
                or _EXCEPTION_RE.match(line)
                or not line.strip()
            ):
                current.append(line.rstrip("\n"))
                if not line.strip() and len(current) > 3:
                    # Kết thúc block khi gặp dòng trống / end on blank line
                    blocks.append({"lines": current, "preview": "".join(current[:3]).strip()})
                    current = []
            elif current:
                blocks.append({"lines": current, "preview": "".join(current[:3]).strip()})
                current = []
        if current:
            blocks.append({"lines": current, "preview": "".join(current[:3]).strip()})
        return {
            "exception_blocks": len(blocks),
            "previews": [b["preview"][:200] for b in blocks[:20]],
        }

    def _op_summary(self, lines: List[str]) -> Dict[str, Any]:
        """Tổng hợp nhanh file log."""
        counts = self._op_count(lines)
        size_bytes = sum(len(l.encode("utf-8", errors="replace")) for l in lines)
        # Phát hiện timestamp đầu & cuối / detect first/last timestamps
        ts_re = re.compile(r"(\d{4}-\d{2}-\d{2}[\dT\s:.:+\-]+\d{2}:\d{2}:\d{2})")
        first_ts = last_ts = None
        for line in lines[:200]:
            m = ts_re.search(line)
            if m:
                first_ts = m.group(1)
                break
        for line in reversed(lines[-200:]):
            m = ts_re.search(line)
            if m:
                last_ts = m.group(1)
                break
        return {
            **counts,
            "file_size_bytes": size_bytes,
            "first_timestamp": first_ts,
            "last_timestamp": last_ts,
        }

    # ---- Execute --------------------------------------------------------

    def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult:
        path = args["path"]
        op = args.get("operation", "summary")

        if not os.path.exists(path):
            return ToolResult(success=False, error=f"Log file không tồn tại: {path}", return_code=1)
        if os.path.isdir(path):
            return ToolResult(success=False, error=f"Path là thư mục, không phải file: {path}", return_code=1)

        try:
            lines = self._read_lines(path)
        except Exception as e:
            return ToolResult(success=False, error=f"Read log failed: {e}", return_code=1)

        if context.dry_run:
            return ToolResult(
                success=True,
                output=f"[dry-run] op='{op}' trên {len(lines)} lines của {path}",
                metadata={"operation": op, "n_lines": len(lines), "dry_run": True},
            )

        try:
            if op == "count":
                result: Any = self._op_count(lines)
            elif op == "filter":
                result = self._op_filter(
                    lines,
                    args.get("level"),
                    args.get("min_level"),
                    args.get("pattern"),
                )
            elif op == "top_errors":
                result = self._op_top_errors(lines, int(args.get("top_k", 10)))
            elif op == "extract_exceptions":
                result = self._op_extract_exceptions(lines)
            elif op == "summary":
                result = self._op_summary(lines)
            elif op == "tail":
                n = int(args.get("last_n", 100))
                result = {"tail_lines": [l.rstrip("\n") for l in lines[-n:]], "count": n}
            elif op == "head":
                n = int(args.get("last_n", 100))
                result = {"head_lines": [l.rstrip("\n") for l in lines[:n]], "count": n}
            else:
                return ToolResult(success=False, error=f"Unknown operation: {op}", return_code=1)

            return ToolResult(
                success=True,
                output=str(result)[:5000],  # Truncate output lớn / truncate huge outputs
                metadata={
                    "operation": op,
                    "file": path,
                    "total_lines": len(lines),
                    "result": result,
                },
            )
        except Exception as e:
            return ToolResult(success=False, error=f"Analysis failed: {e}", return_code=1)