# tools/system.py - Remaining system tools + backward-compatible re-exports.
#
# Shell execution, confirmation IPC, and env setup moved to tools/shell.py.
# Python code execution moved to tools/code_execution.py.
# Screenshot capture moved to tools/screenshots.py.
from __future__ import annotations

import contextvars
import fnmatch
import os
import re
import time
import zipfile
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from threading import Event, Lock, Thread

from langchain_core.tools import tool

from tools._logging import _debug_log, _emit_input_status, emit_chat_event
from tools.live_view_visuals import emit_live_view_preview, image_to_preview_base64
from utils.path_safety import is_dangerous_path, is_plugins_dir_write_blocked, get_user_output_dir

# Backward-compatible re-exports so existing `from tools.system import ...` still works.
from tools.shell import (  # noqa: F401
    _load_app_prefs_tools,
    _mugi_rich_path,
    _safe_subprocess_env,
    complete_shell_confirm,
    complete_shell_confirm_file,
    complete_ytdl_cookies_file,
    get_pending_shell_confirm,
    get_pending_shell_confirm_from_file,
    get_pending_ytdl_cookies_from_file,
    shell_execute,
    wait_for_shell_confirm,
    wait_for_ytdl_cookies_choice,
)
from tools.code_execution import (  # noqa: F401
    _get_agent_venv_python,
    _run_python_code_script,
    run_python_code,
)
from tools.screenshots import capture_screenshot  # noqa: F401


TEXT_PATTERNS = [
    "*.txt", "*.md", "*.markdown", "*.rst", "*.py", "*.js", "*.ts", "*.jsx", "*.tsx",
    "*.html", "*.htm", "*.css", "*.json", "*.yaml", "*.yml", "*.toml", "*.ini",
    "*.csv", "*.log", "*.sql", "*.sh", "*.bat", "*.ps1", "*.xml", "*.cfg", "*.conf",
]


def _coerce_search_filesystem_query_and_pattern(query: str, file_pattern: str) -> tuple[str, str]:
    """LLMs often put shell globs in `query` (regex slot). Normalize obvious cases.

    - Fixes tokenization gaps like ``*. toml`` → ``*.toml``.
    - When ``file_pattern`` is default and ``query`` looks like a glob (``*``/``?``)
      and not like a regex, move it to ``file_pattern`` and set ``query`` to ``.*``.
    """
    q = (query or "").strip()
    fp = (file_pattern or "*").strip() or "*"
    if not q:
        return ".*", fp
    q = re.sub(r"\*\.\s+", "*.", q)
    if fp not in ("*", ""):
        return q, fp
    if "*" not in q and "?" not in q:
        return q, fp
    if re.search(r"[|+^$\\()[\]{}]", q):
        return q, fp
    return ".*", q


@tool
def get_current_directory() -> str:
    """Return the current working directory path."""
    try:
        try:
            from utils.kanban_dispatcher import current_kanban_worker_cwd

            preferred = current_kanban_worker_cwd()
            if preferred and os.path.isdir(preferred):
                return f"Current working directory: {preferred}"
        except Exception:
            pass
        cwd = os.getcwd()
        return f"Current working directory: {cwd}"
    except Exception as e:
        _debug_log(f"[ERROR get_current_directory] {e}", "ERROR")
        return f"Failed to get current directory: {str(e)}"


_DISPLAY_IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp")
_DISPLAY_VIDEO_EXTS = (".mp4", ".webm", ".mov", ".avi", ".m4v")
_DISPLAY_HTML_EXTS = (".html", ".htm")
_DISPLAY_TEXT_EXTS = (".txt",)


def _display_content_type_for_extension(ext: str) -> str | None:
    ext = (ext or "").lower()
    if ext in _DISPLAY_VIDEO_EXTS:
        return "video"
    if ext in _DISPLAY_IMAGE_EXTS:
        return "image"
    if ext in _DISPLAY_HTML_EXTS:
        return "html"
    if ext in _DISPLAY_TEXT_EXTS:
        return "text"
    return None


