# -*- coding: utf-8 -*-
# flake8: noqa: E501
# pylint: disable=line-too-long
import os
from pathlib import Path
from typing import Optional

from agentscope.message import TextBlock, ToolResultState
from agentscope.tool import ToolChunk

from .utils import (
    build_truncation_metadata,
    truncate_text_output,
    read_file_safe,
    DEFAULT_MAX_BYTES,
    TRUNCATION_METADATA_KEY,
)
from ...config.context import (
    get_current_workspace_dir,
    get_current_recent_max_bytes,
)
from ...constant import WORKING_DIR
from ...runtime.tool_registry import tool_descriptor
from ...utils.io_utils import (
    append_text_async,
    get_path_lock,
    write_text_atomic_async,
)

_USER_FILE_MODE = 0o644


def _path_to_file_url(path: str) -> str:
    """Convert a local file path to a ``file://`` URL.

    Does NOT percent-encode non-ASCII characters because agentscope's
    DashScope formatter extracts the local path from the URL without
    ``unquote()``, causing ``FileNotFoundError`` for files with
    non-ASCII names (e.g. Chinese characters).
    """
    abs_path = os.path.abspath(path)
    if os.name == "nt":
        abs_path = abs_path.replace("\\", "/")

    if os.name == "nt":
        if abs_path.startswith("//"):
            return f"file:{abs_path}"
        return f"file:///{abs_path}"
    return f"file://{abs_path}"


def _resolve_file_path(file_path: str) -> str:
    """Resolve file path: use absolute path as-is,
    resolve relative path from current workspace or WORKING_DIR.

    Args:
        file_path: The input file path (absolute or relative).

    Returns:
        The resolved absolute file path as string.
    """
    path = Path(file_path).expanduser()
    if path.is_absolute():
        return str(path)
    else:
        # Use current workspace_dir from context, fallback to WORKING_DIR
        workspace_dir = get_current_workspace_dir() or WORKING_DIR
        return str(workspace_dir / file_path)


def _get_encoding_for_file(file_path: str) -> str:
    """Determine the appropriate encoding for a file based on its type.

    For cross-platform compatibility, especially with Windows Excel/Notepad:
    - CSV/TSV/TXT files: Use UTF-8-BOM (Windows Excel needs BOM to detect UTF-8)
    - All other files: Use UTF-8 (safer default, no BOM)

    Args:
        file_path: Path to the file

    Returns:
        Encoding string: "utf-8-sig" or "utf-8"
    """
    suffix = Path(file_path).suffix.lower()

    # Files that need BOM for Windows compatibility
    if suffix in {".csv", ".tsv", ".tab", ".txt", ".log"}:
        return "utf-8-sig"

    # Default: UTF-8 without BOM (safe for all other files)
    # This includes: .sh, .yaml, .json, .py, .js, .md, etc.
    return "utf-8"


@tool_descriptor(
    requires_sandbox=("file_read",),
    async_execution=True,
    tool_type="file",
    target_param="file_path",
    policy_name="Read",
    default_policy="allow",
    policy_reason="Read-only file access (global)",
    ui_description="Read file contents",
    ui_icon="📄",
)
async def read_file(  # pylint: disable=too-many-return-statements
    file_path: str,
    start_line: Optional[int] = None,
    end_line: Optional[int] = None,
) -> ToolChunk:
    """Read a file. Relative paths resolve from WORKING_DIR.

    Use start_line/end_line to read a specific line range (output includes
    line numbers). Omit both to read the full file.

    Args:
        file_path (`str`):
            Path to the file.
        start_line (`int`, optional):
            First line to read (1-based, inclusive).
        end_line (`int`, optional):
            Last line to read (1-based, inclusive).
    """

    # Convert start_line/end_line to int if they are strings
    if start_line is not None:
        try:
            start_line = int(start_line)
        except (ValueError, TypeError):
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: start_line must be an integer, got {start_line!r}.",
                    ),
                ],
            )

    if end_line is not None:
        try:
            end_line = int(end_line)
        except (ValueError, TypeError):
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: end_line must be an integer, got {end_line!r}.",
                    ),
                ],
            )

    file_path = _resolve_file_path(file_path)

    try:
        content = await read_file_safe(file_path)
        all_lines = content.split("\n")
        total = len(all_lines)

        # Determine read range
        s = max(1, start_line if start_line is not None else 1)
        e = min(total, end_line if end_line is not None else total)

        if s > total:
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: start_line {s} exceeds file length ({total} lines).",
                    ),
                ],
            )

        if s > e:
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: start_line ({s}) > end_line ({e}).",
                    ),
                ],
            )

        # Extract selected lines
        selected_content = "\n".join(all_lines[s - 1 : e])

        # Apply smart truncation (consistent with shell output format)
        max_bytes = get_current_recent_max_bytes() or DEFAULT_MAX_BYTES
        text, metadata = truncate_text_output(
            selected_content,
            start_line=s,
            total_lines=total,
            file_path=file_path,
            file_size_bytes=len(content.encode("utf-8")),
            max_bytes=max_bytes,
        )

        # Add continuation hint if partial read without truncation.
        # Use TRUNCATION_NOTICE_MARKER format so ToolResultPruningMiddleware can
        # re-truncate with the correct start_line when pruning old messages.
        if text == selected_content and e < total:
            content_bytes = len(text.encode("utf-8"))
            metadata = build_truncation_metadata(
                file_path=file_path,
                file_size_bytes=len(content.encode("utf-8")),
                total_lines=total,
                start_line=s,
                max_bytes=content_bytes,
                excerpt_bytes=content_bytes,
                read_from=e + 1,
            )
            text += metadata[TRUNCATION_METADATA_KEY]["0"]["notice"]

        return ToolChunk(
            is_last=True,
            state=ToolResultState.SUCCESS,
            content=[TextBlock(type="text", text=text)],
            metadata=metadata,
        )

    except FileNotFoundError:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text=f"Error: The file {file_path} does not exist.",
                ),
            ],
        )
    except IsADirectoryError:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text=f"Error: The path {file_path} is not a file.",
                ),
            ],
        )
    except Exception as e:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text=f"Error: Read file failed due to \n{e}",
                ),
            ],
        )


