apps/whatsapp_control_api/app/services/web_search.py
text
from __future__ import annotations
import html
import logging
import re
import base64
from dataclasses import dataclass
from urllib.parse import parse_qs, unquote, urlparse
import httpx
from ..core.config import get_settings
logger = logging.getLogger(__name__)
CURRENT_INFO_KEYWORDS = {
"artikel",
"berita",
"cari",
"carikan",
"cek",
"data",
"harga",
"hari ini",
"info",
"informasi",
"jadwal",
"kurs",
"latest",
"real-time",
"realtime",
"saat ini",
"sekarang",
"sumber",
"terbaru",
"update",
}
@dataclass(slots=True)
class SearchResult:
title: str
url: str
snippet: str
class WebSearchService:
@property
def settings(self):
return get_settings()
@property
def is_enabled(self) -> bool:
return self.settings.web_search_enabled
async def search(self, query: str) -> list[SearchResult]:
if not self.is_enabled:
return []
provider = self.settings.web_search_provider
try:
if provider == "duckduckgo":
results = await self._search_duckduckgo(query)
return results or await self._search_bing(query)
if provider == "bing":
return await self._search_bing(query)
if provider == "serper":
return await self._search_serper(query)
if provider == "tavily":
return await self._search_tavily(query)
except httpx.HTTPError as exc:
logger.warning("Web search request failed provider=%s error=%s", provider, exc)
if provider == "duckduckgo":
try:
return await self._search_bing(query)
except Exception as fallback_exc:
logger.warning("Web search fallback failed provider=bing error=%s", fallback_exc)
except Exception as exc:
logger.exception("Web search failed provider=%s error=%s", provider, exc)
return []
async def _search_duckduckgo(self, query: str) -> list[SearchResult]:
base_url = self.settings.web_search_base_url or "https://duckduckgo.com/html/"
headers = {
"User-Agent": "Mozilla/5.0 (compatible; HermesWhatsAppControl/1.0; +https://hermes-assistants.duckdns.org)",
}
async with httpx.AsyncClient(timeout=self.settings.web_search_timeout_seconds, follow_redirects=True) as client:
response = await client.get(base_url, params={"q": query}, headers=headers)
response.raise_for_status()
return parse_duckduckgo_html(response.text, limit=self.settings.web_search_max_results)
async def _search_serper(self, query: str) -> list[SearchResult]:
if not self.settings.web_search_api_key:
return []
base_url = self.settings.web_search_base_url or "https://google.serper.dev/search"
headers = {"X-API-KEY": self.settings.web_search_api_key, "Content-Type": "application/json"}
async with httpx.AsyncClient(timeout=self.settings.web_search_timeout_seconds) as client:
response = await client.post(base_url, json={"q": query, "num": self.settings.web_search_max_results}, headers=headers)
response.raise_for_status()
data = response.json()
return [
SearchResult(
title=str(item.get("title") or "").strip(),
url=str(item.get("link") or "").strip(),
snippet=str(item.get("snippet") or "").strip(),
)
for item in data.get("organic", [])[: self.settings.web_search_max_results]
if item.get("link")
]
async def _search_bing(self, query: str) -> list[SearchResult]:
base_url = self.settings.web_search_base_url or "https://www.bing.com/search"
headers = {
"User-Agent": "Mozilla/5.0 (compatible; HermesWhatsAppControl/1.0; +https://hermes-assistants.duckdns.org)",
}
async with httpx.AsyncClient(timeout=self.settings.web_search_timeout_seconds, follow_redirects=True) as client:
response = await client.get(base_url, params={"q": query}, headers=headers)
response.raise_for_status()
return parse_bing_html(response.text, limit=self.settings.web_search_max_results)
async def _search_tavily(self, query: str) -> list[SearchResult]:
if not self.settings.web_search_api_key:
return []
base_url = self.settings.web_search_base_url or "https://api.tavily.com/search"
payload = {
"api_key": self.settings.web_search_api_key,
"query": query,
"search_depth": "basic",
"max_results": self.settings.web_search_max_results,
}
async with httpx.AsyncClient(timeout=self.settings.web_search_timeout_seconds) as client:
response = await client.post(base_url, json=payload)
response.raise_for_status()
data = response.json()
return [
SearchResult(
title=str(item.get("title") or "").strip(),
url=str(item.get("url") or "").strip(),
snippet=str(item.get("content") or "").strip(),
)
for item in data.get("results", [])[: self.settings.web_search_max_results]
if item.get("url")
]
def should_use_web_search(message: str) -> bool:
lowered = message.strip().lower()
if not lowered:
return False
if any(keyword in lowered for keyword in CURRENT_INFO_KEYWORDS):
return True
return bool(re.search(r"\b20[2-9][0-9]\b", lowered))
def parse_duckduckgo_html(raw_html: str, *, limit: int = 5) -> list[SearchResult]:
results: list[SearchResult] = []
pattern = re.compile(
r'<a[^>]+class="result__a"[^>]+href="(?P<url>[^"]+)"[^>]*>(?P<title>.*?)</a>.*?'
r'(?:<a[^>]+class="result__snippet"[^>]*>(?P<snippet_a>.*?)</a>|<div[^>]+class="result__snippet"[^>]*>(?P<snippet_div>.*?)</div>)',
re.IGNORECASE | re.DOTALL,
)
for match in pattern.finditer(raw_html):
url = clean_duckduckgo_url(match.group("url"))
title = clean_html_text(match.group("title"))
snippet = clean_html_text(match.group("snippet_a") or match.group("snippet_div") or "")
if not url or not title:
continue
results.append(SearchResult(title=title, url=url, snippet=snippet))
if len(results) >= limit:
break
return results
def parse_bing_html(raw_html: str, *, limit: int = 5) -> list[SearchResult]:
results: list[SearchResult] = []
pattern = re.compile(
r'<li[^>]+class="b_algo"[^>]*>.*?<h2[^>]*>\s*<a[^>]+href="(?P<url>[^"]+)"[^>]*>(?P<title>.*?)</a>.*?</h2>.*?'
r'(?:<p[^>]*>(?P<snippet>.*?)</p>)',
re.IGNORECASE | re.DOTALL,
)
for match in pattern.finditer(raw_html):
url = clean_bing_url(match.group("url"))
title = clean_html_text(match.group("title"))
snippet = clean_html_text(match.group("snippet") or "")
if not url or not title:
continue
results.append(SearchResult(title=title, url=url, snippet=snippet))
if len(results) >= limit:
break
return results
def clean_duckduckgo_url(url: str) -> str:
decoded = html.unescape(url)
parsed = urlparse(decoded)
if parsed.path.startswith("/l/"):
target = parse_qs(parsed.query).get("uddg", [""])[0]
if target:
return unquote(target)
return decoded
def clean_bing_url(url: str) -> str:
decoded = html.unescape(url).strip()
parsed = urlparse(decoded)
if "bing.com" not in parsed.netloc or not parsed.path.startswith("/ck/"):
return decoded
target = parse_qs(parsed.query).get("u", [""])[0]
if not target:
return decoded
if target.startswith("a1"):
target = target[2:]
padded = target + "=" * (-len(target) % 4)
try:
return base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8")
except Exception:
return decoded
def clean_html_text(value: str) -> str:
without_tags = re.sub(r"<[^>]+>", " ", value)
normalized = re.sub(r"\s+", " ", html.unescape(without_tags)).strip()
return normalized
def format_search_context(results: list[SearchResult]) -> str:
lines: list[str] = []
for index, result in enumerate(results, start=1):
lines.append(f"[{index}] {result.title}\nURL: {result.url}\nSnippet: {result.snippet}")
return "\n\n".join(lines)