...
 
Commits (3)
"""LICENSE
Copyright 2018 Hermann Krumrey <hermann@krumreyh.com>
This file is part of bokkichat.
bokkichat is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
bokkichat is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
import argparse
from bokkichat.address import Address
from bokkichat.message import TextMessage
from bokkichat.connection import CliConnection, CliSettings
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("mode", choices={"send", "echo"})
parser.add_argument("--message", default="Hello World")
args = parser.parse_args()
settings = CliSettings()
connection = CliConnection(settings)
if args.mode == "send":
message = TextMessage(Address(""), Address(""), args.message)
connection.send(message)
else:
connection.loop(lambda con, msg: print(msg.body))
"""LICENSE
Copyright 2018 Hermann Krumrey <hermann@krumreyh.com>
This file is part of bokkichat.
bokkichat is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
bokkichat is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
import logging
import argparse
from bokkichat.message import TextMessage
from bokkichat.connection import TelegramSettings, TelegramConnection
from bokkichat.address import Address
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
parser = argparse.ArgumentParser()
parser.add_argument("api_key")
parser.add_argument("mode", choices={"echo"})
args = parser.parse_args()
settings = TelegramSettings(args.api_key)
connection = TelegramConnection(settings)
connection.send(
TextMessage(connection.address, Address("207968297"), "Hello Master")
)
if args.mode == "echo":
def echo(con, msg):
receiver = msg.receiver
msg.receiver = msg.sender
msg.sender = receiver
con.send(msg)
connection.loop(echo)
......@@ -16,3 +16,5 @@ GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
from bokkichat.address.Address import Address
"""LICENSE
Copyright 2018 Hermann Krumrey <hermann@krumreyh.com>
This file is part of bokkichat.
bokkichat is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
bokkichat is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
from typing import Callable, List
from bokkichat.address.Address import Address
from bokkichat.message.Message import Message
from bokkichat.message.TextMessage import TextMessage
from bokkichat.connection.Connection import Connection
class CliConnection(Connection):
"""
Class that implements a CLI connection which can be used in testing
"""
@property
def address(self) -> Address:
"""
A CLI connection has no real address, so a dummy address is generated.
:return: The address of the connection
"""
return Address("CLI")
# noinspection PyMethodMayBeStatic
def send(self, message: Message):
"""
Prints a "sent" message
:param message: The message to "send"
:return: None
"""
print(message)
def receive(self) -> List[Message]:
"""
A CLI Connection receives messages by listening to the input
:return: A list of pending Message objects
"""
return [TextMessage(self.address, self.address, input())]
def loop(self, callback: Callable):
"""
Starts a loop that periodically checks for new messages, calling
a provided callback function in the process.
:param callback: The callback function to call for each
received message.
The callback should have the following format:
lambda connection, message: do_stuff()
:return: None
"""
while True:
for message in self.receive():
callback(self, message)
"""LICENSE
Copyright 2018 Hermann Krumrey <hermann@krumreyh.com>
This file is part of bokkichat.
bokkichat is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
bokkichat is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
from bokkichat.connection.Settings import Settings
class CliSettings(Settings):
"""
Class that defines a Settings object for a CLI connection
"""
# noinspection PyMethodMayBeStatic
def serialize(self) -> str:
"""
Serializes the settings to a string
:return: The serialized Settings object
"""
return ""
@classmethod
def deserialize(cls, _: str):
"""
Deserializes a string and generates a Settings object from it
:param _: The serialized string
:return: The deserialized Settings object
"""
return cls()
......@@ -17,7 +17,9 @@ You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
import logging
from typing import Callable, List
from bokkichat.address.Address import Address
from bokkichat.message.Message import Message
from bokkichat.connection.Settings import Settings
......@@ -36,6 +38,15 @@ class Connection:
:param settings: The settings for the connection
"""
self.settings = settings
self.logger = logging.getLogger("bokkichat")
@property
def address(self) -> Address:
"""
A connection must be able to specify its own address
:return: The address of the connection
"""
raise NotImplementedError()
def send(self, message: Message):
"""
......
"""LICENSE
Copyright 2018 Hermann Krumrey <hermann@krumreyh.com>
This file is part of bokkichat.
bokkichat is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
bokkichat is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
import time
# noinspection PyPackageRequirements
import telegram
import requests
from typing import Callable, List
from bokkichat.address.Address import Address
from bokkichat.message.Message import Message
from bokkichat.message.TextMessage import TextMessage
from bokkichat.message.MediaType import MediaType
from bokkichat.message.MediaMessage import MediaMessage
from bokkichat.connection.Connection import Connection
from bokkichat.connection.TelegramSettings import TelegramSettings
class TelegramConnection(Connection):
"""
Class that implements a Telegram connection
"""
def __init__(self, settings: TelegramSettings):
"""
Initializes the connection, with credentials provided by a
Settings object.
:param settings: The settings for the connection
"""
super().__init__(settings)
self.bot = telegram.Bot(settings.api_key)
try:
self.update_id = self.bot.get_updates()[0].update_id
except IndexError:
self.update_id = 0
@property
def address(self) -> Address:
"""
A connection must be able to specify its own address
:return: The address of the connection
"""
return Address(str(self.bot.id))
def send(self, message: Message):
"""
Sends a message. A message may be either a TextMessage
or a MediaMessage.
:param message: The message to send
:return: None
"""
self.logger.info("Sending message to {}".format(message.receiver))
try:
if isinstance(message, TextMessage):
self.bot.send_message(
chat_id=message.receiver.address,
text=message.body
)
elif isinstance(message, MediaMessage):
with open("/tmp/bokkichat-telegram-temp", "wb") as f:
f.write(message.data)
tempfile = open("/tmp/bokkichat-telegram-temp", "rb")
if message.media_type == MediaType.IMAGE:
self.bot.send_photo(
chat_id=message.receiver.address,
photo=tempfile,
caption=message.caption
)
elif message.media_type == MediaType.AUDIO:
self.bot.send_audio(
chat_id=message.receiver.address,
audio=tempfile,
caption=message.caption
)
else:
self.bot.send_video(
chat_id=message.receiver.address,
video=tempfile,
caption=message.caption
)
tempfile.close()
except (telegram.error.Unauthorized, telegram.error.BadRequest):
self.logger.info(
"Failed to send message to {}".format(message.receiver)
)
def receive(self) -> List[Message]:
"""
Receives all pending messages.
:return: A list of pending Message objects
"""
messages = []
try:
for update in self.bot.get_updates(
offset=self.update_id, timeout=10
):
self.update_id = update.update_id + 1
telegram_message = update.message.to_dict()
address = Address(str(telegram_message['chat']['id']))
self.logger.info("Received message from {}".format(address))
if "text" in telegram_message:
body = telegram_message['text']
self.logger.debug("Message Body: {}".format(body))
messages.append(TextMessage(address, self.address, body))
for media_key, media_type in {
"photo": MediaType.IMAGE,
"audio": MediaType.AUDIO,
"video": MediaType.VIDEO,
"voice": MediaType.AUDIO
}.items():
if media_key in telegram_message:
self.logger.debug("Media Type: {}".format(media_key))
media_info = telegram_message[media_key]
if isinstance(media_info, list):
if len(media_info) == 0:
continue
largest = media_info[len(media_info) - 1]
file_id = largest["file_id"]
elif isinstance(media_info, dict):
file_id = media_info["file_id"]
else:
continue
file_info = self.bot.get_file(file_id)
resp = requests.get(file_info["file_path"])
data = resp.content
messages.append(MediaMessage(
address,
self.address,
media_type,
data,
telegram_message.get("caption", "")
))
break
except telegram.error.Unauthorized:
# The self.bot.get_update method may cause an
# Unauthorized Error if the bot is blocked by the user
self.update_id += 1
except telegram.error.TimedOut:
pass
return messages
def loop(self, callback: Callable):
"""
Starts a loop that periodically checks for new messages, calling
a provided callback function in the process.
:param callback: The callback function to call for each
received message.
The callback should have the following format:
lambda connection, message: do_stuff()
:return: None
"""
while True:
for message in self.receive():
callback(self, message)
time.sleep(1)
"""LICENSE
Copyright 2018 Hermann Krumrey <hermann@krumreyh.com>
This file is part of bokkichat.
bokkichat is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
bokkichat is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
import json
from bokkichat.connection.Settings import Settings
class TelegramSettings(Settings):
"""
Class that defines a Settings object for a Telegram connection
"""
def __init__(self, api_key):
"""
Initializes the Telegram Connection.
:param api_key: The API key used for authentication
"""
self.api_key = api_key
# noinspection PyMethodMayBeStatic
def serialize(self) -> str:
"""
Serializes the settings to a string
:return: The serialized Settings object
"""
return json.dumps({
"api_key": self.api_key
})
@classmethod
def deserialize(cls, serialized: str):
"""
Deserializes a string and generates a Settings object from it
:param serialized: The serialized string
:return: The deserialized Settings object
"""
obj = json.loads(serialized)
return cls(obj["api_key"])
......@@ -16,3 +16,10 @@ GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
from bokkichat.connection.Connection import Connection
from bokkichat.connection.Settings import Settings
from bokkichat.connection.CliConnection import CliConnection
from bokkichat.connection.CliSettings import CliSettings
from bokkichat.connection.TelegramConnection import TelegramConnection
from bokkichat.connection.TelegramSettings import TelegramSettings
......@@ -48,3 +48,9 @@ class MediaMessage(Message):
self.media_type = media_type
self.data = data
self.caption = caption
def __str__(self) -> str:
"""
:return: A string representation of the MediaMessage object
"""
return "{}: {}".format(self.media_type.name, self.caption)
......@@ -33,3 +33,9 @@ class Message:
"""
self.sender = sender
self.receiver = receiver
def __str__(self) -> str:
"""
:return: A string representation of the Message object
"""
raise NotImplementedError()
......@@ -46,3 +46,9 @@ class TextMessage(Message):
super().__init__(sender, receiver)
self.body = body
self.title = title
def __str__(self) -> str:
"""
:return: A string representation of the TextMessage object
"""
return "{}: {}".format(self.title, self.body)
......@@ -16,3 +16,8 @@ GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with bokkichat. If not, see <http://www.gnu.org/licenses/>.
LICENSE"""
from bokkichat.message.Message import Message
from bokkichat.message.TextMessage import TextMessage
from bokkichat.message.MediaMessage import MediaMessage
from bokkichat.message.MediaType import MediaType
......@@ -36,8 +36,11 @@ if __name__ == "__main__":
url="https://gitlab.namibsun.net/namibsun/python/bokkichat",
license="GNU GPL3",
packages=find_packages(),
scripts=list(map(lambda x: os.path.join("bin", x), os.listdir("bin"))),
install_requires=[
"typing"
"typing",
"python-telegram-bot",
"requests"
],
test_suite='nose.collector',
tests_require=['nose'],
......