"""
skills/search_skill.py - 搜索技能
===================================
提供网络搜索、URL 读取等功能。

搜索引擎优先级:
  1. DuckDuckGo (HTML版, 无需API Key, 解析真实URL)
  2. Bing (直接请求, 无需API Key)
  3. Google (直接请求, 无需API Key)
"""
from __future__ import annotations

import asyncio
from typing import Optional, List
from urllib.parse import urlparse, parse_qs

from core.logger import get_logger
from core.utils import truncate_str
from aiskills.base import Skill, SkillResult, SkillParameter

logger = get_logger("myagent.skills.search")

_HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/120.0.0.0 Safari/537.36"
    ),
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
}


def _extract_ddg_url(raw_url: str) -> str:
    """从 DuckDuckGo 重定向 URL 中提取真实 URL"""
    if not raw_url:
        return ""
    if "uddg=" in raw_url:
        parsed = urlparse(raw_url)
        params = parse_qs(parsed.query)
        if "uddg" in params:
            return params["uddg"][0]
    return raw_url


class WebSearchSkill(Skill):
    """网络搜索"""
    name = "web_search"
    description = "【搜索互联网信息】输入关键词，返回搜索结果列表（含标题、真实URL、摘要）。适用于：搜索新闻、查询资料、获取最新信息等。搜索后可用 web_read 读取具体网页的详细内容。"
    category = "search"
    parameters = [
        SkillParameter("query", "string", "搜索关键词", required=True),
        SkillParameter("num", "integer", "返回结果数量（默认10，最大20）", required=False, default=10),
    ]

    async def execute(self, query: str = "", num: int = 10, **kwargs) -> SkillResult:
        """
        执行网络搜索。
        多后端自动降级:
          1. DuckDuckGo HTML (无需 API Key)
          2. Bing (无需 API Key)
        """
        num = min(max(num, 1), 20)

        # 确保搜索依赖已安装（requests, beautifulsoup4, lxml）
        try:
            import requests
            from bs4 import BeautifulSoup
        except ImportError as imp_err:
            logger.warning(f"搜索依赖缺失: {imp_err}，尝试自动安装...")
            from core.deps_checker import ensure_skill_deps
            if not ensure_skill_deps("search"):
                return SkillResult(
                    success=False,
                    error=f"搜索依赖安装失败，请手动运行: pip install requests beautifulsoup4 lxml ({imp_err})",
                )
        try:
            import requests  # noqa: F811
            from bs4 import BeautifulSoup  # noqa: F811
        except ImportError as imp_err:
            return SkillResult(
                success=False,
                error=f"搜索依赖不可用，请手动运行: pip install requests beautifulsoup4 lxml ({imp_err})",
            )

        try:
            # 尝试 DuckDuckGo HTML
            results = await self._duckduckgo_html_search(query, num)
            if results:
                logger.info(f"DuckDuckGo HTML 搜索成功: {len(results)} 条结果")
                formatted = self._format_results(query, results)
                return SkillResult(
                    success=True,
                    output=formatted,
                    data={"query": query, "results": results},
                    message=f"找到 {len(results)} 条结果",
                )

            # 降级到 Bing
            results = await self._bing_search(query, num)
            if results:
                logger.info(f"Bing 搜索成功: {len(results)} 条结果")
                formatted = self._format_results(query, results)
                return SkillResult(
                    success=True,
                    output=formatted,
                    data={"query": query, "results": results},
                    message=f"找到 {len(results)} 条结果",
                )

            return SkillResult(
                success=False,
                error="搜索未返回结果，请检查网络连接或尝试其他关键词",
            )
        except Exception as e:
            logger.error(f"搜索异常: {e}", exc_info=True)
            return SkillResult(success=False, error=f"搜索失败: {e}")

    @staticmethod
    def _format_results(query: str, results: List[dict]) -> str:
        """将搜索结果格式化为 LLM 可读的文本"""
        lines = [f"搜索关键词: {query}", f"共找到 {len(results)} 条结果:", ""]
        for i, r in enumerate(results, 1):
            title = r.get("title", "无标题")
            url = r.get("url", "")
            snippet = r.get("snippet", "")
            lines.append(f"{i}. {title}")
            if url:
                lines.append(f"   URL: {url}")
            if snippet:
                lines.append(f"   摘要: {snippet}")
            lines.append("")
        return "\n".join(lines)

    async def _duckduckgo_html_search(self, query: str, num: int) -> List[dict]:
        """DuckDuckGo HTML 搜索（无需 API Key，直接解析 HTML）"""
        try:
            import requests
            from bs4 import BeautifulSoup

            loop = asyncio.get_running_loop()

            def _fetch():
                r = None
                try:
                    r = requests.get("https://html.duckduckgo.com/html/", params={"q": query}, headers=_HEADERS, timeout=15)
                    r.raise_for_status()
                except requests.exceptions.SSLError:
                    logger.warning("DuckDuckGo SSL 验证失败，跳过证书验证重试")
                    try:
                        r = requests.get("https://html.duckduckgo.com/html/", params={"q": query}, headers=_HEADERS, timeout=15, verify=False)
                        r.raise_for_status()
                    except Exception as ssl_retry_err:
                        logger.error(f"DuckDuckGo SSL 重试也失败: {ssl_retry_err}")
                        return ""
                except Exception as req_err:
                    logger.error(f"DuckDuckGo 请求失败: {req_err}")
                    return ""
                if r is not None:
                    try:
                        return r.text or ""
                    except Exception as read_err:
                        logger.error(f"DuckDuckGo 读取响应失败: {read_err}")
                        return ""
                return ""

            html = await asyncio.wait_for(loop.run_in_executor(None, _fetch), timeout=30)
            soup = BeautifulSoup(html, "html.parser")

            results = []
            for result in soup.find_all("div", class_="result"):
                # 跳过广告结果
                if result.get("class") and "result--ad" in result.get("class"):
                    continue
                title_el = result.find("a", class_="result__a")
                snippet_el = result.find("a", class_="result__snippet")
                if title_el:
                    raw_url = title_el.get("href", "")
                    real_url = _extract_ddg_url(raw_url)
                    # 过滤广告和追踪链接
                    ad_keywords = ["ad_domain", "ad_provider", "duckduckgo.com/y.js",
                                   "doubleclick", "googleads"]
                    if not real_url or not real_url.startswith("http"):
                        continue
                    if any(kw in real_url for kw in ad_keywords):
                        continue
                    results.append({
                        "title": title_el.get_text(strip=True),
                        "url": real_url,
                        "snippet": snippet_el.get_text(strip=True) if snippet_el else "",
                    })
                    if len(results) >= num:
                        break

            return results
        except Exception as e:
            logger.warning(f"DuckDuckGo HTML 搜索失败: {e}")
            return []

    async def _bing_search(self, query: str, num: int) -> List[dict]:
        """Bing 搜索（直接请求 bing.com，无需 API Key）"""
        try:
            import requests
            from bs4 import BeautifulSoup

            loop = asyncio.get_running_loop()

            def _fetch():
                r = None
                try:
                    r = requests.get("https://www.bing.com/search", params={"q": query, "count": num}, headers=_HEADERS, timeout=15)
                    r.raise_for_status()
                except requests.exceptions.SSLError:
                    logger.warning("Bing SSL 验证失败，跳过证书验证重试")
                    try:
                        r = requests.get("https://www.bing.com/search", params={"q": query, "count": num}, headers=_HEADERS, timeout=15, verify=False)
                        r.raise_for_status()
                    except Exception as ssl_retry_err:
                        logger.error(f"Bing SSL 重试也失败: {ssl_retry_err}")
                        return ""
                except Exception as req_err:
                    logger.error(f"Bing 请求失败: {req_err}")
                    return ""
                if r is not None:
                    try:
                        return r.text or ""
                    except Exception as read_err:
                        logger.error(f"Bing 读取响应失败: {read_err}")
                        return ""
                return ""

            html = await asyncio.wait_for(loop.run_in_executor(None, _fetch), timeout=30)
            soup = BeautifulSoup(html, "html.parser")

            results = []
            for li in soup.find_all("li", class_="b_algo"):
                h2 = li.find("h2")
                if not h2:
                    continue
                link = h2.find("a")
                if not link:
                    continue

                href = link.get("href", "")
                # 跳过 Bing 重定向链接
                if href.startswith("https://www.bing.com/ck/a"):
                    continue

                # 提取摘要
                snippet = ""
                for p in li.find_all("p"):
                    text = p.get_text(strip=True)
                    if text and len(text) > 20:
                        snippet = text
                        break
                if not snippet:
                    caption = li.find("div", class_="b_caption")
                    if caption:
                        snippet = caption.get_text(strip=True)[:300]

                title = link.get_text(strip=True)
                if title and href.startswith("http"):
                    results.append({
                        "title": title,
                        "url": href,
                        "snippet": snippet,
                    })
                if len(results) >= num:
                    break

            return results
        except Exception as e:
            logger.warning(f"Bing 搜索失败: {e}")
            return []


