GuidoandClaude Opus 4.8
1b7107b9bd
Backend: - Modelle + Migration 0002: accounts, account_identifiers, category_groups, categories, transactions, exchange_rates (alle owner-gescoped). - EZB-Kursdienst (90-Tage/Historie) + Backfill-Skript; Umrechnung EUR<->CHF zum Buchungsdatum mit lokalem Cache und manueller Übersteuerung; EUR-Pivot (amount_eur_cents) eingefroren. - Kontostände + Netto-Vermögen (Umrechnung in Hauptwährung). - Router: accounts (inkl. IBAN-Identifikatoren), categories (Gruppen+Kategorien), transactions (Ausgabe/Einnahme/Transfer, cross-currency), rates. - Gruppierte Standard-Kategorien je Nutzer beim Anlegen. - pytest-Suite: Währung, Salden, Cross-Currency-Transfer, Nutzer-Isolation. Frontend: - Seiten Konten, Kategorien, Buchungen; Icon-Chip, Geldbetrag mit Umrechnung, Icon-/Farbwähler, Modal; erweitertes SVG-Icon-Set (Konto-/Kategorie-Icons). - Währungs-/Datumsformatierung (EUR de-DE, CHF de-CH), i18n DE/EN erweitert. - Live-Umrechnungs-Vorschau im Transfer-Formular. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
"""EZB-Referenzkurse laden und lokal cachen.
|
||
|
||
Die EZB veröffentlicht die Euro-Referenzkurse werktäglich als XML: ein Tages-Feed,
|
||
ein 90-Tage-Feed und die volle Historie seit 1999. Wir brauchen nur CHF. Historische
|
||
Kurse ändern sich nicht – vorhandene Zeilen werden daher nicht überschrieben.
|
||
"""
|
||
|
||
import xml.etree.ElementTree as ET
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
import httpx
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import ExchangeRate
|
||
|
||
FEED_90D = "https://www.ecb.europa.eu/stats/eurofxref/eurofxref-hist-90d.xml"
|
||
FEED_FULL = "https://www.ecb.europa.eu/stats/eurofxref/eurofxref-hist.xml"
|
||
|
||
# Welche Währungen wir speichern (Hauptwährungen des Nutzers: EUR ist Pivot).
|
||
CURRENCIES = {"CHF"}
|
||
|
||
|
||
def _parse(xml_bytes: bytes) -> list[tuple[date, str, str]]:
|
||
"""Extrahiert (Datum, Währung, Kurs) aus dem EZB-XML (namespace-tolerant)."""
|
||
root = ET.fromstring(xml_bytes)
|
||
rows: list[tuple[date, str, str]] = []
|
||
for elem in root.iter():
|
||
t = elem.attrib.get("time")
|
||
if not t:
|
||
continue
|
||
try:
|
||
d = date.fromisoformat(t)
|
||
except ValueError:
|
||
continue
|
||
for child in elem:
|
||
ccy = child.attrib.get("currency")
|
||
rate = child.attrib.get("rate")
|
||
if ccy in CURRENCIES and rate:
|
||
rows.append((d, ccy, rate))
|
||
return rows
|
||
|
||
|
||
def _fetch(url: str, timeout: float = 15.0) -> bytes:
|
||
with httpx.Client(timeout=timeout, follow_redirects=True) as client:
|
||
resp = client.get(url)
|
||
resp.raise_for_status()
|
||
return resp.content
|
||
|
||
|
||
def _upsert(db: Session, rows: list[tuple[date, str, str]]) -> int:
|
||
added = 0
|
||
for d, ccy, rate in rows:
|
||
if db.get(ExchangeRate, (d, ccy)) is None:
|
||
db.add(ExchangeRate(date=d, currency=ccy, rate_per_eur=Decimal(rate), source="ecb"))
|
||
added += 1
|
||
if added:
|
||
db.commit()
|
||
return added
|
||
|
||
|
||
def refresh_recent(db: Session) -> int:
|
||
"""Lädt die letzten ~90 Tage (schnell) und ergänzt fehlende Tage."""
|
||
return _upsert(db, _parse(_fetch(FEED_90D)))
|
||
|
||
|
||
def backfill_full(db: Session) -> int:
|
||
"""Lädt die komplette Historie (einmalig, größer)."""
|
||
return _upsert(db, _parse(_fetch(FEED_FULL)))
|
||
|
||
|
||
def ensure_rates_on_startup(db: Session, enabled: bool) -> None:
|
||
"""Beim App-Start best-effort die aktuellen Kurse nachladen.
|
||
|
||
Blockiert nie hart: Ohne Internet läuft die App weiter, nur ohne frische Kurse.
|
||
Die volle Historie holt das Install-Skript / der systemd-Timer separat."""
|
||
if not enabled:
|
||
return
|
||
try:
|
||
refresh_recent(db)
|
||
except Exception: # noqa: BLE001 – bewusst tolerant (Netz weg etc.)
|
||
pass
|