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

Merge pull request #100 from WebEngineering2/bot

Browse Source 
All commands should work now
This commit is contained in:
NormalParameter 2022-04-28 15:14:58 +02:00 committed by GitHub
commit d01c39300a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 36 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
telegram_bot/api_handling/api_handler.py
View File

@ -2,8 +2,8 @@
script for communicating with webservice to get data from database
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "16.03.2022"
__version__ = "0.0.1"
__date__ = "26.04.2022"
__version__ = "1.0.1"
__license__ = "None"
#side-dependencies: none
@ -13,6 +13,9 @@ import sys
import os
import requests as r
from croniter import croniter # used for checking cron formatting
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#/
@ -287,6 +290,7 @@ class API_Handler:
int: status code
"""
with r.Session() as s:
time = time[:-3] + "Z" # remove last character and add Z to make it a valid date for db
headers = {'Authorization': 'Bearer ' + self.token + ":" + str(user_id)}
transaction = {"comment": str(comment), "count": float(count), "isin": str(isin), "price": float(price), "time": str(time)} # set transaction as JSON with all the attributes needed according to Swagger docs
req = s.post(self.db_adress + "/transaction", json=transaction, headers=headers)
@ -340,21 +344,21 @@ class API_Handler:
Args:
email (string): email of the user
is_admin (String): "true" if user should be Admin, "false" if not
is_admin (bool): "true" if user should be Admin, "false" if not
Returns:
int: status code
"""
with r.Session() as s:
headers = {'Authorization': 'Bearer ' + self.token} # only bot token is needed, user is chosen by email
req = s.put(self.db_adress + "/user/setAdmin", json={"admin": str(is_admin),"email": str(email)})
req = s.put(self.db_adress + "/user/setAdmin", json={"admin": is_admin,"email": str(email)}, headers=headers)
return req.status_code
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", "bot@example.com", "bot")
handler = API_Handler("https://gruppe1.testsites.info/api", 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)

47
telegram_bot/bot.py
View File

@ -3,8 +3,8 @@
script for telegram bot and its functions
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "11.03.2022"
__version__ = "0.0.4"
__date__ = "26.04.2022"
__version__ = "1.2.2"
__license__ = "None"
# side-dependencies: none
@ -14,13 +14,11 @@ __license__ = "None"
# API Documentation https://core.telegram.org/bots/api
# Code examples https://github.com/eternnoir/pyTelegramBotAPI#getting-started
import email
import os
import telebot
import sys
import logging
import json
import re
import news.news_fetcher as news
@ -41,31 +39,6 @@ bot_version = "1.0.1" # version of bot
api_handler = API_Handler("https://gruppe1.testsites.info/api", str(os.getenv("BOT_EMAIL")), str(os.getenv("BOT_PASSWORD"))) # get creds from env vars.
print("Webserver Token: " + str(api_handler.token))
class User: # Currently saving users in this class to test functionality -> later database REMOVABLE
def __init__(self, p_user_id, p_user_name, p_chat_id):
""" Initialize a new user
:type self: User
:param self: object of the 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)
bot = telebot.TeleBot(os.getenv('BOT_API_KEY'))
@bot.message_handler(commands=['start', 'Start']) # /start -> saving as new user and sending welcome
@ -143,7 +116,7 @@ def send_all_users(message):
bot.send_message(chat_id=user_id, text=f'Username: {username}\nEmail: {email}\nID: {id}\nCron: {cron}\nAdmin: {admin}') # format user data into readable message text
@bot.message_handler(commands=['setAdmin', 'SetAdmin']) # set admin rights to user TBD: not working!!
@bot.message_handler(commands=['setAdmin', 'SetAdmin', 'setadmin', 'Setadmin']) # set admin rights to user TBD: not working!!
def set_admin(message):
""" Set admin rights to user
@ -161,7 +134,7 @@ def set_admin(message):
bot.reply_to(message, "You have to be an admin to use this command")
return
bot.send_message(chat_id=user_id, text='send email and "true" if this account should have admin rights, else "false"\n in format: <email>,<is_admin>') # ask for email and admin rights to set
bot.send_message(chat_id=user_id, text='send email and true if this account should have admin rights, else false\n in format: <email>,<is_admin>') # request email and admin rights to change to
bot.register_next_step_handler(message, set_admin_step)
def set_admin_step(message):
@ -174,7 +147,11 @@ def set_admin_step(message):
return
email = args_message[0]
is_admin = args_message[1]
is_admin = False # default: False
if args_message[1].lower() == "true": # if user types true, set is_admin to true
is_admin = True
status = api_handler.set_admin(email, is_admin) # set admin in db
if(status == 200):
@ -303,7 +280,7 @@ def send_share_update(message):
def send_share_price(message):
str_share_price = share_fetcher.get_share_price(str(message.text))
bot.reply_to(message, str_share_price)
bot.reply_to(message, str_share_price) # add dollar symbol etc.
@bot.message_handler(commands=['allnews', 'Allnews']) # /allnews -> get all news
@ -504,8 +481,8 @@ def set_new_transaction_step(message):
amount = float(transaction_data[2])
price = float(transaction_data[3])
time = dt.datetime.now().isoformat()
#print("\n\n\n\n\n")
#print(f"{symbol},{amount},{price},{time}")
print("\n\n\n\n\n")
print(f"{isin},{amount},{price},{time}")
status = api_handler.set_transaction(user_id, desc, isin, amount, price, time)
if status == 200:

