from __future__ import annotations

import calendar
import math
from datetime import date
from types import SimpleNamespace

import pytest

from core.app.entities.app_invoke_entities import InvokeFrom
from core.tools.__base.tool_runtime import ToolRuntime
from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
from core.tools.builtin_tool.providers.audio.audio import AudioToolProvider
from core.tools.builtin_tool.providers.audio.tools.asr import ASRTool
from core.tools.builtin_tool.providers.audio.tools.tts import TTSTool
from core.tools.builtin_tool.providers.code.code import CodeToolProvider
from core.tools.builtin_tool.providers.code.tools.simple_code import SimpleCode
from core.tools.builtin_tool.providers.time.time import WikiPediaProvider
from core.tools.builtin_tool.providers.time.tools.current_time import CurrentTimeTool
from core.tools.builtin_tool.providers.time.tools.localtime_to_timestamp import LocaltimeToTimestampTool
from core.tools.builtin_tool.providers.time.tools.timestamp_to_localtime import TimestampToLocaltimeTool
from core.tools.builtin_tool.providers.time.tools.timezone_conversion import TimezoneConversionTool
from core.tools.builtin_tool.providers.time.tools.weekday import WeekdayTool
from core.tools.builtin_tool.providers.webscraper.tools.webscraper import WebscraperTool
from core.tools.builtin_tool.providers.webscraper.webscraper import WebscraperProvider
from core.tools.builtin_tool.tool import BuiltinTool
from core.tools.entities.common_entities import I18nObject
from core.tools.entities.tool_entities import ToolEntity, ToolIdentity, ToolInvokeMessage
from core.tools.errors import ToolInvokeError
from graphon.file import FileType
from graphon.model_runtime.entities.model_entities import ModelPropertyKey


def _build_builtin_tool(tool_cls: type[BuiltinTool]) -> BuiltinTool:
    entity = ToolEntity(
        identity=ToolIdentity(
            author="author",
            name="tool-a",
            label=I18nObject(en_US="tool-a"),
            provider="provider-a",
        ),
        parameters=[],
    )
    runtime = ToolRuntime(tenant_id="tenant-1", invoke_from=InvokeFrom.DEBUGGER)
    return tool_cls(provider="provider-a", entity=entity, runtime=runtime)


def _raise_runtime_error(*_args: object, **_kwargs: object) -> None:
    raise RuntimeError("boom")


def test_current_time_tool():
    current_tool = _build_builtin_tool(CurrentTimeTool)
    utc_text = list(current_tool.invoke(user_id="u", tool_parameters={"timezone": "UTC"}))[0].message.text
    assert utc_text

    invalid_tz = list(current_tool.invoke(user_id="u", tool_parameters={"timezone": "Invalid/TZ"}))[0].message.text
    assert "Invalid timezone" in invalid_tz


def test_localtime_to_timestamp_tool():
    localtime_tool = _build_builtin_tool(LocaltimeToTimestampTool)
    ts_message = list(
        localtime_tool.invoke(user_id="u", tool_parameters={"localtime": "2024-01-01 10:00:00", "timezone": "UTC"})
    )[0].message.text
    ts_value = float(ts_message.strip())
    assert math.isfinite(ts_value)
    assert ts_value >= 0
    with pytest.raises(ToolInvokeError):
        LocaltimeToTimestampTool.localtime_to_timestamp("bad", "%Y-%m-%d %H:%M:%S", "UTC")


def test_timestamp_to_localtime_tool():
    to_local_tool = _build_builtin_tool(TimestampToLocaltimeTool)
    local_text = list(to_local_tool.invoke(user_id="u", tool_parameters={"timestamp": 1704067200, "timezone": "UTC"}))[
        0
    ].message.text
    assert "2024" in local_text
    with pytest.raises(ToolInvokeError):
        TimestampToLocaltimeTool.timestamp_to_localtime("bad", "UTC")  # type: ignore[arg-type]


def test_timezone_conversion_tool():
    timezone_tool = _build_builtin_tool(TimezoneConversionTool)
    converted = list(
        timezone_tool.invoke(
            user_id="u",
            tool_parameters={
                "current_time": "2024-01-01 08:00:00",
                "current_timezone": "UTC",
                "target_timezone": "Asia/Tokyo",
            },
        )
    )[0].message.text
    assert converted.startswith("2024-01-01")
    with pytest.raises(ToolInvokeError):
        TimezoneConversionTool.timezone_convert("bad", "UTC", "Asia/Tokyo")


def test_weekday_tool():
    weekday_tool = _build_builtin_tool(WeekdayTool)
    valid = list(weekday_tool.invoke(user_id="u", tool_parameters={"year": 2024, "month": 1, "day": 1}))[0].message.text
    expected_date = date(2024, 1, 1)
    expected_message = (
        f"{calendar.month_name[expected_date.month]} "
        f"{expected_date.day}, {expected_date.year} "
        f"is {calendar.day_name[expected_date.weekday()]}."
    )
    assert valid == expected_message
    invalid = list(weekday_tool.invoke(user_id="u", tool_parameters={"year": 2024, "month": 2, "day": 31}))[
        0
    ].message.text
    assert "Invalid date" in invalid
    with pytest.raises(ValueError, match="Month is required"):
        list(weekday_tool.invoke(user_id="u", tool_parameters={"year": 2024, "day": 1}))


