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