myapps/ibo
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
ibo/core.py
timu 650d7f3394 ibo: containerize + Gitea-CI + dev/prod-Deploy + Zitadel-SSO + noVNC-Live-View 
- Dockerfile/compose: Flask-App + selenium/standalone-chrome (noVNC), Healthcheck
- app.py: /healthz + eingebettete noVNC-Live-Ansicht (NOVNC_URL, lokal:7900 / Cluster:/novnc/)
- Helm-Chart infra/k8s/base/ibo: App + Selenium-Sidecar, 2 Ports, Ingress
  (/ + /novnc via stripPrefix), SSO-Middlewares aus dem CD-Repo
- Gitea-CI: version -> test -> build-push -> helm-validate -> deploy-dev (Write-back)
- .gitignore/.dockerignore (Dev-Samples raus)
- infra/RUNBOOK-ibo.md: Zitadel/DNS/Deploy/Prod-Promote

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-14 01:03:32 +02:00

388 lines
14 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
core.py - UI-unabhaengige Logik fuer den Immo-Bot.
Wird sowohl von der Flask-App (app.py) als auch vom Desktop-Script
(immo_bot.py) genutzt. Enthaelt:
- make_driver() Chrome lokal ODER remote (VPS via Selenium-Container)
- fill_contact_form() immowelt-Kontaktformular (React) ausfuellen
- submit_contact_form() Kontaktformular absenden
- extract_form_link() "Formular ausfuellen"-Link aus einer Mail ziehen
- watch_inbox_imap() Postfach per IMAP beobachten
- disposable_address() Wegwerf-Adresse (mail.tm) anlegen
- watch_disposable() Wegwerf-Postfach beobachten
- fill_application_form() Bewerbungsformular (Angular) ausfuellen
"""
import os
import re
import json
import time
import email
import imaplib
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
import autofill # Angular-Bewerbungsformular
import autofill_kontakt as ak # React-Kontaktformular
import mailtm # Wegwerf-E-Mail per API
import config # zentrale config.json
# ---------------------------------------------------------------------------
# Cookie-/DSGVO-Zustimmung dauerhaft merken (Consent-Snapshot)
# ---------------------------------------------------------------------------
#
# Ziel: die Cookie-/DSGVO-Zustimmung nur EINMAL anklicken (per Pre-App
# cookies_setup.py) und danach nie wieder.
#
# Wichtig fuer dieses Projekt: app.py startet die Browser PARALLEL (mehrere
# gleichzeitig). Ein persistenter Chrome-Profilordner (--user-data-dir) geht
# dabei NICHT, weil Chrome ein Profil auf genau eine laufende Instanz sperrt.
# Deshalb wird stattdessen ein Snapshot aus Cookies UND localStorage in
# consent.json gespeichert (moderne Consent-Banner legen die Zustimmung oft im
# localStorage ab, nicht nur in Cookies) und in jeden frischen Browser
# eingespielt. Das funktioniert lokal, parallel und remote (VPS-Container).
#
# Ablauf:
# 1) cookies_setup.py oeffnet die Seite(n), du klickst die Zustimmung an,
# save_consent() schreibt Cookies + localStorage pro Origin in consent.json.
# 2) Der Bot ruft nach driver.get(url) restore_consent(driver) auf -> Banner
# bleibt weg. click_cookie_banner() bleibt als Fallback bestehen.
from urllib.parse import urlparse
HERE = os.path.dirname(os.path.abspath(__file__))
CONSENT_PATH = os.path.join(HERE, "consent.json")
def _origin(url):
p = urlparse(url or "")
return f"{p.scheme}://{p.netloc}" if p.netloc else ""
def _read_consent():
if os.path.exists(CONSENT_PATH):
try:
with open(CONSENT_PATH, encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def save_consent(driver, log=print):
"""Cookies + localStorage der AKTUELL geoeffneten Seite in consent.json ablegen.
Pro Origin (Schema+Host) ein Eintrag; mehrfach aufrufbar fuer mehrere Seiten.
Gibt den gespeicherten Origin zurueck.
"""
origin = _origin(driver.current_url)
if not origin:
log(" !! Keine gueltige Seite offen - nichts gespeichert.")
return None
try:
ls = driver.execute_script(
"var o={};for(var i=0;i<localStorage.length;i++)"
"{var k=localStorage.key(i);o[k]=localStorage.getItem(k);}return o;")
except Exception:
ls = {}
data = _read_consent()
data[origin] = {"cookies": driver.get_cookies(), "localStorage": ls or {}}
with open(CONSENT_PATH, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
log(f" Consent gespeichert fuer {origin}: "
f"{len(data[origin]['cookies'])} Cookies, {len(ls or {})} localStorage-Eintraege.")
return origin
def restore_consent(driver, log=None):
"""Gespeicherten Consent (Cookies + localStorage) fuer die aktuelle Seite einspielen.
NACH driver.get(<url>) aufrufen. Fuehrt am Ende ein refresh() aus, damit die
Seite die Zustimmung uebernimmt. Gibt True zurueck, wenn etwas eingespielt wurde.
"""
data = _read_consent()
if not data:
return False
origin = _origin(driver.current_url)
entry = data.get(origin)
if entry is None: # Fallback: nach Host matchen
host = urlparse(driver.current_url or "").netloc
entry = next((v for k, v in data.items()
if urlparse(k).netloc == host), None)
if not entry:
return False
for c in entry.get("cookies", []):
c = dict(c)
c.pop("sameSite", None) # manche Werte loesen add_cookie-Fehler aus
try:
driver.add_cookie(c)
except Exception:
pass
ls = entry.get("localStorage", {})
if ls:
try:
driver.execute_script(
"var o=arguments[0];for(var k in o)localStorage.setItem(k,o[k]);", ls)
except Exception:
pass
try:
driver.refresh()
except Exception:
pass
if log:
log(f" Consent wiederhergestellt fuer {origin} "
f"({len(entry.get('cookies', []))} Cookies, {len(ls)} localStorage).")
return True
# ---------------------------------------------------------------------------
# WebDriver: lokal oder remote (VPS)
# ---------------------------------------------------------------------------
def make_driver(remote_url=None, headless=None):
"""
remote_url: z.B. "http://VPS-IP:4444" -> nutzt Selenium-Container auf VPS.
Falls None, wird die Env-Var SELENIUM_REMOTE_URL gelesen.
headless: True/False; Default: headless wenn remote oder Env HEADLESS=1.
"""
remote_url = remote_url or os.environ.get("SELENIUM_REMOTE_URL")
if headless is None:
headless = bool(remote_url) or os.environ.get("HEADLESS") == "1"
opts = webdriver.ChromeOptions()
if headless:
opts.add_argument("--headless=new")
opts.add_argument("--no-sandbox")
opts.add_argument("--disable-dev-shm-usage")
opts.add_argument("--window-size=1400,1000")
if remote_url:
return webdriver.Remote(command_executor=remote_url, options=opts)
return webdriver.Chrome(options=opts)
# ---------------------------------------------------------------------------
# Schritt 1: Kontaktformular ausfuellen (+ optional absenden)
# ---------------------------------------------------------------------------
def contact_data(email_override=None):
"""Kontakt-Felder aus config.json holen (Kopie -> thread-sicher)."""
c = dict(config.CFG["contact"])
if email_override:
c["email"] = email_override
return c
def fill_contact_form(driver, log, contact):
"""
contact: dict mit salutation, firstName, lastName, email, phone, message.
Daten kommen als Parameter (kein globaler Zustand) -> parallel-tauglich.
"""
ak.click_cookie_banner(driver)
if not ak.visible(driver, 'input[name="email"]'):
log("E-Mail-Feld nicht sichtbar -> versuche Kontakt-Button ...")
ak.open_contact_form(driver)
log("Warte auf Kontaktformular ...")
for _ in range(25):
if ak.visible(driver, 'input[name="email"]'):
break
time.sleep(1.0)
if not ak.visible(driver, 'input[name="email"]'):
log("!! Kein Kontaktformular gefunden.")
return False
log("Fuelle Kontaktformular ...")
ak.choose_salutation(driver, contact["salutation"])
ak.fill_by_name(driver, "firstName", contact["firstName"])
ak.fill_by_name(driver, "lastName", contact["lastName"])
ak.fill_by_name(driver, "email", contact["email"])
ak.fill_phone(driver, contact["phone"])
ak.fill_by_name(driver, "message", contact["message"])
log(f" -> E-Mail im Formular: {contact['email']}")
return True
def submit_contact_form(driver, log):
labels = ["Nachricht senden", "Anfrage senden", "Kontaktanfrage senden",
"Anfrage absenden", "Absenden", "Senden", "Abschicken"]
for lbl in labels:
try:
btns = driver.find_elements(
By.XPATH, f"//button[contains(normalize-space(.), '{lbl}')]")
for btn in btns:
if btn.is_displayed() and btn.is_enabled():
driver.execute_script(
"arguments[0].scrollIntoView({block:'center'});", btn)
btn.click()
log(f"Kontaktformular abgesendet (Button: '{lbl}').")
return True
except Exception:
pass
log("!! Kein Absenden-Button gefunden - bitte manuell pruefen.")
return False
# ---------------------------------------------------------------------------
# Schritt 2: Link aus Mail extrahieren
# ---------------------------------------------------------------------------
def extract_form_link(html):
"""Sucht in der Mail den Link hinter 'Formular ausfuellen'.
Validiert:
- Must contain "/property/application/form/" (URL-Struktur)
- Linktext sollte "formular" enthalten oder URL passt
- Exclude: withdraw, widerruf, abmeld, unsub, datenschutz, impressum
"""
anchors = re.findall(r'<a\b[^>]*href="([^"]+)"[^>]*>(.*?)</a>', html, re.I | re.S)
exclude = ("withdraw", "widerruf", "abmeld", "unsub", "datenschutz", "impressum")
candidates = []
for href, text in anchors:
# Nur Links mit der richtigen URL-Struktur
if "/property/application/form/" not in href.lower():
continue
# Ausschlusswörter prüfen
if any(x in href.lower() for x in exclude):
continue
candidates.append(href)
if candidates:
return candidates[0]
# Fallback: regulärer Ausdruck als Backup
m = re.search(r'https://[^"\s]*?/property/application/form/[^"\s]*', html, re.I)
return m.group(0) if m else None
# ---- IMAP-Variante (echtes Postfach) --------------------------------------
def _html_from_message(msg):
parts = msg.walk() if msg.is_multipart() else [msg]
for part in parts:
if part.get_content_type() == "text/html":
charset = part.get_content_charset() or "utf-8"
try:
return part.get_payload(decode=True).decode(charset, errors="replace")
except Exception:
return part.get_payload(decode=True).decode("utf-8", errors="replace")
return ""
def watch_inbox_imap(host, user, password, log,
sender_hint="", timeout=300, poll=15, should_cancel=None):
log(f"Verbinde mit {host} als {user} ...")
try:
M = imaplib.IMAP4_SSL(host)
M.login(user, password)
except Exception as e:
log(f"!! IMAP-Login fehlgeschlagen: {e}")
return None
deadline = time.time() + timeout
seen_ids = set()
try:
while time.time() < deadline:
if should_cancel and should_cancel():
log("Warten abgebrochen.")
return None
M.select("INBOX")
typ, data = M.search(None, "UNSEEN")
ids = data[0].split() if typ == "OK" and data[0] else []
if not ids:
typ, data = M.search(None, "ALL")
ids = (data[0].split() if typ == "OK" and data[0] else [])[-10:]
for mid in reversed(ids):
if mid in seen_ids:
continue
seen_ids.add(mid)
typ, msg_data = M.fetch(mid, "(RFC822)")
if typ != "OK":
continue
msg = email.message_from_bytes(msg_data[0][1])
frm = str(msg.get("From", ""))
if sender_hint and sender_hint.lower() not in frm.lower():
continue
link = extract_form_link(_html_from_message(msg))
if link:
log(f"Formular-Link in Mail von: {frm}")
return link
remaining = int(deadline - time.time())
log(f"Noch keine passende Mail. Warte {poll}s (Rest {remaining}s) ...")
slept = 0.0
while slept < poll:
if should_cancel and should_cancel():
log("Warten abgebrochen.")
return None
time.sleep(0.5)
slept += 0.5
finally:
try:
M.logout()
except Exception:
pass
log("!! Timeout - keine Mail mit Formular-Link gefunden.")
return None
# ---- Wegwerf-Variante (mail.tm, ohne Einrichtung) -------------------------
def disposable_address(log):
"""Legt eine Wegwerf-Adresse an. Gibt (address, token)."""
log("Erzeuge Wegwerf-Adresse (mail.tm) ...")
address, _pw, token = mailtm.create_account()
log(f"Wegwerf-Adresse: {address}")
return address, token
def watch_disposable(token, log, sender_hint="", timeout=300, poll=10, should_cancel=None):
return mailtm.wait_for_message(
token, extract_form_link, log,
sender_hint=sender_hint, timeout=timeout, poll=poll, should_cancel=should_cancel)
# ---------------------------------------------------------------------------
# Schritt 3: Bewerbungsformular (Angular) ausfuellen
# ---------------------------------------------------------------------------
def fill_application_form(driver, log):
wait = WebDriverWait(driver, 10)
time.sleep(1.5)
log("Aktiviere zweiten Mieter ...")
autofill.set_checkbox(driver, "formly_5_checkbox_secondTenant_5", True)
time.sleep(0.5)
log("Dropdowns ...")
for fid, opt in autofill.SELECTS.items():
autofill.select_option(driver, wait, fid, opt)
log("Textfelder ...")
for fid, val in autofill.TEXT_FIELDS.items():
autofill.fill_text(driver, wait, fid, val)
log("Zustimmungs-Checkboxen ...")
for cid in autofill.CHECKBOXES:
if cid == "formly_5_checkbox_secondTenant_5":
continue
autofill.set_checkbox(driver, cid, True)
log("Bewerbungsformular ausgefuellt (nicht abgesendet).")
return True
Reference in New Issue View Git Blame Copy Permalink