def test_simple_code_valid_execution(monkeypatch: pytest.MonkeyPatch):
    simple_code = _build_builtin_tool(SimpleCode)

    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.code.tools.simple_code.CodeExecutor.execute_code",
        lambda *a: "ok",
    )
    result = list(
        simple_code.invoke(
            user_id="u",
            tool_parameters={"language": "python3", "code": "print(1)"},
        )
    )[0].message.text
    assert result == "ok"


def test_simple_code_invalid_language():
    simple_code = _build_builtin_tool(SimpleCode)

    with pytest.raises(ValueError, match="Only python3 and javascript"):
        list(simple_code.invoke(user_id="u", tool_parameters={"language": "go", "code": "fmt.Println(1)"}))


def test_simple_code_execution_error(monkeypatch: pytest.MonkeyPatch):
    simple_code = _build_builtin_tool(SimpleCode)

    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.code.tools.simple_code.CodeExecutor.execute_code",
        _raise_runtime_error,
    )
    with pytest.raises(ToolInvokeError, match="boom"):
        list(simple_code.invoke(user_id="u", tool_parameters={"language": "python3", "code": "print(1)"}))


def test_webscraper_empty_url():
    webscraper = _build_builtin_tool(WebscraperTool)
    empty = list(webscraper.invoke(user_id="u", tool_parameters={"url": ""}))[0].message.text
    assert empty == "Please input url"


def test_webscraper_fetch(monkeypatch: pytest.MonkeyPatch):
    webscraper = _build_builtin_tool(WebscraperTool)
    monkeypatch.setattr("core.tools.builtin_tool.providers.webscraper.tools.webscraper.get_url", lambda *a, **k: "page")
    full = list(webscraper.invoke(user_id="u", tool_parameters={"url": "https://example.com"}))[0].message.text
    assert full == "page"


def test_webscraper_summary(monkeypatch: pytest.MonkeyPatch):
    webscraper = _build_builtin_tool(WebscraperTool)
    monkeypatch.setattr("core.tools.builtin_tool.providers.webscraper.tools.webscraper.get_url", lambda *a, **k: "page")
    monkeypatch.setattr(webscraper, "summary", lambda user_id, content: "summary")
    summarized = list(
        webscraper.invoke(
            user_id="u",
            tool_parameters={"url": "https://example.com", "generate_summary": True},
        )
    )[0].message.text
    assert summarized == "summary"


def test_webscraper_fetch_error(monkeypatch: pytest.MonkeyPatch):
    webscraper = _build_builtin_tool(WebscraperTool)
    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.webscraper.tools.webscraper.get_url",
        _raise_runtime_error,
    )
    with pytest.raises(ToolInvokeError, match="boom"):
        list(webscraper.invoke(user_id="u", tool_parameters={"url": "https://example.com"}))


def test_asr_invalid_file():
    asr = _build_builtin_tool(ASRTool)
    file_obj = SimpleNamespace(type=FileType.DOCUMENT)
    invalid_file = list(asr.invoke(user_id="u", tool_parameters={"audio_file": file_obj}))[0].message.text
    assert "not a valid audio file" in invalid_file


def test_asr_valid_file_invocation(monkeypatch: pytest.MonkeyPatch):
    asr = _build_builtin_tool(ASRTool)
    model_instance = type("M", (), {"invoke_speech2text": lambda self, file: "transcript"})()
    model_manager = type("Mgr", (), {"get_model_instance": lambda *a, **k: model_instance})()
    monkeypatch.setattr("core.tools.builtin_tool.providers.audio.tools.asr.download", lambda file: b"audio-bytes")
    captured_manager_kwargs = {}

    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.audio.tools.asr.ModelManager.for_tenant",
        lambda **kwargs: captured_manager_kwargs.update(kwargs) or model_manager,
    )
    audio_file = SimpleNamespace(type=FileType.AUDIO)
    ok = list(asr.invoke(user_id="u", tool_parameters={"audio_file": audio_file, "model": "p#m"}))[0].message.text
    assert ok == "transcript"
    assert captured_manager_kwargs == {"tenant_id": "tenant-1", "user_id": "u"}


def test_asr_available_models_and_runtime_parameters(monkeypatch: pytest.MonkeyPatch):
    asr = _build_builtin_tool(ASRTool)
    provider_model = type("PM", (), {"provider": "p", "models": [type("Model", (), {"model": "m"})()]})()
    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.audio.tools.asr.ModelProviderService.get_models_by_model_type",
        lambda *a, **k: [provider_model],
    )
    assert asr.get_available_models() == [("p", "m")]
    assert asr.get_runtime_parameters()[0].name == "model"


