# -*- coding: utf-8 -*-
import json
import os
from typing import Any, Dict, List, Optional, Tuple

import aiohttp
from mcp.server.fastmcp import Context
from pydantic import BaseModel, Field

from agentscope_bricks.base.component import Component
from agentscope_bricks.components.searches.modelstudio_search import (
    SEARCH_PAGE,
    SEARCH_STRATEGY_SETTING,
    SEARCH_TIMEOUT,
    FieldValidator,
)
from agentscope_bricks.utils.tracing_utils.wrapper import trace
from agentscope_bricks.utils.mcp_util import MCPUtil

SEARCH_URL = os.getenv(
    "SEARCH_URL",
    "https://dashscope.aliyuncs.com/api/v1/indices/plugin/mcp_search",
)
SEARCH_STRATEGY = os.getenv("SEARCH_STRATEGY", "turbo")
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", None)
SEARCH_RULES = {
    "url": {
        "DROPOUT_ENTIRE_IF_MISSING": "DROPOUT_ENTIRE_IF_MISSING",
        "FILTER_ITEMS_FROM_LIST": [
            "uclink://www.uc.cn",
            "qklink://www.quark.cn",
            "https://page.sm.cn",
        ],
    },
    "title": "DROPOUT_ENTIRE_IF_MISSING",
}


class SearchLiteInput(BaseModel):
    """
    Search Input.
    """

    query: str = Field(..., description="user query in the format of string")
    count: int = Field(default=5, description="number of search results")
    ctx: Optional[Context] = Field(
        default=None,
        description="HTTP request context containing headers",
    )


class SearchLiteOutput(BaseModel):
    """
    Search Output.
    """

    status: int = Field(
        ...,
        description="status code for success or failed, 0 is success",
    )
    pages: List[Dict[str, Any]] = Field(
        ...,
        description="Search results with url, title, etc,.",
    )
    tools: List[Dict[str, Any]] = Field(
        ...,
        description="The tool calling result during search",
    )
    request_id: Optional[str] = Field(
        default=None,
        title="Request ID",
        description="请求ID",
    )


