182 lines
6.3 KiB
Python
182 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Einmaliger Import der Mieter-Stammdaten aus der SharePoint-Export-CSV
|
||
(„Master Data V") in die PostgreSQL-Datenbank des Mieterportals.
|
||
|
||
Liest die CSV von stdin, schreibt SQL auf stdout:
|
||
python import-masterdata.py < \"Master Data V.csv\" | \\
|
||
docker exec -i mp-db psql -U mp -d mieterportal
|
||
|
||
ACHTUNG: Ersetzt die Demo-Daten (Häuser, Wohnungen, Mieter, Buchungen,
|
||
Meldungen, Unterhaltungen, Dokumente, Zähler, Kosten, Belege, Codes).
|
||
Verwaltungs-Konten, Ankündigungen und Sessions der Verwaltung bleiben.
|
||
|
||
IDs sind deterministisch (Slug aus Haus + WE) – das Skript ist damit
|
||
fehlerverträglich wiederholbar, solange die CSV unverändert bleibt.
|
||
Die CSV gehört NICHT ins Git (PII) – siehe .gitignore.
|
||
"""
|
||
import csv
|
||
import io
|
||
import re
|
||
import secrets
|
||
import sys
|
||
from datetime import date
|
||
|
||
HEUTE = f'{date.today().day}.{date.today().month}.{date.today().year}'
|
||
|
||
|
||
def q(wert):
|
||
"""SQL-Escaping für Textfelder."""
|
||
if wert is None:
|
||
return 'NULL'
|
||
return "'" + str(wert).replace("'", "''") + "'"
|
||
|
||
|
||
def slug(text):
|
||
s = str(text or '').strip().lower()
|
||
s = (s.replace('ä', 'ae').replace('ö', 'oe').replace('ü', 'ue').replace('ß', 'ss'))
|
||
s = re.sub(r'[^a-z0-9]+', '-', s).strip('-')
|
||
return s or 'x'
|
||
|
||
|
||
def zahl(wert):
|
||
"""Deutsche Zahlenformate: '2.700' -> 2700, '71,4' -> 71.4, '' -> 0"""
|
||
s = str(wert or '').strip()
|
||
if not s:
|
||
return 0.0
|
||
s = s.replace('.', '').replace(',', '.')
|
||
try:
|
||
return float(s)
|
||
except ValueError:
|
||
return 0.0
|
||
|
||
|
||
def jahr(wert):
|
||
"""'1.905' -> 1905, leer/0 -> None"""
|
||
s = str(wert or '').strip()
|
||
if not s:
|
||
return None
|
||
s = s.replace('.', '')
|
||
try:
|
||
j = int(s)
|
||
except ValueError:
|
||
return None
|
||
return j if 1500 < j < 2100 else None
|
||
|
||
|
||
def etage(lage):
|
||
s = (lage or '').strip().lower()
|
||
if s.startswith('eg') or 'hinterhof' in s:
|
||
return 0
|
||
m = re.search(r'(\d+)\.\s*og', s)
|
||
if m:
|
||
return int(m.group(1))
|
||
if s.startswith('dg'):
|
||
return 4
|
||
return 0
|
||
|
||
|
||
def wohnungsnummer(we):
|
||
"""'WE01' -> '1', 'Garage 1' bleibt, 'Zimmer 2' bleibt."""
|
||
s = str(we or '').strip()
|
||
m = re.match(r'^WE0*(\d+)$', s, re.IGNORECASE)
|
||
if m:
|
||
return str(int(m.group(1)))
|
||
return s
|
||
|
||
|
||
def main():
|
||
eingabe = io.TextIOWrapper(sys.stdin.buffer, encoding='utf-8-sig')
|
||
zeilen = list(csv.DictReader(eingabe))
|
||
if not zeilen:
|
||
sys.exit('CSV ist leer oder konnte nicht gelesen werden.')
|
||
|
||
sql = []
|
||
sql.append('BEGIN;')
|
||
# buildings: NOT NULL lockern – die CSV liefert nicht alle Pflichtfelder
|
||
for spalte in ('baujahr', 'stockwerke', 'etagenhoehe', 'wohneinheiten', 'quadratmeter'):
|
||
sql.append(f'ALTER TABLE buildings ALTER COLUMN {spalte} DROP NOT NULL;')
|
||
|
||
# ── Demo-Daten entfernen (Reihenfolge wegen Fremdschlüsseln) ──
|
||
sql += [
|
||
'DELETE FROM threads;',
|
||
'DELETE FROM tickets;',
|
||
'DELETE FROM transactions;',
|
||
'DELETE FROM costs;',
|
||
'DELETE FROM belege;',
|
||
'DELETE FROM documents;',
|
||
'DELETE FROM meters;',
|
||
'DELETE FROM registration_codes;',
|
||
"DELETE FROM sessions WHERE user_id IN (SELECT id FROM users WHERE rolle = 'mieter');",
|
||
"DELETE FROM users WHERE rolle = 'mieter';",
|
||
'DELETE FROM units;',
|
||
'DELETE FROM buildings;',
|
||
]
|
||
|
||
# ── Häuser ──
|
||
haeuser = []
|
||
for z in zeilen:
|
||
t = z['Titel'].strip()
|
||
if t and t not in haeuser:
|
||
haeuser.append(t)
|
||
for t in haeuser:
|
||
zeilen_haus = [z for z in zeilen if z['Titel'].strip() == t]
|
||
we_zeilen = [z for z in zeilen_haus if z['WE'].strip()]
|
||
strasse = zeilen_haus[0]['Straße'].strip()
|
||
plz = zeilen_haus[0]['PLZ'].strip().replace('.', '')
|
||
stadt = zeilen_haus[0]['Stadt'].strip()
|
||
adresse = ', '.join(x for x in (strasse, f'{plz} {stadt}'.strip()) if x)
|
||
bj = jahr(zeilen_haus[0]['Baujahr'])
|
||
qm = sum(zahl(z['m²']) for z in we_zeilen) or None
|
||
wohnungen = len(we_zeilen)
|
||
sql.append(
|
||
'INSERT INTO buildings (name, adresse, baujahr, stockwerke, etagenhoehe, wohneinheiten, quadratmeter)'
|
||
f' VALUES ({q(t)}, {q(adresse)}, {bj if bj else "NULL"}, NULL, NULL, {wohnungen}, {qm if qm else "NULL"});',
|
||
)
|
||
|
||
# ── Wohneinheiten + Mieter ──
|
||
warnungen = []
|
||
for z in zeilen:
|
||
we = z['WE'].strip()
|
||
if not we:
|
||
continue
|
||
titel = z['Titel'].strip()
|
||
einheit_id = f"{slug(titel)}-{slug(we)}"
|
||
nummer = wohnungsnummer(we)
|
||
mieter = z['Name des Mieters'].strip()
|
||
kalt = zahl(z['Nettokaltmiete'])
|
||
warm = kalt + zahl(z['Nebenkosten'])
|
||
sql.append(
|
||
'INSERT INTO units (id, building, number, floor, size, cold_rent, warm_rent, tenant, status)'
|
||
f' VALUES ({q(einheit_id)}, {q(titel)}, {q(nummer)}, {etage(z["Lage im Objekt"])},'
|
||
f' {zahl(z["m²"])}, {kalt}, {warm}, {q(mieter) if mieter else "NULL"},'
|
||
f" '{('vermietet' if mieter else 'frei')}');",
|
||
)
|
||
if mieter:
|
||
nr = z['Nr. Mietpartei'].strip() or slug(we)
|
||
benutzername = f"{slug(titel)}-{nr}"
|
||
# Zufälliger unbekannter Hash -> Mieter kann sich erst nach
|
||
# Passwort-Reset durch die Verwaltung (Button in /benutzer) anmelden.
|
||
passwort_hash = secrets.token_hex(32)
|
||
sql.append(
|
||
'INSERT INTO users (id, benutzername, name, passwort_hash, rolle, email, einheit_id, aktiv, erstellt_am)'
|
||
f" VALUES ({q('b-' + einheit_id)}, {q(benutzername)}, {q(mieter)}, {q(passwort_hash)},"
|
||
f" 'mieter', '', {q(einheit_id)}, TRUE, {q(HEUTE)});",
|
||
)
|
||
# Zähler werden NICHT importiert: die CSV-Zählerspalten sind leer/'0'
|
||
if z['Stromzähler'].strip() not in ('', '0'):
|
||
warnungen.append(f"Stromzähler in {titel}/{we}: {z['Stromzähler'].strip()} (nicht importiert)")
|
||
|
||
sql.append('COMMIT;')
|
||
|
||
sys.stdout.write('\n'.join(sql) + '\n')
|
||
print(f'-- {len(haeuser)} Häuser, '
|
||
f"{sum(1 for z in zeilen if z['WE'].strip())} Wohneinheiten, "
|
||
f"{sum(1 for z in zeilen if z['Name des Mieters'].strip())} Mieter", file=sys.stderr)
|
||
for w in warnungen:
|
||
print('-- HINWEIS: ' + w, file=sys.stderr)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|