class WebReadSkill(Skill):
    """读取网页内容"""
    name = "web_read"
    description = "【读取网页正文内容】输入URL，提取网页的标题、正文文本和链接列表。适用于：已通过 web_search 获得URL后，读取具体网页的详细内容。"
    category = "search"
    parameters = [
        SkillParameter("url", "string", "网页 URL", required=True),
        SkillParameter("extract_text", "boolean", "是否提取纯文本（默认true）", required=False, default=True),
    ]

    async def execute(self, url: str = "", extract_text: bool = True, **kwargs) -> SkillResult:
        if not url or not isinstance(url, str) or not url.startswith("http"):
            return SkillResult(success=False, error=f"无效的URL: {url!r}")

        # 确保搜索依赖已安装
        try:
            import requests  # noqa: F811
            from bs4 import BeautifulSoup  # noqa: F811
        except ImportError as imp_err:
            logger.warning(f"网页读取依赖缺失: {imp_err}，尝试自动安装...")
            from core.deps_checker import ensure_skill_deps
            if not ensure_skill_deps("search"):
                return SkillResult(success=False, error=f"依赖安装失败，请手动运行: pip install requests beautifulsoup4 lxml ({imp_err})")

        try:
            loop = asyncio.get_running_loop()

            def _fetch():
                r = None
                try:
                    r = requests.get(url, headers=_HEADERS, timeout=30)
                    r.raise_for_status()
                except requests.exceptions.SSLError:
                    logger.warning(f"web_read SSL 验证失败 ({url})，跳过证书验证重试")
                    try:
                        r = requests.get(url, headers=_HEADERS, timeout=30, verify=False)
                        r.raise_for_status()
                    except Exception as ssl_retry_err:
                        logger.error(f"web_read SSL 重试也失败 ({url}): {ssl_retry_err}")
                        return None
                except Exception as req_err:
                    logger.error(f"web_read 请求失败 ({url}): {req_err}")
                    return None
                if r is not None:
                    try:
                        r.encoding = r.apparent_encoding
                        return r.text or ""
                    except Exception as read_err:
                        logger.error(f"web_read 读取响应内容失败 ({url}): {read_err}")
                        return None
                return None

            html = await asyncio.wait_for(
                loop.run_in_executor(None, _fetch), timeout=60
            )

            if not html or not isinstance(html, str):
                return SkillResult(success=False, error="网页内容为空")

            soup = BeautifulSoup(html, "html.parser")
            title = soup.title.get_text(strip=True) if soup.title else ""

            # 提取所有链接（去重）
            seen = set()
            links = []
            for a in soup.find_all("a", href=True):
                href = a.get("href", "")
                if href and href.startswith("http") and href not in seen:
                    seen.add(href)
                    links.append({"url": href, "text": a.get_text(strip=True)})

            if extract_text:
                # 移除干扰标签
                for tag in soup(["script", "style", "nav", "footer", "header", "aside"]):
                    tag.decompose()
                # 移除所有 class 含特定关键词的元素
                # 注意: decompose() 会连带销毁子元素，子元素的 attrs 变为 None，
                # 必须同时检查 attrs 和 parent 来跳过已被销毁的标签
                try:
                    for tag in list(soup.find_all(True, class_=True)):
                        if not getattr(tag, 'attrs', None):
                            continue
                        cls_str = " ".join(tag.get("class", []))
                        if any(kw in cls_str.lower() for kw in
                               ["sidebar", "footer", "nav", "header", "advertisement",
                                "cookie", "popup", "modal", "banner", "social", "share"]):
                            tag.decompose()
                except Exception:
                    pass  # 解析异常时跳过清理，不影响主要内容提取

                text = soup.get_text(separator="\n", strip=True)
                text = "\n".join(line for line in text.split("\n") if line.strip())
                content = truncate_str(text, 20000)
            else:
                content = html[:50000]

            return SkillResult(
                success=True,
                output=f"标题: {title}\nURL: {url}\n\n{content}",
                data={
                    "url": url,
                    "title": title,
                    "content": content,
                    "links": links[:50],
                },
                message=f"已读取: {title} ({len(content)} 字符)",
            )
        except asyncio.TimeoutError:
            return SkillResult(success=False, error=f"网页读取超时 (60s): {url}")
        except Exception as e:
            logger.error(f"web_read 异常 ({url}): {e}", exc_info=True)
            return SkillResult(success=False, error=f"网页读取失败: {e}")