@tool_descriptor(
    requires_sandbox=("file_write",),
    async_execution=True,
    tool_type="file",
    target_param="file_path",
    policy_name="Write",
    ui_description="Write content to file",
    ui_icon="✍️",
)
async def write_file(
    file_path: str,
    content: str,
) -> ToolChunk:
    """Create or overwrite a file. Relative paths resolve from WORKING_DIR.

    Args:
        file_path (`str`):
            Path to the file.
        content (`str`):
            Content to write.
    """

    if not file_path:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text="Error: No `file_path` provided.",
                ),
            ],
        )

    file_path = _resolve_file_path(file_path)
    encoding = _get_encoding_for_file(file_path)

    try:
        async with get_path_lock(file_path):
            await write_text_atomic_async(
                file_path,
                content,
                encoding=encoding,
                new_file_mode=_USER_FILE_MODE,
            )
        return ToolChunk(
            is_last=True,
            state=ToolResultState.SUCCESS,
            content=[
                TextBlock(
                    type="text",
                    text=f"Wrote {len(content)} bytes to {file_path}.",
                ),
            ],
        )
    except Exception as e:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text=f"Error: Write file failed due to \n{e}",
                ),
            ],
        )


# pylint: disable=too-many-return-statements
@tool_descriptor(
    requires_sandbox=("file_write",),
    async_execution=True,
    tool_type="file",
    target_param="file_path",
    policy_name="Edit",
    ui_description="Edit file using find-and-replace",
    ui_icon="🖊️",
)
async def edit_file(
    file_path: str,
    old_text: str,
    new_text: str,
) -> ToolChunk:
    """Find-and-replace text in a file. All occurrences of old_text are
    replaced with new_text. Relative paths resolve from WORKING_DIR.

    Args:
        file_path (`str`):
            Path to the file.
        old_text (`str`):
            Exact text to find.
        new_text (`str`):
            Replacement text.
    """

    if not file_path:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text="Error: No `file_path` provided.",
                ),
            ],
        )

    resolved_path = _resolve_file_path(file_path)

    encoding = _get_encoding_for_file(resolved_path)
    async with get_path_lock(resolved_path):
        try:
            content = await read_file_safe(resolved_path)
        except FileNotFoundError:
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=(
                            f"Error: The file {resolved_path} "
                            f"does not exist."
                        ),
                    ),
                ],
            )
        except IsADirectoryError:
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=(
                            f"Error: The path {resolved_path} "
                            f"is not a file."
                        ),
                    ),
                ],
            )
        except Exception as e:
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: Read file failed due to \n{e}",
                    ),
                ],
            )

        if old_text not in content:
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: The text to replace was not found in {file_path}.",
                    ),
                ],
            )

        try:
            new_content = content.replace(old_text, new_text)
            await write_text_atomic_async(
                resolved_path,
                new_content,
                encoding=encoding,
                new_file_mode=_USER_FILE_MODE,
            )
        except Exception as e:
            return ToolChunk(
                is_last=True,
                state=ToolResultState.ERROR,
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: Write file failed due to \n{e}",
                    ),
                ],
            )

    return ToolChunk(
        is_last=True,
        state=ToolResultState.SUCCESS,
        content=[
            TextBlock(
                type="text",
                text=f"Successfully replaced text in {file_path}.",
            ),
        ],
    )


@tool_descriptor(
    requires_sandbox=("file_write",),
    async_execution=True,
    enabled_by_default=False,
    tool_type="file",
    target_param="file_path",
    policy_name="Append",
    ui_description="Append content to a file",
    ui_icon="📎",
)
async def append_file(
    file_path: str,
    content: str,
) -> ToolChunk:
    """Append content to the end of a file. Relative paths resolve from
    WORKING_DIR.

    Args:
        file_path (`str`):
            Path to the file.
        content (`str`):
            Content to append.
    """

    if not file_path:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text="Error: No `file_path` provided.",
                ),
            ],
        )

    file_path = _resolve_file_path(file_path)
    encoding = _get_encoding_for_file(file_path)

    try:
        await append_text_async(
            file_path,
            content,
            encoding=encoding,
        )
        return ToolChunk(
            is_last=True,
            state=ToolResultState.SUCCESS,
            content=[
                TextBlock(
                    type="text",
                    text=f"Appended {len(content)} bytes to {file_path}.",
                ),
            ],
        )
    except Exception as e:
        return ToolChunk(
            is_last=True,
            state=ToolResultState.ERROR,
            content=[
                TextBlock(
                    type="text",
                    text=f"Error: Append file failed due to \n{e}",
                ),
            ],
        )
