69 lines
2.5 KiB
Python
69 lines
2.5 KiB
Python
"""
|
|
collect.py
|
|
|
|
Functions for collecting data for insertion into the databse or pulling data from the database for analysis
|
|
"""
|
|
from datetime import datetime, timedelta
|
|
from enum import Enum
|
|
from time import sleep
|
|
|
|
from sqlalchemy import func, text
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from data.database import Base, dal, MegaMillionsDraw, PowerballDraw
|
|
import util.drawing as drawutil
|
|
import util.environment as envutil
|
|
import util.scrape as scrapeutil
|
|
|
|
project_variables = envutil.load_environment_variables()
|
|
|
|
|
|
def get_record_count(table_name: str) -> int | None:
|
|
"""get the most recent draw record from the specified table"""
|
|
game_dict = drawutil.check_table_name(table_name=table_name)
|
|
if game_dict:
|
|
dal.connect()
|
|
session = dal.Session()
|
|
target_table = Base.metadata.tables.get(game_dict['db_table_name'])
|
|
return session.execute(text(f"select count(draw_date) from {target_table.name}")).first()[0]
|
|
else:
|
|
raise ValueError('An invalid table name was specified.')
|
|
|
|
|
|
def get_recent_draw_date(table_name: str) -> datetime | None:
|
|
"""get the most recent draw date from the specified table"""
|
|
game_dict = drawutil.check_table_name(table_name=table_name)
|
|
if game_dict:
|
|
dal.connect()
|
|
session = dal.Session()
|
|
target_table = Base.metadata.tables.get(game_dict['db_table_name'])
|
|
return session.execute(text(f"select max(draw_date) from {target_table.name}")).first()[0]
|
|
else:
|
|
raise ValueError('An invalid table name was specified.')
|
|
|
|
|
|
def get_missing_records(table_name: str) -> int:
|
|
"""collect and insert missing records, returning a count of insertions"""
|
|
# if the table_name is valid
|
|
game_dict = drawutil.check_table_name(table_name=table_name)
|
|
if game_dict:
|
|
# connect to the database and establish a session
|
|
dal.connect()
|
|
session = dal.Session()
|
|
# get the ORM table value
|
|
target_table = Base.metadata.tables.get(game_dict['db_table_name'])
|
|
record_count = get_record_count(table_name=table_name)
|
|
target_game = game_dict['game_name']
|
|
end_year = datetime.today().year + 1
|
|
start_year = game_dict['inception_year'] if record_count > 0 else get_recent_draw_date(table_name=table_name).year
|
|
|
|
# check to see if there are records in the table
|
|
if get_record_count(table_name) > 0:
|
|
pass
|
|
else:
|
|
pass
|
|
|
|
|
|
else:
|
|
raise ValueError('An invalid table name was specified.')
|