class URLReadSkill(Skill):
    """读取原始 URL 内容"""
    name = "url_read"
    description = "读取指定 URL 的原始内容（API 调用等）"
    category = "search"
    parameters = [
        SkillParameter("url", "string", "URL 地址", required=True),
        SkillParameter("method", "string", "HTTP 方法", required=False, default="GET",
                       enum=["GET", "POST", "PUT", "DELETE"]),
        SkillParameter("headers", "object", "请求头", required=False, default={}),
        SkillParameter("body", "string", "请求体(POST/PUT)", required=False, default=""),
    ]

    async def execute(self, url: str = "", method: str = "GET",
                      headers: dict = None, body: str = "", **kwargs) -> SkillResult:
        if not url:
            return SkillResult(success=False, error="缺少 URL 参数")
        # 确保依赖已安装
        try:
            import requests
        except ImportError as imp_err:
            logger.warning(f"URL读取依赖缺失: {imp_err}，尝试自动安装...")
            from core.deps_checker import ensure_skill_deps
            if not ensure_skill_deps("search"):
                return SkillResult(success=False, error=f"依赖安装失败，请手动运行: pip install requests ({imp_err})")
        try:
            import requests  # noqa: F811

            headers = headers or {}
            loop = asyncio.get_running_loop()

            def _request():
                r = None
                try:
                    r = requests.request(
                        method=method,
                        url=url,
                        headers=headers,
                        data=body,
                        timeout=30,
                    )
                except requests.exceptions.SSLError:
                    logger.warning(f"url_read SSL 验证失败 ({url})，跳过证书验证重试")
                    try:
                        r = requests.request(
                            method=method,
                            url=url,
                            headers=headers,
                            data=body,
                            timeout=30,
                            verify=False,
                        )
                    except Exception as ssl_retry_err:
                        logger.error(f"url_read SSL 重试也失败 ({url}): {ssl_retry_err}")
                        return None
                except Exception as req_err:
                    logger.error(f"url_read 请求失败 ({url}): {req_err}")
                    return None
                if r is not None:
                    return r
                return None

            response = await asyncio.wait_for(loop.run_in_executor(None, _request), timeout=60)

            if response is None:
                return SkillResult(success=False, error=f"请求失败: 无法连接到 {url}")

            try:
                content = response.json()
            except (ValueError, AttributeError):
                content = response.text[:20000]

            # 格式化输出供 LLM 使用
            content_str = json.dumps(content, ensure_ascii=False) if isinstance(content, (dict, list)) else str(content)
            formatted_output = f"HTTP {response.status_code}\nURL: {url}\n\n{truncate_str(content_str, 20000)}"

            return SkillResult(
                success=True,
                output=formatted_output,
                data={
                    "status_code": response.status_code,
                    "headers": dict(response.headers),
                    "content": content,
                },
                message=f"HTTP {response.status_code}",
            )
        except Exception as e:
            logger.error(f"url_read 异常 ({url}): {e}", exc_info=True)
            return SkillResult(success=False, error=f"请求失败: {e}")