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

Migrate from github to gitea + some improvements

Browse Source
This commit is contained in:
Administrator
2022-09-18 00:39:03 +02:00
parent 594b532fe7
commit 7595d42f9a
15 changed files with 44 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
telegram_bot/.env.example
View File

@@ -12,7 +12,7 @@ BOT_EMAIL=
BOT_PASSWORD=
# urls
WEBSITE_URL=
APP_URL=
API_URL=
# timezone

4
telegram_bot/api_handling/api_handler.py
View File

@@ -19,7 +19,7 @@ from dotenv import load_dotenv
load_dotenv() # loads environment vars
# note: for more information about the api visit swagger documentation on https://gruppe1.testsites.info/api/docs#/
# note: for more information about the api visit swagger documentation on os.getenv("API_URL")/docs#/
class API_Handler:
"""class for interacting with the api webservice
@@ -392,7 +392,7 @@ class API_Handler:
if __name__ == "__main__": # editable, just for basic on the go testing of new functions
print("This is a module for the telegram bot. It is not intended to be run directly.")
handler = API_Handler("https://gruppe1.testsites.info/api", str(os.getenv("BOT_EMAIL")), str(os.getenv("BOT_PASSWORD"))) # get creds from env
handler = API_Handler(os.getenv("API_URL"), str(os.getenv("BOT_EMAIL")), str(os.getenv("BOT_PASSWORD"))) # get creds from env
print(handler.token)
keywords = handler.get_user_keywords(user_id=1709356058) # user_id here is currently mine (Linus)
print(keywords)

28
telegram_bot/bot.py
View File

@@ -50,7 +50,7 @@ def send_start(message):
"""
bot.reply_to(message, "Welcome to this share bot project. \
\nType /help to get information on what this bot can do. \
\nAlso see " + os.getenv("WEBSITE_URL") + " \
\nAlso see " + os.getenv("APP_URL") + " \
to start configuring your bot")
@@ -78,7 +78,7 @@ def send_help(message):
:rtype: none
"""
bot.reply_to(message,
"/id or /auth get your user id\n/shares get update on interesting shares\n/setAdmin set admin rights of user (ADMIN)\n/users see all users. (ADMIN)\n/me get my user info\n/news get top article for each keyword.\n/allnews get all news (last 7 days)\n/keywords get all your keywords\n/addkeyword add a keyword\n/removekeyword remove a keyword\n/transactions get all transactions\n/newtransaction create new transaction\n/share get price of specific share\n/portfolio see own stock portfolio\n/removeshare removes share from portfolio\n/interval get update interval\n/setinterval set update interval\n For further details see " + os.getenv("WEBSITE_URL"))
"/id or /auth get your user id\n/shares get update on interesting shares\n/setAdmin set admin rights of user (ADMIN)\n/users see all users. (ADMIN)\n/me get my user info\n/news get top article for each keyword.\n/allnews get all news (last 7 days)\n/keywords get all your keywords\n/addkeyword add a keyword\n/removekeyword remove a keyword\n/transactions get all transactions\n/newtransaction create new transaction\n/share get price of specific share\n/portfolio see own stock portfolio\n/removeshare removes share from portfolio\n/interval get update interval\n/setinterval set update interval\n For further details see " + os.getenv("APP_URL"))
@bot.message_handler(commands=['users', 'Users']) # /users -> sending all users
@@ -170,7 +170,7 @@ def send_user(message):
user_id = int(message.from_user.id)
user_data = api_handler.get_user(user_id)
if not user_data or user_data == None: # true if user is not registered
bot.reply_to(message, "This didn\'t work. Make sure to connect your telegram id (/id) on " + os.getenv("WEBSITE_URL"))
bot.reply_to(message, "This didn\'t work. Make sure to connect your telegram id (/id) on " + os.getenv("APP_URL"))
return
username = user_data['username']
email = user_data['email']
@@ -190,7 +190,7 @@ def send_id(message):
:rtype: none
"""
answer = 'Your ID/Authentication Code is: [' + str(message.from_user.id) + ']. Enter this code in the settings on ' + os.getenv("WEBSITE_URL") + ' to get updates on your shares.'
answer = 'Your ID/Authentication Code is: [' + str(message.from_user.id) + ']. Enter this code in the settings on ' + os.getenv("APP_URL") + ' to get updates on your shares.'
bot.reply_to(message, answer)
@@ -219,7 +219,7 @@ def update_for_user(message):
my_portfolio = p_my_handler.get_user_portfolio(p_user_id)
if my_portfolio == None: # true if user is not registered
bot.send_message(chat_id=p_user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("WEBSITE_URL"))
bot.send_message(chat_id=p_user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("APP_URL"))
return
my_user = p_my_handler.get_user(p_user_id)
@@ -240,7 +240,7 @@ def update_for_user(message):
my_update_message = f'{share_fetcher.get_share_information_markdown(share_symbols[i])}\ncount: {amounts}\nTotal: {hf.make_markdown_proof(round(float(my_price) * float(share_amounts[i]), 2))} EUR'
bot.send_message(chat_id=p_user_id, text=my_update_message, parse_mode="MARKDOWNV2")
else:
send_to_user("No shares found for your account. Check " + os.getenv("WEBSITE_URL") + " to change your settings and add shares.", pUser_id=p_user_id)
send_to_user("No shares found for your account. Check " + os.getenv("APP_URL") + " to change your settings and add shares.", pUser_id=p_user_id)
def send_to_user(pText, pUser_id):
@@ -294,7 +294,7 @@ def send_all_news(message):
keywords = api_handler.get_user_keywords(user_id) # get keywords of user
if keywords == None: # true if user is not registered
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("WEBSITE_URL"))
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("APP_URL"))
return
if not keywords: # true if user is registered but does not have any keywords
@@ -330,7 +330,7 @@ def send_news(message):
keywords = api_handler.get_user_keywords(user_id) # get keywords of user
if keywords == None: # true if user is not registered
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("WEBSITE_URL"))
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("APP_URL"))
return
if not keywords: # true if user is registered but does not have any keywords
@@ -375,7 +375,7 @@ def store_keyword(message):
if status == 200: # statuscode 200 means keyword was added successfully without errors
bot.send_message(chat_id=user_id, text=f'Keyword "{keyword}" added.') # duplicate keywords are denied by Database, so no need to check for that here
else:
bot.send_message(chat_id=user_id, text=f'Keyword "{keyword}" could not be stored. Make sure to connect your telegram id (/id) on https://gruppe1.testsites.info (statuscode {status})')
bot.send_message(chat_id=user_id, text=f'Keyword "{keyword}" could not be stored. Make sure to connect your telegram id (/id) on {os.getenv("APP_URL")} (statuscode {status})')
@bot.message_handler(commands=['removekeyword', 'Removekeyword']) # /removekeyword -> remove keyword from user
@@ -417,7 +417,7 @@ def send_keywords(message):
keywords = api_handler.get_user_keywords(user_id) # get keywords of user
if keywords == None: # true if user is not registered
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("WEBSITE_URL"))
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("APP_URL"))
return
if not keywords: # true if user is registered but does not have any keywords
@@ -513,7 +513,7 @@ def send_interval(message):
user_id = int(message.from_user.id)
user_data = api_handler.get_user(user_id) # get cron interval of user (stored in user data)
if user_data == None: # true if user is not registered in DB
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("WEBSITE_URL") + ' and set an interval with /setinterval')
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("APP_URL") + ' and set an interval with /setinterval')
return
else: # send interval
interval = str(user_data['cron']) # get cron from user data
@@ -538,7 +538,7 @@ def send_transactions(message):
transactions = api_handler.get_user_transactions(user_id) # get transactions of user
if transactions == None: # true if user does not exist
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("WEBSITE_URL"))
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("APP_URL"))
return
if not transactions: # true if user has no transactions
@@ -571,9 +571,9 @@ def send_shares(message):
shares = api_handler.get_user_shares(user_id) # get shares of user
if shares == None: # true if user does not exist
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("WEBSITE_URL"))
bot.send_message(chat_id=user_id, text='This didn\'t work. Make sure to connect your telegram id (/id) on ' + os.getenv("APP_URL"))
elif not shares: # true if user has no shares
bot.send_message(chat_id=user_id, text='You do not have any shares. Add shares on ' + os.getenv("WEBSITE_URL"))
bot.send_message(chat_id=user_id, text='You do not have any shares. Add shares on ' + os.getenv("APP_URL"))
else:
for element in shares:
bot.send_message(chat_id=user_id, text=share_fetcher.get_share_information_markdown(element), parse_mode="MARKDOWNV2")

2
telegram_bot/bot_updates.py
View File

@@ -156,7 +156,7 @@ def update_for_user(p_user_id, p_my_handler):
my_update_message = f'{share_fetcher.get_share_information_markdown(share_symbols[i])}\ncount: {hf.make_markdown_proof(share_amounts[i])}\nTotal: {hf.make_markdown_proof(round(float(my_price) * float(share_amounts[i]), 2))} EUR'
bot.send_message(chat_id=p_user_id, text=my_update_message, parse_mode="MARKDOWNV2")
else:
send_to_user("No shares found for your account. Check " + os.getenv("WEBSITE_URL") + " to change your settings and add shares.", pUser_id=p_user_id)
send_to_user("No shares found for your account. Check " + os.getenv("APP_URL") + " to change your settings and add shares.", pUser_id=p_user_id)
if len(shares) != 0: # Send updates on watchlist shares if existing
send_to_user("Your watchlist shares:", pUser_id=p_user_id)