default show Fehlerliste and insert neue Buchungen via no. selektion
This commit is contained in:
trnq1ll0 committed
1 parent
2b40bd2756
commit
6b7e344031
1 file changed
+455
-2
+455
-2
@@ -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'<input[^>]+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'<div\s+id=["\']dfhl["\'][^>]*>(.*?)</div>',
|
||||
html_text,
|
||||
flags=re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
if not dfhl_match:
|
||||
return []
|
||||
|
||||
rows = re.findall(
|
||||
r"<tr\b[^>]*>(.*?)</tr>",
|
||||
dfhl_match.group(1),
|
||||
flags=re.IGNORECASE | re.DOTALL,
|
||||
)
|
||||
entries: list[FehlerlisteEntry] = []
|
||||
for row in rows:
|
||||
cells = re.findall(
|
||||
r"<td\b[^>]*>(.*?)</td>",
|
||||
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 '<leer>'}")
|
||||
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 '<leer>'}")
|
||||
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 '<leer>'}")
|
||||
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())
|
||||
Reference in new issue
Block a user