@tool
def show_display_window(filename: str, title: str = "Display") -> str:
    """Open a dedicated display window to show a preview file to the user.

    Use this when you want to present a visualization, HTML report, text output,
    or video prominently rather than only inline in the chat.
    The file must be in ~/Documents/Mugi (e.g. output from run_python_code).
    Supports images, video, HTML (.html, .htm), and plain text (.txt).

    Args:
        filename: Basename of the file in ~/Documents/Mugi (e.g. 'chart.png', 'report.html').
        title: Optional title for the window (e.g. 'Network Traffic').
    """
    _debug_log(f"[show_display_window] filename={filename!r}, title={title!r}", "INFO")
    if not filename or not isinstance(filename, str):
        return "Invalid input: filename must be a non-empty string."
    base = os.path.basename(filename.strip())
    if not base:
        return "Invalid filename."
    _file_output_dir = get_user_output_dir()
    path = os.path.join(_file_output_dir, base)
    if not os.path.isfile(path):
        return f"File not found: {base}. Save to ~/Documents/Mugi first (e.g. run_python_code with plt.savefig(FILE_OUTPUT_DIR + '/name.png'))."
    ext = os.path.splitext(base)[1].lower()
    content_type = _display_content_type_for_extension(ext)
    if content_type is None:
        return f"Unsupported format: {ext}. Use image, video, HTML (.html, .htm), or text (.txt)."
    emit_chat_event({"type": "display_window", "files": [base], "title": title or "Display", "content_type": content_type})
    return f"Display window opened: {title}"


