""" 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 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 = DATA_DIR / "icon.png" 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 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="CartoDB positron", ) 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 # ── 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._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()) # ── 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: """Zeichnet ein einfaches rotes Kreuz als Fenster-Icon (ohne Extra-Abhängigkeit).""" try: 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 if not ICON_PATH.exists(): try: img.write(str(ICON_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.") 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: # 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