AmineTMIRI/backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update cron job

Browse Source
This commit is contained in:
MuslemRahimi 2025-01-29 00:14:53 +01:00
parent 80e7e4d863
commit aa14555076
1 changed files with 37 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
app/cron_potus_tracker.py
View File

@ -11,10 +11,23 @@ import asyncio
import orjson import orjson
from dotenv import load_dotenv from dotenv import load_dotenv
import os import os
import sqlite3
import pandas as pd
load_dotenv() load_dotenv()
geolocator = Nominatim(user_agent="myGeocodingApp/1.0 (your-email@example.com)") geolocator = Nominatim(user_agent="myGeocodingApp/1.0 (your-email@example.com)")
query_template = """
SELECT
date, close
FROM
"{symbol}"
WHERE
date BETWEEN ? AND ?
"""
def save_json(data): def save_json(data):
path = "json/tracker/potus" path = "json/tracker/potus"
os.makedirs(path, exist_ok=True) os.makedirs(path, exist_ok=True)
@ -78,13 +91,13 @@ def get_bills():
read_more_buttons[0].click() read_more_buttons[0].click()
# Wait for the popup to become visible # Wait for the popup to become visible
print("Waiting for the popup to appear...") #print("Waiting for the popup to appear...")
WebDriverWait(driver, 10).until( WebDriverWait(driver, 10).until(
EC.visibility_of_element_located((By.ID, "popup-container")) # Wait until popup is visible EC.visibility_of_element_located((By.ID, "popup-container")) # Wait until popup is visible
) )
# Extract content from the popup # Extract content from the popup
print("Popup appeared, extracting content...") #print("Popup appeared, extracting content...")
popup_title = driver.find_element(By.ID, "popup-title").text popup_title = driver.find_element(By.ID, "popup-title").text
popup_content = driver.find_element(By.ID, "popup-content").text popup_content = driver.find_element(By.ID, "popup-content").text
@ -95,7 +108,7 @@ def get_bills():
# Close the popup (optional) # Close the popup (optional)
close_button = driver.find_element(By.ID, "popup-close-button") close_button = driver.find_element(By.ID, "popup-close-button")
close_button.click() close_button.click()
print("Popup closed.") #print("Popup closed.")
# Append data to list # Append data to list
data.append({ data.append({
@ -116,6 +129,19 @@ def get_bills():
async def get_data(): async def get_data():
bill_data = get_bills() bill_data = get_bills()
query = query_template.format(symbol='SPY')
etf_con = sqlite3.connect('etf.db')
etf_cursor = etf_con.cursor()
etf_cursor.execute("PRAGMA journal_mode = wal")
df = pd.read_sql_query(query, etf_con, params=("2025-01-20", '2025-01-27'))
if not df.empty:
df['changesPercentage'] = (df['close'].pct_change() * 100).round(2)
sp500_list = df.dropna().to_dict(orient="records") # Drop NaN values and convert to list
etf_con.close()
return_since = round((sp500_list[-1]['close']/sp500_list[0]['close']-1)*100,2)
url = "https://media-cdn.factba.se/rss/json/trump/calendar-full.json" url = "https://media-cdn.factba.se/rss/json/trump/calendar-full.json"
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
@ -159,8 +185,13 @@ async def get_data():
print(f"Latitude: {latitude}, Longitude: {longitude}") print(f"Latitude: {latitude}, Longitude: {longitude}")
break break
res_dict = {'city': city, 'lon': longitude, 'lat': latitude, 'history': data, 'billData': bill_data} for item in data:
for price_item in sp500_list:
if item['date'] == price_item['date']:
item['changesPercentage'] = price_item['changesPercentage']
break
res_dict = {'returnSince': return_since,'city': city, 'lon': longitude, 'lat': latitude, 'history': data, 'billData': bill_data}
save_json(res_dict) save_json(res_dict)
asyncio.run(get_data()) asyncio.run(get_data())