@tool
def search_filesystem(
    root_path: str,
    query: str = ".*",
    max_results: int = 20,
    case_sensitive: bool = False,
    max_depth: int = 8,
    include_hidden: bool = False,
    list_only: bool = False,
    file_pattern: str = "*",
) -> list:
    """Search for files or text inside files from a root path. Use list_only=True for fast listing. file_pattern e.g. '*.py','*.md'. root_path must be specific (e.g. ~/Documents)."""
    query, file_pattern = _coerce_search_filesystem_query_and_pattern(query, file_pattern)
    _debug_log(f"[search_filesystem] query={query!r}, root={root_path}, list_only={list_only}, pattern={file_pattern}")
    abs_root = os.path.expanduser(root_path)
    if not os.path.isdir(abs_root):
        return [{"error": f"Not a directory: {abs_root}"}]
    broad_roots = ["~", os.path.expanduser("~"), "/", "/Users", "/Volumes"]
    resolved_root = str(Path(abs_root).resolve())
    if resolved_root in [str(Path(b).resolve()) for b in broad_roots]:
        return [{"error": "Search root is too broad. Specify a narrower folder like ~/Documents or ~/Projects."}]
    if is_dangerous_path(abs_root):
        return [{"error": f"Blocked: root path is in a protected location: {abs_root}"}]
    emit_live_view_preview(
        f"Searching in {os.path.basename(abs_root) or abs_root}…",
        tool_name="search_filesystem",
    )
    try:
        pattern = re.compile(query, 0 if case_sensitive else re.IGNORECASE)
    except re.error:
        pattern = re.compile(re.escape(query), 0 if case_sensitive else re.IGNORECASE)

    def should_search_file(p: Path) -> bool:
        if not include_hidden and p.name.startswith("."):
            return False
        if file_pattern and file_pattern.strip() and file_pattern != "*":
            patterns = [pat.strip() for pat in file_pattern.split(",") if pat.strip()]
            if patterns:
                return any(fnmatch.fnmatch(p.name, pat) for pat in patterns)
        return any(fnmatch.fnmatch(p.name, pat) for pat in TEXT_PATTERNS)

    results = []
    lock = Lock()
    stop_event = Event()
    processed_files = 0
    total_files_estimate = 0
    start_time = time.time()
    last_update_time = start_time

    def update_progress():
        nonlocal processed_files, total_files_estimate, last_update_time
        while not stop_event.is_set():
            now = time.time()
            if now - last_update_time >= 2.0:
                pct = min(99, int((processed_files / max(1, total_files_estimate)) * 100)) if total_files_estimate else 0
                elapsed = now - start_time
                status = f"Scanning... {pct}% ({processed_files} files) | {int(elapsed)}s"
                _emit_input_status(status)
                last_update_time = now
            time.sleep(0.5)

    progress_thread = Thread(target=update_progress, daemon=True)
    progress_thread.start()
    candidates = []
    file_count = 0
    for root, dirs, files in os.walk(abs_root, topdown=True):
        depth = len(Path(root).relative_to(abs_root).parts)
        if depth > max_depth:
            del dirs[:]
            continue
        if not include_hidden:
            dirs[:] = [d for d in dirs if not d.startswith(".")]
        for file in files:
            fp = os.path.join(root, file)
            p = Path(fp)
            if should_search_file(p):
                candidates.append(fp)
                file_count += 1
                if file_count % 500 == 0:
                    _emit_input_status(f"Discovering files... {file_count} found")
                if len(candidates) > 1500:
                    break
        if len(candidates) > 1500:
            break
    total_files_estimate = max(1, len(candidates))

    def search_one_file(file_path: str):
        nonlocal processed_files
        try:
            p = Path(file_path)
            if pattern.search(p.name):
                with lock:
                    if len(results) < max_results and not stop_event.is_set():
                        results.append({"path": str(p), "line": 0, "snippet": "[filename match]", "match_type": "filename"})
                        if len(results) >= max_results:
                            stop_event.set()
            if list_only:
                with lock:
                    processed_files += 1
                return
            with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
                for line_num, line in enumerate(f, 1):
                    if stop_event.is_set():
                        return
                    if pattern.search(line):
                        snippet = line.strip()[:180]
                        if len(snippet) < len(line.strip()):
                            snippet += "..."
                        with lock:
                            if len(results) < max_results:
                                results.append({"path": str(p), "line": line_num, "snippet": snippet, "match_type": "content"})
                                if len(results) >= max_results:
                                    stop_event.set()
                                    return
            with lock:
                processed_files += 1
        except Exception:
            pass

    max_workers = min(12, (os.cpu_count() or 4) * 2)
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        for future in as_completed(
            executor.submit(contextvars.copy_context().run, search_one_file, fp)
            for fp in candidates
        ):
            try:
                future.result()
            except Exception as e:
                _debug_log(f"[WARN search_filesystem] Worker error: {e}")
    stop_event.set()
    progress_thread.join(timeout=1.0)
    elapsed = time.time() - start_time
    _emit_input_status(f"Search complete: {len(results)} matches in {processed_files} files ({elapsed:.1f}s)")
    emit_live_view_preview(
        f"Found {len(results)} match(es) in {processed_files} files ({elapsed:.1f}s)",
        tool_name="search_filesystem",
    )
    results.sort(key=lambda x: (0 if x.get("match_type") == "filename" else 1, x.get("path", ""), x.get("line", 0)))
    return results[:max_results]


@tool
def compress_files(files: list[str], output_zip: str) -> str:
    """Compress a list of files into a ZIP archive. Input: list of file paths and output ZIP path."""
    if not files or not isinstance(files, list) or not all(isinstance(f, str) for f in files):
        return "Invalid input: 'files' must be a non-empty list of strings."
    abs_out = os.path.abspath(os.path.expanduser(output_zip.strip()))
    if is_dangerous_path(abs_out):
        return f"Blocked: output path is in a protected location: {abs_out}"
    if is_plugins_dir_write_blocked(abs_out):
        return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
    valid_files = []
    for f in files:
        abs_f = os.path.abspath(os.path.expanduser(f.strip()))
        if is_dangerous_path(abs_f):
            continue
        if not os.path.isfile(abs_f):
            continue
        valid_files.append(abs_f)
    if not valid_files:
        return "No valid files to compress."
    n_files = len(valid_files)
    emit_live_view_preview(
        f"Compressing {n_files} file(s)…",
        tool_name="compress_files",
        progress_pct=0,
    )
    try:
        with zipfile.ZipFile(abs_out, "w", zipfile.ZIP_DEFLATED) as zf:
            for idx, f in enumerate(valid_files):
                zf.write(f, os.path.basename(f))
                pct = int(100 * (idx + 1) / n_files) if n_files else 100
                emit_live_view_preview(
                    f"Adding file {idx + 1}/{n_files}…",
                    tool_name="compress_files",
                    progress_pct=pct,
                )
        emit_live_view_preview(
            f"Compressed to: {os.path.basename(abs_out)}",
            tool_name="compress_files",
            progress_pct=100,
        )
        return f"Compressed {len(valid_files)} files to: {abs_out}"
    except Exception as e:
        _debug_log(f"[ERROR compress_files] {e}", "ERROR")
        return f"Compression failed: {str(e)}"


