330 lines
12 KiB
Python
330 lines
12 KiB
Python
import requests
|
|
import os
|
|
import csv
|
|
from datetime import datetime, timedelta
|
|
import sys
|
|
|
|
# Přidání adresáře src do sys.path, aby bylo možné importovat moduly
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
|
|
|
|
import database
|
|
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"
|
|
|
|
# Export cesty k databázovému souboru
|
|
DB_PATH = database.DB_PATH
|
|
|
|
def check_yearly_data_consistency(year, output_dir="data"):
|
|
"""
|
|
Zkontroluje, zda roční data pro zadaný rok obsahují záznamy za poslední 3 pracovní dny.
|
|
|
|
:param year: Rok, pro který se má zkontrolovat konzistence dat.
|
|
:param output_dir: Adresář, kde se nachází CSV soubor s ročními daty.
|
|
:return: True pokud jsou data konzistentní, jinak False.
|
|
"""
|
|
filename = f"{year}.csv"
|
|
filepath = os.path.join(output_dir, filename)
|
|
|
|
# Zkontrolujeme, zda soubor existuje
|
|
if not os.path.exists(filepath):
|
|
print(f"Soubor {filepath} neexistuje.")
|
|
return False
|
|
|
|
# Načteme data z CSV souboru
|
|
try:
|
|
with open(filepath, 'r', encoding='utf-8') as csvfile:
|
|
reader = csv.reader(csvfile)
|
|
lines = list(reader)
|
|
except IOError as e:
|
|
print(f"Chyba při čtení souboru {filepath}: {e}")
|
|
return False
|
|
|
|
if len(lines) < 2:
|
|
print(f"Soubor {filepath} je prázdný nebo obsahuje pouze hlavičku.")
|
|
return False
|
|
|
|
# Získáme seznam dat z CSV (první sloupec)
|
|
dates_in_file = []
|
|
for line in lines[1:]: # Přeskočíme hlavičku
|
|
if line and line[0]: # Zkontrolujeme, že řádek a první sloupec nejsou prázdné
|
|
dates_in_file.append(line[0])
|
|
|
|
if not dates_in_file:
|
|
print(f"Soubor {filepath} neobsahuje žádná data.")
|
|
return False
|
|
|
|
# Zkontrolujeme poslední 3 pracovní dny
|
|
today = datetime.now()
|
|
checked_dates = []
|
|
|
|
# Projdeme posledních 7 dní (pro jistotu) a vybereme 3 pracovní dny
|
|
current_date = today - timedelta(days=1) # Začneme včerejškem
|
|
days_checked = 0
|
|
|
|
while len(checked_dates) < 3 and days_checked < 7:
|
|
date_str = current_date.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):
|
|
# Zkontrolujeme, zda je rok správný
|
|
try:
|
|
date_obj = datetime.strptime(date_str, "%d.%m.%Y")
|
|
if date_obj.year == year:
|
|
checked_dates.append(date_str)
|
|
except ValueError:
|
|
pass
|
|
|
|
current_date -= timedelta(days=1)
|
|
days_checked += 1
|
|
|
|
# Pokud nemáme žádné pracovní dny k ověření, vrátíme True (nemáme co kontrolovat)
|
|
if not checked_dates:
|
|
print(f"Pro rok {year} nebyly nalezeny žádné pracovní dny k ověření v posledním týdnu.")
|
|
return True
|
|
|
|
# Zkontrolujeme, zda všechny pracovní dny jsou v souboru
|
|
missing_dates = []
|
|
for date_str in checked_dates:
|
|
if date_str not in dates_in_file:
|
|
missing_dates.append(date_str)
|
|
|
|
if missing_dates:
|
|
print(f"V souboru {filepath} chybí záznamy pro následující pracovní dny: {', '.join(missing_dates)}")
|
|
return False
|
|
else:
|
|
print(f"Roční data pro rok {year} jsou konzistentní.")
|
|
return True
|
|
|
|
def download_yearly_data(year, output_dir="data", force=False):
|
|
"""
|
|
Stáhne roční data z CNB a uloží je do CSV souboru a databáze.
|
|
|
|
:param year: Rok, pro který se mají stáhnout data (např. 2020).
|
|
:param output_dir: Adresář, kam se má CSV soubor uložit.
|
|
: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 konzistenci existujících dat
|
|
if not force:
|
|
if check_yearly_data_consistency(year, output_dir):
|
|
print(f"Roční data pro rok {year} jsou aktuální. Není nutné je stahovat znovu.")
|
|
return os.path.join(output_dir, f"{year}.csv")
|
|
|
|
url = f"{YEARLY_DATA_URL}?rok={year}"
|
|
print(f"Stahuji data z: {url}")
|
|
|
|
try:
|
|
response = requests.get(url)
|
|
response.raise_for_status() # Vyvolá výjimku pro chybné HTTP kódy
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Chyba při stahování dat: {e}")
|
|
return None
|
|
|
|
# Vytvoření jména souboru
|
|
filename = f"{year}.csv"
|
|
filepath = os.path.join(output_dir, filename)
|
|
|
|
# Zpracování dat a uložení do CSV
|
|
lines = response.text.strip().split('\n')
|
|
if not lines:
|
|
print("Stažený soubor je prázdný.")
|
|
return None
|
|
|
|
# První řádek obsahuje hlavičku
|
|
header = lines[0].split('|')
|
|
# Převod názvů sloupců do formátu vhodného pro CSV (např. nahrazení mezer podtržítky)
|
|
header = [col.replace(' ', '_').replace('-', '_') for col in header]
|
|
|
|
# Zpracování datových řádků
|
|
data_rows = []
|
|
for line in lines[1:]:
|
|
if not line.strip():
|
|
continue
|
|
# Nahrazení českých desetinných čárek za tečky
|
|
row = [item.replace(',', '.') for item in line.split('|')]
|
|
data_rows.append(row)
|
|
|
|
# Uložení do databáze
|
|
date_str = row[0] # Datum je vždy v prvním sloupci
|
|
for i in range(1, len(header)):
|
|
col_name = header[i]
|
|
if col_name == "Datum":
|
|
continue
|
|
|
|
# Název sloupce je ve formátu "1_USD", "100_JPY" atd.
|
|
# Rozdělíme ho na množství a kód měny
|
|
parts = col_name.split('_')
|
|
if len(parts) >= 2:
|
|
try:
|
|
amount = int(parts[0])
|
|
currency_code = '_'.join(parts[1:]) # Pro případ měn se složeným názvem
|
|
rate = float(row[i])
|
|
database.insert_rate(date_str, currency_code, amount, rate)
|
|
except (ValueError, IndexError):
|
|
# Přeskočit sloupce, které nelze parsovat
|
|
continue
|
|
|
|
# Zápis do CSV souboru
|
|
try:
|
|
with open(filepath, 'w', newline='', encoding='utf-8') as csvfile:
|
|
writer = csv.writer(csvfile)
|
|
writer.writerow(header)
|
|
writer.writerows(data_rows)
|
|
print(f"Data byla úspěšně uložena do: {filepath}")
|
|
return filepath
|
|
except IOError as e:
|
|
print(f"Chyba při zápisu do souboru: {e}")
|
|
return None
|
|
|
|
def download_monthly_data(currency_code, start_date, end_date, output_dir="data"):
|
|
"""
|
|
Stáhne měsíční data pro zadanou měnu a časové období z CNB a uloží je do CSV souboru a databáze.
|
|
|
|
:param currency_code: Kód měny (např. USD).
|
|
:param start_date: Počáteční datum ve formátu DD.MM.YYYY.
|
|
:param end_date: Koncové datum ve formátu DD.MM.YYYY.
|
|
:param output_dir: Adresář, kam se má CSV soubor uložit.
|
|
:return: Cesta k vytvořenému CSV souboru.
|
|
"""
|
|
url = f"{MONTHLY_DATA_URL}?od={start_date}&do={end_date}&mena={currency_code}&format=txt"
|
|
print(f"Stahuji data z: {url}")
|
|
|
|
try:
|
|
response = requests.get(url)
|
|
response.raise_for_status()
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Chyba při stahování dat: {e}")
|
|
return None
|
|
|
|
# Vytvoření jména souboru
|
|
filename = f"{start_date.replace('.', '_')}_to_{end_date.replace('.', '_')}_{currency_code}.csv"
|
|
filepath = os.path.join(output_dir, filename)
|
|
|
|
# Zpracování dat a uložení do CSV
|
|
lines = response.text.strip().split('\n')
|
|
if not lines:
|
|
print("Stažený soubor je prázdný.")
|
|
return None
|
|
|
|
# První řádek obsahuje informace o měně, druhý hlavičku dat
|
|
currency_info = lines[0]
|
|
header = lines[1].split('|')
|
|
header = [col.replace(' ', '_').replace('-', '_') for col in header]
|
|
|
|
# Zpracování datových řádků
|
|
data_rows = []
|
|
for line in lines[2:]:
|
|
if not line.strip():
|
|
continue
|
|
row = [item.replace(',', '.') for item in line.split('|')]
|
|
data_rows.append(row)
|
|
|
|
# Uložení do databáze
|
|
# Formát řádku: Datum|Kurz
|
|
if len(row) >= 2:
|
|
date_str = row[0]
|
|
try:
|
|
# Z currency_info získáme množství
|
|
# Příklad: "Měna: USD|Množství: 1"
|
|
parts = currency_info.split('|')
|
|
amount_part = next((p for p in parts if p.startswith("Množství:")), None)
|
|
if amount_part:
|
|
amount = int(amount_part.split(':')[1].strip())
|
|
rate = float(row[1])
|
|
database.insert_rate(date_str, currency_code, amount, rate)
|
|
except (ValueError, IndexError):
|
|
print(f"Chyba při parsování řádku: {line}")
|
|
|
|
# Zápis do CSV souboru
|
|
try:
|
|
with open(filepath, 'w', newline='', encoding='utf-8') as csvfile:
|
|
csvfile.write(f"{currency_info}\n") # Uložení informací o měně
|
|
writer = csv.writer(csvfile)
|
|
writer.writerow(header)
|
|
writer.writerows(data_rows)
|
|
print(f"Data byla úspěšně uložena do: {filepath}")
|
|
return filepath
|
|
except IOError as e:
|
|
print(f"Chyba při zápisu do souboru: {e}")
|
|
return None
|
|
|
|
def download_daily_data(date, output_dir="data"):
|
|
"""
|
|
Stáhne denní data z CNB pro zadané datum a uloží je do CSV souboru a databáze.
|
|
|
|
:param date: Datum ve formátu DD.MM.YYYY.
|
|
:param output_dir: Adresář, kam se má CSV soubor uložit.
|
|
:return: Cesta k vytvořenému CSV souboru.
|
|
"""
|
|
url = f"{DAILY_DATA_URL}?date={date}"
|
|
print(f"Stahuji data z: {url}")
|
|
|
|
try:
|
|
response = requests.get(url)
|
|
response.raise_for_status()
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Chyba při stahování dat: {e}")
|
|
return None
|
|
|
|
# Vytvoření jména souboru
|
|
filename = f"daily_{date.replace('.', '_')}.csv"
|
|
filepath = os.path.join(output_dir, filename)
|
|
|
|
# Zpracování dat a uložení do CSV
|
|
lines = response.text.strip().split('\n')
|
|
if not lines:
|
|
print("Stažený soubor je prázdný.")
|
|
return None
|
|
|
|
# První řádek obsahuje datum a číslo, druhý hlavičku
|
|
date_info = lines[0]
|
|
header = lines[1].split('|')
|
|
header = [col.replace(' ', '_').replace('-', '_') for col in header]
|
|
|
|
# Zpracování datových řádků
|
|
data_rows = []
|
|
for line in lines[2:]:
|
|
if not line.strip():
|
|
continue
|
|
row = [item.replace(',', '.') for item in line.split('|')]
|
|
data_rows.append(row)
|
|
|
|
# Uložení do databáze
|
|
# Formát řádku: země|měna|množství|kód|kurz
|
|
if len(row) >= 5:
|
|
try:
|
|
# country = row[0]
|
|
# currency_name = row[1]
|
|
amount = int(row[2])
|
|
currency_code = row[3]
|
|
rate = float(row[4])
|
|
database.insert_rate(date, currency_code, amount, rate)
|
|
except (ValueError, IndexError):
|
|
print(f"Chyba při parsování řádku: {line}")
|
|
|
|
# Zápis do CSV souboru
|
|
try:
|
|
with open(filepath, 'w', newline='', encoding='utf-8') as csvfile:
|
|
csvfile.write(f"{date_info}\n") # Uložení informací o datu
|
|
writer = csv.writer(csvfile)
|
|
writer.writerow(header)
|
|
writer.writerows(data_rows)
|
|
print(f"Data byla úspěšně uložena do: {filepath}")
|
|
return filepath
|
|
except IOError as e:
|
|
print(f"Chyba při zápisu do souboru: {e}")
|
|
return None
|
|
|
|
# Příklad použití
|
|
if __name__ == "__main__":
|
|
# Pro testování lze spustit tento skript samostatně
|
|
database.init_db()
|
|
downloaded_file = download_yearly_data(2020)
|
|
if downloaded_file:
|
|
print(f"Stažený soubor: {downloaded_file}")
|
|
else:
|
|
print("Stažení se nezdařilo.") |