From 2a94c4c24611274d11762832764fe85516ae96f5 Mon Sep 17 00:00:00 2001 From: H4CK3R-01 Date: Thu, 12 May 2022 09:09:05 +0200 Subject: [PATCH 1/2] Reformatting --- telegram_bot/bot_updates.py | 67 ++++++++++++++-------------- telegram_bot/shares/share_fetcher.py | 45 +++++++++---------- 2 files changed, 56 insertions(+), 56 deletions(-) diff --git a/telegram_bot/bot_updates.py b/telegram_bot/bot_updates.py index 8f391ef..95396f1 100644 --- a/telegram_bot/bot_updates.py +++ b/telegram_bot/bot_updates.py @@ -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 @@ -63,30 +63,30 @@ def update_crontab(p_my_handler): global user_crontab global user_ids - - all_users = p_my_handler.get_all_users() # get all users so crontabs can update for everybody - + + all_users = p_my_handler.get_all_users() # get all users so crontabs can update for everybody + 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: user_crontab.append(str(element["cron"])) - except: - user_ids.pop() # if something goes wrong with cron I have to delete matching user id - except: continue - - + except: + 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 @@ -102,19 +102,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 @@ -129,9 +130,9 @@ def update_for_user(p_user_id, p_my_handler): """ share_symbols = [] share_amounts = [] - - my_portfolio = p_my_handler.get_user_portfolio(p_user_id) # get all existing shares for user - + + my_portfolio = p_my_handler.get_user_portfolio(p_user_id) # get all existing shares for user + for element in my_portfolio: if element["count"] != '' and element["isin"] != '': print(element["count"], element["isin"]) @@ -139,11 +140,11 @@ def update_for_user(p_user_id, p_my_handler): share_amounts.append(element["count"]) my_user = p_my_handler.get_user(p_user_id) - send_to_user("Hello %s this is your share update for today:"%str(my_user["username"]), pUser_id=p_user_id) - - shares = p_my_handler.get_user_shares(p_user_id) # all interest shares - - if len(share_symbols) != 0: # iterate through all shares + send_to_user("Hello %s this is your share update for today:" % str(my_user["username"]), pUser_id=p_user_id) + + 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' @@ -164,10 +165,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 diff --git a/telegram_bot/shares/share_fetcher.py b/telegram_bot/shares/share_fetcher.py index 66cf0a4..e283444 100644 --- a/telegram_bot/shares/share_fetcher.py +++ b/telegram_bot/shares/share_fetcher.py @@ -23,23 +23,23 @@ def get_share_price(str_search_for): try: search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], n_results=1, countries=['germany']) - currency = str(search_result.retrieve_currency()) # retrieve currency from data + currency = str(search_result.retrieve_currency()) # retrieve currency from data # should always be Euro because of countries=['germany'] - - recent_data = pandas.DataFrame(search_result.retrieve_recent_data()) # stock prices of last few days - - stock_price = recent_data.iloc[-1]["Close"] # retrieve latest stock price - + + recent_data = pandas.DataFrame(search_result.retrieve_recent_data()) # stock prices of last few days + + stock_price = recent_data.iloc[-1]["Close"] # retrieve latest stock price + stock_price = round(float(stock_price), 2) - - str_return =str(stock_price) + " " + str(currency) # return + currency - + + str_return = str(stock_price) + " " + str(currency) # return + currency + return str_return - - except RuntimeError: # if no shares are found for germany (e.g. isin: US.....) + + except RuntimeError: # if no shares are found for germany (e.g. isin: US.....) try: - my_Converter = CurrencyConverter() # need a currency converter - + my_Converter = CurrencyConverter() # need a currency converter + search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], n_results=1) currency = str(search_result.retrieve_currency()) @@ -47,10 +47,10 @@ def get_share_price(str_search_for): recent_data = pandas.DataFrame(search_result.retrieve_recent_data()) stock_price = recent_data.iloc[-1]["Close"] - - #convert stock price from currency to EUR - stock_price = my_Converter.convert(float(stock_price), str(currency), 'EUR') - + + # convert stock price from currency to EUR + stock_price = my_Converter.convert(float(stock_price), str(currency), 'EUR') + stock_price = round(float(stock_price), 2) str_return = str(stock_price) + " EUR" @@ -109,17 +109,16 @@ def get_share_information(str_search_for): def get_share_information_markdown(str_search_for): - try: search_result = investpy.search_quotes(text=str_search_for, products=['stocks'], - countries=['germany'], n_results=1) - + countries=['germany'], n_results=1) + except RuntimeError as e: - return hf.make_markdown_proof(f"no shares found for \"{str_search_for}\"") # if no shares are found, make error message markdown proof and return + return hf.make_markdown_proof(f"no shares found for \"{str_search_for}\"") # if no shares are found, make error message markdown proof and return except ConnectionError as e: - return hf.make_markdown_proof(f"connection not possible. Try again later.") # if no connection, make error message markdown proof and return - + return hf.make_markdown_proof(f"connection not possible. Try again later.") # if no connection, make error message markdown proof and return + str_return = f'*{hf.make_markdown_proof(search_result.name)}*\n_{hf.make_markdown_proof(search_result.symbol)}_\nworth: {hf.make_markdown_proof(get_share_price(str_search_for))}' return str_return -- 2.45.2 From d715734353ade9ca5043ee545dccfeeb2dbb5aa1 Mon Sep 17 00:00:00 2001 From: Linus E <75929322+Rripped@users.noreply.github.com> Date: Thu, 12 May 2022 09:28:37 +0200 Subject: [PATCH 2/2] added more markdown symbols --- telegram_bot/helper_functions.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/telegram_bot/helper_functions.py b/telegram_bot/helper_functions.py index 1262d54..ca6d4d2 100644 --- a/telegram_bot/helper_functions.py +++ b/telegram_bot/helper_functions.py @@ -57,6 +57,8 @@ def make_markdown_proof(text): # used to avoid errors related to markdown parse text = text.replace("^", "\\^") text = text.replace("$", "\\$") text = text.replace("%", "\\%") + text = text.replace("=", "\\=") + text = text.replace("@", "\\@") return text -- 2.45.2