Add: Booking function to correct days with missing bookings
This commit is contained in:
trnq1ll0 committed
1 parent
bd47371f88
commit
b05fdfb071
1 file changed
+708
Executable
+708
@@ -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'<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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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'<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]]:
|
||||
"""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 '<leer>'}")
|
||||
|
||||
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 '<leer>'}")
|
||||
|
||||
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())
|
||||
Reference in new issue
Block a user