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

Merge branch 'main' into bot

Browse Source
This commit is contained in:
Florian Kellermann
2022-05-13 12:32:58 +02:00
committed by GitHub
parent e06c43b762 3777ed76f0
commit aadfba74f0
146 changed files with 129 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
telegram_bot/bot_updates.py
View File

@@ -10,7 +10,7 @@ import os
import sys
import time
from apscheduler.schedulers.background import BackgroundScheduler # scheduler for cron
from apscheduler.schedulers.background import BackgroundScheduler # scheduler for cron
from dotenv import load_dotenv
import helper_functions as hf
@@ -70,12 +70,12 @@ def update_crontab(p_my_handler):
print("No users found, trying again.")
update_crontab(p_my_handler)
return
user_ids = []
user_crontab = []
for element in all_users:
if element["cron"] != '' and element["telegram_user_id"] != '': # check if both values are existing so I have consistent data
if element["cron"] != '' and element["telegram_user_id"] != '': # check if both values are existing so I have consistent data
try:
user_ids.append(int(element["telegram_user_id"]))
try:
@@ -84,15 +84,13 @@ def update_crontab(p_my_handler):
user_ids.pop() # if something goes wrong with cron I have to delete matching user id
except: continue
print(user_ids)
update_based_on_crontab(user_ids, user_crontab, p_my_handler)
update_crontab(p_my_handler) # restart the update after time sleep
update_crontab(p_my_handler) # restart the update after time sleep
def update_based_on_crontab(p_user_ids, p_user_crontab, p_my_handler):
""" Check all the crontab codes and add jobs to start in time
:type p_user_ids: array
@@ -108,19 +106,20 @@ def update_based_on_crontab(p_user_ids, p_user_crontab, p_my_handler):
:rtype: none
"""
my_scheduler = BackgroundScheduler() # schedule sends based on cron
my_scheduler = BackgroundScheduler() # schedule sends based on cron
for i in range(len(p_user_ids)):
cron_split = p_user_crontab[i].split(" ") # split it up to use in scheduler
cron_split = p_user_crontab[i].split(" ") # split it up to use in scheduler
print(cron_split[4], cron_split[1], cron_split[0], cron_split[3], cron_split[2])
my_scheduler.add_job(update_for_user, 'cron', day_of_week=cron_split[4], hour=cron_split[1], minute=cron_split[0], month=cron_split[3], day=cron_split[2], args=(p_user_ids[i], p_my_handler))
my_scheduler.start()
time.sleep( 600 ) # scheduler runs in background and I wait 10mins
my_scheduler.shutdown() # after this the new crontabs will be loaded
time.sleep(600) # scheduler runs in background and I wait 10mins
my_scheduler.shutdown() # after this the new crontabs will be loaded
def update_for_user(p_user_id, p_my_handler):
""" Pull shares and send updates for specific user id
:type p_user_id: integer
@@ -137,7 +136,6 @@ def update_for_user(p_user_id, p_my_handler):
share_amounts = []
my_portfolio = p_my_handler.get_user_portfolio(p_user_id) # get all existing shares for user
my_portfolio = None
if my_portfolio!=None:
for element in my_portfolio:
@@ -152,6 +150,7 @@ def update_for_user(p_user_id, p_my_handler):
shares = p_my_handler.get_user_shares(p_user_id) # all interest shares
if len(share_symbols) != 0: # iterate through all shares
for i in range(len(share_symbols)):
my_price = share_fetcher.get_share_price_no_currency(share_symbols[i])
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'
@@ -172,10 +171,10 @@ def update_for_user(p_user_id, p_my_handler):
news = news_fetcher.get_top_news_by_keyword(keyword)["articles"]
keyword = hf.make_markdown_proof(keyword)
if not news: # if empty news array
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)
elif news == None: # if news is none
elif 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