feat: Add JSON output support and auto-download functionality

- Add --json CLI flag for structured JSON output across all commands
- Implement JSON formatting functions for different data types:
  * Single rate lookups with fallback information
  * Unified tax rates (single year and multi-year)
  * Last available rates
  * Error responses with codes and details
- Add auto-download functionality for missing monthly data in tax calculations
- Modify calculate_tax_yearly_average to automatically fetch missing months
- Add rate limiting (1s delay) between API calls to be respectful
- Update CLI argument parsing and output logic for JSON/text modes
- Maintain full backward compatibility - existing commands work unchanged
- Enhance documentation with JSON usage examples and schema
- Update help text to include new --json option

Features:
- JSON output for programmatic consumption
- Automatic data fetching for incomplete years
- Structured error handling
- Comprehensive documentation updates

Breaking changes: None (fully backward compatible)
This commit is contained in:
kdusek
2026-01-12 22:48:59 +01:00
parent 51193ab933
commit ed5d126d77
3 changed files with 652 additions and 222 deletions

View File

@@ -3,6 +3,7 @@
import argparse
import sys
import os
import json
from datetime import datetime
# Přidání adresáře src do sys.path, aby bylo možné importovat moduly
@@ -17,16 +18,18 @@ import rate_reporter
# Global debug flag
DEBUG = False
def debug_print(*args, **kwargs):
"""Print debug messages only if debug mode is enabled."""
if DEBUG:
print(*args, **kwargs)
def set_debug_mode(debug):
"""Set the debug mode for this module."""
global DEBUG
DEBUG = debug
# Nastavíme debug mód pro všechny moduly
database.set_debug_mode(DEBUG)
data_fetcher.set_debug_mode(DEBUG)
@@ -34,18 +37,91 @@ def set_debug_mode(debug):
rate_finder.set_debug_mode(DEBUG)
rate_reporter.set_debug_mode(DEBUG)
def format_single_rate_json(
currency, rate, requested_date, actual_date=None, fallback=False
):
"""Format single rate lookup as JSON."""
data = {
"currency": currency,
"rate": float(rate) if rate is not None else None,
"date": requested_date,
"timestamp": datetime.now().isoformat() + "Z",
}
if fallback and actual_date:
data["actual_date"] = actual_date
data["fallback"] = True
return data
def format_tax_rate_json(currency, year, rate, monthly_rates=None):
"""Format unified tax rate as JSON."""
data = {
"currency": currency,
"year": year,
"unified_rate": float(rate) if rate is not None else None,
"calculation_date": datetime.now().isoformat() + "Z",
}
if monthly_rates:
data["monthly_rates"] = monthly_rates
return data
def format_multi_year_json(currency, year_results):
"""Format multi-year stats as JSON."""
data = {
"currency": currency,
"results": [
{"year": year, "unified_rate": float(rate) if rate is not None else None}
for year, rate in year_results
],
"calculation_date": datetime.now().isoformat() + "Z",
}
return data
def format_last_rate_json(currency, rate, date):
"""Format last available rate as JSON."""
data = {
"currency": currency,
"rate": float(rate) if rate is not None else None,
"date": date,
"timestamp": datetime.now().isoformat() + "Z",
}
return data
def format_error_json(error_msg, error_code=None, details=None):
"""Format error response as JSON."""
data = {"error": error_msg, "timestamp": datetime.now().isoformat() + "Z"}
if error_code:
data["code"] = error_code
if details:
data["details"] = details
return data
def output_json(data):
"""Output data as formatted JSON."""
print(json.dumps(data, indent=2, ensure_ascii=False))
def check_and_update_yearly_data():
"""
Zkontroluje konzistenci ročních dat pro aktuální rok a případně je aktualizuje.
"""
current_year = datetime.now().year
debug_print(f"Kontroluji konzistenci ročních dat pro rok {current_year}...")
# Zkontrolujeme konzistenci dat
is_consistent = data_fetcher.check_yearly_data_consistency(current_year, output_dir="data")
is_consistent = data_fetcher.check_yearly_data_consistency(
current_year, output_dir="data"
)
if not is_consistent:
debug_print(f"Roční data pro rok {current_year} nejsou konzistentní. Stahuji aktualizovaná data...")
debug_print(
f"Roční data pro rok {current_year} nejsou konzistentní. Stahuji aktualizovaná data..."
)
# Ujistěme se, že adresář data existuje
os.makedirs("data", exist_ok=True)
# Stáhneme roční data znovu
@@ -53,81 +129,83 @@ def check_and_update_yearly_data():
else:
debug_print(f"Roční data pro rok {current_year} jsou aktuální.")
def main():
global DEBUG
# Inicializace databáze
database.init_db()
parser = argparse.ArgumentParser(description="Stahování a správa kurzů měn z ČNB.")
parser.add_argument(
"--year",
type=int,
help="Rok, pro který se mají stáhnout data (např. 2020)."
"--year", type=int, help="Rok, pro který se mají stáhnout data (např. 2020)."
)
parser.add_argument(
"-c", "--currency",
"-c",
"--currency",
type=str,
help="Kód měny (např. USD) pro měsíční stahování, vyhledání kurzu nebo generování reportu."
help="Kód měny (např. USD) pro měsíční stahování, vyhledání kurzu nebo generování reportu.",
)
parser.add_argument(
"--start-date",
type=str,
help="Počáteční datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY."
help="Počáteční datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY.",
)
parser.add_argument(
"--end-date",
type=str,
help="Koncové datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY."
help="Koncové datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY.",
)
parser.add_argument(
"--date",
type=str,
help="Datum pro stažení denních kurzů ve formátu DD.MM.YYYY."
help="Datum pro stažení denních kurzů ve formátu DD.MM.YYYY.",
)
parser.add_argument(
"--get-rate", "-d",
"--get-rate",
"-d",
type=str,
help="Vyhledá kurz pro zadané datum. Formát: DD.MM.YYYY"
help="Vyhledá kurz pro zadané datum. Formát: DD.MM.YYYY",
)
parser.add_argument(
"--auto-download",
action="store_true",
help="Automaticky stáhne denní data, pokud je po 14:30 a kurz pro dnešní datum není k dispozici."
help="Automaticky stáhne denní data, pokud je po 14:30 a kurz pro dnešní datum není k dispozici.",
)
parser.add_argument(
"--report-year",
type=int,
help="Rok, pro který se má vygenerovat report kurzů."
"--report-year", type=int, help="Rok, pro který se má vygenerovat report kurzů."
)
parser.add_argument(
"--report-month",
type=int,
help="Měsíc, pro který se má vygenerovat report kurzů (1-12). Vyžaduje --report-year."
help="Měsíc, pro který se má vygenerovat report kurzů (1-12). Vyžaduje --report-year.",
)
parser.add_argument(
"--report-period",
nargs=2,
metavar=('START_DATE', 'END_DATE'),
help="Období, pro které se má vygenerovat report kurzů. Formát: DD.MM.YYYY DD.MM.YYYY"
metavar=("START_DATE", "END_DATE"),
help="Období, pro které se má vygenerovat report kurzů. Formát: DD.MM.YYYY DD.MM.YYYY",
)
parser.add_argument(
"--stats",
nargs='?',
nargs="?",
const=True,
type=int,
help="Vygeneruje 'Jednotný kurz' pro daňové účely podle metodiky ČNB. "
"Pokud je zadán rok, vytvoří kurz pro konkrétní rok. "
"Pokud není rok zadán, vytvoří kurzy pro všechny roky s dostupnými daty."
"Pokud je zadán rok, vytvoří kurz pro konkrétní rok. "
"Pokud není rok zadán, vytvoří kurzy pro všechny roky s dostupnými daty.",
)
parser.add_argument(
"--debug",
"--debug", action="store_true", help="Zobrazí podrobné ladicí informace."
)
parser.add_argument(
"--json",
action="store_true",
help="Zobrazí podrobné ladicí informace."
help="Výstup ve formátu JSON místo prostého textu pro programové zpracování.",
)
args = parser.parse_args()
# Pokud nebyly zadány žádné argumenty, vytiskneme nápovědu a seznam dostupných měn
if len(sys.argv) == 1:
parser.print_help()
@@ -138,25 +216,34 @@ def main():
else:
print("Žádné měny nejsou v databázi k dispozici.")
sys.exit(0)
# Nastavíme debug mód
DEBUG = args.debug
set_debug_mode(DEBUG)
# Převedeme kód měny na velká písmena, pokud byl zadán
if args.currency:
args.currency = args.currency.upper()
# Kontrola a případná aktualizace ročních dat pro aktuální rok (pouze v debug módu)
if DEBUG:
check_and_update_yearly_data()
else:
# V normálním módu zkontrolujeme pouze při stahování dat
if args.year or args.start_date or args.end_date or args.date or args.get_rate or args.report_year or args.report_period or args.stats:
if (
args.year
or args.start_date
or args.end_date
or args.date
or args.get_rate
or args.report_year
or args.report_period
or args.stats
):
current_year = datetime.now().year
# Pro jednoduchost v normálním módu nebudeme kontrolovat konzistenci automaticky
pass
# Zde bude logika pro zpracování argumentů
if args.year:
debug_print(f"Stahuji roční data pro rok {args.year}...")
@@ -166,11 +253,23 @@ def main():
data_fetcher.download_yearly_data(args.year, output_dir="data")
elif args.currency and args.start_date and args.end_date and not args.report_period:
# Měsíční stahování dat
debug_print(f"Stahuji měsíční data pro měnu {args.currency} od {args.start_date} do {args.end_date}...")
debug_print(
f"Stahuji měsíční data pro měnu {args.currency} od {args.start_date} do {args.end_date}..."
)
# Ujistěme se, že adresář data existuje
os.makedirs("data", exist_ok=True)
# Volání funkce pro stažení měsíčních dat
data_fetcher.download_monthly_data(args.currency, args.start_date, args.end_date, output_dir="data")
data_fetcher.download_monthly_data(
args.currency, args.start_date, args.end_date, output_dir="data"
)
elif args.report_period and args.currency:
start_date, end_date = args.report_period
debug_print(
f"Generuji report pro měnu {args.currency} od {start_date} do {end_date}..."
)
rate_reporter.generate_period_report(
start_date, end_date, args.currency, output_dir="data"
)
elif args.date:
debug_print(f"Stahuji denní data pro datum {args.date}...")
# Ujistěme se, že adresář data existuje
@@ -183,64 +282,38 @@ def main():
debug_print(f"Vyhledávám kurz pro {currency_code} na datum {date_str}...")
rate = rate_finder.get_rate_for_date(date_str, currency_code)
if rate:
# Pro --get-rate v normálním režimu zobrazíme pouze kurz
if not DEBUG:
print(rate)
if args.json:
json_data = format_single_rate_json(currency_code, rate, date_str)
output_json(json_data)
else:
print(f"Kurz {currency_code} na datum {date_str} (nebo nejbližší pracovní den): {rate}")
# Pro --get-rate v normálním režimu zobrazíme pouze kurz
if not DEBUG:
print(rate)
else:
print(
f"Kurz {currency_code} na datum {date_str} (nebo nejbližší pracovní den): {rate}"
)
else:
# Pokud nebyl kurz nalezen a je aktivní přepínač --auto-download, zkusíme stáhnout denní data
if args.auto_download:
try:
requested_date = datetime.strptime(date_str, "%d.%m.%Y")
today = datetime.now()
# Zkontrolujeme, zda je požadované datum dnešní
if requested_date.date() == today.date():
# Zkontrolujeme, zda je čas po 14:30
if today.time() >= datetime.strptime("14:30", "%H:%M").time():
debug_print("Automaticky stahuji denní data...")
# Ujistěme se, že adresář data existuje
os.makedirs("data", exist_ok=True)
# Stáhneme denní data pro dnešní datum
today_str = today.strftime("%d.%m.%Y")
data_fetcher.download_daily_data(today_str, output_dir="data")
# Zkusíme znovu vyhledat kurz
rate = rate_finder.get_rate_for_date(date_str, currency_code)
if rate:
if not DEBUG:
print(rate)
else:
print(f"Kurz {currency_code} na datum {date_str} (nebo nejbližší pracovní den): {rate}")
else:
if not DEBUG:
print("Kurz nenalezen")
else:
print(f"Kurz {currency_code} na datum {date_str} (ani v předchozích dnech) nebyl nalezen ani po stažení denních dat.")
else:
if not DEBUG:
print("Kurz nenalezen")
else:
print(f"Chyba: Automatické stahování nelze provést, protože čas ještě není po 14:30. Aktuální čas je {today.strftime('%H:%M')}.")
else:
if not DEBUG:
print("Kurz nenalezen")
else:
print("Automatické stahování denních dat je možné pouze pro dnešní datum.")
except ValueError:
if not DEBUG:
print("Kurz nenalezen")
else:
print(f"Neplatný formát data: {date_str}")
# Rate not found
if args.json:
error_data = format_error_json(
f"Kurz {currency_code} na datum {date_str} nebyl nalezen",
"RATE_NOT_FOUND",
)
output_json(error_data)
else:
if not DEBUG:
print("Kurz nenalezen")
else:
print(f"Kurz {currency_code} na datum {date_str} (ani v předchozích dnech) nebyl nalezen.")
print(
f"Kurz {currency_code} na datum {date_str} (ani v předchozích dnech) nebyl nalezen."
)
elif args.get_rate is not None and not args.currency:
# Pokud je zadán --get-rate bez data a bez měny
if DEBUG:
print("Chyba: Pro použití --get-rate musí být zadána měna pomocí -c/--currency.")
print(
"Chyba: Pro použití --get-rate musí být zadána měna pomocí -c/--currency."
)
sys.exit(1)
# DŮLEŽITÉ: Pořadí následujících elif podmínek je důležité!
# Nejprve zpracujeme --stats, pak teprve "poslední dostupný kurz"
@@ -249,8 +322,10 @@ def main():
currency_code = args.currency
if args.stats is True:
# Pokud je --stats zadán bez roku, vytvoříme kurzy pro všechny roky s dostupnými daty
debug_print(f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} pro všechny roky...")
debug_print(
f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} pro všechny roky..."
)
# Získáme seznam všech roků s daty
years = database.get_years_with_data()
if not years:
@@ -259,77 +334,117 @@ def main():
else:
print("Databáze neobsahuje žádná data.")
return
# Pro každý rok vypočítáme 'Jednotný kurz'
year_results = []
for year in years:
# Zkontrolujeme, zda databáze obsahuje data pro daný rok
if not rate_finder.check_year_data_in_db(year):
debug_print(f"Databáze neobsahuje data pro rok {year}. Stahuji roční data...")
debug_print(
f"Databáze neobsahuje data pro rok {year}. Stahuji roční data..."
)
# Ujistěme se, že adresář data existuje
os.makedirs("data", exist_ok=True)
# Stáhneme roční data s vynuceným stažením
data_fetcher.download_yearly_data(year, output_dir="data", force=True)
data_fetcher.download_yearly_data(
year, output_dir="data", force=True
)
# Vypočítáme 'Jednotný kurz' podle metodiky ČNB
tax_rate = rate_reporter.calculate_tax_yearly_average(year, currency_code, output_dir="data")
if tax_rate:
# Pro --stats v normálním režimu zobrazíme pouze 'Jednotný kurz' zaokrouhlený na 2 desetinná místa
if not DEBUG:
print(f"{year}: {tax_rate:.2f}")
tax_rate = rate_reporter.calculate_tax_yearly_average(
year, currency_code, output_dir="data"
)
year_results.append((year, tax_rate))
if not args.json:
if tax_rate:
# Pro --stats v normálním režimu zobrazíme pouze 'Jednotný kurz' zaokrouhlený na 2 desetinná místa
if not DEBUG:
print(f"{year}: {tax_rate:.2f}")
else:
print(
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}"
)
else:
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}")
else:
if not DEBUG:
print(f"{year}: 'Jednotný kurz' nenalezen")
else:
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen.")
if not DEBUG:
print(f"{year}: 'Jednotný kurz' nenalezen")
else:
print(
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen."
)
# Output JSON for multi-year results
if args.json:
json_data = format_multi_year_json(currency_code, year_results)
output_json(json_data)
else:
# Pokud je --stats zadán s konkrétním rokem
year = args.stats
debug_print(f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}...")
debug_print(
f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}..."
)
# Zkontrolujeme, zda databáze obsahuje data pro daný rok
if not rate_finder.check_year_data_in_db(year):
debug_print(f"Databáze neobsahuje data pro rok {year}. Stahuji roční data...")
debug_print(
f"Databáze neobsahuje data pro rok {year}. Stahuji roční data..."
)
# Ujistěme se, že adresář data existuje
os.makedirs("data", exist_ok=True)
# Stáhneme roční data s vynuceným stažením
data_fetcher.download_yearly_data(year, output_dir="data", force=True)
# Vypočítáme 'Jednotný kurz' podle metodiky ČNB
tax_rate = rate_reporter.calculate_tax_yearly_average(year, currency_code, output_dir="data")
if tax_rate:
# Pro --stats v normálním režimu zobrazíme pouze 'Jednotný kurz' zaokrouhlený na 2 desetinná místa
if not DEBUG:
print(f"{tax_rate:.2f}")
else:
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}")
tax_rate = rate_reporter.calculate_tax_yearly_average(
year, currency_code, output_dir="data"
)
if args.json:
json_data = format_tax_rate_json(currency_code, year, tax_rate)
output_json(json_data)
else:
if not DEBUG:
print("'Jednotný kurz' nenalezen")
if tax_rate:
# Pro --stats v normálním režimu zobrazíme pouze 'Jednotný kurz' zaokrouhlený na 2 desetinná místa
if not DEBUG:
print(f"{tax_rate:.2f}")
else:
print(
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}"
)
else:
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen.")
if not DEBUG:
print("'Jednotný kurz' nenalezen")
else:
print(
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen."
)
elif args.currency and not args.get_rate:
# Pokud je zadána měna, ale není zadán --get-rate, vytiskneme poslední dostupný kurz
# Toto musí být až po --stats, jinak by se --stats nikdy nevykonalo
currency_code = args.currency
debug_print(f"Vyhledávám poslední dostupný kurz pro {currency_code}...")
rate, date = database.get_last_rate_for_currency(currency_code)
if rate and date:
# Pro normální režim zobrazíme kurz ve formátu "11.11 # dated: dd.mm.yyyy"
if not DEBUG:
print(f"{rate} # dated: {date}")
else:
print(f"Poslední dostupný kurz {currency_code}: {rate} # dated: {date}")
if args.json:
json_data = format_last_rate_json(currency_code, rate, date)
output_json(json_data)
else:
if not DEBUG:
print("Kurz nenalezen")
if rate and date:
# Pro normální režim zobrazíme kurz ve formátu "11.11 # dated: dd.mm.yyyy"
if not DEBUG:
print(f"{rate} # dated: {date}")
else:
print(
f"Poslední dostupný kurz {currency_code}: {rate} # dated: {date}"
)
else:
print(f"Poslední dostupný kurz pro {currency_code} nebyl nalezen.")
if not DEBUG:
print("Kurz nenalezen")
else:
print(f"Poslední dostupný kurz pro {currency_code} nebyl nalezen.")
else:
if DEBUG:
parser.print_help()
sys.exit(1)
if __name__ == "__main__":
main()
main()