From 6b7e344031e16803e27d3e36b8cadf2c3f95344b Mon Sep 17 00:00:00 2001 From: trnq1ll0 Date: Mon, 8 Jun 2026 09:31:00 +0200 Subject: [PATCH] default show Fehlerliste and insert neue Buchungen via no. selektion --- ezicom_login.py | 457 +++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 455 insertions(+), 2 deletions(-) diff --git a/ezicom_login.py b/ezicom_login.py index 31d64b8..7f4f633 100755 --- a/ezicom_login.py +++ b/ezicom_login.py @@ -8,9 +8,12 @@ interactive prompts. They are never written to disk by this script. 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 @@ -24,6 +27,19 @@ DEFAULT_USER_AGENT = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:151.0) " "Gecko/20100101 Firefox/151.0" ) +BOOKING_TYPES = { + "kommen": {"art": "1000", "status": "0", "tp": "0", "doppel": "0"}, + "pause": {"art": "0", "status": "1751", "tp": "0", "doppel": "0"}, + "gehen": {"art": "2000", "status": "0", "tp": "0", "doppel": "0"}, +} +MAX_CORRECTION_AGE_DAYS = 14 +CORRECTION_BOOKINGS = (("10:00", "1000"), ("14:00", "2000")) + + +def clean_html_text(value: str) -> str: + value = re.sub(r"<[^>]+>", " ", value) + value = html.unescape(value).replace("\xa0", " ") + return " ".join(value.split()) def load_dotenv(path: Path = Path(".env")) -> None: @@ -90,6 +106,336 @@ def request( return error.code, error.geturl(), raw.decode(charset, "replace") +def find_single_quoted_js_value(html: str, pattern: str) -> str | None: + match = re.search(pattern, html) + if not match: + return None + return match.group(1) + + +def follow_next_page( + opener: urllib.request.OpenerDirector, + base_url: str, + html: str, +) -> tuple[int, str, str] | None: + cust_info_url = find_single_quoted_js_value( + html, + r"url:\s*'([^']+/isCustInfoPnl)'", + ) + build_info_url = find_single_quoted_js_value( + html, + r"\$\.get\('([^']+/buildCustInfoPnl)'", + ) + frame_url = find_single_quoted_js_value( + html, + 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: str, url: str) -> str | None: + sid = find_single_quoted_js_value( + html, + 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 + + +def mask_secret(value: str) -> str: + if len(value) <= 8: + return "***" + return value[:4] + "..." + value[-4:] + + +def build_booking_payload(sid: str, booking_type: str) -> dict[str, str]: + now = dt.datetime.now() + payload = { + "page": "buch", + "sid": sid, + "kstbuch": "", + "datum": now.strftime("%d.%m.%Y"), + "zeit": now.strftime("%H:%M:%S"), + } + payload.update(BOOKING_TYPES[booking_type]) + return payload + + +def submit_booking( + opener: urllib.request.OpenerDirector, + base_url: str, + payload: dict[str, str], +) -> tuple[int, str, str]: + return request( + opener, + base_url.rstrip("/") + "/user", + method="POST", + data=payload, + headers={ + "Origin": base_url.rstrip("/"), + "Referer": base_url.rstrip("/") + f"/user/{payload['sid']}/rahmen.html", + }, + ) + + +def fetch_fehlerliste( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, +) -> tuple[int, str, str]: + 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)) + + +FehlerlisteEntry = tuple[dt.date, str, str] + + +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") + + +def parse_fehlerliste(html_text: str) -> list[FehlerlisteEntry]: + 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]]: + 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: + 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]: + if not entries: + print("Fehlerliste: keine fehlerhaften Tage gefunden.") + return [] + + print(f"Fehlerliste: {len(entries)} fehlerhafte Tage") + recent, too_old = split_fehlerliste_entries(entries) + if recent: + print("Buchbar (juenger als 15 Tage):") + for number, entry in enumerate(recent, start=1): + print(f" {number}) {format_fehlerliste_entry(entry)}") + else: + print("Buchbar (juenger als 15 Tage): keine Eintraege") + + if too_old: + print("--- Aelter als 14 Tage, keine Aenderung mehr moeglich ---") + for entry in too_old: + print(f" {format_fehlerliste_entry(entry)}") + + return recent + + +def ask_for_correction_entry(entries: list[FehlerlisteEntry]) -> FehlerlisteEntry | None: + if not entries: + return None + + answer = input("Nummer fuer automatische Buchung eingeben, sonst Enter: ").strip() + if not answer: + return None + if not answer.isdecimal(): + print("Keine gueltige Nummer eingegeben.", file=sys.stderr) + return None + + index = int(answer) - 1 + if index < 0 or index >= len(entries): + print("Nummer ist nicht in der Liste.", file=sys.stderr) + return None + + return entries[index] + + +def submit_correction_booking( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, + date_value: dt.date, + time_value: str, + art_value: str, +) -> tuple[int, str, str]: + params = { + "lnr": "0", + "typ": "mem", + "xplnr": "0", + "date": format_ezicom_date(date_value), + "time": time_value, + "art": art_value, + "awn": "", + "comm": "Automatisch nachgetragen", + } + url = ( + base_url.rstrip("/") + + f"/call/{urllib.parse.quote(sid)}/mem/js/corrbuchdo?" + + urllib.parse.urlencode(params) + ) + print(f" Debug request URL: {url}") + status, final_url, response_text = request(opener, url) + print(f" Debug response: HTTP {status} ({final_url})") + print(f" Debug response body: {response_text.strip() or ''}") + return status, final_url, response_text + + +def init_correction_form( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, +) -> tuple[int, str, str]: + params = { + "lnr": "0", + "bantr": "0", + "xplnr": "0", + "typ": "mem", + } + url = ( + base_url.rstrip("/") + + f"/call/{urllib.parse.quote(sid)}/mem/js/corrbuch?" + + urllib.parse.urlencode(params) + ) + print(f"Debug corrbuch request URL: {url}") + status, final_url, response_text = request(opener, url) + print(f"Debug corrbuch response: HTTP {status} ({final_url})") + preview = clean_html_text(response_text)[:500] + print(f"Debug corrbuch response body preview: {preview or ''}") + return status, final_url, response_text + + +def fetch_new_correction_page( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, +) -> tuple[int, str, str]: + params = { + "page": "corrnew", + "sid": sid, + "bt": "1", + } + url = base_url.rstrip("/") + "/user?" + urllib.parse.urlencode(params) + print(f"Debug corrnew request URL: {url}") + status, final_url, response_text = request(opener, url) + print(f"Debug corrnew response: HTTP {status} ({final_url})") + preview = clean_html_text(response_text)[:500] + print(f"Debug corrnew response body preview: {preview or ''}") + return status, final_url, response_text + + +def submit_corrections_for_day( + opener: urllib.request.OpenerDirector, + base_url: str, + sid: str, + entry: FehlerlisteEntry, +) -> bool: + date_value, _, _ = entry + print(f"Buchungen fuer {format_ezicom_date(date_value)} werden gesendet:") + init_status, init_url, _ = init_correction_form(opener, base_url, sid) + if init_status >= 400: + print( + f"Konnte Buchungsformular nicht initialisieren: HTTP {init_status} ({init_url})", + file=sys.stderr, + ) + return False + + corrnew_status, corrnew_url, _ = fetch_new_correction_page(opener, base_url, sid) + if corrnew_status >= 400: + print( + f"Neue Buchung konnte nicht geoeffnet werden: HTTP {corrnew_status} ({corrnew_url})", + file=sys.stderr, + ) + return False + + for time_value, art_value in CORRECTION_BOOKINGS: + status, url, _ = submit_correction_booking( + opener, + base_url, + sid, + date_value, + time_value, + art_value, + ) + if status >= 400: + print(f" {time_value}: fehlgeschlagen HTTP {status} ({url})", file=sys.stderr) + return False + print(f" {time_value}: gesendet") + return True + + def login( base_url: str, username: str, @@ -97,6 +443,10 @@ def login( *, insecure: bool, save_response: Path | None, + save_next_response: Path | None, + booking_type: str | None, + execute_booking: bool, + save_booking_response: Path | None, ) -> int: base_url = base_url.rstrip("/") opener = make_opener(insecure) @@ -148,6 +498,78 @@ def login( print(f"Login looks successful: HTTP {status} ({final_url})") if save_response: print(f"Saved response HTML to {save_response}") + + next_response = follow_next_page(opener, base_url, html) + if not next_response: + print("No follow-up page URL found in login response.", file=sys.stderr) + return 2 + + next_status, next_url, next_html = next_response + if next_status >= 400: + print(f"Follow-up page failed: HTTP {next_status} ({next_url})", file=sys.stderr) + return 2 + + print(f"Follow-up page loaded: HTTP {next_status} ({next_url})") + if save_next_response: + save_next_response.write_text(next_html, encoding="utf-8") + print(f"Saved follow-up HTML to {save_next_response}") + + sid = extract_sid(next_html, next_url) + if not sid: + print("No session id found in follow-up page.", file=sys.stderr) + return 2 + + fehlerliste_status, fehlerliste_url, fehlerliste_html = fetch_fehlerliste( + opener, + base_url, + sid, + ) + if fehlerliste_status >= 400: + print( + f"Fehlerliste failed: HTTP {fehlerliste_status} ({fehlerliste_url})", + file=sys.stderr, + ) + return 2 + recent_entries = print_fehlerliste(parse_fehlerliste(fehlerliste_html)) + correction_entry = ask_for_correction_entry(recent_entries) + if correction_entry and not submit_corrections_for_day( + opener, + base_url, + sid, + correction_entry, + ): + return 2 + + if booking_type: + payload = build_booking_payload(sid, booking_type) + printable_payload = dict(payload) + printable_payload["sid"] = mask_secret(printable_payload["sid"]) + + if not execute_booking: + print(f"Dry run: would submit {booking_type!r} booking with:") + for key, value in printable_payload.items(): + print(f" {key}={value}") + print("Add --execute-booking to actually send this booking.") + return 0 + + booking_status, booking_url, booking_html = submit_booking( + opener, + base_url, + payload, + ) + if save_booking_response: + save_booking_response.write_text(booking_html, encoding="utf-8") + print(f"Saved booking response HTML to {save_booking_response}") + + if booking_status >= 400: + print( + f"Booking failed: HTTP {booking_status} ({booking_url})", + file=sys.stderr, + ) + return 2 + + print(f"Booking submitted: {booking_type} HTTP {booking_status} ({booking_url})") + return 0 @@ -163,13 +585,40 @@ def parse_args() -> argparse.Namespace: parser.add_argument( "--insecure", action="store_true", - help="Disable TLS certificate verification if the server certificate cannot be verified.", + default=True, + help="Disable TLS certificate verification. This is the default for this eZICOM server.", + ) + parser.add_argument( + "--verify-tls", + action="store_false", + dest="insecure", + help="Enable TLS certificate verification.", ) parser.add_argument( "--save-response", type=Path, help="Optional path for saving the post-login HTML response for inspection.", ) + parser.add_argument( + "--save-next-response", + type=Path, + help="Optional path for saving the HTML after following the Weiter step.", + ) + parser.add_argument( + "--book", + choices=sorted(BOOKING_TYPES), + help="Prepare a time booking. Defaults to dry-run unless --execute-booking is set.", + ) + parser.add_argument( + "--execute-booking", + action="store_true", + help="Actually submit the booking selected with --book.", + ) + parser.add_argument( + "--save-booking-response", + type=Path, + help="Optional path for saving the HTML response after a submitted booking.", + ) return parser.parse_args() @@ -183,8 +632,12 @@ def main() -> int: password, insecure=args.insecure, save_response=args.save_response, + save_next_response=args.save_next_response, + booking_type=args.book, + execute_booking=args.execute_booking, + save_booking_response=args.save_booking_response, ) if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file