def test_tts_invoke_returns_messages(monkeypatch: pytest.MonkeyPatch):
    tts = _build_builtin_tool(TTSTool)
    captured_manager_kwargs = {}
    voices_model_instance = type(
        "TTSM",
        (),
        {
            "get_tts_voices": lambda self: [{"value": "voice-1"}],
            "invoke_tts": lambda self, **kwargs: [b"a", b"b"],
        },
    )()
    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.audio.tools.tts.ModelManager.for_tenant",
        lambda **kwargs: (
            captured_manager_kwargs.update(kwargs)
            or type("M", (), {"get_model_instance": lambda *a, **k: voices_model_instance})()
        ),
    )
    messages = list(tts.invoke(user_id="u", tool_parameters={"model": "p#m", "text": "hello"}))
    assert [m.type for m in messages] == [ToolInvokeMessage.MessageType.TEXT, ToolInvokeMessage.MessageType.BLOB]
    assert captured_manager_kwargs == {"tenant_id": "tenant-1", "user_id": "u"}


def test_tts_get_available_models_requires_runtime():
    tts = _build_builtin_tool(TTSTool)
    tts.runtime = None
    with pytest.raises(ValueError, match="Runtime is required"):
        tts.get_available_models()


def test_tts_tool_raises_when_runtime_missing():
    tts = _build_builtin_tool(TTSTool)
    tts.runtime = None
    with pytest.raises(ValueError, match="Runtime is required"):
        list(tts.invoke(user_id="u", tool_parameters={"model": "p#m", "text": "hello"}))


@pytest.mark.parametrize(
    "voices",
    [[{"value": None}], []],
)
def test_tts_tool_raises_when_voice_unavailable(monkeypatch, voices):
    tts = _build_builtin_tool(TTSTool)
    tts.runtime = ToolRuntime(tenant_id="tenant-1", invoke_from=InvokeFrom.DEBUGGER)
    model_without_voice = type(
        "TTSModelNoVoice",
        (),
        {
            "get_tts_voices": lambda self: voices,
            "invoke_tts": lambda self, **kwargs: [b"x"],
        },
    )()
    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.audio.tools.tts.ModelManager.for_tenant",
        lambda **_: type("Manager", (), {"get_model_instance": lambda *args, **kwargs: model_without_voice})(),
    )
    with pytest.raises(ValueError, match="no voice available"):
        list(tts.invoke(user_id="u", tool_parameters={"model": "p#m", "text": "hello"}))


def test_tts_tool_get_available_models_and_runtime_parameters(monkeypatch: pytest.MonkeyPatch):
    tts = _build_builtin_tool(TTSTool)

    model_1 = SimpleNamespace(
        model="model-a",
        model_properties={ModelPropertyKey.VOICES: [{"mode": "v1", "name": "Voice 1"}]},
    )
    model_2 = SimpleNamespace(model="model-b", model_properties={})
    provider_models = [SimpleNamespace(provider="provider-a", models=[model_1, model_2])]
    monkeypatch.setattr(
        "core.tools.builtin_tool.providers.audio.tools.tts.ModelProviderService.get_models_by_model_type",
        lambda *args, **kwargs: provider_models,
    )

    available_models = tts.get_available_models()
    assert available_models == [
        ("provider-a", "model-a", [{"mode": "v1", "name": "Voice 1"}]),
        ("provider-a", "model-b", []),
    ]

    runtime_parameters = tts.get_runtime_parameters()
    assert runtime_parameters[0].name == "model"
    assert runtime_parameters[0].required is True
    assert runtime_parameters[0].options[0].value == "provider-a#model-a"
    assert runtime_parameters[1].name == "voice#provider-a#model-a"


def test_provider_classes_and_builtin_sort(monkeypatch: pytest.MonkeyPatch):
    # Use object.__new__ to avoid YAML-loading __init__; only pass-through validation is exercised.
    # Ensure pass-through _validate_credentials methods are executed.
    AudioToolProvider._validate_credentials(object.__new__(AudioToolProvider), "u", {})
    CodeToolProvider._validate_credentials(object.__new__(CodeToolProvider), "u", {})
    WikiPediaProvider._validate_credentials(object.__new__(WikiPediaProvider), "u", {})
    WebscraperProvider._validate_credentials(object.__new__(WebscraperProvider), "u", {})

    providers = [SimpleNamespace(name="b"), SimpleNamespace(name="a")]
    monkeypatch.setattr(BuiltinToolProviderSort, "_position", {})
    monkeypatch.setattr(
        "core.tools.builtin_tool.providers._positions.get_tool_position_map",
        lambda _: {"a": 0, "b": 1},
    )
    monkeypatch.setattr(
        "core.tools.builtin_tool.providers._positions.sort_by_position_map",
        lambda position, values, name_func: sorted(values, key=lambda x: name_func(x)),
    )
    sorted_providers = BuiltinToolProviderSort.sort(providers)
    assert [p.name for p in sorted_providers] == ["a", "b"]
