web_tools.py

"""
وحدة البحث في الويب وقراءة الروابط
- بحث عبر DuckDuckGo (بدون API key)
- بحث في Wikipedia
- استخراج محتوى الصفحات
"""
import asyncio
import logging
import re
from typing import List, Dict
from urllib.parse import quote_plus, urlparse

import httpx

logger = logging.getLogger(__name__)

USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36"
)


async def search_duckduckgo(query: str, max_results: int = 8) -> List[Dict]:
    """
    بحث في DuckDuckGo HTML (بدون API key).
    يعيد قائمة نتائج: [{"title", "url", "snippet"}]
    """
    results: List[Dict] = []
    async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
        url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
        try:
            resp = await client.get(url, headers={"User-Agent": USER_AGENT})
            if resp.status_code != 200:
                logger.warning(f"DDG returned {resp.status_code}")
                return results

            text = resp.text
            # استخراج النتائج بنمط بسيط
            # كل نتيجة في <a class="result__a" href="...">title</a>
            # وsnippet في <a class="result__snippet" ...>
            results_raw = re.findall(
                r'<a[^>]*class="result__a"[^>]*href="([^"]+)"[^>]*>(.+?)</a>'
                r'.*?<a[^>]*class="result__snippet"[^>]*>(.+?)</a>',
                text, re.DOTALL,
            )
            for url_raw, title_html, snippet_html in results_raw[:max_results]:
                # DuckDuckGo يضع الرابط بـ //duckduckgo.com/l/?uddg=
                if "uddg=" in url_raw:
                    from urllib.parse import parse_qs, unquote
                    qs = urlparse(url_raw).query
                    params = parse_qs(qs)
                    real_url = unquote(params.get("uddg", [url_raw])[0])
                else:
                    real_url = url_raw

                title = _strip_html(title_html).strip()
                snippet = _strip_html(snippet_html).strip()
                if title and real_url:
                    results.append({
                        "title": title,
                        "url": real_url,
                        "snippet": snippet,
                    })
        except Exception as e:
            logger.error(f"DDG search failed: {e}")
    return results


async def search_wikipedia(query: str, lang: str = "ar", max_results: int = 3) -> List[Dict]:
    """بحث في Wikipedia عبر API الرسمي"""
    results: List[Dict] = []
    async with httpx.AsyncClient(timeout=15) as client:
        try:
            # بحث
            search_url = f"https://{lang}.wikipedia.org/w/api.php"
            params = {
                "action": "query",
                "list": "search",
                "srsearch": query,
                "srlimit": max_results,
                "format": "json",
            }
            resp = await client.get(search_url, params=params)
            if resp.status_code != 200:
                return results

            data = resp.json()
            for item in data.get("query", {}).get("search", []):
                title = item.get("title", "")
                snippet = _strip_html(item.get("snippet", "")).strip()
                page_url = f"https://{lang}.wikipedia.org/wiki/{quote_plus(title.replace(' ', '_'))}"
                results.append({
                    "title": title,
                    "url": page_url,
                    "snippet": snippet,
                })
        except Exception as e:
            logger.error(f"Wikipedia search failed: {e}")
    return results


async def fetch_url_content(url: str, max_chars: int = 5000) -> Dict:
    """
    استخراج المحتوى النصي من صفحة ويب.
    يعيد {"url", "title", "content", "error" (optional)}
    """
    result = {"url": url, "title": "", "content": "", "error": ""}
    async with httpx.AsyncClient(
        timeout=20, follow_redirects=True,
        headers={"User-Agent": USER_AGENT},
    ) as client:
        try:
            resp = await client.get(url)
            if resp.status_code != 200:
                result["error"] = f"HTTP {resp.status_code}"
                return result

            html = resp.text

            # استخراج العنوان
            title_match = re.search(r"<title[^>]*>(.+?)</title>", html, re.DOTALL | re.IGNORECASE)
            if title_match:
                result["title"] = _strip_html(title_match.group(1)).strip()

            # إزالة الـ scripts و styles
            html = re.sub(r"<script[^>]*>.*?</script>", "", html, flags=re.DOTALL | re.IGNORECASE)
            html = re.sub(r"<style[^>]*>.*?</style>", "", html, flags=re.DOTALL | re.IGNORECASE)
            html = re.sub(r"<!--.*?-->", "", html, flags=re.DOTALL)

            # استخراج النص من الفقرات أولاً
            paragraphs = re.findall(r"<(?:p|div|li|h[1-6])[^>]*>(.+?)</(?:p|div|li|h[1-6])>", html, re.DOTALL)
            text_parts = []
            for p in paragraphs:
                clean = _strip_html(p).strip()
                if len(clean) > 30:  # تجاهل الفقرات القصيرة
                    text_parts.append(clean)

            content = "\n\n".join(text_parts)
            if len(content) > max_chars:
                content = content[:max_chars] + "\n\n... [محتوى مقطوع]"

            result["content"] = content

        except httpx.TimeoutException:
            result["error"] = "انتهت المهلة"
        except Exception as e:
            result["error"] = str(e)

    return result


def _strip_html(text: str) -> str:
    """إزالة وسوم HTML وتنظيف النص"""
    # إزالة الوسوم
    text = re.sub(r"<[^>]+>", "", text)
    # تحويل الكيانات الشائعة
    text = text.replace("&nbsp;", " ").replace("&amp;", "&")
    text = text.replace("&lt;", "<").replace("&gt;", ">")
    text = text.replace("&quot;", '"').replace("&#39;", "'")
    text = text.replace("&laquo;", "«").replace("&raquo;", "»")
    # ضغط المسافات
    text = re.sub(r"\s+", " ", text)
    return text.strip()


def format_search_results(results: List[Dict], query: str) -> str:
    """تنسيق نتائج البحث كرسالة تيليجرام"""
    if not results:
        return f"🔍 لا توجد نتائج لـ: `{query}`"

    lines = [f"🔍 **نتائج البحث:** `{query}`\n"]
    for i, r in enumerate(results[:6], 1):
        lines.append(f"**{i}. {r['title']}**")
        lines.append(f"   {r['url']}")
        if r.get("snippet"):
            snippet = r["snippet"][:200] + ("..." if len(r["snippet"]) > 200 else "")
            lines.append(f"   _{snippet}_")
        lines.append("")
    return "\n".join(lines)