forked from wiai/doorbot
76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
import logging
|
|
|
|
import requests
|
|
|
|
from matrix_client.client import MatrixClient as MatrixApiClient
|
|
from matrix_client.errors import MatrixError
|
|
|
|
class Client:
|
|
def send_image(self, target, path, name, content_type="image/png"):
|
|
raise NotImplementedError()
|
|
|
|
def send_text(self, target, text):
|
|
raise NotImplementedError()
|
|
|
|
|
|
class MatrixClient(Client):
|
|
|
|
def __init__(self, host, username, password):
|
|
self.client = MatrixApiClient(host)
|
|
self.client.login_with_password_no_sync(username=username, password=password)
|
|
self.log = logging.getLogger(__name__)
|
|
|
|
def _get_room(self, target):
|
|
if target not in self.client.rooms:
|
|
try:
|
|
self.client.join_room(target)
|
|
except MatrixError as e:
|
|
self.log.error("could not join room '" + target + "'")
|
|
self.log.exception(e)
|
|
return None
|
|
return self.client.rooms[target]
|
|
|
|
def send_image(self, target, path, name, content_type="image/png"):
|
|
"""Send an image to a room
|
|
|
|
Args:
|
|
target (str): The internal room id to post into
|
|
path (str/Path-like): The path to the image file
|
|
name (str): The name for the file in the room
|
|
content_type (str): Content-type of the image
|
|
"""
|
|
with open(path, "rb") as src:
|
|
data = src.read()
|
|
try:
|
|
mxc = self.client.api.media_upload(data, content_type)
|
|
except MatrixError as e:
|
|
self.log.exception(e)
|
|
return
|
|
room = self._get_room(target)
|
|
if room:
|
|
room.send_image(mxc['content_uri'], name)
|
|
|
|
def send_text(self, target, text):
|
|
room = self._get_room(target)
|
|
if room:
|
|
room.send_text(text)
|
|
|
|
class TelegramClient(Client):
|
|
|
|
MESSAGE_URL = "https://api.telegram.org/bot{token}/sendMessage"
|
|
IMAGE_URL = "https://api.telegram.org/bot{token}/sendPhoto"
|
|
|
|
def __init__(self, token):
|
|
self.text_url = self.MESSAGE_URL.format(token=token)
|
|
self.image_url = self.IMAGE_URL.format(token=token)
|
|
self.log = logging.getLogger(__name__)
|
|
|
|
def send_image(self, target, path, name, content_type="image/png"):
|
|
files = {'photo': open(path, "rb")}
|
|
values = {"chat_id": target}
|
|
r = requests.post(self.image_url, files=files, data=values)
|
|
self.log.info("post photo: %s", r.status_code)
|
|
|
|
def send_text(self, target, text):
|
|
response = requests.post(self.text_url, data={'chat_id': target, "text": text})
|
|
self.log.info("post message: %s", response.status_code) |