94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
"""Client for the WhatsApp Bridge add-on API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
ADDON_NAME = "WhatsApp Bridge"
|
|
ADDON_SLUG = "whatsapp_bridge"
|
|
ADDON_PORT = 3000
|
|
LEGACY_BASE_URL = "http://866fd2eb-whatsapp-bridge:3000"
|
|
SUPERVISOR_ADDONS_URL = "http://supervisor/addons"
|
|
|
|
|
|
def _addon_base_url(addons: list[dict[str, Any]]) -> str | None:
|
|
"""Return the internal URL for the installed WhatsApp Bridge add-on."""
|
|
for addon in addons:
|
|
slug = str(addon.get("slug", ""))
|
|
name = str(addon.get("name", ""))
|
|
if name.casefold() == ADDON_NAME.casefold() or slug == ADDON_SLUG or slug.endswith(
|
|
f"_{ADDON_SLUG}"
|
|
):
|
|
return f"http://{slug.replace('_', '-')}:{ADDON_PORT}"
|
|
return None
|
|
|
|
|
|
class WhatsAppBridgeApi:
|
|
"""Resolve and communicate with the installed add-on."""
|
|
|
|
def __init__(self, hass) -> None:
|
|
self._session = async_get_clientsession(hass)
|
|
self._base_url: str | None = None
|
|
|
|
async def _async_resolve_base_url(self) -> str:
|
|
if self._base_url is not None:
|
|
return self._base_url
|
|
|
|
token = os.environ.get("SUPERVISOR_TOKEN")
|
|
if token:
|
|
try:
|
|
async with self._session.get(
|
|
SUPERVISOR_ADDONS_URL,
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
timeout=aiohttp.ClientTimeout(total=5),
|
|
) as response:
|
|
response.raise_for_status()
|
|
payload = await response.json()
|
|
data = payload.get("data", payload)
|
|
resolved = _addon_base_url(data.get("addons", []))
|
|
if resolved:
|
|
self._base_url = resolved
|
|
_LOGGER.debug("WhatsApp Bridge add-on found at %s", resolved)
|
|
return resolved
|
|
_LOGGER.warning("WhatsApp Bridge add-on was not found by Supervisor")
|
|
except (aiohttp.ClientError, TimeoutError, ValueError, TypeError) as err:
|
|
_LOGGER.warning("Could not discover WhatsApp Bridge add-on: %s", err)
|
|
|
|
return LEGACY_BASE_URL
|
|
|
|
async def async_get_status(self) -> dict[str, Any]:
|
|
"""Fetch the current bridge status."""
|
|
base_url = await self._async_resolve_base_url()
|
|
async with self._session.get(
|
|
f"{base_url}/api/status",
|
|
timeout=aiohttp.ClientTimeout(total=8),
|
|
) as response:
|
|
response.raise_for_status()
|
|
data = await response.json()
|
|
if not isinstance(data, dict):
|
|
raise ValueError("Status response is not an object")
|
|
return data
|
|
|
|
async def async_send_message(self, number: str, message: str) -> None:
|
|
"""Send one WhatsApp message and surface add-on errors to Home Assistant."""
|
|
base_url = await self._async_resolve_base_url()
|
|
async with self._session.post(
|
|
f"{base_url}/send",
|
|
json={"number": number, "message": message},
|
|
timeout=aiohttp.ClientTimeout(total=15),
|
|
) as response:
|
|
if response.status >= 400:
|
|
try:
|
|
detail = (await response.json()).get("error")
|
|
except (aiohttp.ContentTypeError, ValueError, AttributeError):
|
|
detail = await response.text()
|
|
raise RuntimeError(detail or f"HTTP {response.status}")
|