@tool
def extract_archive(archive_path: str, output_dir: str) -> str:
    """Extract a ZIP archive to a directory. Creates output dir if needed. Overwrites existing files."""
    abs_arch = os.path.abspath(os.path.expanduser(archive_path.strip()))
    abs_out = os.path.abspath(os.path.expanduser(output_dir.strip()))
    if is_dangerous_path(abs_arch) or is_dangerous_path(abs_out):
        return "Blocked: archive or output path is in a protected location."
    if is_plugins_dir_write_blocked(abs_out):
        return "Blocked: cannot write to the plugins directory. Use create_plugin (evolution mode) to add plugins."
    if not os.path.isfile(abs_arch):
        return f"Not a file or not found: {abs_arch}"
    emit_live_view_preview(
        f"Extracting {os.path.basename(abs_arch)}…",
        tool_name="extract_archive",
        progress_pct=0,
    )
    os.makedirs(abs_out, exist_ok=True)
    try:
        with zipfile.ZipFile(abs_arch, "r") as zf:
            members = zf.infolist()
            n = len(members) if members else 1
            real_out = os.path.realpath(abs_out)
            from utils.user_model import enforce_low_level_user_model_write_guard

            for idx, member in enumerate(members):
                target = os.path.realpath(os.path.join(abs_out, member.filename))
                if not target.startswith(real_out + os.sep) and target != real_out:
                    continue
                _um_err = enforce_low_level_user_model_write_guard(target, mode="overwrite")
                if _um_err:
                    return _um_err
                zf.extract(member, abs_out)
                pct = int(100 * (idx + 1) / n) if n else 100
                emit_live_view_preview(
                    f"Extracting {idx + 1}/{n}…",
                    tool_name="extract_archive",
                    progress_pct=pct,
                )
        _IMAGE_EXTENSIONS = (".png", ".jpg", ".jpeg", ".gif", ".webp")
        first_image_path = None
        for _root, _dirs, files in os.walk(abs_out):
            for name in sorted(files):
                if os.path.splitext(name)[1].lower() in _IMAGE_EXTENSIONS:
                    first_image_path = os.path.join(_root, name)
                    break
            if first_image_path:
                break
        b64 = None
        if first_image_path and os.path.isfile(first_image_path) and os.path.getsize(first_image_path) < 10 * 1024 * 1024:
            try:
                with open(first_image_path, "rb") as _f:
                    b64 = image_to_preview_base64(_f.read(), max_width=400)
            except Exception:
                pass
        emit_live_view_preview(
            f"Extracted to: {os.path.basename(abs_out)}",
            tool_name="extract_archive",
            progress_pct=100,
            preview_image_base64=b64,
        )
        return f"Extracted to: {abs_out}"
    except zipfile.BadZipFile:
        return "Invalid or corrupt ZIP file."
    except Exception as e:
        _debug_log(f"[ERROR extract_archive] {e}", "ERROR")
        return f"Extraction failed: {str(e)}"


@tool
def get_current_time(include_timezone: bool = True) -> str:
    """Return the current local date and time. Use when the user asks for the time, or when reasoning about schedules, deadlines, reminders, or heartbeat timing (e.g. 'in 2 hours', 'by end of day', 'is it morning?')."""
    now = datetime.now().astimezone()
    if include_timezone:
        tz_name = now.strftime("%Z") or now.strftime("%z")
        return now.strftime(f"%Y-%m-%d %H:%M:%S {tz_name}")
    return now.strftime("%Y-%m-%d %H:%M:%S")


