1  from __future__ import annotations
   2  
   3  import re
   4  from html import unescape
   5  from typing import Any
   6  from urllib.parse import unquote
   7  
   8  from .exceptions import DuckDuckGoSearchException
   9  
  10  try:
  11      HAS_ORJSON = True
  12      import orjson
  13  except ImportError:
  14      HAS_ORJSON = False
  15      import json
  16  
  17  REGEX_STRIP_TAGS = re.compile("<.*?>")
  18  
  19  
  20  def json_dumps(obj: Any) -> str:
  21      try:
  22          return (
  23              orjson.dumps(obj, option=orjson.OPT_INDENT_2).decode()
  24              if HAS_ORJSON
  25              else json.dumps(obj, ensure_ascii=False, indent=2)
  26          )
  27      except Exception as ex:
  28          raise DuckDuckGoSearchException(f"{type(ex).__name__}: {ex}") from ex
  29  
  30  
  31  def json_loads(obj: str | bytes) -> Any:
  32      try:
  33          return orjson.loads(obj) if HAS_ORJSON else json.loads(obj)
  34      except Exception as ex:
  35          raise DuckDuckGoSearchException(f"{type(ex).__name__}: {ex}") from ex
  36  
  37  
  38  def _extract_vqd(html_bytes: bytes, keywords: str) -> str:
  39      """Extract vqd from html bytes."""
  40      for c1, c1_len, c2 in (
  41          (b'vqd="', 5, b'"'),
  42          (b"vqd=", 4, b"&"),
  43          (b"vqd='", 5, b"'"),
  44      ):
  45          try:
  46              start = html_bytes.index(c1) + c1_len
  47              end = html_bytes.index(c2, start)
  48              return html_bytes[start:end].decode()
  49          except ValueError:
  50              pass
  51      raise DuckDuckGoSearchException(f"_extract_vqd() {keywords=} Could not extract vqd.")
  52  
  53  
  54  def _normalize(raw_html: str) -> str:
  55      """Strip HTML tags from the raw_html string."""
  56      return unescape(REGEX_STRIP_TAGS.sub("", raw_html)) if raw_html else ""
  57  
  58  
  59  def _normalize_url(url: str) -> str:
  60      """Unquote URL and replace spaces with '+'."""
  61      return unquote(url).replace(" ", "+") if url else ""
  62  
  63  
  64  def _expand_proxy_tb_alias(proxy: str | None) -> str | None:
  65      """Expand "tb" to a full proxy URL if applicable."""
  66      return "socks5://127.0.0.1:9150" if proxy == "tb" else proxy
  67