From b05fdfb0718085b45d088e1421bc9445a5c97ff3 Mon Sep 17 00:00:00 2001 From: trnq1ll0 Date: Tue, 23 Jun 2026 06:46:31 +0200 Subject: [PATCH] Add: Booking function to correct days with missing bookings --- ezicom_book.py | 708 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 708 insertions(+) create mode 100755 ezicom_book.py diff --git a/ezicom_book.py b/ezicom_book.py new file mode 100755 index 0000000..8fe3358 --- /dev/null +++ b/ezicom_book.py @@ -0,0 +1,708 @@ +#!/usr/bin/env python3 +"""eZICOM5 Zeitbuchungs-Korrektur-Tool. + +Nimmt Korrekturbuchungen für fehlerhafte Tage aus der Fehlerliste vor. +Verwendet die gleichen Login-Schritte wie ezicom_login.py. + +Credentials werden aus Umgebungsvariablen, Kommandozeilenargumenten oder +interaktiven Eingaben gelesen und niemals auf die Festplatte geschrieben. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import getpass +import html +import http.cookiejar +import os +import re +import ssl +import sys +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + +# --------------------------------------------------------------------------- +# Konstanten +# --------------------------------------------------------------------------- + +DEFAULT_BASE_URL = "https://zm.sufw.de:31443" +DEFAULT_USER_AGENT = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:151.0) " + "Gecko/20100101 Firefox/151.0" +) + +# Standard-Korrekturbuchungen: (Uhrzeit, Art) +# 1000 = Kommen, 2000 = Gehen +CORRECTION_BOOKINGS = (("10:00", "1000"), ("14:00", "2000")) + +# Maximale Anzahl Tage, für die Korrekturen möglich sind +MAX_CORRECTION_AGE_DAYS = 14 + +# --------------------------------------------------------------------------- +# Hilfsfunktionen +# --------------------------------------------------------------------------- + + +def clean_html_text(value: str) -> str: + """Entfernt HTML-Tags und normalisiert Whitespace.""" + value = re.sub(r"<[^>]+>", " ", value) + value = html.unescape(value).replace("\xa0", " ") + return " ".join(value.split()) + + +def load_dotenv(path: Path = Path(".env")) -> None: + """Lädt Umgebungsvariablen aus einer .env-Datei.""" + if not path.exists(): + return + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, value = line.split("=", 1) + key = key.strip() + value = value.strip().strip("'\"") + if key and key not in os.environ: + os.environ[key] = value + + +def make_opener(insecure: bool) -> urllib.request.OpenerDirector: + """Erstellt einen URL-Opener mit Cookie-Jar.""" + cookie_jar = http.cookiejar.CookieJar() + handlers: list[urllib.request.BaseHandler] = [ + urllib.request.HTTPCookieProcessor(cookie_jar) + ] + if insecure: + context = ssl._create_unverified_context() + handlers.append(urllib.request.HTTPSHandler(context=context)) + return urllib.request.build_opener(*handlers) + + +def request( + opener: urllib.request.OpenerDirector, + url: str, + *, + method: str = "GET", + data: dict[str, str] | None = None, + headers: dict[str, str] | None = None, + timeout: float = 20.0, +) -> tuple[int, str, str]: + """Führt einen HTTP-Request aus und gibt (status, url, body) zurück.""" + body = None + request_headers = { + "User-Agent": DEFAULT_USER_AGENT, + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + } + if headers: + request_headers.update(headers) + if data is not None: + body = urllib.parse.urlencode(data).encode("ascii") + request_headers["Content-Type"] = "application/x-www-form-urlencoded" + + req = urllib.request.Request(url, data=body, headers=request_headers, method=method) + try: + with opener.open(req, timeout=timeout) as response: + raw = response.read() + charset = response.headers.get_content_charset() or "iso-8859-1" + return response.status, response.geturl(), raw.decode(charset, "replace") + except urllib.error.HTTPError as error: + raw = error.read() + charset = error.headers.get_content_charset() or "iso-8859-1" + return error.code, error.geturl(), raw.decode(charset, "replace") + + +def find_single_quoted_js_value(html_text: str, pattern: str) -> str | None: + """Sucht einen in einfachen Anführungszeichen stehenden JS-Wert.""" + match = re.search(pattern, html_text) + if not match: + return None + return match.group(1) + + +def mask_secret(value: str) -> str: + """Maskiert einen sensiblen Wert für die Anzeige.""" + if len(value) <= 8: + return "***" + return value[:4] + "..." + value[-4:] + + +def _save_debug_response(debug_dir: Path, label: str, content: str) -> None: + """Speichert eine Server-Antwort im Debug-Ordner mit Zeitstempel.""" + timestamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") + filename = debug_dir / f"{timestamp}-{label}.html" + filename.write_text(content, encoding="utf-8") + print(f"[DEBUG] → {filename}") + + +# --------------------------------------------------------------------------- +# Datumshilfen +# --------------------------------------------------------------------------- + + +def parse_ezicom_date(value: str) -> dt.date: + return dt.datetime.strptime(value, "%d.%m.%Y").date() + + +def format_ezicom_date(value: dt.date) -> str: + return value.strftime("%d.%m.%Y") + + +# --------------------------------------------------------------------------- +# Login (gleiche Schritte wie ezicom_login.py) +# --------------------------------------------------------------------------- + + +def login( + base_url: str, + username: str, + password: str, + *, + insecure: bool, + debug: bool = False, +) -> tuple[urllib.request.OpenerDirector, str] | None: + """Führt den Login durch und gibt (opener, sid) zurück. + + Bei Fehlschlag wird None zurückgegeben. + """ + base_url = base_url.rstrip("/") + opener = make_opener(insecure) + + # 1. Initiale Seite laden + try: + initial_status, _, _ = request(opener, base_url + "/") + except urllib.error.URLError as error: + print(f"Konnte Startseite nicht öffnen: {error}", file=sys.stderr) + if "CERTIFICATE_VERIFY_FAILED" in str(error): + print("Hinweis: versuche --insecure oder installiere das CA-Zertifikat.", file=sys.stderr) + return None + if initial_status >= 400: + print(f"Startseite lieferte HTTP {initial_status}", file=sys.stderr) + return None + + # 2. Login-Formular absenden + try: + status, final_url, html_text = request( + opener, + base_url + "/user", + method="POST", + data={"action": "login", "uid": username, "pw0": password}, + headers={ + "Origin": base_url, + "Referer": base_url + "/", + }, + ) + except urllib.error.URLError as error: + print(f"Login-Request fehlgeschlagen: {error}", file=sys.stderr) + if "CERTIFICATE_VERIFY_FAILED" in str(error): + print("Hinweis: versuche --insecure oder installiere das CA-Zertifikat.", file=sys.stderr) + return None + + lower_html = html_text.lower() + failed = ( + status in {401, 403} + or "benutzername und/oder kennwort" in lower_html + or "error 403" in lower_html + or "403 - verboten" in lower_html + ) + if failed: + print(f"Login fehlgeschlagen: HTTP {status} ({final_url})", file=sys.stderr) + return None + + if debug: + print(f"[DEBUG] Login erfolgreich: HTTP {status} ({final_url})") + + # 3. Weiter-Seite folgen (rahmen.html) + next_result = _follow_next_page(opener, base_url, html_text) + if next_result is None: + print("Keine Folge-URL in Login-Antwort gefunden.", file=sys.stderr) + return None + + next_status, next_url, next_html = next_result + if next_status >= 400: + print(f"Folgeseite fehlgeschlagen: HTTP {next_status} ({next_url})", file=sys.stderr) + return None + + if debug: + print(f"[DEBUG] Folgeseite geladen: HTTP {next_status} ({next_url})") + + # 4. Session-ID extrahieren + sid = _extract_sid(next_html, next_url) + if not sid: + print("Keine Session-ID in der Folgeseite gefunden.", file=sys.stderr) + return None + + if debug: + print(f"[DEBUG] Session-ID: {mask_secret(sid)}") + + return opener, sid + + +def _follow_next_page( + opener: urllib.request.OpenerDirector, + base_url: str, + html_text: str, +) -> tuple[int, str, str] | None: + """Folgt der 'Weiter'-Seite nach dem Login.""" + cust_info_url = find_single_quoted_js_value( + html_text, r"url:\s*'([^']+/isCustInfoPnl)'" + ) + build_info_url = find_single_quoted_js_value( + html_text, r"\$\.get\('([^']+/buildCustInfoPnl)'" + ) + frame_url = find_single_quoted_js_value( + html_text, r"location\.href='([^']+rahmen\.html)'" + ) + + if not frame_url: + return None + + if cust_info_url: + status, _, response = request( + opener, urllib.parse.urljoin(base_url + "/", cust_info_url) + ) + if status < 400 and response.strip() == "-1" and build_info_url: + return request( + opener, urllib.parse.urljoin(base_url + "/", build_info_url) + ) + + return request(opener, urllib.parse.urljoin(base_url + "/", frame_url)) + + +def _extract_sid(html_text: str, url: str) -> str | None: + """Extrahiert die Session-ID aus HTML oder URL.""" + sid = find_single_quoted_js_value( + html_text, r']+name="sid"[^>]+value="([^"]+)"' + ) + if sid: + return sid + match = re.search(r"/user/([A-Fa-f0-9]+)/rahmen\.html", url) + if match: + return match.group(1) + return None + + +# --------------------------------------------------------------------------- +# Fehlerliste +# --------------------------------------------------------------------------- + +FehlerlisteEntry = tuple[dt.date, str, str] # (Datum, Wochentag, Grund) + + +def fetch_fehlerliste( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, +) -> tuple[int, str, str]: + """Ruft die Fehlerliste ab.""" + params = { + "page": "corrindex", + "go": "dfhl", + "sid": sid, + "Btn": "9", + "nc": str(int(dt.datetime.now().timestamp() * 1000)), + } + return request( + opener, base_url.rstrip("/") + "/user?" + urllib.parse.urlencode(params) + ) + + +def parse_fehlerliste(html_text: str) -> list[FehlerlisteEntry]: + """Parst die HTML-Fehlerliste und gibt Einträge zurück.""" + dfhl_match = re.search( + r']*>(.*?)', + html_text, + flags=re.IGNORECASE | re.DOTALL, + ) + if not dfhl_match: + return [] + + rows = re.findall( + r"]*>(.*?)", + dfhl_match.group(1), + flags=re.IGNORECASE | re.DOTALL, + ) + entries: list[FehlerlisteEntry] = [] + for row in rows: + cells = re.findall( + r"]*>(.*?)", + row, + flags=re.IGNORECASE | re.DOTALL, + ) + if len(cells) < 4: + continue + + date_text = clean_html_text(cells[2]) + date_match = re.search(r"\b(\d{2}\.\d{2}\.\d{4})\b", date_text) + if not date_match: + continue + + weekday = date_text[: date_match.start()].strip() + reason = re.sub(r"\[([^\]]+)\]", r"\1", clean_html_text(cells[3])) + entries.append((parse_ezicom_date(date_match.group(1)), weekday, reason)) + + return entries + + +def split_fehlerliste_entries( + entries: list[FehlerlisteEntry], +) -> tuple[list[FehlerlisteEntry], list[FehlerlisteEntry]]: + """Teilt Einträge in 'aktuell korrigierbar' und 'zu alt'.""" + today = dt.date.today() + recent: list[FehlerlisteEntry] = [] + too_old: list[FehlerlisteEntry] = [] + for entry in entries: + age_days = (today - entry[0]).days + if 0 <= age_days <= MAX_CORRECTION_AGE_DAYS: + recent.append(entry) + else: + too_old.append(entry) + return recent, too_old + + +def format_fehlerliste_entry(entry: FehlerlisteEntry) -> str: + """Formatiert einen Fehlerlisteneintrag für die Anzeige.""" + date_value, weekday, reason = entry + day = ( + f"{weekday} {format_ezicom_date(date_value)}" + if weekday + else format_ezicom_date(date_value) + ) + return f"{day}: {reason}" + + +def print_fehlerliste(entries: list[FehlerlisteEntry]) -> list[FehlerlisteEntry]: + """Gibt die Fehlerliste aus und gibt die korrigierbaren Einträge zurück.""" + if not entries: + print("Fehlerliste: keine fehlerhaften Tage gefunden.") + return [] + + print(f"\nFehlerliste: {len(entries)} fehlerhafte Tage") + recent, too_old = split_fehlerliste_entries(entries) + + if recent: + print(f"\n📋 Buchbar (jünger als {MAX_CORRECTION_AGE_DAYS + 1} Tage):") + for number, entry in enumerate(recent, start=1): + print(f" {number:>2}) {format_fehlerliste_entry(entry)}") + else: + print(f"\n📋 Buchbar (jünger als {MAX_CORRECTION_AGE_DAYS + 1} Tage): keine Einträge") + + if too_old: + print(f"\n⚠️ Älter als {MAX_CORRECTION_AGE_DAYS} Tage – keine Änderung mehr möglich:") + for entry in too_old: + print(f" {format_fehlerliste_entry(entry)}") + + return recent + + +# --------------------------------------------------------------------------- +# Korrekturbuchungen +# --------------------------------------------------------------------------- + + +def fetch_corrnew_page( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, + *, + debug: bool = False, + debug_dir: Path | None = None, +) -> bool: + """Öffnet die 'Neue Buchung'-Seite (corrnew). + + Diese Seite wird über den Button 'Neue Buchung' auf der Fehlerliste + aufgerufen. Sie bereitet das System auf eine neue Korrekturbuchung vor. + + URL: /user?page=corrnew&sid={sid}&bt=1 + """ + params = { + "page": "corrnew", + "sid": sid, + "bt": "1", + } + url = base_url.rstrip("/") + "/user?" + urllib.parse.urlencode(params) + if debug: + print(f"[DEBUG] corrnew URL: {url}") + + status, final_url, response_text = request(opener, url) + if debug: + print(f"[DEBUG] corrnew response: HTTP {status} ({final_url})") + preview = clean_html_text(response_text)[:500] + print(f"[DEBUG] corrnew body preview: {preview or ''}") + + if debug_dir: + _save_debug_response(debug_dir, "corrnew", response_text) + + if status >= 400: + print( + f"Neue Buchung konnte nicht geöffnet werden: HTTP {status}", + file=sys.stderr, + ) + return False + return True + + +def submit_correction_booking( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, + date_value: dt.date, + time_value: str, + art_value: str, + *, + comment: str = "Automatisch nachgetragen", + debug: bool = False, + debug_dir: Path | None = None, +) -> bool: + """Sendet eine einzelne Korrekturbuchung – entspricht postbook(1) im Browser. + + Endpoint: /call/{sid}/mem/htm/corrnewdo + Antwort: JSON {val: "ok"|"err", msg: "..."} + """ + params = { + "datum": format_ezicom_date(date_value), + "zeit": time_value, + "art": art_value, + "awn": "0", + "msg": comment, + } + url = ( + base_url.rstrip("/") + + f"/call/{urllib.parse.quote(sid)}/mem/htm/corrnewdo?" + + urllib.parse.urlencode(params) + ) + if debug: + print(f"[DEBUG] corrnewdo URL: {url}") + + status, final_url, response_text = request(opener, url) + if debug: + print(f"[DEBUG] corrnewdo response: HTTP {status}") + print(f"[DEBUG] corrnewdo body: {response_text.strip() or ''}") + + if debug_dir: + label = f"corrnewdo-{time_value.replace(':', '')}" + _save_debug_response(debug_dir, label, response_text) + + if status >= 400: + return False + + # corrnewdo liefert JSON: {"val":"ok","msg":"..."} oder {"val":"err","msg":"..."} + try: + import json + result = json.loads(response_text) + if result.get("val") == "ok": + return True + else: + print(f" ↳ Server meldet: {result.get('msg', 'Unbekannter Fehler')}", file=sys.stderr) + return False + except json.JSONDecodeError: + # Fallback: HTML-Response wie bisher auswerten + lower_resp = response_text.lower() + if "fehler" in lower_resp or "error" in lower_resp: + if debug: + print(f"[DEBUG] Möglicher Fehler im Response: {response_text.strip()}") + return False + return True + + +def submit_corrections_for_day( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, + entry: FehlerlisteEntry, + *, + debug: bool = False, + debug_dir: Path | None = None, +) -> bool: + """Führt alle Korrekturbuchungen für einen Tag durch. + + Ablauf (entspricht dem Browser-Workflow): + 1. corrnew – 'Neue Buchung'-Seite aufrufen (wie der Button) + 2. corrnewdo – postbook(1) für jeden Eintrag (10:00 Kommen, 14:00 Gehen) + """ + date_value, _, reason = entry + print(f"\n🔧 Korrektur für {format_ezicom_date(date_value)} ({reason}):") + + # Schritt 1: corrnew Seite aufrufen (entspricht dem Button "Neue Buchung") + if not fetch_corrnew_page(opener, base_url, sid, debug=debug, debug_dir=debug_dir): + return False + + # Schritt 2: Einzelne Buchungen senden (postbook(1) für jeden Eintrag) + success = True + for time_value, art_value in CORRECTION_BOOKINGS: + art_name = "Kommen" if art_value == "1000" else "Gehen" + if submit_correction_booking( + opener, base_url, sid, date_value, time_value, art_value, + debug=debug, debug_dir=debug_dir, + ): + print(f" ✅ {time_value} Uhr: {art_name} – gesendet") + else: + print(f" ❌ {time_value} Uhr: {art_name} – fehlgeschlagen", file=sys.stderr) + success = False + + return success + + +# --------------------------------------------------------------------------- +# Interaktive Auswahl +# --------------------------------------------------------------------------- + + +def ask_for_correction_entries( + entries: list[FehlerlisteEntry], +) -> list[FehlerlisteEntry]: + """Fragt den Benutzer, welche Einträge korrigiert werden sollen. + + Unterstützt: + - Einzelne Nummer (z.B. "3") + - Mehrere Nummern durch Komma getrennt (z.B. "1,3,5") + - Bereich mit Bindestrich (z.B. "1-4") + - "a" oder "alle" für alle Einträge + - Leere Eingabe zum Abbrechen + """ + if not entries: + return [] + + print("\n📝 Auswahl:") + print(" Nummer(n) eingeben (z.B. '1', '1,3,5', '1-4', 'a' für alle, Enter für Abbruch):") + answer = input(" > ").strip() + + if not answer: + return [] + + if answer.lower() in ("a", "alle", "all"): + return list(entries) + + selected: list[FehlerlisteEntry] = [] + parts = [p.strip() for p in answer.split(",")] + + for part in parts: + if "-" in part: + try: + start_s, end_s = part.split("-", 1) + start = int(start_s.strip()) + end = int(end_s.strip()) + for i in range(start, end + 1): + idx = i - 1 + if 0 <= idx < len(entries): + if entries[idx] not in selected: + selected.append(entries[idx]) + except ValueError: + print(f"Ungültiger Bereich: {part}", file=sys.stderr) + else: + try: + idx = int(part) - 1 + if 0 <= idx < len(entries): + if entries[idx] not in selected: + selected.append(entries[idx]) + else: + print(f"Nummer {part} ist nicht in der Liste.", file=sys.stderr) + except ValueError: + print(f"Ungültige Nummer: {part}", file=sys.stderr) + + return selected + + +# --------------------------------------------------------------------------- +# Hauptprogramm +# --------------------------------------------------------------------------- + + +def parse_args() -> argparse.Namespace: + load_dotenv() + + parser = argparse.ArgumentParser( + description="eZICOM5 Zeitbuchungs-Korrektur-Tool – trägt fehlende Buchungen nach.", + ) + parser.add_argument("--base-url", default=DEFAULT_BASE_URL) + parser.add_argument("-u", "--username", default=os.getenv("EZICOM_USERNAME")) + parser.add_argument("-p", "--password", default=os.getenv("EZICOM_PASSWORD")) + parser.add_argument( + "--debug", + action="store_true", + help="Debug-Ausgaben aktivieren (zeigt URLs und Responses).", + ) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + + # Debug-Ordner vorbereiten + debug_dir = Path("Research Data") if args.debug else None + if debug_dir: + debug_dir.mkdir(parents=True, exist_ok=True) + + # Credentials ermitteln + username = args.username or input("eZICOM Benutzername: ") + password = args.password or getpass.getpass("eZICOM Passwort: ") + + # Login + print("🔐 Melde an...") + result = login( + args.base_url, + username, + password, + insecure=True, + debug=args.debug, + ) + if result is None: + return 1 + + opener, sid = result + print(f"✅ Login erfolgreich (Session: {mask_secret(sid)})") + + # Fehlerliste abrufen + print("\n📋 Rufe Fehlerliste ab...") + fl_status, fl_url, fl_html = fetch_fehlerliste(opener, args.base_url, sid) + if fl_status >= 400: + print(f"Fehlerliste konnte nicht abgerufen werden: HTTP {fl_status}", file=sys.stderr) + return 2 + + if debug_dir: + _save_debug_response(debug_dir, "fehlerliste", fl_html) + + entries = parse_fehlerliste(fl_html) + recent = print_fehlerliste(entries) + + if not recent: + print("\n✅ Keine korrigierbaren Einträge – alles in Ordnung!") + return 0 + + # Interaktive Auswahl + selected = ask_for_correction_entries(recent) + if not selected: + print("\n👋 Keine Einträge ausgewählt – beende.") + return 0 + + # Korrekturbuchungen durchführen + success_count = 0 + fail_count = 0 + + for entry in selected: + if submit_corrections_for_day( + opener, + args.base_url, + sid, + entry, + debug=args.debug, + debug_dir=debug_dir, + ): + success_count += 1 + else: + fail_count += 1 + + # Zusammenfassung + print(f"\n{'─' * 40}") + print(f"📊 Zusammenfassung:") + print(f" Erfolgreich: {success_count} Tage") + if fail_count: + print(f" Fehlgeschlagen: {fail_count} Tage") + print(f"{'─' * 40}") + + return 0 if fail_count == 0 else 2 + + +if __name__ == "__main__": + raise SystemExit(main())