class ModelstudioSearchLite(Component[SearchLiteInput, SearchLiteOutput]):
    """
    Search Component that calling dashscope for llm search result.
    """

    description = "搜索可用于查询百科知识、时事新闻、天气等信息"
    name = "modelstudio_web_search"

    @trace(trace_type="SEARCH", trace_name="modelstudio_search_lite")
    async def _arun(
        self,
        args: SearchLiteInput,
        **kwargs: Any,
    ) -> SearchLiteOutput:
        """
        Run the search component with SearchLiteInput and return
        SearchLiteOutput.
        Args:
            args: in SearchLiteInput format that generated by from llm
            ctx: Optional MCP context containing request information
            **kwargs: should include other arguments if needed,
            'trace_event' in kwargs can be used for tracing_utils

        Returns:SearchLiteOutput

        """

        if DASHSCOPE_API_KEY is None:
            raise ValueError(
                "DASHSCOPE_API_KEY is required for search component, "
                "please get your dashscope api-key on Modelstudio platform",
            )
        trace_event = kwargs.pop("trace_event", None)
        request_id = MCPUtil._get_mcp_dash_request_id(args.ctx)

        # call search engine to get search result
        payload = ModelstudioSearchLite.generate_search_payload(
            search_input=args,
            request_id=request_id,
            rid=request_id,
        )
        header = {
            "Content-Type": "application/json",
            "Accept-Encoding": "utf-8",
            "Authorization": "Bearer " + DASHSCOPE_API_KEY,
            "x-acs-req-uuid": request_id,
        }
        kwargs["context"] = {
            "payload": payload,
            "search_strategy": SEARCH_STRATEGY,
            "timeout": SEARCH_TIMEOUT,
        }

        trace_event.on_log(
            "",
            **{
                "step_suffix": "test_header_inject",
                "payload": {
                    "request_id": request_id,
                    "dash_request_id": request_id,
                    "search_header": header,
                    "search_payload": payload,
                    "search_url": SEARCH_URL,
                },
            },
        )
        try:
            search_result, extra_tool_info, status, result_message = (
                await ModelstudioSearchLite.dashscope_search_kernel(
                    url=SEARCH_URL,
                    payload=payload,
                    headers=header,
                    timeout=SEARCH_TIMEOUT,
                )
            )
            if trace_event:
                trace_event.on_log(
                    "",
                    **{
                        "step_suffix": "results",
                        "payload": {
                            "request_id": request_id,
                            "search_status": status,
                            "search_query_result": result_message,
                            "search_result": search_result,
                            "extra_tool_info": extra_tool_info,
                        },
                    },
                )

        except Exception as e:
            print(f"Error: {e}")
            return SearchLiteOutput(
                status=1,
                pages=[],
                tools=[],
                request_id=request_id,
            )

        # post process search results
        search_items, tool_items = (
            ModelstudioSearchLite.post_process_search_detail(
                search_results=search_result,
                extra_tool_info=extra_tool_info,
            )
        )

        return SearchLiteOutput(
            status=0,
            pages=search_items,
            tools=tool_items,
            request_id=request_id,
        )

    @staticmethod
    def generate_search_payload(
        search_input: SearchLiteInput,
        request_id: str,
        **kwargs: Any,
    ) -> str:
        search_strategy = SEARCH_STRATEGY

        payload = {
            "scene": SEARCH_STRATEGY_SETTING[search_strategy]["scene"],
            "uq": search_input.query,
            "rid": request_id,
            "fields": [],
            "page": int(SEARCH_PAGE),
            "rows": search_input.count,
            "headers": {
                "__d_head_qto": SEARCH_STRATEGY_SETTING[search_strategy][
                    "timeout"
                ],
            },
        }
        result = json.dumps(payload)
        return result

    @staticmethod
    async def dashscope_search_kernel(
        url: str,
        payload: str,
        headers: Dict,
        timeout: int,
        **kwargs: Any,
    ) -> Tuple[List, List, int, str]:

        extra_tool_info = []
        results_list = []
        results = {}
        try:
            # 使用异步HTTP客户端替代同步requests
            timeout_config = aiohttp.ClientTimeout(total=timeout)
            async with aiohttp.ClientSession(
                timeout=timeout_config,
            ) as session:
                async with session.post(
                    url,
                    headers=headers,
                    data=payload,
                ) as response:
                    results = await response.json()

            result_message = results["message"]
            if results["status"] == 0:
                extra_tool_info = results["data"]["extras"].get(
                    "toolResult",
                    [],
                )
                results_list = results["data"]["docs"]
        except Exception as e:
            print(f"Error: {e}")
            raise RuntimeError(
                f"Unable to reach search source with error: {e} and result: "
                f"{results}",
            )

        return results_list, extra_tool_info, results["status"], result_message

    @staticmethod
    def post_process_search_detail(
        search_results: List,
        extra_tool_info: List,
        **kwargs: Any,
    ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
        field_validator = FieldValidator(SEARCH_RULES)
        search_items = []
        tool_items = []

        try:
            for doc in search_results:
                tmp_search_result = {
                    "url": doc.get("url", "") or "",
                    "title": doc.get("title", "") or "",
                    "icon": doc.get("hostlogo", "") or "",
                    "site_name": doc.get("hostname", "") or "",
                    "image": doc.get("image", "") or "",
                }
                filtered_search_result = field_validator.validate(
                    tmp_search_result,
                )
                if filtered_search_result:
                    url = doc.get("url")
                    url = url.replace(" ", "%20").strip() or "expired_url"
                    url = url.replace(
                        "chatm6.sm.cn",
                        "quark.sm.cn",
                    )  # noqa E501
                    search_items.append(
                        {
                            "snippet": doc.get("snippet", ""),
                            "title": doc.get("title", ""),
                            "url": url,
                            "hostname": doc.get("hostname", ""),
                            "hostlogo": doc.get("hostlogo", ""),
                        },
                    )
            for item in extra_tool_info:
                if item.get("tool"):
                    tool_items.append(
                        {
                            "tool": item.get("tool"),
                            "result": item.get("result", ""),
                        },
                    )
        except Exception as e:
            print(f"Error: {e}")
            return [], []

        return search_items, tool_items


if __name__ == "__main__":

    import asyncio

    input_data = SearchLiteInput(query="魅族最新Flyme介绍", count=5)
    search_component = ModelstudioSearchLite()

    async def main() -> None:
        image_gent_output = await search_component.arun(input_data)
        print(image_gent_output)

    asyncio.run(main())
