86 lines
3.6 KiB
Python
86 lines
3.6 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 doctest import script_from_examples
|
|
from time import sleep
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from data.database import Base, dal
|
|
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.query(func.count(target_table.columns.draw_date)).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.query(func.max(target_table.columns.draw_date)).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:
|
|
insert_count = 0
|
|
# connect to the database and establish a session
|
|
dal.connect()
|
|
session = dal.Session()
|
|
|
|
date_format = project_variables['LP_DATE_PASS_FORMAT']
|
|
|
|
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
|
|
recent_draw_date = (get_recent_draw_date(table_name=table_name) + timedelta(days=1)).strftime(date_format)
|
|
following_draw_date = datetime.strptime(drawutil.get_next_draw_date(game=target_game,
|
|
from_date=recent_draw_date), date_format)
|
|
start_year = game_dict['inception_year'] if record_count == 0 else following_draw_date.year
|
|
check_date = datetime.strptime(drawutil.get_previous_draw_date(game=target_game, from_date=
|
|
drawutil.get_next_draw_date(game=target_game)), date_format).date()
|
|
|
|
if session.query(target_table).filter_by(draw_date=check_date).first() is None:
|
|
# loop over the range of years and insert missing records
|
|
for year in range(start_year, end_year):
|
|
scrape_data = scrapeutil.scrape_game_history(game=target_game, year=year)
|
|
# convert and insert records, discarding duplicate dates
|
|
for draw_date, draw_info in scrape_data.items():
|
|
record = game_dict['record_type'](draw_date=draw_date, draw_result=draw_info)
|
|
try:
|
|
session.add(record)
|
|
session.commit()
|
|
insert_count += 1
|
|
except IntegrityError:
|
|
session.rollback()
|
|
sleep(30) # sleep between calls to the site for data to avoid being blocked
|
|
|
|
return insert_count
|
|
else:
|
|
raise ValueError('An invalid table name was specified.')
|