709 lines
22 KiB
Python
Executable File
709 lines
22 KiB
Python
Executable File
#!/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())
|