mirror of
https://github.com/sysadminsmedia/homebox.git
synced 2025-12-21 13:23:14 +01:00
060062b484
Retries + logging: Set timestamped errors on 5xx/429 or JSON failures. Empty-array guard: script exits non-zero if the API returns [], so won’t commit an empty file. Exit codes: any API or I/O error yields sys.exit(1), triggering “no commit on failure” logic.
100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from requests.adapters import HTTPAdapter, Retry
|
|
|
|
API_URL = 'https://restcountries.com/v3.1/all?fields=name,common,currencies'
|
|
SAVE_PATH = Path('backend/internal/core/currencies/currencies.json')
|
|
TIMEOUT = 10 # seconds
|
|
|
|
def setup_logging():
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s %(levelname)s: %(message)s'
|
|
)
|
|
|
|
def fetch_currencies():
|
|
session = requests.Session()
|
|
retries = Retry(
|
|
total=3,
|
|
backoff_factor=1,
|
|
status_forcelist=[429, 500, 502, 503, 504],
|
|
allowed_methods=frozenset(['GET'])
|
|
)
|
|
session.mount('https://', HTTPAdapter(max_retries=retries))
|
|
|
|
try:
|
|
resp = session.get(API_URL, timeout=TIMEOUT)
|
|
resp.raise_for_status()
|
|
except requests.exceptions.RequestException as e:
|
|
logging.error("API request failed: %s", e)
|
|
return None
|
|
|
|
try:
|
|
countries = resp.json()
|
|
except ValueError as e:
|
|
logging.error("Failed to parse JSON response: %s", e)
|
|
return None
|
|
|
|
result = []
|
|
for country in countries:
|
|
country_name = country.get('name', {}).get('common') or "Unknown"
|
|
for code, info in country.get('currencies', {}).items():
|
|
result.append({
|
|
'code': code,
|
|
'local': country_name,
|
|
'symbol': info.get('symbol', ''),
|
|
'name': info.get('name', '')
|
|
})
|
|
|
|
# sort by country name for consistency
|
|
return sorted(result, key=lambda x: x['local'].lower())
|
|
|
|
def load_existing(path: Path):
|
|
if not path.exists():
|
|
return []
|
|
try:
|
|
with path.open('r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except (IOError, ValueError) as e:
|
|
logging.warning("Could not load existing file (%s): %s", path, e)
|
|
return []
|
|
|
|
def save_currencies(data, path: Path):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
with path.open('w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=4)
|
|
logging.info("Wrote %d entries to %s", len(data), path)
|
|
except IOError as e:
|
|
logging.error("Failed to write file: %s", e)
|
|
sys.exit(1)
|
|
|
|
def main():
|
|
setup_logging()
|
|
logging.info("Starting currency update")
|
|
|
|
existing = load_existing(SAVE_PATH)
|
|
new = fetch_currencies()
|
|
|
|
if new is None:
|
|
logging.error("Aborting: failed to fetch or parse API data.")
|
|
sys.exit(1)
|
|
|
|
if not new:
|
|
logging.error("Aborting: API returned an empty list.")
|
|
sys.exit(1)
|
|
|
|
if new == existing:
|
|
logging.info("No changes detected; skipping write.")
|
|
return
|
|
|
|
save_currencies(new, SAVE_PATH)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|