#!/usr/bin/env python3 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 sys.path.insert(0, os.path.join(os.path.dirname(__file__))) import database import data_fetcher import holidays import rate_finder import rate_reporter import data_validator # 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) holidays.set_debug_mode(DEBUG) rate_finder.set_debug_mode(DEBUG) rate_reporter.set_debug_mode(DEBUG) data_validator.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" ) if not is_consistent: 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 data_fetcher.download_yearly_data(current_year, output_dir="data", force=True) 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)." ) parser.add_argument( "-c", "--currency", type=str, 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.", ) 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.", ) parser.add_argument( "--date", type=str, help="Datum pro stažení denních kurzů ve formátu DD.MM.YYYY.", ) parser.add_argument( "--get-rate", "-d", type=str, 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.", ) parser.add_argument( "--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.", ) 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", ) parser.add_argument( "--stats", 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.", ) parser.add_argument( "--validate", action="store_true", help="Validuje data pro měnu nebo všechny měny. Zkontroluje konzistenci kurzů a detekuje možné chyby.", ) parser.add_argument( "--record-counts", action="store_true", help="Zobrazí počet záznamů podle časových období (týden, měsíc, čtvrtletí, pololetí, rok).", ) parser.add_argument( "--change-threshold", type=float, default=1.0, help="Práh pro detekci změn kurzů v procentech (výchozí: 1.0).", ) parser.add_argument( "--gap-threshold", type=int, default=3, help="Maximální přijatelná mezera v pracovních dnech (výchozí: 3).", ) parser.add_argument( "--debug", action="store_true", help="Zobrazí podrobné ladicí informace." ) parser.add_argument( "--json", action="store_true", help="Výstup ve formátu JSON místo prostého textu pro programové zpracování.", ) parser.add_argument( "--no-adaptive", action="store_true", help="Vypne adaptivní učení prahů na základě historických dat.", ) args = parser.parse_args() # 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 ): 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ů # Zde bude logika pro zpracování argumentů if args.validate: # Validation command base_threshold = args.change_threshold adaptive = not args.no_adaptive max_gap_days = getattr(args, "gap_threshold", 3) # Default to 3 if not defined if args.currency: # Validate specific currency debug_print(f"Validuji data pro měnu {args.currency}...") results = data_validator.validate_currency_data( args.currency, args.year, base_threshold, adaptive, max_gap_days ) if args.json: output_json(results) else: text_output = data_validator.format_validation_text(results) print(text_output) else: # Validate all currencies debug_print("Validuji data pro všechny měny...") results = data_validator.validate_all_currencies( args.year, base_threshold, adaptive, max_gap_days ) if args.json: output_json(results) else: text_output = data_validator.format_validation_text(results) print(text_output) elif args.record_counts: # Record counts command if not args.currency: print( "Chyba: Pro --record-counts je nutné zadat měnu pomocí -c/--currency." ) sys.exit(1) debug_print(f"Získávám počty záznamů pro měnu {args.currency}...") record_counts = data_validator.get_record_counts_by_period( args.currency, args.year ) if args.json: output_json({"currency": args.currency, "record_counts": record_counts}) else: print(f"Record Counts for {args.currency}:") print("=" * 50) for year_key, periods in record_counts.items(): print(f"\nYear {year_key}:") print(f" Total records: {periods.get('year', 0)}") # Half years half_years = periods.get("half_year", {}) if half_years: print( f" Half years: H1={half_years.get('H1', 0)}, H2={half_years.get('H2', 0)}" ) # Quarters quarters = periods.get("quarter", {}) if quarters: quarter_str = ", ".join( [f"Q{q}={quarters.get(f'Q{q}', 0)}" for q in range(1, 5)] ) print(f" Quarters: {quarter_str}") # Months months = periods.get("month", {}) if months: month_list = [] for month in range(1, 13): month_key = f"{month:02d}" count = months.get(month_key, 0) month_list.append(f"{month}={count}") print(f" Months: {', '.join(month_list)}") # Weeks summary weeks = periods.get("week", {}) if weeks: total_weeks = len(weeks) if total_weeks <= 10: week_list = sorted([f"{w}={weeks[w]}" for w in weeks.keys()]) print(f" Weeks: {', '.join(week_list)}") else: sample_weeks = sorted(list(weeks.keys())[:5]) week_sample = [f"{w}={weeks[w]}" for w in sample_weeks] print( f" Weeks: {', '.join(week_sample)}... ({total_weeks} total weeks)" ) elif args.year: # Validation command base_threshold = args.change_threshold adaptive = not args.no_adaptive if args.currency: # Validate specific currency debug_print(f"Validuji data pro měnu {args.currency}...") results = data_validator.validate_currency_data( args.currency, args.year, base_threshold, adaptive ) if args.json: output_json(results) else: text_output = data_validator.format_validation_text(results) print(text_output) else: # Validate all currencies debug_print("Validuji data pro všechny měny...") results = data_validator.validate_all_currencies( args.year, base_threshold, adaptive ) if args.json: output_json(results) else: text_output = data_validator.format_validation_text(results) print(text_output) return # elif args.currency and args.start_date and args.end_date and not args.report_period: # Měsíční stahování dat debug_print("HIT: Monthly download condition") 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" ) elif args.report_period and args.currency: start_date, end_date = args.report_period debug_print("HIT: Report period condition") 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("HIT: Daily data condition") debug_print(f"Stahuji denní data pro datum {args.date}...") # Ujistěme se, že adresář data existuje os.makedirs("data", exist_ok=True) # Volání funkce pro stažení denních dat data_fetcher.download_daily_data(args.date, output_dir="data") elif args.get_rate and args.currency: debug_print("HIT: Get rate condition") date_str = args.get_rate currency_code = args.currency 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: if args.json: json_data = format_single_rate_json(currency_code, rate, date_str) output_json(json_data) else: # 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: # 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." ) elif args.get_rate is not None and not args.currency: debug_print("HIT: Get rate without currency condition") # 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." ) 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" elif args.stats is not None and args.currency: debug_print("HIT: Stats condition") 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..." ) # Získáme seznam všech roků s daty years = database.get_years_with_data() if not years: if not DEBUG: print("Žádná data nenalezena") 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..." ) # 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" ) 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: 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}..." ) # 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..." ) # 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 args.json: json_data = format_tax_rate_json(currency_code, year, tax_rate) output_json(json_data) else: 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: 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 args.json: json_data = format_last_rate_json(currency_code, rate, date) output_json(json_data) else: 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: 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()