diff --git a/src/data_fetcher.py b/src/data_fetcher.py index 1766a96..7f98dbb 100644 --- a/src/data_fetcher.py +++ b/src/data_fetcher.py @@ -13,7 +13,7 @@ import holidays # Definice URL pro stahování dat YEARLY_DATA_URL = "https://www.cnb.cz/cs/financni-trhy/devizovy-trh/kurzy-devizoveho-trhu/kurzy-devizoveho-trhu/rok.txt" MONTHLY_DATA_URL = "https://www.cnb.cz/cs/financni-trhy/devizovy-trh/kurzy-devizoveho-trhu/kurzy-devizoveho-trhu/vybrane.txt" -DAILY_DATA_URL = "https://www.cnb.cz/cs/financni-trhy/devizovy-trh/kurzy-devizoveho-trhu/kurzy-devizoveho-trhu/denni_kurz.txt" +DAILY_DATA_URL = "https://www.cnb.cz/cs/financni-trhy/devizovy-trhu/kurzy-devizoveho-trhu/kurzy-devizoveho-trhu/denni_kurz.txt" # Export cesty k databázovému souboru DB_PATH = database.DB_PATH @@ -31,10 +31,34 @@ def set_debug_mode(debug): global DEBUG DEBUG = debug +def get_last_working_day_of_year(year): + """ + Vrátí poslední pracovní den daného roku (ne víkend, ne svátek). + + :param year: Rok + :return: Datum posledního pracovního dne ve formátu DD.MM.YYYY + """ + # Začneme 31.12.rok a jdeme zpět, dokud nenajdeme pracovní den + last_day = datetime(year, 12, 31) + + # Odečítáme dny, dokud nenajdeme pracovní den + # Omezení na 10 dní zpět, abychom se nezacyklili + for _ in range(10): + date_str = last_day.strftime("%d.%m.%Y") + + # Zkontrolujeme, zda je to pracovní den (není víkend ani svátek) + if not holidays.is_weekend(date_str) and not holidays.is_holiday(date_str): + return date_str + + last_day -= timedelta(days=1) + + # Pokud se nepodařilo najít pracovní den, vrátíme None + return None + def is_yearly_data_complete(year, output_dir="data"): """ Zkontroluje, zda roční data pro zadaný rok jsou kompletní podle kritérií: - 1. Poslední řádek obsahuje datum 31.12.rok + 1. Poslední řádek obsahuje datum posledního pracovního dne roku (ne nutně 31.12.rok) 2. První řádek obsahuje datum, který je ne-víkend a ne-svátek (1.1, 2.1, 3.1 nebo 4.1) :param year: Rok, který se má zkontrolovat. @@ -83,7 +107,7 @@ def is_yearly_data_complete(year, output_dir="data"): try: first_date_str = first_data_line[0] # Datum je vždy v prvním sloupci - # Zkontrolujeme, zda první datum je 1.1., 2.1., 3.1. nebo 4.1. a je to pracovní den + # Zkontrolujeme, zda první datum je 1.1., 2.1., 3.1., nebo 4.1. a je to pracovní den first_date = datetime.strptime(first_date_str, "%d.%m.%Y") if first_date.year != year: debug_print(f"První datum v souboru {filepath} není z roku {year}.") @@ -95,7 +119,7 @@ def is_yearly_data_complete(year, output_dir="data"): try: check_date = datetime(year, 1, day) check_date_str = check_date.strftime("%d.%m.%Y") - # Zkontrolujeme, zda je to pracovní den (ne víkend a ne svátek) + # Zkontrolujeme, zda je to pracovní den (není víkend a ne svátek) if not holidays.is_weekend(check_date_str) and not holidays.is_holiday(check_date_str): valid_first_dates.append(check_date_str) except ValueError: @@ -114,12 +138,11 @@ def is_yearly_data_complete(year, output_dir="data"): try: last_date_str = last_data_line[0] # Datum je vždy v prvním sloupci - # Zkontrolujeme, zda poslední datum je 31.12.rok - last_date = datetime.strptime(last_date_str, "%d.%m.%Y") - expected_last_date = datetime(year, 12, 31) + # Zkontrolujeme, zda poslední datum je poslední pracovní den roku + expected_last_date_str = get_last_working_day_of_year(year) - if last_date != expected_last_date: - debug_print(f"Poslední datum v souboru {filepath} ({last_date_str}) není 31.12.{year}.") + if last_date_str != expected_last_date_str: + debug_print(f"Poslední datum v souboru {filepath} ({last_date_str}) není poslední pracovní den roku {year} ({expected_last_date_str}).") return False except (ValueError, IndexError) as e: @@ -219,14 +242,11 @@ def download_yearly_data(year, output_dir="data", force=False): :param force: Pokud je True, data se stáhnou i v případě, že již existují a jsou konzistentní. :return: Cesta k vytvořenému CSV souboru. """ - # Pokud není vynucené stahování, zkontrolujeme, zda existující data jsou kompletní + # Pokud není vynucené stahování, zkontrolujeme konzistenci existujících dat if not force: - # Zkontrolujeme, zda soubor existuje a je kompletní if is_yearly_data_complete(year, output_dir): debug_print(f"Roční data pro rok {year} jsou kompletní. Není nutné je stahovat znovu.") return os.path.join(output_dir, f"{year}.csv") - else: - debug_print(f"Roční data pro rok {year} nejsou kompletní. Stahuji znovu...") url = f"{YEARLY_DATA_URL}?rok={year}" debug_print(f"Stahuji data z: {url}")