24
telegram_bot/bot_updates.py
View File

@ -2,21 +2,16 @@
script for regularly sending updates on shares and news based on user interval
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "05.04.2022"
__version__ = "1.0.1"
__date__ = "26.04.2022"
__version__ = "1.0.2"
__license__ = "None"
from calendar import month
from symtable import Symbol
from dotenv import load_dotenv
from shares.share_fetcher import get_share_price
import news.news_fetcher as news_fetcher
import time
import datetime
import os
from bot import bot
import sys
from multiprocessing import Process
from apscheduler.schedulers.background import BackgroundScheduler
from api_handling.api_handler import API_Handler
@ -146,16 +141,23 @@ def update_for_user(p_user_id, p_my_handler):
my_update_message = f'Symbol: {share_symbols[i]}\nCurrent Price per Share: {share_courses[i]}\nAmount owned: {share_amounts[i]}\nTotal Investment: {float(share_courses[i]) * float(share_amounts[i])}'
send_to_user(my_update_message, pUser_id=p_user_id)
else:
send_to_user("No shares found for your account. Check https://gruppe1.testsites.info/api to change your settings and add shares.", pUser_id=p_user_id)
send_to_user("No shares found for your account. Check https://gruppe1.testsites.info to change your settings and add shares.", pUser_id=p_user_id)
keywords = p_my_handler.get_user_keywords(p_user_id) # get keywords as array
if(keywords): # if keywords exist and array is not empty
send_to_user("If you haven't read yet: \nHere are some interesting news according to your keywords:", pUser_id=p_user_id)
for keyword in keywords:
news = news_fetcher.get_top_news_by_keyword(keyword)["articles"][0] # only use the most popular news
news_formatted = news_fetcher.format_article(news) # format for message
send_to_user(f"_keyword: {keyword}_\n\n{news_formatted}", pUser_id=p_user_id, md_mode=True) # send news with related keyword in Markdown
news = news_fetcher.get_top_news_by_keyword(keyword)["articles"]
if not news: # if empty news array
send_to_user(f"No news found for keyword _{keyword}_.", pUser_id=p_user_id, md_mode=True)
if news == None: # if news is none
send_to_user(f"Server error for keyword _{keyword}_.", pUser_id=p_user_id, md_mode=True)
else:
news_formatted = news_fetcher.format_article(news[0]) # format for message, only use the most popular article
send_to_user(f"_keyword: {keyword}_\n\n{news_formatted}", pUser_id=p_user_id, md_mode=True) # send news with related keyword in Markdown

6
telegram_bot/news/news_fetcher.py
View File

@ -2,15 +2,13 @@
script for news fetching (by keywords)
"""
__author__ = "Florian Kellermann, Linus Eickhoff"
__date__ = "15.03.2022"
__version__ = "0.0.1"
__date__ = "26.04.2022"
__version__ = "1.0.0"
__license__ = "None"
import sys
import os
import json
import requests
import datetime as dt
from newsapi import NewsApiClient
from dotenv import load_dotenv

1
telegram_bot/shares/share_fetcher.py
View File

@ -7,7 +7,6 @@ __version__ = "0.0.2"
__license__ = "None"
import yfinance
import json
def get_share_price(str_symbol):