-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
122 lines (93 loc) · 3.93 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
# MIT License
#
# Copyright (c) 2022 Resha Dwika Hefni Al-Fahsi
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
#
# Modified from:
# - https://github.com/liuhh02/python-telegram-bot-heroku/blob/master/bot.py
# ==============================================================================
import logging
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters
import os
import requests
import numpy as np
import cv2
import uuid
from ykfbot import YourKindFriendBot
bot = YourKindFriendBot()
# Enable logging
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)
# Define a few command handlers. These usually take the two arguments update and
# context. Error handlers also receive the raised TelegramError object in error.
def start(update, context):
"""Send a message when the command /start is issued."""
update.message.reply_text(
"Your typical kind friend who talk nonsense just to kill time."
)
def help(update, context):
"""Send a message when the command /help is issued."""
update.message.reply_text("Just type or send an image, it will reply.")
def reply_text(update, context):
"""Reply text input from the user message."""
update.message.reply_text(bot.reply(update.message.text))
def reply_photo(update, context):
"""Reply image input from the user message."""
logger.info("Receive image from user.")
filename = "image-{}.jpg".format(uuid.uuid4())
image = update.message.photo[-1].get_file()
image = image.download(filename)
files = {'file': open(filename, 'rb')}
logger.info("Send image to API.")
response = requests.post("https://wpir-dnjf-8439.herokuapp.com/predict", files=files)
logger.info("Process the response.")
image = np.asarray(bytearray(response.content), dtype=np.uint8)
image = cv2.imdecode(image, -1)
logger.info("Send image to API.")
files = {'file': open(filename, 'rb')}
response = requests.post("https://image-captioning-69420.herokuapp.com/predict", files=files)
logger.info("Process the response.")
caption = response.content.decode("utf-8").replace('"', '')
logger.info("Send result to user.")
cv2.imwrite(filename, image)
update.message.reply_photo(photo=open(filename, 'rb'), caption=caption)
os.remove(filename)
def error(update, context):
"""Log Errors caused by Updates."""
logger.warning('Update "%s" caused error "%s"', update, context.error)
def main():
"""Start the bot."""
# Create the Updater and pass it your bot's token.
# Make sure to set use_context=True to use the new context based callbacks
# Post version 12 this will no longer be necessary
PORT = int(os.environ.get("PORT", "5000"))
TOKEN = os.environ.get("TOKEN")
updater = Updater(TOKEN, use_context=True)
# Get the dispatcher to register handlers
dp = updater.dispatcher
# on different commands - answer in Telegram
dp.add_handler(CommandHandler("start", start))
dp.add_handler(CommandHandler("help", help))
# on noncommand i.e message - echo the message on Telegram
dp.add_handler(MessageHandler(Filters.text, reply_text))
dp.add_handler(MessageHandler(Filters.photo, reply_photo))
# log all errors
dp.add_error_handler(error)
# Start the Bot
updater.start_webhook(
listen="0.0.0.0",
port=int(PORT),
url_path=TOKEN,
webhook_url="https://ykfbot.herokuapp.com/" + TOKEN,
)
if __name__ == "__main__":
main()