evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bandage telegram bot (#23022)

Browse source 
* Bandage

* lint

* move everything into __init__.py

* fix lint
This commit is contained in:
Diogo Gomes 2019-04-12 22:28:59 +01:00 • committed by Paulus Schoutsen
parent 0a3e11aa12
commit 51508d69ad
4 changed files with 57 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
homeassistant/components/telegram_bot/__init__.py
View file

@ -1,5 +1,6 @@
"""Support to send and receive Telegram messages."""
import io
from ipaddress import ip_network
from functools import partial
import importlib
import logging
@ -12,7 +13,7 @@ from homeassistant.components.notify import (
ATTR_DATA, ATTR_MESSAGE, ATTR_TITLE)
from homeassistant.const import (
ATTR_COMMAND, ATTR_LATITUDE, ATTR_LONGITUDE, CONF_API_KEY,
CONF_PLATFORM, CONF_TIMEOUT, HTTP_DIGEST_AUTHENTICATION)
CONF_PLATFORM, CONF_TIMEOUT, HTTP_DIGEST_AUTHENTICATION, CONF_URL)
import homeassistant.helpers.config_validation as cv
from homeassistant.exceptions import TemplateError
@ -51,6 +52,7 @@ ATTR_VERIFY_SSL = 'verify_ssl'
CONF_ALLOWED_CHAT_IDS = 'allowed_chat_ids'
CONF_PROXY_URL = 'proxy_url'
CONF_PROXY_PARAMS = 'proxy_params'
CONF_TRUSTED_NETWORKS = 'trusted_networks'
DOMAIN = 'telegram_bot'
@ -73,17 +75,34 @@ EVENT_TELEGRAM_TEXT = 'telegram_text'
PARSER_HTML = 'html'
PARSER_MD = 'markdown'
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA.extend({
vol.Required(CONF_PLATFORM): vol.In(('broadcast', 'polling', 'webhooks')),
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_ALLOWED_CHAT_IDS):
vol.All(cv.ensure_list, [vol.Coerce(int)]),
vol.Optional(ATTR_PARSER, default=PARSER_MD): cv.string,
vol.Optional(CONF_PROXY_URL): cv.string,
vol.Optional(CONF_PROXY_PARAMS): dict,
})
DEFAULT_TRUSTED_NETWORKS = [
ip_network('149.154.167.197/32'),
ip_network('149.154.167.198/31'),
ip_network('149.154.167.200/29'),
ip_network('149.154.167.208/28'),
ip_network('149.154.167.224/29'),
ip_network('149.154.167.232/31')
]
PLATFORM_SCHEMA_BASE = cv.PLATFORM_SCHEMA_BASE.extend(PLATFORM_SCHEMA.schema)
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.All(cv.ensure_list, [
vol.Schema({
vol.Required(CONF_PLATFORM): vol.In(
('broadcast', 'polling', 'webhooks')),
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_ALLOWED_CHAT_IDS):
vol.All(cv.ensure_list, [vol.Coerce(int)]),
vol.Optional(ATTR_PARSER, default=PARSER_MD): cv.string,
vol.Optional(CONF_PROXY_URL): cv.string,
vol.Optional(CONF_PROXY_PARAMS): dict,
# webhooks
vol.Optional(CONF_URL): cv.url,
vol.Optional(CONF_TRUSTED_NETWORKS,
default=DEFAULT_TRUSTED_NETWORKS):
vol.All(cv.ensure_list, [ip_network])
})
])
}, extra=vol.ALLOW_EXTRA)
BASE_SERVICE_SCHEMA = vol.Schema({
vol.Optional(ATTR_TARGET): vol.All(cv.ensure_list, [vol.Coerce(int)]),
@ -213,33 +232,33 @@ async def async_setup(hass, config):
if not config[DOMAIN]:
return False
p_config = config[DOMAIN][0]
for p_config in config[DOMAIN]:
p_type = p_config.get(CONF_PLATFORM)
p_type = p_config.get(CONF_PLATFORM)
platform = importlib.import_module('.{}'.format(config[CONF_PLATFORM]),
__name__)
platform = importlib.import_module(
'.{}'.format(p_config[CONF_PLATFORM]), __name__)
_LOGGER.info("Setting up %s.%s", DOMAIN, p_type)
try:
receiver_service = await \
platform.async_setup_platform(hass, p_config)
if receiver_service is False:
_LOGGER.error(
"Failed to initialize Telegram bot %s", p_type)
_LOGGER.info("Setting up %s.%s", DOMAIN, p_type)
try:
receiver_service = await \
platform.async_setup_platform(hass, p_config)
if receiver_service is False:
_LOGGER.error(
"Failed to initialize Telegram bot %s", p_type)
return False
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error setting up platform %s", p_type)
return False
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error setting up platform %s", p_type)
return False
bot = initialize_bot(p_config)
notify_service = TelegramNotificationService(
hass,
bot,
p_config.get(CONF_ALLOWED_CHAT_IDS),
p_config.get(ATTR_PARSER)
)
bot = initialize_bot(p_config)
notify_service = TelegramNotificationService(
hass,
bot,
p_config.get(CONF_ALLOWED_CHAT_IDS),
p_config.get(ATTR_PARSER)
)
async def async_send_telegram_message(service):
"""Handle sending Telegram Bot message service calls."""

4
homeassistant/components/telegram_bot/broadcast.py
View file

@ -1,12 +1,10 @@
"""Support for Telegram bot to send messages only."""
import logging
from . import PLATFORM_SCHEMA as TELEGRAM_PLATFORM_SCHEMA, initialize_bot
from . import initialize_bot
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = TELEGRAM_PLATFORM_SCHEMA
async def async_setup_platform(hass, config):
"""Set up the Telegram broadcast platform."""

6
homeassistant/components/telegram_bot/polling.py
View file

@ -5,14 +5,10 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
from homeassistant.core import callback
from . import (
CONF_ALLOWED_CHAT_IDS, PLATFORM_SCHEMA as TELEGRAM_PLATFORM_SCHEMA,
BaseTelegramBotEntity, initialize_bot)
from . import (CONF_ALLOWED_CHAT_IDS, BaseTelegramBotEntity, initialize_bot)
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = TELEGRAM_PLATFORM_SCHEMA
async def async_setup_platform(hass, config):
"""Set up the Telegram polling platform."""

29
homeassistant/components/telegram_bot/webhooks.py
View file

@ -1,43 +1,20 @@
"""Support for Telegram bots using webhooks."""
import datetime as dt
from ipaddress import ip_network
import logging
import voluptuous as vol
from homeassistant.components.http import HomeAssistantView
from homeassistant.components.http.const import KEY_REAL_IP
from homeassistant.const import (
CONF_URL, EVENT_HOMEASSISTANT_STOP, HTTP_BAD_REQUEST, HTTP_UNAUTHORIZED)
import homeassistant.helpers.config_validation as cv
EVENT_HOMEASSISTANT_STOP, HTTP_BAD_REQUEST, HTTP_UNAUTHORIZED)
from . import (
CONF_ALLOWED_CHAT_IDS, PLATFORM_SCHEMA, BaseTelegramBotEntity,
initialize_bot)
from . import (CONF_ALLOWED_CHAT_IDS, CONF_TRUSTED_NETWORKS, CONF_URL,
BaseTelegramBotEntity, initialize_bot)
_LOGGER = logging.getLogger(__name__)
TELEGRAM_HANDLER_URL = '/api/telegram_webhooks'
REMOVE_HANDLER_URL = ''
CONF_TRUSTED_NETWORKS = 'trusted_networks'
DEFAULT_TRUSTED_NETWORKS = [
ip_network('149.154.167.197/32'),
ip_network('149.154.167.198/31'),
ip_network('149.154.167.200/29'),
ip_network('149.154.167.208/28'),
ip_network('149.154.167.224/29'),
ip_network('149.154.167.232/31')
]
# pylint: disable=no-value-for-parameter
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_URL): vol.Url(),
vol.Optional(CONF_TRUSTED_NETWORKS, default=DEFAULT_TRUSTED_NETWORKS):
vol.All(cv.ensure_list, [ip_network])
})
async def async_setup_platform(hass, config):
"""Set up the Telegram webhooks platform."""