OpenClaw
239 lines
7.1 KiB
Python
239 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Importiert 5 Kredite mit den aktuellen Werten vom User via API.
|
|
"""
|
|
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import datetime
|
|
|
|
# API Konfiguration
|
|
API_BASE = "http://localhost:3001/api"
|
|
|
|
# Die 5 Kredite mit den Werten vom User
|
|
# Hinweis: Zinssätze werden aus der Excel berechnet oder auf 0 gesetzt
|
|
KREDITE = [
|
|
{
|
|
"name": "DSL Bank",
|
|
"ursprungsschuld": 64656.88,
|
|
"restschuld": 64656.88,
|
|
"monatsrate": 513.80,
|
|
"zinssatz": 0, # Werde aus Excel berechnen
|
|
"start_datum": "2026-04-01",
|
|
"kreditgeber": "DSL Bank",
|
|
"person": "Rene",
|
|
"richtung": "ausgehend"
|
|
},
|
|
{
|
|
"name": "PSD Nord",
|
|
"ursprungsschuld": 50384.50,
|
|
"restschuld": 50384.50,
|
|
"monatsrate": 237.35,
|
|
"zinssatz": 0,
|
|
"start_datum": "2026-04-01",
|
|
"kreditgeber": "PSD Nord",
|
|
"person": "Rene",
|
|
"richtung": "ausgehend"
|
|
},
|
|
{
|
|
"name": "Zingelstr. 14 DSL",
|
|
"ursprungsschuld": 24382.38,
|
|
"restschuld": 24382.38,
|
|
"monatsrate": 350.00,
|
|
"zinssatz": 0,
|
|
"start_datum": "2026-04-01",
|
|
"kreditgeber": "DSL Bank",
|
|
"person": "Rene",
|
|
"notizen": "Zingelstraße 14 - DSL Teil",
|
|
"richtung": "ausgehend"
|
|
},
|
|
{
|
|
"name": "Zingelstr. 14 Sparkasse",
|
|
"ursprungsschuld": 8140.11,
|
|
"restschuld": 8140.11,
|
|
"monatsrate": 350.00,
|
|
"zinssatz": 0,
|
|
"start_datum": "2026-04-01",
|
|
"kreditgeber": "Sparkasse",
|
|
"person": "Rene",
|
|
"notizen": "Zingelstraße 14 - Sparkasse Teil",
|
|
"richtung": "ausgehend"
|
|
},
|
|
{
|
|
"name": "PVCreditplus",
|
|
"ursprungsschuld": 1666.53,
|
|
"restschuld": 1666.53,
|
|
"monatsrate": 1666.53, # Einmalige Zahlung?
|
|
"zinssatz": 0,
|
|
"start_datum": "2026-04-01",
|
|
"kreditgeber": "CreditPlus",
|
|
"person": "Rene",
|
|
"notizen": "PV-Anlage CreditPlus",
|
|
"richtung": "ausgehend"
|
|
}
|
|
]
|
|
|
|
def api_call(endpoint, method='GET', data=None):
|
|
"""Fuehrt einen API-Call aus"""
|
|
url = f"{API_BASE}{endpoint}"
|
|
headers = {'Content-Type': 'application/json'}
|
|
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(data).encode() if data else None,
|
|
headers=headers,
|
|
method=method
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as response:
|
|
return json.loads(response.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
print(f"HTTP Error {e.code}: {e.read().decode()}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
return None
|
|
|
|
def get_current_kredite():
|
|
"""Holt alle aktuellen Kredite"""
|
|
return api_call('/kredite')
|
|
|
|
def create_backup():
|
|
"""Erstellt ein Backup der aktuellen Kredite"""
|
|
print("="*60)
|
|
print("BACKUP ERSTELLEN")
|
|
print("="*60)
|
|
|
|
current = get_current_kredite()
|
|
if current:
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M')
|
|
backup_file = f"backup_vor_kredit_import_{timestamp}.json"
|
|
|
|
with open(backup_file, 'w', encoding='utf-8') as f:
|
|
json.dump(current, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"[OK] Backup erstellt: {backup_file}")
|
|
print(f" {len(current)} bestehende Kredite gesichert")
|
|
return backup_file
|
|
else:
|
|
print("[WARN] Keine Kredite gefunden oder API-Fehler")
|
|
return None
|
|
|
|
def import_kredit(kredit_data):
|
|
"""Importiert einen einzelnen Kredit"""
|
|
print(f"\n Importiere: {kredit_data['name']}")
|
|
print(f" Restschuld: {kredit_data['restschuld']:.2f} EUR")
|
|
print(f" Monatsrate: {kredit_data['monatsrate']:.2f} EUR")
|
|
print(f" Zinssatz: {kredit_data['zinssatz']}%")
|
|
|
|
result = api_call('/kredite', method='POST', data=kredit_data)
|
|
|
|
if result:
|
|
print(f" [OK] Kredit erstellt mit ID: {result.get('id')}")
|
|
return result.get('id')
|
|
else:
|
|
print(f" [ERR] Fehler beim Erstellen")
|
|
return None
|
|
|
|
def verify_import(kredit_id, expected_restschuld):
|
|
"""Verifiziert den Import"""
|
|
result = api_call(f'/kredite/{kredit_id}')
|
|
if result:
|
|
db_restschuld = float(result.get('restschuld', 0))
|
|
diff = abs(db_restschuld - expected_restschuld)
|
|
if diff < 0.01:
|
|
print(f" [OK] Restschuld stimmt: {db_restschuld:.2f} EUR")
|
|
return True
|
|
else:
|
|
print(f" [WARN] Restschuld weicht ab: {db_restschuld:.2f} vs {expected_restschuld:.2f}")
|
|
return False
|
|
return False
|
|
|
|
def verify_buchungen(kredit_id):
|
|
"""Prüft die automatisch erstellten Buchungen"""
|
|
result = api_call(f'/kredite/{kredit_id}/buchungen')
|
|
if result:
|
|
print(f" [OK] {len(result)} Buchungen erstellt")
|
|
return len(result)
|
|
return 0
|
|
|
|
def main():
|
|
print("="*60)
|
|
print("IMPORT: 5 KREDITE AUS EXCEL (via API)")
|
|
print("="*60)
|
|
print(f"Start: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
|
# 1. Backup erstellen
|
|
print("\n1. Erstelle Backup...")
|
|
backup_file = create_backup()
|
|
|
|
# 2. API Health Check
|
|
print("\n2. Prüfe API-Verbindung...")
|
|
health = api_call('/health')
|
|
if not health:
|
|
print(" [ERR] API nicht erreichbar!")
|
|
return False
|
|
print(" [OK] API erreichbar")
|
|
|
|
# 3. Bestehende Kredite anzeigen
|
|
print("\n3. Bestehende Kredite:")
|
|
current = get_current_kredite()
|
|
if current:
|
|
for k in current:
|
|
print(f" - {k.get('name')}: {float(k.get('restschuld', 0)):.2f} EUR")
|
|
else:
|
|
print(" Keine bestehenden Kredite")
|
|
|
|
# 4. Import der 5 Kredite
|
|
print("\n4. Importiere 5 neue Kredite:")
|
|
print("-"*60)
|
|
|
|
imported_ids = []
|
|
for kredit in KREDITE:
|
|
kredit_id = import_kredit(kredit)
|
|
if kredit_id:
|
|
imported_ids.append((kredit_id, kredit['restschuld'], kredit['name']))
|
|
|
|
# 5. Verifikation
|
|
print("\n5. Verifikation:")
|
|
print("-"*60)
|
|
|
|
all_ok = True
|
|
for kredit_id, expected_restschuld, name in imported_ids:
|
|
print(f"\n Prüfe {name}:")
|
|
verify_import(kredit_id, expected_restschuld)
|
|
buchungen_count = verify_buchungen(kredit_id)
|
|
if buchungen_count == 0:
|
|
all_ok = False
|
|
|
|
# 6. Zusammenfassung
|
|
print("\n" + "="*60)
|
|
print("ZUSAMMENFASSUNG")
|
|
print("="*60)
|
|
print(f"Importierte Kredite: {len(imported_ids)}")
|
|
print(f"Backup: {backup_file}")
|
|
|
|
total_restschuld = sum(k['restschuld'] for k in KREDITE)
|
|
total_rate = sum(k['monatsrate'] for k in KREDITE)
|
|
|
|
print(f"\nGesamt-Restschuld: {total_restschuld:,.2f} EUR")
|
|
print(f"Gesamt-Monatsrate: {total_rate:,.2f} EUR")
|
|
|
|
if all_ok:
|
|
print("\n[OK] Alle Kredite erfolgreich importiert!")
|
|
else:
|
|
print("\n[WARN] Einige Kredite haben Probleme")
|
|
|
|
print("\n" + "="*60)
|
|
print("HINWEIS: Backend hat automatisch Tilgungsbuchungen erstellt!")
|
|
print("Überprüfe im Frontend den Kontoauszug für jeden Kredit.")
|
|
print("="*60)
|
|
|
|
return all_ok
|
|
|
|
if __name__ == '__main__':
|
|
success = main()
|
|
exit(0 if success else 1)
|