"""
DRK Blutspende – Arbeitsorte Logger
Protokolliert Arbeitseinsätze mit Datum und Ort (mit PLZ).
Visualisiert Einsatzorte auf einer interaktiven Karte.
"""
from __future__ import annotations
import json
import logging
import math
import os
import re
import shutil
import subprocess
import sys
import threading
import time
import webbrowser
from datetime import datetime
from logging.handlers import RotatingFileHandler
from pathlib import Path
import folium
import pandas as pd
import ttkbootstrap as ttk
from geopy.exc import GeocoderServiceError, GeocoderTimedOut
from geopy.geocoders import Nominatim
from ttkbootstrap.constants import *
import tkinter as tk
import tkinter.font as tkfont
from tkinter import messagebox
# ── Pfade ────────────────────────────────────────────────────────────────────
BASE_DIR = Path(__file__).parent
DATA_DIR = BASE_DIR / "data"
CSV_PATH = DATA_DIR / "orte.csv"
MAP_PATH = DATA_DIR / "karte.html"
BACKUP_DIR = DATA_DIR / "backups"
LOG_PATH = DATA_DIR / "app.log"
ICON_PATH = BASE_DIR / "icon.png" # mitgeliefert (für Fenster + Startmenü)
ICON_FALLBACK_PATH = DATA_DIR / "icon.png" # zur Laufzeit gezeichnet, falls obiges fehlt
WINDOW_STATE_PATH = DATA_DIR / "window.json"
CSV_COLUMNS = ["date", "city", "postal_code", "lat", "lon"]
#: Wie viele automatische Backups von orte.csv aufbewahrt werden.
MAX_BACKUPS = 20
#: Schriftgröße der Oberfläche (größer = besser lesbar auf einem Laptop).
UI_FONT_SIZE = 11
#: DRK-Rot – für Akzente in der Oberfläche und die Kartenpunkte.
DRK_RED = "#CC0000"
#: Regionen, die bei der Geokodierung bevorzugt durchsucht werden (Reihenfolge zählt).
GEOCODE_REGIONS = ["Baden-Württemberg", "Hessen"]
#: Anfangsansicht der Karte (Mitte + Zoom), falls keine Punkte vorhanden sind.
MAP_DEFAULT_CENTER = [49.0, 9.0]
MAP_DEFAULT_ZOOM = 8
#: Kartenkacheln – Esri "World Light Gray" (heller, ruhiger Hintergrund).
#: Braucht keinen API-Schlüssel und keinen Referer (anders als tile.openstreetmap.org,
#: das lokale Dateien mit 403 blockt, und "CartoDB positron", das einen Schlüssel will).
#: Alternativen ohne Schlüssel als Ein-Zeilen-Tausch:
#: Esri Straßenkarte: …/World_Street_Map/MapServer/tile/{z}/{y}/{x}
#: Esri Topo: …/World_Topo_Map/MapServer/tile/{z}/{y}/{x}
_ESRI = "https://server.arcgisonline.com/ArcGIS/rest/services"
MAP_TILES = f"{_ESRI}/Canvas/World_Light_Gray_Base/MapServer/tile/{{z}}/{{y}}/{{x}}"
MAP_TILES_LABELS = f"{_ESRI}/Canvas/World_Light_Gray_Reference/MapServer/tile/{{z}}/{{y}}/{{x}}"
MAP_TILES_ATTR = "Kacheln © Esri"
log = logging.getLogger("arbeitsorte")
# ── Datum ────────────────────────────────────────────────────────────────────
#: Eingabeformate, die beim Parsen akzeptiert werden. Gespeichert wird immer ISO.
_DATE_INPUT_FORMATS = ["%Y-%m-%d", "%d.%m.%Y", "%d.%m.%y", "%Y/%m/%d"]
def parse_date(text: str) -> str | None:
"""Normalisiert eine Datumseingabe auf ``JJJJ-MM-TT``.
Gibt ``None`` zurück, wenn der Text kein gültiges Datum ist.
"""
text = (text or "").strip()
if not text:
return None
for fmt in _DATE_INPUT_FORMATS:
try:
return datetime.strptime(text, fmt).date().isoformat()
except ValueError:
continue
return None
def _setup_logging() -> None:
DATA_DIR.mkdir(exist_ok=True)
handler = RotatingFileHandler(
LOG_PATH, maxBytes=512_000, backupCount=3, encoding="utf-8"
)
handler.setFormatter(
logging.Formatter("%(asctime)s %(levelname)-7s %(message)s")
)
root = logging.getLogger("arbeitsorte")
root.setLevel(logging.INFO)
root.addHandler(handler)
def _files_equal(a: Path, b: Path) -> bool:
try:
return a.read_bytes() == b.read_bytes()
except OSError:
return False
# ── Autovervollständigung (von Eingabe- und Bearbeiten-Formular genutzt) ──────
_CITY_PLZ_RE = re.compile(r"^(.+?)\s*\((\d{4,5})\)\s*$")
def filter_locations(typed: str, all_locs: list[str], limit: int = 30) -> list[str]:
typed = typed.strip().lower()
matches = [s for s in all_locs if typed in s.lower()] if typed else all_locs
return matches[:limit]
def split_city_plz(value: str) -> tuple[str, str] | None:
"""Zerlegt ``"Stuttgart (70173)"`` in ``("Stuttgart", "70173")``."""
m = _CITY_PLZ_RE.match(value)
if m:
return m.group(1).strip(), m.group(2).strip()
return None
def _is_blank(value) -> bool:
"""True für leere / fehlende Zellwerte ("", NaN, "nan", None)."""
if value is None:
return True
try:
if pd.isna(value):
return True
except (TypeError, ValueError):
pass
return str(value).strip().lower() in ("", "nan")
def _coord_str(value) -> str:
"""Formatiert einen gespeicherten Koordinatenwert für ein Eingabefeld."""
num = parse_coord("" if value is None else str(value))
return "" if num is None else f"{num:.5f}"
def parse_coord(text: str) -> float | None:
"""Parst eine Koordinateneingabe (akzeptiert Komma als Dezimaltrennzeichen)."""
text = (text or "").strip().replace(",", ".")
if not text:
return None
try:
value = float(text)
except ValueError:
return None
return value if math.isfinite(value) else None
# ── DataStore ─────────────────────────────────────────────────────────────────
class DataStore:
"""Verwaltet die CSV-Datei und die bekannten Orte.
Schreibvorgänge laufen immer über :meth:`_save_csv`: erst ein rotierendes
Backup, dann ein atomarer Schreibvorgang (Temp-Datei + ``os.replace``), damit
ein Absturz mitten im Speichern die Daten nicht beschädigt.
"""
def __init__(self) -> None:
DATA_DIR.mkdir(exist_ok=True)
BACKUP_DIR.mkdir(exist_ok=True)
if not CSV_PATH.exists():
self._write_atomic(pd.DataFrame(columns=CSV_COLUMNS))
self._load()
self._make_backup() # Sicherung des Standes beim Programmstart
def _load(self) -> None:
try:
self.df = pd.read_csv(
CSV_PATH, dtype={"postal_code": str, "lat": str, "lon": str}
)
except (pd.errors.EmptyDataError, FileNotFoundError):
log.warning("orte.csv fehlt oder ist leer – starte mit leerer Tabelle.")
self.df = pd.DataFrame(columns=CSV_COLUMNS)
except Exception:
log.exception("orte.csv konnte nicht gelesen werden – nutze letztes Backup.")
self.df = self._load_from_newest_backup()
for col in CSV_COLUMNS:
if col not in self.df.columns:
self.df[col] = ""
self.df = self.df[CSV_COLUMNS]
self._refresh_known()
def _load_from_newest_backup(self) -> pd.DataFrame:
backups = sorted(BACKUP_DIR.glob("orte-*.csv"), reverse=True)
for path in backups:
try:
return pd.read_csv(
path, dtype={"postal_code": str, "lat": str, "lon": str}
)
except Exception:
continue
return pd.DataFrame(columns=CSV_COLUMNS)
# ── Schreiben (atomar + Backup) ──────────────────────────────────────────
@staticmethod
def _write_atomic(df: pd.DataFrame) -> None:
"""Schreibt ``df`` atomar nach CSV_PATH (Temp-Datei + os.replace)."""
tmp = CSV_PATH.with_name(CSV_PATH.name + ".tmp")
df.to_csv(tmp, index=False)
with open(tmp, "rb") as fh:
os.fsync(fh.fileno())
os.replace(tmp, CSV_PATH)
def _make_backup(self) -> None:
"""Kopiert die aktuelle CSV nach data/backups/ und hält MAX_BACKUPS Stück."""
if not CSV_PATH.exists() or CSV_PATH.stat().st_size == 0:
return
backups = sorted(BACKUP_DIR.glob("orte-*.csv"), reverse=True)
if backups and _files_equal(CSV_PATH, backups[0]):
return # nichts geändert seit dem letzten Backup
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
dest = BACKUP_DIR / f"orte-{stamp}.csv"
try:
shutil.copy2(CSV_PATH, dest)
except Exception:
log.exception("Backup konnte nicht erstellt werden.")
return
for old in sorted(BACKUP_DIR.glob("orte-*.csv"), reverse=True)[MAX_BACKUPS:]:
old.unlink(missing_ok=True)
def _save_csv(self) -> None:
self._make_backup()
self._write_atomic(self.df)
self._refresh_known()
log.info("orte.csv gespeichert (%d Einträge).", len(self.df))
def _refresh_known(self) -> None:
pairs = (
self.df[["city", "postal_code"]]
.dropna(subset=["city"])
.drop_duplicates()
)
self._known: list[tuple[str, str]] = [
(str(r.city).strip(), str(r.postal_code).strip() if pd.notna(r.postal_code) else "")
for _, r in pairs.iterrows()
if str(r.city).strip()
]
def get_autocomplete_strings(self) -> list[str]:
result = []
for city, plz in self._known:
result.append(f"{city} ({plz})" if plz else city)
return sorted(set(result), key=str.lower)
def append_rows(self, rows: list[dict]) -> None:
save_rows = [{k: v for k, v in r.items() if k in CSV_COLUMNS} for r in rows]
new_df = pd.DataFrame(save_rows, columns=CSV_COLUMNS)
self.df = pd.concat([self.df, new_df], ignore_index=True)
self._save_csv()
def update_row(self, idx: int, values: dict) -> None:
"""Aktualisiert einen bestehenden Eintrag und speichert die CSV."""
for col, val in values.items():
if col in CSV_COLUMNS:
# Columns are str-typed; convert floats/None to string
self.df.at[idx, col] = "" if (val is None or val == "") else str(val)
self._save_csv()
def delete_rows(self, indices: list[int]) -> None:
"""Löscht Einträge anhand ihrer DataFrame-Indizes."""
self.df = self.df.drop(index=indices).reset_index(drop=True)
self._save_csv()
def find_duplicates(self, rows: list[dict]) -> list[dict]:
dupes = []
for row in rows:
mask = (
(self.df["date"].astype(str) == str(row["date"]))
& (self.df["city"].astype(str).str.strip().str.lower() == str(row["city"]).strip().lower())
& (self.df["postal_code"].astype(str).str.strip() == str(row["postal_code"]).strip())
)
if mask.any():
dupes.append(row)
return dupes
def total_count(self) -> int:
return len(self.df)
def get_map_data(self) -> pd.DataFrame:
df = self.df.copy()
df["lat"] = pd.to_numeric(df["lat"], errors="coerce")
df["lon"] = pd.to_numeric(df["lon"], errors="coerce")
df = df.dropna(subset=["lat", "lon"])
if df.empty:
return pd.DataFrame(columns=["city", "lat", "lon", "count"])
return df.groupby(["city", "lat", "lon"]).size().reset_index(name="count")
# ── Geocodierung ──────────────────────────────────────────────────────────────
_geolocator = Nominatim(user_agent="drk_blutspende_orte/1.0", timeout=10)
class GeocodingUnavailable(Exception):
"""Alle Geokodier-Anfragen sind an einem Dienst-/Netzwerkfehler gescheitert."""
def geocode_city(city: str, postal_code: str) -> tuple[float, float] | None:
"""Sucht Koordinaten für ``city``/``postal_code``.
Rückgabe ``None`` bedeutet "Ort nicht gefunden". Ein
:class:`GeocodingUnavailable` wird geworfen, wenn *jede* Anfrage an einem
Dienst- oder Netzwerkfehler gescheitert ist (z. B. keine Internetverbindung).
"""
queries: list[str] = []
if postal_code:
queries.append(f"{postal_code} {city}, Germany")
queries += [f"{city}, {region}, Germany" for region in GEOCODE_REGIONS]
queries.append(f"{city}, Germany")
had_result = False
for q in queries:
try:
result = _geolocator.geocode(q)
had_result = True
if result:
return result.latitude, result.longitude
except (GeocoderTimedOut, GeocoderServiceError) as exc:
log.warning("Geokodierung fehlgeschlagen für %r: %s", q, exc)
continue
except Exception:
log.exception("Unerwarteter Fehler bei der Geokodierung von %r", q)
continue
if not had_result:
raise GeocodingUnavailable(city)
return None
class GeocoderWorker(threading.Thread):
"""Hintergrund-Thread für Geokodierung (Nominatim-Rate-Limit: 1 req/s).
Der Thread bleibt auch dann am Leben, wenn eine einzelne Anfrage oder ein
Callback eine Exception wirft – sonst würden alle folgenden Einträge
dauerhaft in "wird gesucht…" hängen bleiben.
"""
#: Ab so vielen Fehlschlägen in Folge wird ``on_service_error`` ausgelöst.
SERVICE_ERROR_THRESHOLD = 3
def __init__(self, on_service_error=None) -> None:
super().__init__(daemon=True)
self._items: list[tuple[str, str, str, object]] = []
self._lock = threading.Lock()
self._event = threading.Event()
self._stopped = False
self._on_service_error = on_service_error
self._consecutive_errors = 0
def enqueue(self, row_id: str, city: str, plz: str, callback) -> None:
with self._lock:
self._items.append((row_id, city, plz, callback))
self._event.set()
def run(self) -> None:
while not self._stopped:
self._event.wait()
self._event.clear()
while not self._stopped:
with self._lock:
if not self._items:
break
row_id, city, plz, callback = self._items.pop(0)
coords: tuple[float, float] | None = None
try:
coords = geocode_city(city, plz)
self._consecutive_errors = 0
except GeocodingUnavailable:
self._consecutive_errors += 1
log.warning(
"Standortdienst nicht erreichbar (%d in Folge) für %r.",
self._consecutive_errors, city,
)
self._maybe_report_service_error()
except Exception:
log.exception("Geokodierung von %r abgebrochen.", city)
try:
callback(row_id, coords)
except Exception:
log.exception("Geocode-Callback für %r fehlgeschlagen.", row_id)
time.sleep(1)
def _maybe_report_service_error(self) -> None:
if (
self._on_service_error is not None
and self._consecutive_errors == self.SERVICE_ERROR_THRESHOLD
):
try:
self._on_service_error()
except Exception:
log.exception("on_service_error-Callback fehlgeschlagen.")
def stop(self) -> None:
self._stopped = True
self._event.set()
# ── Kartenerstellung ──────────────────────────────────────────────────────────
def generate_map(store: DataStore) -> Path:
freq = store.get_map_data()
if freq.empty:
raise ValueError("Keine Koordinaten vorhanden. Bitte zuerst Einträge speichern.")
m = folium.Map(
location=MAP_DEFAULT_CENTER, zoom_start=MAP_DEFAULT_ZOOM, tiles=None,
)
folium.TileLayer(
MAP_TILES, attr=MAP_TILES_ATTR, name="Karte", control=False,
).add_to(m)
folium.TileLayer(
MAP_TILES_LABELS, attr=MAP_TILES_ATTR, name="Beschriftung",
control=False, overlay=True,
).add_to(m)
max_count = int(freq["count"].max())
for _, row in freq.iterrows():
count = int(row["count"])
radius = 5 + (count / max_count) * 20
folium.CircleMarker(
location=[float(row["lat"]), float(row["lon"])],
radius=radius,
popup=folium.Popup(
f"{row['city']}
{count} Einsatz/Einsätze", max_width=200
),
tooltip=str(row["city"]),
color=DRK_RED,
fill=True,
fill_color=DRK_RED,
fill_opacity=0.65,
weight=2,
).add_to(m)
if len(freq) > 1:
m.fit_bounds([
[freq["lat"].min(), freq["lon"].min()],
[freq["lat"].max(), freq["lon"].max()],
])
m.save(str(MAP_PATH))
log.info("Karte erzeugt mit %d Orten.", len(freq))
return MAP_PATH
# ── Selbst-Update (git) ──────────────────────────────────────────────────────
class UpdateError(Exception):
"""Ein Schritt beim Aktualisieren ist fehlgeschlagen (Text ist benutzertauglich)."""
class Updater:
"""Aktualisiert die App per ``git`` – nur wenn sie aus einem Git-Klon läuft.
Fährt bewusst konservativ: es wird ausschließlich per *fast-forward* auf den
Stand des Servers gezogen. Lokale Änderungen oder abweichende Historie führen
zu einer klaren Fehlermeldung statt zu einem automatischen Merge.
"""
def __init__(self, base_dir: Path) -> None:
self.base_dir = base_dir
self.available = bool(shutil.which("git")) and (base_dir / ".git").is_dir()
def _git(self, *args: str, timeout: int = 30) -> subprocess.CompletedProcess:
return subprocess.run(
["git", "-C", str(self.base_dir), *args],
capture_output=True, text=True, timeout=timeout,
)
def _upstream_ref(self) -> str:
r = self._git("rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
return r.stdout.strip() or "origin/main"
def current_version(self) -> str:
r = self._git("log", "-1", "--format=%h · %cs")
return r.stdout.strip() or "unbekannt"
def check(self) -> int:
"""Anzahl neuer Commits auf dem Server. Wirft ``UpdateError`` bei Problemen."""
f = self._git("fetch", "--quiet", timeout=60)
if f.returncode != 0:
raise UpdateError(f.stderr.strip() or "Server nicht erreichbar (git fetch).")
ref = self._upstream_ref()
c = self._git("rev-list", "--count", f"HEAD..{ref}")
if c.returncode != 0:
raise UpdateError(c.stderr.strip() or "Versionsvergleich fehlgeschlagen.")
return int(c.stdout.strip() or "0")
def update(self) -> bool:
"""Zieht den neuen Stand (fast-forward) und installiert ggf. neue Pakete.
Gibt ``True`` zurück, wenn tatsächlich etwas aktualisiert wurde.
"""
f = self._git("fetch", "--quiet", timeout=60)
if f.returncode != 0:
raise UpdateError(f.stderr.strip() or "Server nicht erreichbar (git fetch).")
ref = self._upstream_ref()
before = self._git("rev-parse", "HEAD").stdout.strip()
m = self._git("merge", "--ff-only", ref, timeout=60)
if m.returncode != 0:
raise UpdateError(
(m.stderr.strip() or "Automatisches Update nicht möglich.")
+ "\n\nBitte Patrick Bescheid geben."
)
after = self._git("rev-parse", "HEAD").stdout.strip()
if before == after:
return False
changed = self._git("diff", "--name-only", before, after).stdout.split()
if "requirements.txt" in changed:
self._install_requirements()
log.info("Update: %s → %s", before[:9], after[:9])
return True
def _install_requirements(self) -> None:
req = self.base_dir / "requirements.txt"
if not req.exists():
return
r = subprocess.run(
[sys.executable, "-m", "pip", "install", "-r", str(req)],
capture_output=True, text=True, timeout=600,
)
if r.returncode != 0:
raise UpdateError(
"Neue Programmbibliotheken konnten nicht installiert werden:\n"
+ (r.stderr.strip()[-400:] or "pip-Fehler")
)
# ── Bearbeiten-Dialog ─────────────────────────────────────────────────────────
class EditDialog(tk.Toplevel):
"""Modal-Dialog zum Bearbeiten eines bestehenden Eintrags."""
def __init__(self, parent: App, df_idx: int, row: pd.Series) -> None:
super().__init__(parent)
self.parent = parent
self.df_idx = df_idx
self.result: dict | None = None
self.title("Eintrag bearbeiten")
self.resizable(False, False)
self.grab_set() # modal
pad = {"padx": 8, "pady": 4}
ttk.Label(self, text="Datum:").grid(row=0, column=0, sticky=W, **pad)
start = parse_date(str(row.get("date", "")))
start_dt = datetime.strptime(start, "%Y-%m-%d") if start else None
self._date_entry = ttk.DateEntry(
self, dateformat="%Y-%m-%d", width=12, startdate=start_dt
)
self._date_entry.grid(row=0, column=1, sticky=W, **pad)
if start:
self._date_entry.entry.delete(0, END)
self._date_entry.entry.insert(0, start)
ttk.Label(self, text="Ort:").grid(row=1, column=0, sticky=W, **pad)
self._city_var = tk.StringVar(value=str(row.get("city", "")))
self._city_cb = ttk.Combobox(self, textvariable=self._city_var, width=24)
self._city_cb["values"] = parent.store.get_autocomplete_strings()
self._city_cb.grid(row=1, column=1, sticky=W, **pad)
self._city_cb.bind("", self._on_city_key)
self._city_cb.bind("<>", self._on_city_selected)
ttk.Label(self, text="PLZ:").grid(row=2, column=0, sticky=W, **pad)
self._plz_var = tk.StringVar(value=str(row.get("postal_code", "")))
ttk.Entry(self, textvariable=self._plz_var, width=10).grid(row=2, column=1, sticky=W, **pad)
self._regeocode_var = tk.BooleanVar(value=False)
ttk.Checkbutton(
self, text="Koordinaten automatisch neu suchen", variable=self._regeocode_var,
bootstyle="round-toggle", command=self._on_regeocode_toggle,
).grid(row=3, column=0, columnspan=2, sticky=W, **pad)
coord_frame = ttk.LabelFrame(self, text=" Koordinaten (manuell) ", padding=(8, 4))
coord_frame.grid(row=4, column=0, columnspan=2, sticky=EW, padx=8, pady=(4, 4))
ttk.Label(coord_frame, text="Breitengrad:").grid(row=0, column=0, sticky=W, padx=(0, 4), pady=2)
self._lat_var = tk.StringVar(value=_coord_str(row.get("lat", "")))
self._lat_entry = ttk.Entry(coord_frame, textvariable=self._lat_var, width=14)
self._lat_entry.grid(row=0, column=1, sticky=W, pady=2)
ttk.Label(coord_frame, text="Längengrad:").grid(row=1, column=0, sticky=W, padx=(0, 4), pady=2)
self._lon_var = tk.StringVar(value=_coord_str(row.get("lon", "")))
self._lon_entry = ttk.Entry(coord_frame, textvariable=self._lon_var, width=14)
self._lon_entry.grid(row=1, column=1, sticky=W, pady=2)
ttk.Label(
coord_frame, text="Leer lassen = kein Kartenpunkt", bootstyle="secondary",
).grid(row=2, column=0, columnspan=2, sticky=W, pady=(2, 0))
btn_frame = ttk.Frame(self)
btn_frame.grid(row=5, column=0, columnspan=2, pady=(8, 6))
ttk.Button(btn_frame, text="Speichern", bootstyle="success",
command=self._on_save).pack(side=LEFT, padx=6)
ttk.Button(btn_frame, text="Abbrechen", bootstyle="secondary-outline",
command=self.destroy).pack(side=LEFT, padx=6)
self.bind("", lambda _: self._on_save())
self.bind("", lambda _: self.destroy())
# Center over parent
self.update_idletasks()
px, py = parent.winfo_x(), parent.winfo_y()
pw, ph = parent.winfo_width(), parent.winfo_height()
w, h = self.winfo_width(), self.winfo_height()
self.geometry(f"+{px + (pw - w) // 2}+{py + (ph - h) // 2}")
def _on_regeocode_toggle(self) -> None:
state = DISABLED if self._regeocode_var.get() else NORMAL
self._lat_entry.configure(state=state)
self._lon_entry.configure(state=state)
def _on_city_key(self, _event) -> None:
self._city_cb["values"] = filter_locations(
self._city_var.get(), self.parent.store.get_autocomplete_strings()
)
def _on_city_selected(self, _event) -> None:
parts = split_city_plz(self._city_var.get())
if parts:
self._city_var.set(parts[0])
self._plz_var.set(parts[1])
def _on_save(self) -> None:
iso = parse_date(self._date_entry.entry.get())
if iso is None:
messagebox.showwarning(
"Ungültiges Datum",
"Bitte ein gültiges Datum im Format JJJJ-MM-TT eingeben.",
parent=self,
)
return
city = self._city_var.get().strip()
if not city:
messagebox.showwarning("Ort fehlt", "Bitte einen Ort eingeben.", parent=self)
return
regeocode = self._regeocode_var.get()
lat = lon = ""
if not regeocode:
lat_val = parse_coord(self._lat_var.get())
lon_val = parse_coord(self._lon_var.get())
if (self._lat_var.get().strip() or self._lon_var.get().strip()) and (
lat_val is None or lon_val is None
):
messagebox.showwarning(
"Ungültige Koordinaten",
"Breitengrad und Längengrad müssen Zahlen sein – "
"oder beide leer bleiben.",
parent=self,
)
return
lat = "" if lat_val is None else f"{lat_val:.5f}"
lon = "" if lon_val is None else f"{lon_val:.5f}"
self.result = {
"date": iso,
"city": city,
"postal_code": self._plz_var.get().strip(),
"regeocode": regeocode,
"lat": lat,
"lon": lon,
}
self.destroy()
# ── Haupt-App ─────────────────────────────────────────────────────────────────
class App(ttk.Window):
#: Statusmeldungs-Art → ttkbootstrap-Style der Statusleiste.
_STATUS_STYLES = {
"info": "secondary", "success": "success",
"warning": "warning", "error": "danger",
}
def __init__(self) -> None:
super().__init__(themename="litera")
self.minsize(820, 620)
self._scale_fonts()
self._install_icon()
self.store = DataStore()
self.geocoder = GeocoderWorker(
on_service_error=lambda: self.after(0, self._on_geocode_service_error)
)
self.geocoder.start()
self._service_error_shown = False
self.updater = Updater(BASE_DIR)
self._update_busy = False
self._queue: list[dict] = []
self._next_id = 0
self._status_after_id: str | None = None
self._build_ui()
self._update_title()
self._restore_window_state()
self._reset_status()
self.protocol("WM_DELETE_WINDOW", self._on_close)
log.info("App gestartet (%d Einträge).", self.store.total_count())
if self.updater.available:
threading.Thread(target=self._auto_check_update, daemon=True).start()
# ── Aussehen ──────────────────────────────────────────────────────────────
def _scale_fonts(self) -> None:
"""Vergrößert die Standard-Schriftarten – deutlich besser lesbar."""
for name in ("TkDefaultFont", "TkTextFont", "TkMenuFont", "TkHeadingFont"):
try:
tkfont.nametofont(name).configure(size=UI_FONT_SIZE)
except tk.TclError:
pass
self._ui_family = tkfont.nametofont("TkDefaultFont").actual("family")
self.style.configure(".", font=(self._ui_family, UI_FONT_SIZE))
self.style.configure(
"Treeview", font=(self._ui_family, UI_FONT_SIZE),
rowheight=UI_FONT_SIZE * 2 + 8,
)
self.style.configure(
"Treeview.Heading", font=(self._ui_family, UI_FONT_SIZE, "bold")
)
def _install_icon(self) -> None:
"""Setzt das Fenster-Icon: mitgelieferte icon.png, sonst ein gezeichnetes Kreuz."""
try:
if ICON_PATH.exists():
self._icon_img = tk.PhotoImage(file=str(ICON_PATH))
self.iconphoto(True, self._icon_img)
return
size = 64
img = tk.PhotoImage(width=size, height=size)
img.put("white", to=(0, 0, size, size))
arm = size // 3
img.put(DRK_RED, to=(arm, 7, size - arm, size - 7))
img.put(DRK_RED, to=(7, arm, size - 7, size - arm))
self.iconphoto(True, img)
self._icon_img = img # Referenz halten
try:
img.write(str(ICON_FALLBACK_PATH), format="png")
except tk.TclError:
pass # Tk < 8.6 kann kein PNG schreiben – nicht kritisch
except tk.TclError:
log.warning("Fenster-Icon konnte nicht gesetzt werden.")
def _update_title(self) -> None:
n = self.store.total_count()
self.title(f"DRK Blutspende – Arbeitsorte · {n} Einsätze")
# ── Fensterposition merken ────────────────────────────────────────────────
def _restore_window_state(self) -> None:
try:
geom = json.loads(WINDOW_STATE_PATH.read_text()).get("geometry")
except (OSError, ValueError):
return
if isinstance(geom, str) and re.match(r"^\d+x\d+\+-?\d+\+-?\d+$", geom):
try:
self.geometry(geom)
except tk.TclError:
pass
def _save_window_state(self) -> None:
try:
WINDOW_STATE_PATH.write_text(json.dumps({"geometry": self.geometry()}))
except OSError:
log.warning("Fensterposition konnte nicht gespeichert werden.")
# ── Menüleiste & Selbst-Update ───────────────────────────────────────────
def _build_menubar(self) -> None:
menubar = tk.Menu(self)
help_menu = tk.Menu(menubar, tearoff=0)
if self.updater.available:
self._update_menu = help_menu
self._update_menu_index = 0
help_menu.add_command(
label="Nach Update suchen", command=lambda: self._check_update(manual=True)
)
help_menu.add_separator()
else:
self._update_menu = None
help_menu.add_command(label="Version…", command=self._show_version)
menubar.add_cascade(label="Hilfe", menu=help_menu)
self.configure(menu=menubar)
def _show_version(self) -> None:
if self.updater.available:
info = f"Installierte Version:\n{self.updater.current_version()}"
else:
info = "Version: unbekannt (kein Git-Klon – Update über das Menü nicht möglich)."
messagebox.showinfo("DRK Blutspende – Arbeitsorte", info, parent=self)
def _auto_check_update(self) -> None:
"""Stiller Hintergrund-Check beim Start – meldet sich nur, wenn es etwas gibt."""
time.sleep(2)
try:
n = self.updater.check()
except Exception as exc: # offline o. Ä. – nicht stören
log.info("Automatischer Update-Check übersprungen: %s", exc)
return
if n > 0:
self.after(0, lambda: self._on_update_available(n, manual=False))
def _check_update(self, *, manual: bool) -> None:
if self._update_busy:
return
if not self.updater.available:
messagebox.showinfo(
"Update", "Diese Installation kann sich nicht selbst aktualisieren.",
parent=self,
)
return
self._update_busy = True
self._set_status("Suche nach Updates…", sticky=True)
def work() -> None:
try:
n = self.updater.check()
err = None
except Exception as exc:
n, err = 0, str(exc)
self.after(0, self._on_update_checked, n, err, manual)
threading.Thread(target=work, daemon=True).start()
def _on_update_checked(self, n: int, err: str | None, manual: bool) -> None:
self._update_busy = False
if err is not None:
self._set_status("Update-Prüfung fehlgeschlagen.", "warning")
if manual:
messagebox.showwarning(
"Update",
f"Es konnte nicht nach Updates gesucht werden:\n\n{err}\n\n"
"Ist der Computer mit dem Internet verbunden?",
parent=self,
)
return
if n == 0:
self._set_status("Die App ist auf dem neuesten Stand.", "success")
if manual:
messagebox.showinfo("Update", "Die App ist auf dem neuesten Stand.", parent=self)
return
self._on_update_available(n, manual=manual)
def _on_update_available(self, n: int, *, manual: bool) -> None:
if self._update_menu is not None:
self._update_menu.entryconfigure(
self._update_menu_index, label=f"Update installieren ({n} verfügbar)"
)
self._set_status(
f"{n} Aktualisierung(en) verfügbar – Menü „Hilfe“ › Update installieren.",
"info", sticky=True,
)
count = f"{n} Aktualisierung" + ("en" if n != 1 else "")
if messagebox.askyesno(
"Update verfügbar",
f"Es gibt {count} für diese App.\n\n"
"Jetzt installieren? Die App wird dabei kurz neu gestartet.\n"
"Deine gespeicherten Einträge bleiben unverändert.",
parent=self,
):
self._do_update()
def _do_update(self) -> None:
if self._update_busy:
return
self._update_busy = True
self._set_status("App wird aktualisiert – bitte warten…", sticky=True)
def work() -> None:
try:
changed = self.updater.update()
err = None
except Exception as exc:
changed, err = False, str(exc)
self.after(0, self._on_update_done, changed, err)
threading.Thread(target=work, daemon=True).start()
def _on_update_done(self, changed: bool, err: str | None) -> None:
self._update_busy = False
if err is not None:
self._set_status("Update fehlgeschlagen.", "error")
messagebox.showerror("Update fehlgeschlagen", err, parent=self)
return
if not changed:
self._set_status("Nichts zu aktualisieren.", "info")
return
messagebox.showinfo(
"Update installiert",
"Die Aktualisierung wurde installiert. Die App startet jetzt neu.",
parent=self,
)
self._restart()
def _restart(self) -> None:
log.info("Neustart nach Update.")
self._save_window_state()
self.geocoder.stop()
try:
os.chdir(BASE_DIR)
os.execv(sys.executable, [sys.executable, str(Path(__file__).resolve())])
except OSError:
log.exception("Neustart fehlgeschlagen.")
messagebox.showwarning(
"Neustart nötig",
"Bitte die App einmal schließen und neu öffnen, "
"damit das Update wirksam wird.",
parent=self,
)
def _on_geocode_service_error(self) -> None:
"""Einmalige Warnung, wenn der Standortdienst wiederholt nicht erreichbar ist."""
self._set_status(
"Standortdienst nicht erreichbar – Koordinaten werden übersprungen. "
"Bitte Internetverbindung prüfen.",
"error",
)
if not self._service_error_shown:
self._service_error_shown = True
messagebox.showwarning(
"Keine Verbindung zum Standortdienst",
"Die Koordinatensuche (OpenStreetMap) ist gerade nicht erreichbar.\n\n"
"Einträge werden trotzdem gespeichert – nur ohne Kartenpunkt. "
"Die Koordinaten lassen sich später über \"Bearbeiten\" nachtragen.",
parent=self,
)
# ── UI-Aufbau ─────────────────────────────────────────────────────────────
def _build_ui(self) -> None:
self._build_menubar()
# Kopfzeile mit DRK-Rot
header = ttk.Frame(self, bootstyle="danger", padding=(14, 8))
header.pack(fill=X, side=TOP)
ttk.Label(
header, text="DRK Blutspende – Arbeitsorte",
bootstyle="inverse-danger",
font=(self._ui_family, UI_FONT_SIZE + 4, "bold"),
).pack(side=LEFT)
# Statusleiste (ganz unten, vor Notebook packen)
self._status_var = ttk.StringVar()
self._status_label = ttk.Label(
self, textvariable=self._status_var, bootstyle="secondary",
anchor=W, padding=(12, 5),
)
self._status_label.pack(fill=X, side=BOTTOM)
# Notebook mit zwei Tabs
self.notebook = ttk.Notebook(self)
self.notebook.pack(fill=BOTH, expand=True, padx=8, pady=8)
tab1 = ttk.Frame(self.notebook, padding=4)
tab2 = ttk.Frame(self.notebook, padding=4)
self.notebook.add(tab1, text=" Neuer Eintrag ")
self.notebook.add(tab2, text=" Einträge verwalten ")
self._build_entry_tab(tab1)
self._build_history_tab(tab2)
# Beim Wechsel zu Tab 2: Tabelle aktualisieren
self.notebook.bind("<>", self._on_tab_changed)
# ── Tab 1: Neuer Eintrag ──────────────────────────────────────────────────
def _build_entry_tab(self, parent) -> None:
input_frame = ttk.LabelFrame(parent, text=" Eintrag ", padding=10)
input_frame.pack(fill=X, padx=4, pady=(4, 6))
plz_ok = (self.register(lambda P: P == "" or (P.isdigit() and len(P) <= 5)), "%P")
ttk.Label(input_frame, text="Datum:").grid(row=0, column=0, sticky=W, padx=(0, 4), pady=4)
self._date_entry = ttk.DateEntry(input_frame, dateformat="%Y-%m-%d", width=12)
self._date_entry.grid(row=0, column=1, sticky=W, padx=(0, 16), pady=4)
ttk.Label(input_frame, text="Ort:").grid(row=0, column=2, sticky=W, padx=(0, 4), pady=4)
self._city_var = ttk.StringVar()
self.city_cb = ttk.Combobox(input_frame, textvariable=self._city_var, width=24)
self.city_cb["values"] = self.store.get_autocomplete_strings()
self.city_cb.grid(row=0, column=3, sticky=W, padx=(0, 16), pady=4)
self.city_cb.bind("", self._on_city_key)
self.city_cb.bind("<>", self._on_city_selected)
ttk.Label(input_frame, text="PLZ:").grid(row=0, column=4, sticky=W, padx=(0, 4), pady=4)
self._plz_var = ttk.StringVar()
plz_entry = ttk.Entry(
input_frame, textvariable=self._plz_var, width=8,
validate="key", validatecommand=plz_ok,
)
plz_entry.grid(row=0, column=5, sticky=W, padx=(0, 16), pady=4)
add_btn = ttk.Button(
input_frame, text="+ Hinzufügen", bootstyle="success",
command=self._on_add_row,
)
add_btn.grid(row=0, column=6, pady=4)
for w in (self._date_entry.entry, self.city_cb, plz_entry):
w.bind("", lambda _e: self._on_add_row())
# Warteschlange
queue_frame = ttk.LabelFrame(parent, text=" Warteschlange (noch nicht gespeichert) ", padding=(8, 4))
queue_frame.pack(fill=BOTH, expand=True, padx=4, pady=4)
cols = ("date", "city", "postal_code", "coords", "del")
self.tree = ttk.Treeview(queue_frame, columns=cols, show="headings", height=9)
self.tree.heading("date", text="Datum")
self.tree.heading("city", text="Ort")
self.tree.heading("postal_code", text="PLZ")
self.tree.heading("coords", text="Koordinaten")
self.tree.heading("del", text="")
self.tree.column("date", width=100, minwidth=90, stretch=False)
self.tree.column("city", width=180, minwidth=120)
self.tree.column("postal_code", width=65, minwidth=50, stretch=False)
self.tree.column("coords", width=185, minwidth=130)
self.tree.column("del", width=32, minwidth=32, stretch=False, anchor=CENTER)
sb = ttk.Scrollbar(queue_frame, orient=VERTICAL, command=self.tree.yview)
self.tree.configure(yscrollcommand=sb.set)
self.tree.pack(side=LEFT, fill=BOTH, expand=True)
sb.pack(side=RIGHT, fill=Y)
self.tree.bind("", self._on_tree_click)
self.tree.bind("", self._on_queue_right_click)
self.tree.bind("", lambda _e: self._delete_selected_queue_rows())
# Aktionsleiste
action_frame = ttk.Frame(parent, padding=(4, 6))
action_frame.pack(fill=X)
ttk.Button(action_frame, text="Alle speichern", bootstyle="success",
command=self._on_save_all).pack(side=LEFT, padx=(0, 6))
ttk.Button(action_frame, text="Auswahl entfernen", bootstyle="secondary-outline",
command=self._delete_selected_queue_rows).pack(side=LEFT, padx=(0, 6))
ttk.Button(action_frame, text="Warteschlange leeren", bootstyle="secondary-outline",
command=self._on_clear_queue).pack(side=LEFT, padx=(0, 6))
ttk.Button(action_frame, text="Karte öffnen", bootstyle="info",
command=self._on_open_map).pack(side=RIGHT)
# ── Tab 2: Einträge verwalten ─────────────────────────────────────────────
def _build_history_tab(self, parent) -> None:
# Suchleiste
search_frame = ttk.Frame(parent, padding=(4, 4, 4, 2))
search_frame.pack(fill=X)
ttk.Label(search_frame, text="Suche:").pack(side=LEFT, padx=(0, 6))
self._search_var = ttk.StringVar()
self._search_var.trace_add("write", lambda *_: self._refresh_history())
ttk.Entry(search_frame, textvariable=self._search_var, width=30).pack(side=LEFT)
ttk.Button(search_frame, text="✕", bootstyle="secondary-link", width=3,
command=lambda: self._search_var.set("")).pack(side=LEFT)
ttk.Button(search_frame, text="Karte öffnen", bootstyle="info",
command=self._on_open_map).pack(side=RIGHT)
self._only_missing_var = ttk.BooleanVar(value=False)
ttk.Checkbutton(
search_frame, text="Nur ohne Koordinaten",
variable=self._only_missing_var, bootstyle="round-toggle",
command=self._refresh_history,
).pack(side=LEFT, padx=(12, 0))
# Tabelle
hist_frame = ttk.Frame(parent, padding=(4, 2))
hist_frame.pack(fill=BOTH, expand=True)
self._hist_headings = {
"date": "Datum", "city": "Ort", "postal_code": "PLZ", "coords": "Karte",
}
cols = tuple(self._hist_headings)
self.hist_tree = ttk.Treeview(hist_frame, columns=cols, show="headings", height=16)
for key in cols:
self.hist_tree.heading(
key, text=self._hist_headings[key],
command=lambda k=key: self._sort_history(k),
)
self.hist_tree.column("date", width=115, minwidth=95, stretch=False)
self.hist_tree.column("city", width=260, minwidth=140)
self.hist_tree.column("postal_code", width=80, minwidth=55, stretch=False, anchor=CENTER)
self.hist_tree.column("coords", width=90, minwidth=70, stretch=False, anchor=CENTER)
sb2 = ttk.Scrollbar(hist_frame, orient=VERTICAL, command=self.hist_tree.yview)
self.hist_tree.configure(yscrollcommand=sb2.set)
self.hist_tree.pack(side=LEFT, fill=BOTH, expand=True)
sb2.pack(side=RIGHT, fill=Y)
self.hist_tree.tag_configure("missing", foreground=DRK_RED)
self.hist_tree.bind("", self._on_hist_double_click)
self.hist_tree.bind("", self._on_hist_right_click)
self._sort_col = "date"
self._sort_asc = False # neueste zuerst
self._apply_sort_headings()
# Aktionsleiste
action_frame = ttk.Frame(parent, padding=(4, 4))
action_frame.pack(fill=X)
ttk.Button(action_frame, text="Bearbeiten", bootstyle="primary-outline",
command=self._on_hist_edit).pack(side=LEFT, padx=(0, 6))
ttk.Button(action_frame, text="Löschen", bootstyle="danger-outline",
command=self._on_hist_delete).pack(side=LEFT)
self._hist_hint_var = ttk.StringVar()
ttk.Label(action_frame, textvariable=self._hist_hint_var,
bootstyle="secondary").pack(side=RIGHT)
def _on_tab_changed(self, _event) -> None:
if self.notebook.index("current") == 1:
self._refresh_history()
def _refresh_history(self) -> None:
"""Füllt die Verlaufstabelle neu aus dem DataStore."""
for item in self.hist_tree.get_children():
self.hist_tree.delete(item)
df = self.store.df.copy()
missing = df["lat"].map(_is_blank) | df["lon"].map(_is_blank)
total = len(df)
total_missing = int(missing.sum())
query = self._search_var.get().strip().lower()
if query:
mask = (
df["date"].astype(str).str.lower().str.contains(query) |
df["city"].astype(str).str.lower().str.contains(query) |
df["postal_code"].astype(str).str.lower().str.contains(query)
)
df = df[mask]
if self._only_missing_var.get():
df = df[missing.reindex(df.index, fill_value=False)]
if self._sort_col == "coords":
order = missing.reindex(df.index, fill_value=True)
df = df.assign(_ord=order).sort_values("_ord", ascending=self._sort_asc)
else:
df = df.sort_values(
self._sort_col, ascending=self._sort_asc, na_position="last"
)
for idx, row in df.iterrows():
row_missing = _is_blank(row.get("lat", "")) or _is_blank(row.get("lon", ""))
self.hist_tree.insert(
"", END, iid=str(idx),
tags=("missing",) if row_missing else (),
values=(
row.get("date", ""),
row.get("city", ""),
row.get("postal_code", ""),
"fehlt" if row_missing else "✓",
),
)
parts = [f"{total} Einsätze gespeichert"]
if total_missing:
parts.append(
f"{total_missing} ohne Koordinaten (rot) – Rechtsklick › Koordinaten suchen"
)
else:
parts.append("Doppelklick zum Bearbeiten")
self._hist_hint_var.set(" · ".join(parts))
def _apply_sort_headings(self) -> None:
"""Zeigt einen Pfeil an der aktuell sortierten Spalte."""
arrow = " ▲" if self._sort_asc else " ▼"
for key, label in self._hist_headings.items():
self.hist_tree.heading(
key, text=label + (arrow if key == self._sort_col else "")
)
def _sort_history(self, col: str) -> None:
if self._sort_col == col:
self._sort_asc = not self._sort_asc
else:
self._sort_col = col
self._sort_asc = True
self._apply_sort_headings()
self._refresh_history()
def _on_hist_double_click(self, event) -> None:
if self.hist_tree.identify_region(event.x, event.y) == "cell":
self._on_hist_edit()
def _on_hist_right_click(self, event) -> None:
iid = self.hist_tree.identify_row(event.y)
if not iid:
return
self.hist_tree.selection_set(iid)
menu = tk.Menu(self, tearoff=0)
menu.add_command(label="Bearbeiten", command=self._on_hist_edit)
menu.add_command(label="Koordinaten suchen", command=self._on_hist_geocode)
menu.add_separator()
menu.add_command(label="Löschen", command=self._on_hist_delete)
try:
menu.tk_popup(event.x_root, event.y_root)
finally:
menu.grab_release()
def _on_hist_geocode(self) -> None:
sel = self.hist_tree.selection()
if not sel:
self._set_status("Bitte einen Eintrag auswählen.")
return
df_idx = int(sel[0])
row = self.store.df.loc[df_idx]
city = str(row.get("city", "")).strip()
plz = str(row.get("postal_code", "")).strip()
if not city:
self._set_status("Eintrag hat keinen Ort – Koordinatensuche nicht möglich.")
return
self.geocoder.enqueue(
f"edit_{df_idx}", city, plz,
lambda _rid, coords, idx=df_idx: self.after(0, self._apply_edit_geocode, idx, coords),
)
self._set_status(f"Koordinaten werden gesucht für {city}…")
def _on_hist_edit(self) -> None:
sel = self.hist_tree.selection()
if not sel:
self._set_status("Bitte einen Eintrag auswählen.")
return
df_idx = int(sel[0])
row = self.store.df.loc[df_idx]
dlg = EditDialog(self, df_idx, row)
self.wait_window(dlg)
if dlg.result is None:
return # Abgebrochen
new_vals = {
"date": dlg.result["date"],
"city": dlg.result["city"],
"postal_code": dlg.result["postal_code"],
}
if dlg.result["regeocode"]:
# Zuerst speichern (ohne Koordinaten), dann im Hintergrund geocodieren
new_vals["lat"] = ""
new_vals["lon"] = ""
self.store.update_row(df_idx, new_vals)
self.geocoder.enqueue(
f"edit_{df_idx}",
dlg.result["city"],
dlg.result["postal_code"],
lambda _rid, coords, idx=df_idx: self.after(0, self._apply_edit_geocode, idx, coords),
)
self._set_status(f"Koordinaten werden gesucht für {new_vals['city']}…")
else:
new_vals["lat"] = dlg.result["lat"]
new_vals["lon"] = dlg.result["lon"]
self.store.update_row(df_idx, new_vals)
self._set_status(
f"Eintrag aktualisiert: {new_vals['date']} {new_vals['city']}.",
"success",
)
self.city_cb["values"] = self.store.get_autocomplete_strings()
self._refresh_history()
def _apply_edit_geocode(self, df_idx: int, coords: tuple | None) -> None:
"""Koordinaten nach Hintergrund-Geocodierung in bestehenden Eintrag schreiben."""
if df_idx not in self.store.df.index:
return # Zeile wurde zwischenzeitlich gelöscht
if coords:
self.store.update_row(df_idx, {
"lat": round(coords[0], 5),
"lon": round(coords[1], 5),
})
self._set_status("Koordinaten aktualisiert.", "success")
else:
self._set_status(
"Koordinaten konnten nicht gefunden werden – ggf. manuell eintragen.",
"warning",
)
self._refresh_history()
def _on_hist_delete(self) -> None:
sel = self.hist_tree.selection()
if not sel:
self._set_status("Bitte mindestens einen Eintrag auswählen.")
return
count = len(sel)
label = f"{count} Eintrag/Einträge" if count > 1 else "diesen Eintrag"
if not messagebox.askyesno(
"Löschen bestätigen",
f"Soll {label} wirklich gelöscht werden? Diese Aktion kann nicht rückgängig gemacht werden.",
parent=self,
):
return
indices = [int(iid) for iid in sel]
self.store.delete_rows(indices)
self._update_title()
self._refresh_history()
self._set_status(
f"{count} Eintrag/Einträge gelöscht · {self.store.total_count()} insgesamt.",
"warning",
)
# ── Autovervollständigung ─────────────────────────────────────────────────
def _on_city_key(self, _event) -> None:
self.city_cb["values"] = filter_locations(
self._city_var.get(), self.store.get_autocomplete_strings()
)
def _on_city_selected(self, _event) -> None:
parts = split_city_plz(self._city_var.get())
if parts:
self._city_var.set(parts[0])
self._plz_var.set(parts[1])
# ── Warteschlange ─────────────────────────────────────────────────────────
def _on_add_row(self) -> None:
city = self._city_var.get().strip()
plz = self._plz_var.get().strip()
date_str = parse_date(self._date_entry.entry.get())
if not city:
self._set_status("Bitte einen Ort eingeben.", "warning")
return
if date_str is None:
self._set_status("Bitte ein gültiges Datum (JJJJ-MM-TT) eingeben.", "warning")
messagebox.showwarning(
"Ungültiges Datum",
"Bitte ein gültiges Datum im Format JJJJ-MM-TT eingeben "
"oder den Kalender-Knopf benutzen.",
parent=self,
)
return
row_id = str(self._next_id)
self._next_id += 1
row = {"_id": row_id, "date": date_str, "city": city,
"postal_code": plz, "lat": "", "lon": ""}
self._queue.append(row)
self.tree.insert("", END, iid=row_id,
values=(date_str, city, plz, "⏳ wird gesucht…", "✕"))
self.geocoder.enqueue(row_id, city, plz, self._geocode_done)
self._city_var.set("")
self._plz_var.set("")
self.city_cb.focus_set()
self._set_status(
f"{len(self._queue)} Einträge in der Warteschlange.", sticky=True
)
def _geocode_done(self, row_id: str, coords: tuple | None) -> None:
self.after(0, self._apply_geocode_result, row_id, coords)
def _apply_geocode_result(self, row_id: str, coords: tuple | None) -> None:
for row in self._queue:
if row["_id"] == row_id:
if coords:
row["lat"] = round(coords[0], 5)
row["lon"] = round(coords[1], 5)
coord_str = f"{row['lat']:.4f} / {row['lon']:.4f}"
else:
coord_str = "⚠ nicht gefunden"
break
else:
return
if self.tree.exists(row_id):
vals = list(self.tree.item(row_id, "values"))
vals[3] = coord_str
self.tree.item(row_id, values=vals)
def _on_tree_click(self, event) -> None:
if self.tree.identify_region(event.x, event.y) != "cell":
return
if self.tree.identify_column(event.x) == "#5":
row_id = self.tree.identify_row(event.y)
if row_id:
self._delete_queue_row(row_id)
def _on_queue_right_click(self, event) -> None:
row_id = self.tree.identify_row(event.y)
if not row_id:
return
menu = tk.Menu(self, tearoff=0)
menu.add_command(label="Löschen", command=lambda: self._delete_queue_row(row_id))
menu.add_command(label="Koordinaten erneut suchen",
command=lambda: self._retry_geocode(row_id))
try:
menu.tk_popup(event.x_root, event.y_root)
finally:
menu.grab_release()
def _delete_queue_row(self, row_id: str) -> None:
self._queue = [r for r in self._queue if r["_id"] != row_id]
if self.tree.exists(row_id):
self.tree.delete(row_id)
self._set_status(
f"{len(self._queue)} Einträge in der Warteschlange.", sticky=True
)
def _delete_selected_queue_rows(self) -> None:
sel = self.tree.selection()
if not sel:
self._set_status("Bitte zuerst eine Zeile in der Warteschlange auswählen.")
return
for row_id in sel:
self._delete_queue_row(row_id)
def _retry_geocode(self, row_id: str) -> None:
for row in self._queue:
if row["_id"] == row_id:
if self.tree.exists(row_id):
vals = list(self.tree.item(row_id, "values"))
vals[3] = "⏳ wird gesucht…"
self.tree.item(row_id, values=vals)
self.geocoder.enqueue(row_id, row["city"], row["postal_code"],
self._geocode_done)
break
def _on_clear_queue(self) -> None:
if self._queue and not messagebox.askyesno(
"Warteschlange leeren",
f"{len(self._queue)} noch nicht gespeicherte Einträge verwerfen?",
parent=self,
):
return
self._queue.clear()
for item in self.tree.get_children():
self.tree.delete(item)
self._set_status("Warteschlange geleert.")
# ── Speichern ─────────────────────────────────────────────────────────────
def _on_save_all(self) -> None:
if not self._queue:
self._set_status("Warteschlange ist leer.")
return
dupes = self.store.find_duplicates(self._queue)
if dupes:
listed = "\n".join(f" • {d['date']} {d['city']}" for d in dupes[:8])
more = f"\n … und {len(dupes) - 8} weitere" if len(dupes) > 8 else ""
if not messagebox.askyesno(
"Mögliche Duplikate",
f"{len(dupes)} Eintrag/Einträge scheint es schon zu geben:\n\n"
f"{listed}{more}\n\nTrotzdem alle speichern?",
parent=self,
):
self._set_status("Speichern abgebrochen – Warteschlange unverändert.")
return
self.store.append_rows(self._queue)
count = len(self._queue)
self._queue.clear()
for item in self.tree.get_children():
self.tree.delete(item)
self.city_cb["values"] = self.store.get_autocomplete_strings()
self._update_title()
self._refresh_history() # Tab 2 sofort aktualisieren
self._set_status(
f"{count} Eintrag/Einträge gespeichert · {self.store.total_count()} insgesamt.",
"success",
)
# ── Karte ─────────────────────────────────────────────────────────────────
def _on_open_map(self) -> None:
try:
path = generate_map(self.store)
except ValueError as e:
messagebox.showinfo("Karte", str(e), parent=self)
self._set_status(str(e), "warning")
return
except Exception as e:
log.exception("Karte konnte nicht erstellt werden.")
messagebox.showerror(
"Karte", f"Die Karte konnte nicht erstellt werden:\n{e}", parent=self
)
self._set_status(f"Fehler beim Erstellen der Karte: {e}", "error")
return
try:
webbrowser.open(path.as_uri())
except Exception:
log.exception("Karte konnte nicht im Browser geöffnet werden.")
self._set_status("Karte im Browser geöffnet.", "success")
# ── Hilfsmethoden ─────────────────────────────────────────────────────────
def _set_status(self, msg: str, kind: str = "info", *, sticky: bool = False) -> None:
"""Zeigt eine Statusmeldung, farblich nach Art, und blendet sie nach 8 s aus."""
self._status_var.set(msg)
self._status_label.configure(
bootstyle=self._STATUS_STYLES.get(kind, "secondary")
)
if self._status_after_id is not None:
self.after_cancel(self._status_after_id)
self._status_after_id = None
if not sticky:
self._status_after_id = self.after(8000, self._reset_status)
def _reset_status(self) -> None:
"""Setzt die Statusleiste auf den neutralen Dauerzustand zurück."""
self._status_after_id = None
self._status_label.configure(bootstyle="secondary")
if self._queue:
self._status_var.set(
f"{len(self._queue)} Einträge in der Warteschlange – noch nicht gespeichert."
)
else:
self._status_var.set("Bereit.")
def _on_close(self) -> None:
log.info("App wird beendet.")
self._save_window_state()
self.geocoder.stop()
self.destroy()
# ── Start ─────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
_setup_logging()
try:
app = App()
app.mainloop()
except Exception:
log.exception("Nicht abgefangener Fehler – App wird beendet.")
raise