DHBW/TelegramAktienBot
1
0
Fork 0
Code Issues 6 Pull Requests 26 Actions Packages Projects Releases Wiki Activity

Merge pull request #2 from WebEngineering2/bot-features

Browse Source 
implemented /news and deleted debugbot (edit BOT_API_KEY instead)
This commit is contained in:
Rripped 2022-03-16 10:08:43 +01:00 committed by GitHub
commit 48e95a534a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 339 additions and 127 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.env.example
View File

@ -6,8 +6,8 @@ MYSQL_USER=
MYSQL_PASSWORD=
# Telegram bot api key
BOT_API_KEY=
BOT_API_KEY=""
NEWS_API_KEY=""
# Flask secret key
SECRET_KEY=

6
.gitignore vendored
View File

@ -1,2 +1,8 @@
.idea/*
.env
.env.example
env
Lib
Include
*/*/__pycache__/*
*/__pycache__/*

7
README.md
View File

@ -17,3 +17,10 @@ WebEngineering2 Projekt: Aktien und News Bot für Telegram
## Dokumentation
- Postman-API -> docs/postman.json
- Datenbank -> database/*
## Local setup for telegram bot
0. optional: build virtual env by ``python -m venv venv``
``env/Scripts/activate``
2. create .env and set API keys etc. (use .env.example as a layout)
3. install required libs via ``pip install -r ./telegram_bot/requirements.txt``
4. run bot.py via ``python ./telegram_bot/bot.py``

99
debugbot.py
View File

@ -1,99 +0,0 @@
# 5108535940:AAF5FpPHNV96WxGCDt8aMrGGKke1VILYib4
# https://t.me/mynewdebugbot
import os
import telebot
import time
import sys
import logging
from telebot import types
user_list = []
version = "1.5"
class User: # Currently saving users in this class to test functionality -> later database
def __init__(self, p_user_id, p_user_name, p_chat_id):
self.user_id = int(p_user_id)
self.chat_id = int(p_chat_id)
self.user_name = str(p_user_name)
bot = telebot.TeleBot('5108535940:AAF5FpPHNV96WxGCDt8aMrGGKke1VILYib4')
@bot.message_handler(commands=['start']) # /start -> saving as new user and sending welcome
def send_start(message):
new_user = User(int(message.from_user.id), message.from_user.first_name, int(message.chat.id))
existing_already = False
for known_user in user_list:
if known_user.user_id == new_user.user_id:
existing_already = True
if existing_already == False:
user_list.append(new_user)
bot.reply_to(message, "Welcome to this share bot project. Type /help to get information on what this bot can do")
@bot.message_handler(commands=['version'])
def send_version(message):
bot.reply_to(message, version)
@bot.message_handler(commands=['help']) # /help -> sending all functions
def send_welcome(message):
bot.reply_to(message, "/id or /auth for authentication. /update to get updates on your shares. /users to see all users. For further details see aktienbot.flokaiser.com")
@bot.message_handler(commands=['users'])
def send_all_users(message):
print('Debug: users command')
user_id = int(message.from_user.id)
answer = 'Current number of users: ' + str(len(user_list))
bot.send_message(chat_id = user_id, text=answer)
for known_user in user_list:
answer = str(known_user.user_id) + ' : ' + known_user.user_name
bot.send_message(chat_id=user_id, text=answer)
@bot.message_handler(commands=['id', 'auth']) # /auth or /id -> Authentication with user_id over web tool
def send_id(message):
answer = 'Your ID/Authentication Code is: [' + str(message.from_user.id) + ']. Enter this code in the settings on aktienbot.flokaiser.com to get updates on your shares.'
bot.reply_to(message, answer)
@bot.message_handler(commands=['update']) # /update -> send static update via user_id to this user, later fetch from database
def send_update(message):
user_id = int(message.from_user.id)
#Get Information for user with this id
bot.send_message(chat_id=user_id, text='This is your update')
@bot.message_handler(func=lambda message: True) # Returning that command is unkown for any other statement
def echo_all(message):
answer = 'Do not know this command or text: ' + message.text
bot.reply_to(message, answer)
telebot.logger.setLevel(logging.DEBUG)
@bot.inline_handler(lambda query: query.query == 'text') # inline prints for debugging
def query_text(inline_query):
try:
r = types.InlineQueryResultArticle('1', 'Result1', types.InputTextMessageContent('hi'))
r2 = types.InlineQueryResultArticle('2', 'Result2', types.InputTextMessageContent('hi'))
bot.answer_inline_query(inline_query.id, [r, r2])
except Exception as e:
print(e)
def main_loop():
bot.infinity_polling()
while 1:
time.sleep(3)
if __name__ == '__main__':
try:
main_loop()
except KeyboardInterrupt:
print('\nExiting by user request.\n')
sys.exit(0)

177
telegram_bot/bot.py
View File

@ -4,12 +4,12 @@ script for telegram bot and its functions
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "11.03.2022"
__version__ = "0.0.3"
__version__ = "0.0.4"
__license__ = "None"
# side-dependencies: none
# Work in Progress
# Api-Key: 5228016873:AAGFrh0P6brag7oD3gxXjCh5gnLLE8JMvMs
# Api-Key: 5228016873:AAGFrh0P6brag7oD3gxXjCh5gnLLE8JMvMs /debugAPI Key: 5108535940:AAF5FpPHNV96WxGCDt8aMrGGKke1VILYib4 (https://t.me/mynewdebugbot)
# text bot at t.me/projektaktienbot
# API Documentation https://core.telegram.org/bots/api
# Code examples https://github.com/eternnoir/pyTelegramBotAPI#getting-started
@ -21,13 +21,39 @@ import time
import sys
import logging
from telebot import types
import news.news_fetcher as news
import shares.share_fetcher as share_fetcher
bot_version = "0.0.1"
from telebot import types
from dotenv import load_dotenv
load_dotenv()
bot_version = "0.1.1"
user_list = []
class User: # Currently saving users in this class to test functionality -> later database
def __init__(self, p_user_id, p_user_name, p_chat_id):
""" Initialize a new user
:type self:
:param self: for class
:type p_user_id: int
:param p_user_id: telegram user id
:type p_user_name: str
:param p_user_name: first name of user
:type p_chat_id: int
:param p_chat_id: telegram chat id
:raises:
:rtype:
"""
self.user_id = int(p_user_id)
self.chat_id = int(p_chat_id)
self.user_name = str(p_user_name)
@ -36,6 +62,15 @@ bot = telebot.TeleBot(os.getenv('BOT_API_KEY'))
@bot.message_handler(commands=['start']) # /start -> saving as new user and sending welcome
def send_start(message):
""" Description
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/start'
:raises: none
:rtype: none
"""
new_user = User(int(message.from_user.id), message.from_user.first_name, int(message.chat.id))
existing_already = False
for known_user in user_list:
@ -46,19 +81,51 @@ def send_start(message):
bot.reply_to(message, "Welcome to this share bot project. Type /help to get information on what this bot can do")
@bot.message_handler(commands=['version'])
def send_version(message):
""" Sending programm version
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/version'
:raises: none
:rtype:none
"""
bot.reply_to(message, bot_version)
@bot.message_handler(commands=['help']) # /help -> sending all functions
def send_welcome(message):
bot.reply_to(message, "/id or /auth for authentication. /update to get updates on your shares. /users to see all users. For further details see aktienbot.flokaiser.com")
""" Send all functions
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/help'
:raises: none
:rtype: none
"""
bot.reply_to(message, "/id or /auth for authentication. /update to get updates on your shares. /users to see all users. /news to get current use for your keywords. /share to get price of specific share. For further details see aktienbot.flokaiser.com")
@bot.message_handler(commands=['users'])
def send_all_users(message):
""" Send all users, only possible for admins
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/users'
:raises: none
:rtype: none
"""
print('Debug: users command')
user_id = int(message.from_user.id)
# tbd check if user is admin
answer = 'Current number of users: ' + str(len(user_list))
bot.send_message(chat_id = user_id, text=answer)
for known_user in user_list:
@ -68,19 +135,100 @@ def send_all_users(message):
@bot.message_handler(commands=['id', 'auth']) # /auth or /id -> Authentication with user_id over web tool
def send_id(message):
""" Send user id for authentication with browser
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/id' or '/auth'
:raises: none
:rtype: none
"""
answer = 'Your ID/Authentication Code is: [' + str(message.from_user.id) + ']. Enter this code in the settings on aktienbot.flokaiser.com to get updates on your shares.'
bot.reply_to(message, answer)
@bot.message_handler(commands=['update']) # /update -> send static update via user_id to this user, later fetch from database
@bot.message_handler(commands=['update'])
def send_update(message):
""" Send update on shares
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/help'
:raises: none
:rtype: none
"""
user_id = int(message.from_user.id)
share_fetcher = share_fetcher.Share_Handler()
#Get Information for user with this id
#call Share_Handler
bot.send_message(chat_id=user_id, text='This is your update')
@bot.message_handler(commands=['share'])
def send_update(message):
""" Send price of a specific share
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/share'
:raises: none
:rtype: none
"""
user_id = int(message.from_user.id)
#Get Information for user with this id
bot.send_message(chat_id=user_id, text='Send symbol of share:')
#str_share_price = shares.wait_for_share.main_loop(bot)
bot.register_next_step_handler(message, send_share_price)
def send_share_price(message):
share_fetcher_obj = share_fetcher.Share_Handler()
str_share_price = share_fetcher_obj.get_share_price(str(message.text))
bot.reply_to(message, str_share_price)
@bot.message_handler(commands=['news'])
def send_news(message):
""" Get news for keywords of user
:type message: message object bot
:param message: message that was reacted to, in this case always containing '/news'
:raises: none
:rtype: none
"""
keyword = "business"
user_id = int(message.from_user.id)
#Get Information for user with this id
bot.send_message(chat_id=user_id, text='This is your update')
articles = news.get_top_news_by_keyword(keyword) #tbd: get keyword from db
try:
formatted_article = news.format_article(articles["articles"][0])
except IndexError:
bot.send_message(chat_id=user_id, text=f"no news currently available for keyword: {keyword}")
return
bot.send_message(chat_id=user_id, text=f"_keyword: {keyword}_\n\n" + formatted_article, parse_mode="MARKDOWN")
@bot.message_handler(func=lambda message: True) # Returning that command is unkown for any other statement
def echo_all(message):
""" Tell that command is not known if it is no known command
:type message: message object bot
:param message: message that was reacted to, if no other command handler gets called
:raises: none
:rtype: none
"""
answer = 'Do not know this command or text: ' + message.text
bot.reply_to(message, answer)
@ -90,6 +238,15 @@ telebot.logger.setLevel(logging.DEBUG)
@bot.inline_handler(lambda query: query.query == 'text') # inline prints for debugging
def query_text(inline_query):
""" Output in the console about current user actions and status of bot
:type inline_query:
:param inline_query:
:raises: none
:rtype: none
"""
try:
r = types.InlineQueryResultArticle('1', 'Result1', types.InputTextMessageContent('hi'))
r2 = types.InlineQueryResultArticle('2', 'Result2', types.InputTextMessageContent('hi'))
@ -99,6 +256,12 @@ def query_text(inline_query):
def main_loop():
""" Get Information about bot status every 3 seconds
:raises: none
:rtype: none
"""
bot.infinity_polling()
while 1:
time.sleep(3)

32
telegram_bot/news/article_example.json Normal file
View File

@ -0,0 +1,32 @@
{
"status": "ok",
"totalResults": 1,
"articles": [
{
"source": {
"id": "the-verge",
"name": "The Verge"
},
"author": "Justine Calma",
"title": "EU Parliament backs off plans to phase out energy-hungry cryptocurrencies",
"description": "EU Parliament abandoned a measure in its proposed legislative framework for regulating cryptocurrencies that would have amounted to a de facto ban on energy-hungry networks like Bitcoin.",
"url": "https://www.theverge.com/2022/3/14/22977132/bitcoin-european-union-parliament-ban-proof-of-work-cryptocurrencies",
"urlToImage": "https://cdn.vox-cdn.com/thumbor/8bE-uBwwu-eXg-CcB6cOqcAGVDw=/0x286:4000x2380/fit-in/1200x630/cdn.vox-cdn.com/uploads/chorus_asset/file/23315944/834392892.jpg",
"publishedAt": "2022-03-14T23:40:25Z",
"content": "But Bitcoin is still under scrutiny \r\nPower cords for bitcoin mining machines are plugged into electrical outlets at a mining facility operated by Bitmain Technologies Ltd. in Ordos, Inner Mongolia, \u2026 [+5797 chars]"
},
{
"source": {
"id": "the-verge",
"name": "The Verge"
},
"author": "Justine Calma",
"title": "EU Parliament backs off plans to phase out energy-hungry cryptocurrencies",
"description": "EU Parliament abandoned a measure in its proposed legislative framework for regulating cryptocurrencies that would have amounted to a de facto ban on energy-hungry networks like Bitcoin.",
"url": "https://www.theverge.com/2022/3/14/22977132/bitcoin-european-union-parliament-ban-proof-of-work-cryptocurrencies",
"urlToImage": "https://cdn.vox-cdn.com/thumbor/8bE-uBwwu-eXg-CcB6cOqcAGVDw=/0x286:4000x2380/fit-in/1200x630/cdn.vox-cdn.com/uploads/chorus_asset/file/23315944/834392892.jpg",
"publishedAt": "2022-03-14T23:40:25Z",
"content": "But Bitcoin is still under scrutiny \r\nPower cords for bitcoin mining machines are plugged into electrical outlets at a mining facility operated by Bitmain Technologies Ltd. in Ordos, Inner Mongolia, \u2026 [+5797 chars]"
}
]
}

60
telegram_bot/news/news_fetcher.py Normal file
View File

@ -0,0 +1,60 @@
"""
script for news fetching (by keywords)
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "15.03.2022"
__version__ = "0.0.1"
__license__ = "None"
import sys
import os
import json
import pandas as pd
from newsapi import NewsApiClient
from dotenv import load_dotenv
load_dotenv()
# Init
newsapi = NewsApiClient(api_key=os.getenv('NEWS_API_KEY'))
# /v2/top-headlines/sources
sources = newsapi.get_sources()
def get_top_news_by_keyword(keyword):
"""get top news to keyword
Args:
keyword (String): keyword for search
Returns:
JSON/dict: dict containing articles
"""
top_headlines = newsapi.get_top_headlines(q=keyword, sources='bbc-news,the-verge,cnn', language='en')
return top_headlines
def format_article(article):
"""format article for messaging (using markdown syntax)
Args:
article (dict): article to format for messaging
Returns:
String: formatted article
"""
sourcename = article["source"]["name"]
headline = article["title"]
url = article["url"]
formatted_article = f"_{sourcename}_\n*{headline}*\n\n{url}"
return formatted_article
if __name__ == '__main__':
print("fetching top news by keyword business...")
articles = get_top_news_by_keyword("business")
formatted_article = format_article(articles["articles"][0])
print(formatted_article)

7
telegram_bot/news_fetcher.py
View File

@ -1,7 +0,0 @@
"""
script for news fetching (by keywords)
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "15.03.2022"
__version__ = "0.0.1"
__license__ = "None"

5
telegram_bot/requirements.txt
View File

@ -1,4 +1,5 @@
pyTelegramBotAPI~=4.4.0
beautifulsoup4==4.10.0
Markdown==3.3.6
Pillow==9.0.1
yfinance==0.1.70
newsapi-python
python-dotenv~=0.19.2

7
telegram_bot/share_fetcher.py
View File

@ -1,7 +0,0 @@
"""
script for share fetching (by symbols (e.g. AAPL, TSLA etc.))
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "15.03.2022"
__version__ = "0.0.1"
__license__ = "None"

56
telegram_bot/shares/share_fetcher.py Normal file
View File

@ -0,0 +1,56 @@
"""
script for share fetching (by symbols (e.g. AAPL, TSLA etc.))
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "15.03.2022"
__version__ = "0.0.2"
__license__ = "None"
import yfinance
class Share_Handler:
def __init__(self):
return
def all_share_prices_for_user(self, int_user_id):
""" Get all share prices for a certain user with his id
:type int_user_id: integer
:param int_user_id: user_id to get all share prices for
:raises: none
:rtype: ###tbd### (maybe dictonary)
"""
return
def get_share_price(self, str_symbol):
""" get current share price for a certain symbol
:type str_symbol: string
:param str_symbol: share symbol to get price for
:raises:
:rtype:
"""
my_share_info = yfinance.Ticker(str_symbol)
my_share_data = my_share_info.info
my_return_string = f'{my_share_data["regularMarketPrice"]} {my_share_data["currency"]}'
return my_return_string
if __name__ == '__main__':
""" test object and get share price
:raises: none
:rtype: none
"""
new_handler = Share_Handler()
print(new_handler.get_share_price("TL0.DE"))