@tool
def inspect_system_activity(focus: str = "overview") -> str:
    """Inspect live machine activity to diagnose slowness, busy network, memory pressure, or full disks.

    Read-only and safe (psutil + macOS nettop/pmset, no sudo). Takes a short live
    measurement (~1-2s), so results reflect what is happening RIGHT NOW.

    focus options:
    - "network": per-process download/upload rates — answers "what is downloading/uploading right now?"
    - "cpu": top CPU consumers, load average vs cores, thermal throttling — answers "why is my computer slow?"
    - "memory": memory pressure, swap usage, top RAM consumers
    - "disk": current read/write throughput and per-volume usage
    - "overview": combined CPU + memory + disk snapshot (no per-process network)
    """
    try:
        from dashboard_bundled.system_inspect import inspect as _inspect

        return _inspect(focus)
    except ImportError:
        return "System inspection requires the psutil package. Re-run setup to update the backend environment."
    except Exception as e:
        _debug_log(f"[ERROR inspect_system_activity] {e}", "ERROR")
        return f"System inspection failed: {e}"


@tool
def run_host_diagnostics(domain: str, target: str = "", since_minutes: int = 10) -> str:
    """Diagnose a broken thing on this Mac and name the failing layer with a fix.

    Read-only and safe (no sudo): walks layered checks and returns findings plus a
    plain-language verdict. Use when something is BROKEN (vs. inspect_system_activity
    for slow/busy questions).

    domain options:
    - "network": "why isn't my internet working?" — checks Wi-Fi/link, DHCP, router,
      DNS, captive portal, and external reachability; stops at the first broken layer.
    - "app": "why won't this app launch / why does it crash?" — set target to the app
      name; reads recent crash reports and error logs. If nothing is found, ask the
      user to reproduce the failure (launch the app again) and then call this again.
    - "printing": "why won't this print?" — checks for paused queues, stuck jobs, and
      printer reachability. target optionally names a printer (default: system default).
    - "power": "why won't my Mac sleep / battery worries?" — names the app holding a
      sleep assertion; battery health, cycle count, and charging state.
    - "bluetooth": "my AirPods/mouse/keyboard won't connect" — set target to the device
      name; distinguishes Bluetooth-off vs paired-but-disconnected vs never-paired.
    - "audio": "I have no sound" — mute/volume state and where sound is actually routed
      (Bluetooth headphones, virtual loopback device, speakers).
    - "displays": "my external monitor isn't working" — not-detected vs mirrored vs
      healthy; target optionally names the monitor.
    - "storage": "my disk is full / what's eating my space?" — capacity, biggest
      files, known space hogs, and Time Machine snapshots holding freed space.
      Set target to a folder path (e.g. "~/Movies") to drill one level into it.

    Fixes that change state (resume a print queue, cancel a job, renew DHCP) must go
    through shell_execute so the user approves them first.
    """
    try:
        from dashboard_bundled.host_diagnostics import diagnose

        return diagnose(domain, target=target, since_minutes=since_minutes)
    except Exception as e:
        _debug_log(f"[ERROR run_host_diagnostics] {e}", "ERROR")
        return f"Diagnostics failed: {e}"


__all__ = [
    "_load_app_prefs_tools",
    "get_pending_shell_confirm",
    "complete_shell_confirm",
    "get_pending_shell_confirm_from_file",
    "complete_shell_confirm_file",
    "get_pending_ytdl_cookies_from_file",
    "complete_ytdl_cookies_file",
    "wait_for_shell_confirm",
    "wait_for_ytdl_cookies_choice",
    "shell_execute",
    "get_current_directory",
    "run_python_code",
    "capture_screenshot",
    "show_display_window",
    "search_filesystem",
    "compress_files",
    "extract_archive",
    "get_current_time",
    "inspect_system_activity",
    "run_host_diagnostics",
]
