myapps/ibo
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

ibo: containerize + Gitea-CI + dev/prod-Deploy + Zitadel-SSO + noVNC-Live-View

Browse Source 
- 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>
This commit is contained in:
timu 2026-07-14 01:03:32 +02:00
commit 650d7f3394
39 changed files with 6651 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.claude/settings.json Normal file
View File

@ -0,0 +1,7 @@
{
"permissions": {
"allow": [
"Bash(python3 -m py_compile app.py)"
]
}
}

17
.dockerignore Normal file
View File

@ -0,0 +1,17 @@
# Build-Kontext schlank halten.
.git
.gitignore
__pycache__
*.pyc
*.pyo
.DS_Store
*.log
.venv
venv
# Dev-Samples (~10 MB, zur Laufzeit NICHT gebraucht — in keinem .py referenziert)
index_email.html
index_formular.html
index_inserat.html
# Infra-/Deploy-Artefakte gehoeren nicht ins App-Image
infra/
.gitea/

127
.gitea/workflows/momentrum-ci.yml Normal file
View File

@ -0,0 +1,127 @@
# Momentrum CI (Contabo-Gitea) — build · test · push · deploy(dev)
# Zielt auf das Gitea auf momentrum-k3s-01 (org: momentrum).
#
# Umgebungs-Modell fuer ibo (bewusst NUR dev+prod, keine PR-Previews):
# - dev = deployt AUTOMATISCH bei jedem main-Push (dieser Deploy-Step bumpt
# apps/internal/ibo/dev.yaml im CD-Repo -> ArgoCD synct dev).
# - prod = wird MANUELL promotet (Tag aus dev.yaml -> app.yaml committen,
# siehe infra/RUNBOOK-ibo.md). Die CI fasst prod NICHT an.
#
# Image-Tag ist immutable pro Commit:
# - Pull Request: pr-<nummer>-<shortsha>
# - Push auf main: pr-<id>-<sha> (aus Merge-Message) bzw. main-<shortsha>
name: Momentrum CI
on:
push:
branches: [main]
pull_request:
permissions:
contents: read
packages: write
env:
# :3000 = direkt zu Gitea (HTTP). Host-Docker hat diese Registry in insecure-registries.
REGISTRY: momentrum-k3s-01.tailcc1bcf.ts.net:3000
IMAGE: momentrum/ibo-web
CHART_PATH: infra/k8s/base/ibo
jobs:
version:
name: Version (aus PR-ID / SHA)
runs-on: ubuntu-latest
outputs:
tag: ${{ steps.v.outputs.tag }}
steps:
- id: v
env:
HEAD_MSG: ${{ github.event.head_commit.message }}
run: |
if [ "${{ github.event_name }}" = "pull_request" ]; then
short=$(echo "${{ github.event.pull_request.head.sha }}" | cut -c1-7)
tag="pr-${{ github.event.number }}-${short}"
else
short=$(echo "${GITHUB_SHA}" | cut -c1-7)
# Push auf main = PR-Merge: PR-ID aus der Merge-Commit-Message ziehen.
# `|| true`: kein #N in der Message darf unter set -e -o pipefail killen.
pr=$(printf '%s' "$HEAD_MSG" | grep -oE '#[0-9]+' | head -1 | tr -d '#' || true)
if [ -n "$pr" ]; then tag="pr-${pr}-${short}"; else tag="main-${short}"; fi
fi
echo "tag=$tag" >> "$GITHUB_OUTPUT"
echo "Image-Tag: $tag"
test:
name: Test (compile + pytest)
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install --no-cache-dir -r requirements.txt pytest
- name: Compile-Check (alle Module)
run: python -m compileall -q app.py core.py config.py immo_bot.py mailtm.py autofill.py autofill_kontakt.py cookies_setup.py
- name: Unit-Tests (falls vorhanden)
run: |
if ls test_*.py >/dev/null 2>&1; then pytest -q; else echo "keine Tests - skip"; fi
build-push:
name: Build & Push image
runs-on: ubuntu-latest
needs: [version, test]
steps:
- uses: actions/checkout@v4
with:
ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || '' }}
# Plain docker build/push -> Host-Docker-Daemon (gemounteter Socket) inkl.
# dessen insecure-registries. buildx wuerde die daemon-Config ignorieren.
- name: Login to Gitea Registry
run: echo "${{ secrets.REGISTRY_TOKEN }}" | docker login "${{ env.REGISTRY }}" -u "${{ secrets.REGISTRY_USER }}" --password-stdin
- name: Build & push
run: |
IMG="${{ env.REGISTRY }}/${{ env.IMAGE }}:${{ needs.version.outputs.tag }}"
ok=""
for attempt in 1 2 3; do
echo "== Build-Versuch $attempt =="
if DOCKER_BUILDKIT=1 docker build -t "$IMG" -f Dockerfile .; then ok=1; break; fi
echo "Build fehlgeschlagen (Versuch $attempt), neuer Versuch in 10s..."; sleep 10
done
[ "$ok" = "1" ] || { echo "Build nach 3 Versuchen fehlgeschlagen"; exit 1; }
docker push "$IMG"
helm-validate:
name: Helm lint + template
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: azure/setup-helm@v4
with:
version: v3.17.3
- run: |
helm lint "${{ env.CHART_PATH }}"
helm template ibo "${{ env.CHART_PATH }}" >/dev/null
# CD-Write-back: nur auf main. Bumpt den Image-Tag in apps/internal/ibo/dev.yaml
# im CD-Repo -> ArgoCD deployt DEV automatisch. Prod bleibt unberuehrt (Promote
# ist ein bewusster manueller Commit, siehe RUNBOOK). DEPLOY_TOKEN = Gitea-Token
# mit Write auf momentrum-deploy.
deploy-dev:
name: Deploy DEV (Image-Tag -> CD-Repo)
runs-on: ubuntu-latest
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
needs: [version, build-push, helm-validate]
steps:
- name: Write-back image.tag nach momentrum-deploy (dev.yaml)
run: |
TAG="${{ needs.version.outputs.tag }}"
git clone -q "http://ci:${{ secrets.DEPLOY_TOKEN }}@100.111.27.99:3000/momentrum/momentrum-deploy.git" /tmp/deploy
cd /tmp/deploy
APP_FILE="apps/internal/ibo/dev.yaml"
grep -qE '^ tag: ' "$APP_FILE" || { echo "FEHLER: 'tag:'-Zeile nicht in $APP_FILE gefunden"; exit 1; }
sed -i "s|^ tag: .*| tag: ${TAG}|" "$APP_FILE"
git config user.email "ci@momentrum.local"
git config user.name "Momentrum CI"
git commit -am "ibo: deploy dev ${TAG}" || { echo "kein Change - Tag identisch"; exit 0; }
git push origin HEAD:main
echo "Deployed dev tag ${TAG} -> momentrum-deploy"

12
.gitignore vendored Normal file
View File

@ -0,0 +1,12 @@
__pycache__/
*.pyc
*.pyo
.DS_Store
*.log
.env
.env.local
# Lokaler Lauf-State (wird zur Laufzeit geschrieben, nicht versionieren)
config_run.json
# venv / lokale Tool-Ordner
.venv/
venv/

23
Dockerfile Normal file
View File

@ -0,0 +1,23 @@
# Image fuer die Immo-Bot Web-App (Flask).
# Chrome selbst laeuft NICHT hier, sondern im selenium-Container / -Sidecar
# (SELENIUM_REMOTE_URL, siehe docker-compose.yml bzw. das Helm-Chart).
#
# Single-Process bewusst: die App haelt In-Memory-State + SSE-Streams und startet
# eigene Threads fuer parallel_workers -> KEIN gunicorn-Multi-Worker.
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 5000
# Container-/Compose-Healthcheck gegen den leichten /healthz-Endpoint.
HEALTHCHECK --interval=10s --timeout=3s --start-period=15s --retries=5 \
CMD python -c "import urllib.request,sys; sys.exit(0 if urllib.request.urlopen('http://localhost:5000/healthz',timeout=2).status==200 else 1)"
CMD ["python", "app.py"]

2124
app.py Normal file
View File

File diff suppressed because it is too large Load Diff

195
autofill.py Normal file
View File

@ -0,0 +1,195 @@
#!/usr/bin/env python3
"""
Autofill-Script fuer das IBO-Angular-Formular (ng-zorro / Formly).
Fuellt alle Inputfelder mit Mock-Daten. Weil es ein Angular-Formular ist,
reicht ein einfaches value-Setzen nicht -> es werden echte input/change-Events
via Selenium ausgeloest, damit Angular die Werte uebernimmt.
Setup:
pip install selenium
(Chrome + passender chromedriver muessen installiert sein;
bei aktuellem Selenium 4 wird der Driver automatisch verwaltet.)
Benutzung:
python autofill.py "http://localhost:4200/dein/formular"
# oder eine lokale Datei:
python autofill.py "file:///Users/timu/programming/momentrum/projekte/ibo/index.html"
"""
import sys
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# ---------------------------------------------------------------------------
# Felder werden aus config.json (application) gebaut. Siehe config.py.
# Die formly-IDs stammen aus dem gerenderten Formular (data-cy / id).
# ---------------------------------------------------------------------------
import config as _config
def _build_fields(app):
t1, t2 = app["tenant1"], app["tenant2"]
text_fields = {
# Vorgang
"formly_2_input_reserveNumber_0": app["reserveNumber"],
"formly_2_input_validUntil_1": app["validUntil"],
# Mieter 1
"formly_3_input_firstName_0": t1["firstName"],
"formly_3_input_lastName_1": t1["lastName"],
"formly_3_input_dateOfBirth_2": t1["dateOfBirth"],
"formly_4_input_street_0": t1["street"],
"formly_4_input_housenumber_1": t1["housenumber"],
"formly_4_input_zipcode_2": t1["zipcode"],
"formly_4_input_city_3": t1["city"],
"formly_5_input_phoneNumberPrivate_0": t1["phonePrivate"],
"formly_5_input_phoneNumberBusiness_1": t1["phoneBusiness"],
# Mieter 2
"formly_6_input_firstName2_0": t2["firstName"],
"formly_6_input_lastName2_1": t2["lastName"],
"formly_6_input_dateOfBirth2_2": t2["dateOfBirth"],
"formly_7_input_street2_0": t2["street"],
"formly_7_input_housenumber2_1": t2["housenumber"],
"formly_7_input_zipcode2_2": t2["zipcode"],
"formly_7_input_city2_3": t2["city"],
"formly_8_input_phoneNumberPrivate2_0": t2["phonePrivate"],
"formly_8_input_phoneNumberBusiness2_1": t2["phoneBusiness"],
}
checkboxes = []
if app.get("secondTenant", True):
checkboxes.append("formly_5_checkbox_secondTenant_5")
if app.get("consent", True):
checkboxes += ["formly_18_checkbox_dataCorrectness_20",
"formly_18_checkbox_dataPrivacy_21"]
selects = {"formly_2_select_numberRooms_2": str(app["numberRooms"])}
return text_fields, checkboxes, selects
TEXT_FIELDS, CHECKBOXES, SELECTS = _build_fields(_config.CFG["application"])
def reload_fields():
"""Uebernimmt geaenderte config.json-Werte."""
global TEXT_FIELDS, CHECKBOXES, SELECTS
TEXT_FIELDS, CHECKBOXES, SELECTS = _build_fields(
_config.reload()["application"])
# ---------------------------------------------------------------------------
# Fuell-Logik
# ---------------------------------------------------------------------------
def fill_text(driver, wait, field_id, value):
"""Fuellt ein Textfeld und loest Angular-Events aus."""
try:
el = wait.until(EC.presence_of_element_located((By.ID, field_id)))
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", el)
el.clear()
el.click()
el.send_keys(value)
# blur -> Angular validiert / uebernimmt
el.send_keys(Keys.TAB)
print(f" [ok] {field_id} = {value}")
except Exception as e:
print(f" [skip] {field_id}: {e.__class__.__name__}")
def set_checkbox(driver, field_id, checked=True):
try:
el = driver.find_element(By.ID, field_id)
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", el)
if el.is_selected() != checked:
driver.execute_script("arguments[0].click();", el)
print(f" [ok] checkbox {field_id} -> {checked}")
except Exception as e:
print(f" [skip] checkbox {field_id}: {e.__class__.__name__}")
def select_option(driver, wait, field_id, option_text):
"""Bedient ein @ng-select/ng-select Dropdown (nicht ng-zorro!).
Struktur: Input role=combobox oeffnet ein .ng-dropdown-panel mit
Optionen als <div class="ng-option">...</div>.
"""
try:
trigger = wait.until(EC.element_to_be_clickable((By.ID, field_id)))
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", trigger)
# Den umschliessenden .ng-select-Container klicken -> oeffnet Panel zuverlaessig
container = driver.execute_script(
"return arguments[0].closest('.ng-select');", trigger) or trigger
container.click()
time.sleep(0.4)
# Passende Option im Dropdown-Panel finden (erst exakt, dann enthaelt)
opt_xpath = (
"//div[contains(@class,'ng-option')]"
f"[normalize-space()='{option_text}']"
" | //div[contains(@class,'ng-option')]"
f"[contains(normalize-space(),'{option_text}')]"
)
option = wait.until(EC.element_to_be_clickable((By.XPATH, opt_xpath)))
option.click()
print(f" [ok] select {field_id} = {option_text}")
except Exception as e:
# Fallback: ng-select ist durchsuchbar -> tippen + Enter
try:
trigger.send_keys(option_text)
time.sleep(0.4)
trigger.send_keys(Keys.ENTER)
print(f" [ok] select {field_id} = {option_text} (getippt)")
except Exception:
print(f" [skip] select {field_id}: {e.__class__.__name__}")
def main():
url = sys.argv[1] if len(sys.argv) > 1 else \
"file:///Users/timu/programming/momentrum/projekte/ibo/index.html"
options = webdriver.ChromeOptions()
# options.add_argument("--headless=new") # bei Bedarf ohne sichtbares Fenster
driver = webdriver.Chrome(options=options)
wait = WebDriverWait(driver, 10)
try:
print(f"Oeffne {url}")
driver.get(url)
time.sleep(1.5) # Angular-Bootstrap abwarten
# Erst zweiten Mieter aktivieren (schaltet dessen Felder frei)
print("Checkbox 'zweiter Mieter':")
set_checkbox(driver, "formly_5_checkbox_secondTenant_5", True)
time.sleep(0.5)
print("Dropdowns:")
for fid, opt in SELECTS.items():
select_option(driver, wait, fid, opt)
print("Textfelder:")
for fid, val in TEXT_FIELDS.items():
fill_text(driver, wait, fid, val)
print("Zustimmungs-Checkboxen:")
for cid in CHECKBOXES:
if cid == "formly_5_checkbox_secondTenant_5":
continue # bereits gesetzt
set_checkbox(driver, cid, True)
print("\nFertig. Fenster bleibt offen. Programm laeuft weiter - "
"mit Strg+C beenden (oder Browser-Fenster schliessen).")
# Programm bewusst NICHT beenden -> Browser bleibt offen.
while True:
time.sleep(3600)
except KeyboardInterrupt:
print("\nBeende ...")
driver.quit()
if __name__ == "__main__":
main()

243
autofill_kontakt.py Normal file
View File

@ -0,0 +1,243 @@
#!/usr/bin/env python3
"""
Autofill-Script fuer das Kontaktformular (React / react-aria, z.B. ImmoScout24).
Fuellt Anrede, Vorname, Nachname, E-Mail, Telefon und Nachricht mit Mock-Daten.
Weil es ein React-Formular ist, werden echte Tastatur-Events via Selenium
gesendet (send_keys) - nur so uebernimmt React die Werte.
Die Feld-IDs sind dynamisch und aendern sich bei jedem Laden. Deshalb wird
ueber die stabilen name-Attribute selektiert. Felder kommen doppelt vor
(zwei gerenderte Varianten) -> es werden nur die SICHTBAREN gefuellt.
Setup:
pip install selenium
Benutzung:
python autofill_kontakt.py "https://www.immobilienscout24.de/..../kontakt"
# oder lokal (nur zum Ansehen, React laeuft dann nicht):
python autofill_kontakt.py
"""
import sys
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
# ---------------------------------------------------------------------------
# Werte kommen aus config.json (siehe config.py)
# ---------------------------------------------------------------------------
import config as _config
# Kopie, damit die App pro Lauf z.B. die E-Mail ueberschreiben kann,
# ohne die config.json zu veraendern.
MOCK = dict(_config.CFG["contact"])
def reload_mock():
"""Uebernimmt geaenderte config.json-Werte in MOCK."""
MOCK.clear()
MOCK.update(_config.reload()["contact"])
return MOCK
# ---------------------------------------------------------------------------
# Hilfsfunktionen (arbeiten nur auf sichtbaren Elementen)
# ---------------------------------------------------------------------------
def visible(driver, css):
"""Alle sichtbaren Elemente zu einem CSS-Selektor."""
els = driver.find_elements(By.CSS_SELECTOR, css)
return [e for e in els if e.is_displayed() and e.is_enabled()]
REACT_SET_JS = """
const el = arguments[0], val = arguments[1];
const proto = el.tagName === 'TEXTAREA'
? window.HTMLTextAreaElement.prototype
: window.HTMLInputElement.prototype;
const setter = Object.getOwnPropertyDescriptor(proto, 'value').set;
el.focus();
setter.call(el, ''); // erst leeren
el.dispatchEvent(new Event('input', {bubbles: true}));
setter.call(el, val); // dann Wert setzen
el.dispatchEvent(new Event('input', {bubbles: true})); // React onChange
el.dispatchEvent(new Event('change', {bubbles: true}));
el.dispatchEvent(new Event('blur', {bubbles: true})); // Validierung
"""
def type_into(driver, el, value, label=""):
"""React-sicher setzen: nativer value-Setter + input-Event (kein Tippen)."""
try:
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", el)
driver.execute_script(REACT_SET_JS, el, value)
print(f" [ok] {label or el.get_attribute('name')} = {value}")
except Exception as e:
print(f" [skip] {label}: {e.__class__.__name__}")
def fill_by_name(driver, name, value):
"""Fuellt NUR das erste sichtbare Feld dieses Namens (verhindert Dopplung)."""
els = visible(driver, f'input[name="{name}"], textarea[name="{name}"]')
if els:
type_into(driver, els[0], value, label=name)
else:
print(f" [skip] {name}: kein sichtbares Feld")
def choose_salutation(driver, value):
# react-aria Radios sind optisch versteckt (is_displayed==False) -> NICHT filtern.
els = driver.find_elements(
By.CSS_SELECTOR, f'input[name="salutation"][value="{value}"]')
for el in els:
try:
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", el)
# zugehoeriges Label klicken (react-aria reagiert darauf); sonst Input selbst
driver.execute_script("""
const el = arguments[0];
const lbl = el.closest('label')
|| (el.id && document.querySelector('label[for="'+el.id+'"]'));
(lbl || el).click();
el.checked = true;
el.dispatchEvent(new Event('click', {bubbles: true}));
el.dispatchEvent(new Event('change', {bubbles: true}));
""", el)
label = {"mr": "Herr", "mrs": "Frau"}.get(value, value)
print(f" [ok] Anrede = {label}")
return
except Exception as e:
print(f" [skip] Anrede: {e.__class__.__name__}")
print(" [skip] Anrede: kein Radio gefunden")
def fill_phone(driver, value):
"""Telefon: sichtbares tel-Feld (autocomplete='tel', ohne name)."""
for el in visible(driver, 'input[autocomplete="tel"]'):
type_into(driver, el, value, label="phone")
return
# Fallback: irgendein Feld mit inputmode text, das nicht schon belegt ist
print(" [info] kein dediziertes Telefon-Feld gefunden")
def click_cookie_banner(driver):
"""Cookie-Zustimmung wegklicken - auch in iframes (best effort)."""
texts = ["Alle akzeptieren", "Akzeptieren", "Alle zulassen",
"Zustimmen", "Einverstanden", "Accept all", "Accept"]
def try_click(ctx):
for txt in texts:
try:
btn = ctx.find_element(
By.XPATH, f"//button[contains(normalize-space(.), '{txt}')]")
if btn.is_displayed():
btn.click()
return True
except Exception:
pass
return False
if try_click(driver):
time.sleep(0.8)
return
# Consent liegt oft in einem iframe
for frame in driver.find_elements(By.TAG_NAME, "iframe"):
try:
driver.switch_to.frame(frame)
ok = try_click(driver)
except Exception:
ok = False
finally:
driver.switch_to.default_content()
if ok:
time.sleep(0.8)
return
def open_contact_form(driver):
"""Versucht, ueber einen Button das Kontaktformular zu oeffnen."""
labels = ["Nachricht", "Kontakt", "Anfrage", "Kontaktieren",
"Anbieter kontaktieren", "Nachricht senden", "Senden"]
for lbl in labels:
try:
btn = driver.find_element(
By.XPATH,
f"//button[contains(normalize-space(.), '{lbl}')] "
f"| //a[contains(normalize-space(.), '{lbl}')]")
if btn.is_displayed():
driver.execute_script("arguments[0].scrollIntoView({block:'center'});", btn)
btn.click()
print(f" Kontakt-Button geklickt: '{lbl}'")
time.sleep(2.0)
if visible(driver, 'input[name="email"]'):
return
except Exception:
pass
# ---------------------------------------------------------------------------
def main():
url = sys.argv[1] if len(sys.argv) > 1 else \
"file:///Users/timu/programming/momentrum/projekte/ibo/index2.html"
options = webdriver.ChromeOptions()
# options.add_argument("--headless=new")
driver = webdriver.Chrome(options=options)
try:
print(f"Oeffne {url}")
driver.get(url)
time.sleep(2.0) # React-Bootstrap abwarten
# Cookie-Banner wegklicken (best effort, auch in iframes)
click_cookie_banner(driver)
# Kontaktformular ggf. erst per Button oeffnen
if not visible(driver, 'input[name="email"]'):
print("E-Mail-Feld noch nicht sichtbar -> versuche Kontakt-Button ...")
open_contact_form(driver)
# Aktiv auf das E-Mail-Feld warten (bis 25s)
print("Warte auf Kontaktformular ...")
deadline = 25
while deadline > 0 and not visible(driver, 'input[name="email"]'):
time.sleep(1.0)
deadline -= 1
found = visible(driver, 'input[name="email"]')
print(f"Sichtbare E-Mail-Felder gefunden: {len(found)}")
if not found:
print("!! Kein Kontaktformular sichtbar. Ist es evtl. in einem iframe")
print(" oder hinter einem weiteren Klick? Fenster bleibt offen -")
print(" oeffne das Formular manuell und starte das Script erneut,")
print(" ODER sag mir, wie man das Formular auf der Seite oeffnet.")
print("Anrede:")
choose_salutation(driver, MOCK["salutation"])
print("Textfelder:")
fill_by_name(driver, "firstName", MOCK["firstName"])
fill_by_name(driver, "lastName", MOCK["lastName"])
fill_by_name(driver, "email", MOCK["email"])
print("Telefon:")
fill_phone(driver, MOCK["phone"])
print("Nachricht:")
fill_by_name(driver, "message", MOCK["message"])
print("\nFertig. Formular ausgefuellt (nicht abgesendet).")
print("Fenster bleibt offen. Mit Strg+C beenden.")
while True:
time.sleep(3600)
except KeyboardInterrupt:
print("\nBeende ...")
driver.quit()
if __name__ == "__main__":
main()

53
config.json Normal file
View File

@ -0,0 +1,53 @@
{
"_hinweis": "ALTE Gesamt-Config (wird NICHT mehr aktiv benutzt, nur Fallback beim allerersten Start). Die richtigen Configs liegen in config/prod/ bzw. config/mock/: step1.json = Kontaktformular, step2.json = Mail-Abruf, step3.json = Bewerbungsformular, run.json = App-Einstellungen.",
"application": {
"consent": true,
"numberRooms": "3",
"reserveNumber": "RES-12345",
"secondTenant": true,
"tenant1": {
"city": "Muenchen",
"dateOfBirth": "01.01.1990",
"firstName": "Max",
"housenumber": "12",
"lastName": "Mustermann",
"phoneBusiness": "0891234567",
"phonePrivate": "01761234567",
"street": "Hauptstrasse",
"zipcode": "80331"
},
"tenant2": {
"city": "Muenchen",
"dateOfBirth": "02.02.1992",
"firstName": "Erika",
"housenumber": "12",
"lastName": "Mustermann",
"phoneBusiness": "0899876543",
"phonePrivate": "01769876543",
"street": "Hauptstrasse",
"zipcode": "80331"
},
"validUntil": "31.12.2026"
},
"contact": {
"email": "",
"firstName": "Max",
"lastName": "Arno",
"message": "Hallo, ich interessiere mich fuer das Objekt und wuerde mich ueber einen Besichtigungstermin freuen. Bitte kontaktieren Sie mich unter den angegebenen Daten. Vielen Dank!",
"phone": "0172123123",
"salutation": "mr"
},
"email": {
"imap_host": "imap.gmail.com",
"imap_password": "",
"imap_user": "",
"mode": "mailtm",
"sender_filter": "muenchner-wohnen"
},
"run": {
"headless": false,
"parallel_workers": 3,
"auto_submit": false,
"keep_browser_open": true
}
}

278
config.py Normal file
View File

@ -0,0 +1,278 @@
#!/usr/bin/env python3
"""
config.py - laedt die Config und stellt die Werte bereit.
WICHTIG - was der Modus (test|prod) aendert:
NUR die FORMULARDATEN sind pro Modus verschieden:
contact -> Step 1 (Kontaktformular-Daten: Name, Telefon, Nachricht)
application -> Step 3 (Bewerbungsformular / Mieter-Selbstauskunft)
ALLES ANDERE ist GETEILT und in BEIDEN Modi ECHT (wird NICHT gemockt):
email -> Step 2 (mail.tm / IMAP) -> echte Adresse + echter Token
run -> Browser/Parallelitaet (headless, parallel_workers, ...)
Es wird also NICHTS an der Funktionalitaet gemockt. E-Mail und Token sind
IMMER echt. Der Modus bestimmt lediglich, welche Formular-Testdaten bzw.
echten Bewerbungsdaten eingetragen werden.
test -> Formulardaten = Dummy/Testdaten
prod -> Formulardaten = deine echten Bewerbungsdaten
Dateien:
config/email.json -> "email" (GETEILT)
config/run.json -> "run" (GETEILT)
config/<modus>/contact.json -> "contact" (pro Modus)
config/<modus>/application.json -> "application" (pro Modus)
Aktiver Modus: ENV IBO_MODE (test|prod) hat Vorrang, sonst config/active_mode.txt
(via UI umschaltbar), sonst 'prod'.
"""
import os
import json
HERE = os.path.dirname(os.path.abspath(__file__))
CONFIG_DIR = os.path.join(HERE, "config")
MODES = ("test", "prod")
DEFAULT_MODE = "prod"
MODE_STATE_PATH = os.path.join(CONFIG_DIR, "active_mode.txt")
# Geteilte Sektionen (fuer ALLE Modi gleich, echt): Datei im CONFIG_DIR.
SHARED_FILE = {
"email": "email.json", # Step 2 (Mail) - echt
"run": "run.json", # Browser/Parallelitaet
}
# Pro-Modus-Sektionen (nur Formulardaten): Datei im Modus-Ordner.
MODE_FILE = {
"contact": "contact.json", # Step 1
"application": "application.json", # Step 3
}
# Alte Ablagen: fuer Migration/Seed. (a) frueheres config/<mode>/stepN.json,
# (b) noch aeltere flache config_stepN.json, (c) alte gemeinsame config.json.
LEGACY_MODE_STEPFILE = { # frueheres Schema (stepN pro Modus)
"contact": "step1.json", "email": "step2.json",
"application": "step3.json", "run": "run.json",
}
LEGACY_FLAT = {
"contact": os.path.join(HERE, "config_step1.json"),
"email": os.path.join(HERE, "config_step2.json"),
"application": os.path.join(HERE, "config_step3.json"),
"run": os.path.join(HERE, "config_run.json"),
}
LEGACY_COMBINED = os.path.join(HERE, "config.json")
_HINWEIS = {
"contact": "Step 1 – Kontaktformular (PRO MODUS). 'email' wird pro URL "
"automatisch (mail.tm) erzeugt - leer lassen.",
"email": "Step 2 – Mail (GETEILT, in test UND prod identisch/echt). "
"mode: 'mailtm' (Wegwerf) oder 'imap'. sender_filter = erwarteter Absender.",
"application": "Step 3 – Bewerbungsformular (PRO MODUS). reserveNumber, validUntil, "
"numberRooms, secondTenant, consent, tenant1/tenant2 (Selbstauskunft).",
"run": "App-Einstellungen (GETEILT). headless, parallel_workers, auto_submit, "
"keep_browser_open.",
}
# ---------------------------------------------------------------------------
# Eingebaute Defaults
# ---------------------------------------------------------------------------
# GETEILT (echt) - identisch fuer test und prod.
SHARED_DEFAULTS = {
"email": {
"mode": "mailtm", "sender_filter": "muenchner-wohnen",
"imap_host": "imap.gmail.com", "imap_user": "", "imap_password": "",
},
"run": {"headless": False, "parallel_workers": 3,
"auto_submit": False, "keep_browser_open": True},
}
# PRO MODUS - nur Formulardaten (contact + application).
FORM_DEFAULTS = {
"prod": {
"contact": {
"salutation": "mr", "firstName": "Max", "lastName": "Mustermann",
"phone": "", "email": "",
"message": "Hallo, ich interessiere mich fuer das Objekt und wuerde mich "
"ueber einen Besichtigungstermin freuen. Bitte kontaktieren Sie "
"mich unter den angegebenen Daten. Vielen Dank!",
},
"application": {
"consent": True, "numberRooms": "3", "reserveNumber": "RES-12345",
"secondTenant": True, "validUntil": "31.12.2026",
"tenant1": {"firstName": "Max", "lastName": "Mustermann",
"dateOfBirth": "01.01.1990", "street": "Hauptstrasse",
"housenumber": "12", "zipcode": "80331", "city": "Muenchen",
"phonePrivate": "", "phoneBusiness": "0891234567"},
"tenant2": {"firstName": "Erika", "lastName": "Mustermann",
"dateOfBirth": "02.02.1992", "street": "Hauptstrasse",
"housenumber": "12", "zipcode": "80331", "city": "Muenchen",
"phonePrivate": "", "phoneBusiness": ""},
},
},
"test": {
"contact": {
"salutation": "mr", "firstName": "Test", "lastName": "Tester",
"phone": "0000000000", "email": "",
"message": "TEST – automatische Testnachricht.",
},
"application": {
"consent": True, "numberRooms": "2", "reserveNumber": "TEST-000",
"secondTenant": True, "validUntil": "31.12.2099",
"tenant1": {"firstName": "Test", "lastName": "Eins",
"dateOfBirth": "01.01.2000", "street": "Teststrasse",
"housenumber": "1", "zipcode": "10000", "city": "Teststadt",
"phonePrivate": "0000000001", "phoneBusiness": "0000000002"},
"tenant2": {"firstName": "Test", "lastName": "Zwei",
"dateOfBirth": "02.02.2000", "street": "Teststrasse",
"housenumber": "2", "zipcode": "10000", "city": "Teststadt",
"phonePrivate": "0000000003", "phoneBusiness": "0000000004"},
},
},
}
# ---------------------------------------------------------------------------
# Hilfsfunktionen
# ---------------------------------------------------------------------------
def _read_json(path):
try:
with open(path, encoding="utf-8") as f:
return json.load(f)
except Exception:
return {}
def _mode_dir(mode):
return os.path.join(CONFIG_DIR, mode)
def _valid_mode(mode):
mode = (mode or "").strip().lower()
if mode == "mock": # alter Name -> neuer Name
mode = "test"
return mode if mode in MODES else DEFAULT_MODE
def current_mode():
"""Aktiver Modus: ENV IBO_MODE hat Vorrang, sonst persistierte Datei, sonst Default."""
env = os.environ.get("IBO_MODE")
if env:
return _valid_mode(env)
if os.path.exists(MODE_STATE_PATH):
try:
with open(MODE_STATE_PATH, encoding="utf-8") as f:
return _valid_mode(f.read())
except Exception:
pass
return DEFAULT_MODE
def _write(path, section, value):
os.makedirs(os.path.dirname(path), exist_ok=True)
out = {"_hinweis": _HINWEIS.get(section, ""), section: value}
with open(path, "w", encoding="utf-8") as f:
json.dump(out, f, indent=2, ensure_ascii=False)
def _seed_from_chain(section, sources, default):
"""Erste Quelle nehmen, die die Sektion enthaelt; sonst default."""
for src in sources:
data = _read_json(src)
if section in data:
return data[section]
return default
def _seed_shared():
"""Geteilte Dateien (email, run) sicherstellen. Seed-Reihenfolge:
fruehere Modus-Datei (prod) -> flache Legacy -> alte config.json -> Default."""
os.makedirs(CONFIG_DIR, exist_ok=True)
for section, fname in SHARED_FILE.items():
path = os.path.join(CONFIG_DIR, fname)
if os.path.exists(path):
continue
sources = [
os.path.join(_mode_dir("prod"), LEGACY_MODE_STEPFILE[section]),
LEGACY_FLAT[section],
LEGACY_COMBINED,
]
_write(path, section, _seed_from_chain(section, sources, SHARED_DEFAULTS[section]))
def _seed_mode(mode):
"""Pro-Modus-Dateien (contact, application) sicherstellen."""
os.makedirs(_mode_dir(mode), exist_ok=True)
for section, fname in MODE_FILE.items():
path = os.path.join(_mode_dir(mode), fname)
if os.path.exists(path):
continue
sources = [os.path.join(_mode_dir(mode), LEGACY_MODE_STEPFILE[section])]
if mode == "prod": # prod darf auch aus den flachen Legacy-Dateien seeden
sources += [LEGACY_FLAT[section], LEGACY_COMBINED]
default = FORM_DEFAULTS[mode][section]
_write(path, section, _seed_from_chain(section, sources, default))
# ---------------------------------------------------------------------------
# Laden / Speichern
# ---------------------------------------------------------------------------
MODE = current_mode()
def load():
"""Geteilte (email, run) + Pro-Modus (contact, application) zu einer CFG."""
_seed_shared()
_seed_mode(MODE)
cfg = {}
for section, fname in SHARED_FILE.items():
data = _read_json(os.path.join(CONFIG_DIR, fname))
cfg[section] = data.get(section, SHARED_DEFAULTS[section])
for section, fname in MODE_FILE.items():
data = _read_json(os.path.join(_mode_dir(MODE), fname))
cfg[section] = data.get(section, FORM_DEFAULTS[MODE][section])
return cfg
def save(cfg):
"""email/run -> geteilte Dateien; contact/application -> Modus-Dateien."""
for section, fname in SHARED_FILE.items():
if section in cfg:
_write(os.path.join(CONFIG_DIR, fname), section, cfg[section])
for section, fname in MODE_FILE.items():
if section in cfg:
_write(os.path.join(_mode_dir(MODE), fname), section, cfg[section])
CFG = load()
def reload():
global MODE, CFG
MODE = current_mode()
CFG = load()
return CFG
def set_mode(mode):
"""Aktiven Modus umschalten (persistiert) und Config neu laden.
ENV IBO_MODE hat weiterhin Vorrang."""
global MODE, CFG
mode = _valid_mode(mode)
os.makedirs(CONFIG_DIR, exist_ok=True)
with open(MODE_STATE_PATH, "w", encoding="utf-8") as f:
f.write(mode)
MODE = current_mode()
CFG = load()
return MODE
def mode_locked_by_env():
"""True, wenn der Modus per ENV festgelegt ist (UI-Umschalten dann gesperrt)."""
return bool(os.environ.get("IBO_MODE"))

10
config/email.json Normal file
View File

@ -0,0 +1,10 @@
{
"_hinweis": "Step 2 – Mail (GETEILT, in test UND prod identisch/echt). mode: 'mailtm' (Wegwerf) oder 'imap'. sender_filter = erwarteter Absender.",
"email": {
"imap_host": "imap.gmail.com",
"imap_password": "",
"imap_user": "",
"mode": "mailtm",
"sender_filter": "muenchner-wohnen"
}
}

32
config/prod/application.json Normal file
View File

@ -0,0 +1,32 @@
{
"_hinweis": "Step 3 – Bewerbungsformular (PRO MODUS). reserveNumber, validUntil, numberRooms, secondTenant, consent, tenant1/tenant2 (Selbstauskunft).",
"application": {
"consent": true,
"numberRooms": "3",
"reserveNumber": "RES-12345",
"secondTenant": true,
"tenant1": {
"city": "Muenchen",
"dateOfBirth": "01.01.1990",
"firstName": "Max",
"housenumber": "12",
"lastName": "Vortf",
"phoneBusiness": "0891234567",
"phonePrivate": "",
"street": "Hauptstrasse",
"zipcode": "80331"
},
"tenant2": {
"city": "Muenchen",
"dateOfBirth": "02.02.1992",
"firstName": "Erika",
"housenumber": "12",
"lastName": "Vortf",
"phoneBusiness": "",
"phonePrivate": "",
"street": "Hauptstrasse",
"zipcode": "80331"
},
"validUntil": "31.12.2026"
}
}

11
config/prod/contact.json Normal file
View File

@ -0,0 +1,11 @@
{
"_hinweis": "Step 1 – Kontaktformular (PRO MODUS). 'email' wird pro URL automatisch (mail.tm) erzeugt - leer lassen.",
"contact": {
"email": "",
"firstName": "Felix",
"lastName": "Vortf",
"message": "Hallo, ich interessiere mich fuer das Objekt und wuerde mich ueber einen Besichtigungstermin freuen. Bitte kontaktieren Sie mich unter den angegebenen Daten. Vielen Dank!",
"phone": "",
"salutation": "mr"
}
}

9
config/run.json Normal file
View File

@ -0,0 +1,9 @@
{
"_hinweis": "App-Einstellungen (GETEILT). headless, parallel_workers, auto_submit, keep_browser_open.",
"run": {
"headless": false,
"parallel_workers": 3,
"auto_submit": false,
"keep_browser_open": true
}
}

32
config/test/application.json Normal file
View File

@ -0,0 +1,32 @@
{
"_hinweis": "Step 3 – Bewerbungsformular (PRO MODUS). reserveNumber, validUntil, numberRooms, secondTenant, consent, tenant1/tenant2 (Selbstauskunft).",
"application": {
"consent": true,
"numberRooms": "2",
"reserveNumber": "TEST-000",
"secondTenant": true,
"validUntil": "31.12.2099",
"tenant1": {
"firstName": "Test",
"lastName": "Eins",
"dateOfBirth": "01.01.2000",
"street": "Teststrasse",
"housenumber": "1",
"zipcode": "10000",
"city": "Teststadt",
"phonePrivate": "0000000001",
"phoneBusiness": "0000000002"
},
"tenant2": {
"firstName": "Test",
"lastName": "Zwei",
"dateOfBirth": "02.02.2000",
"street": "Teststrasse",
"housenumber": "2",
"zipcode": "10000",
"city": "Teststadt",
"phonePrivate": "0000000003",
"phoneBusiness": "0000000004"
}
}
}

11
config/test/contact.json Normal file
View File

@ -0,0 +1,11 @@
{
"_hinweis": "Step 1 – Kontaktformular (PRO MODUS). 'email' wird pro URL automatisch (mail.tm) erzeugt - leer lassen.",
"contact": {
"salutation": "mr",
"firstName": "Test",
"lastName": "Tester",
"phone": "0000000000",
"email": "",
"message": "TEST – automatische Testnachricht."
}
}

11
config_step1.json Normal file
View File

@ -0,0 +1,11 @@
{
"_hinweis": "Step 1 – Kontaktformular. Default-Werte fuer das Ausfuellen des Kontaktformulars. Hier direkt bearbeiten ODER in der Web-UI unter 'Step 1 -> Config' aendern (wird hierher gespeichert). 'email' wird pro URL automatisch (mail.tm) erzeugt und ueberschrieben - leer lassen.",
"contact": {
"email": "",
"firstName": "Felix",
"lastName": "Vortf",
"message": "Hallo, ich interessiere mich fuer das Objekt und wuerde mich ueber einen Besichtigungstermin freuen. Bitte kontaktieren Sie mich unter den angegebenen Daten. Vielen Dank!",
"phone": "",
"salutation": "mr"
}
}

10
config_step2.json Normal file
View File

@ -0,0 +1,10 @@
{
"_hinweis": "Step 2 – Mail. Einstellungen zum Empfangen der Antwort-Mail. mode: 'mailtm' (Wegwerf-Adresse, keine Einrichtung) oder 'imap' (eigenes Postfach). sender_filter grenzt die Suche auf den erwarteten Absender ein (Teilstring, z.B. 'muenchner-wohnen').",
"email": {
"imap_host": "imap.gmail.com",
"imap_password": "",
"imap_user": "",
"mode": "mailtm",
"sender_filter": "muenchner-wohnen"
}
}

32
config_step3.json Normal file
View File

@ -0,0 +1,32 @@
{
"_hinweis": "Step 3 – NUR das Bewerbungsformular. reserveNumber = Vorgangsnummer (aus der E-Mail), validUntil = Gueltig bis (TT.MM.JJJJ), numberRooms = Anzahl Wohnraeume, secondTenant = Mitmieter/in vorhanden (dann gilt tenant2), consent = Haekchen 'Richtigkeit der Daten', tenant1/tenant2 = Selbstauskunft Mieter 1/2.",
"application": {
"consent": true,
"numberRooms": "3",
"reserveNumber": "RES-12345",
"secondTenant": true,
"tenant1": {
"city": "Muenchen",
"dateOfBirth": "01.01.1990",
"firstName": "Max",
"housenumber": "12",
"lastName": "Vortf",
"phoneBusiness": "0891234567",
"phonePrivate": "",
"street": "Hauptstrasse",
"zipcode": "80331"
},
"tenant2": {
"city": "Muenchen",
"dateOfBirth": "02.02.1992",
"firstName": "Erika",
"housenumber": "12",
"lastName": "Vortf",
"phoneBusiness": "",
"phonePrivate": "",
"street": "Hauptstrasse",
"zipcode": "80331"
},
"validUntil": "31.12.2026"
}
}

293
consent.json Normal file
View File

File diff suppressed because one or more lines are too long

59
cookies_setup.py Normal file
View File

@ -0,0 +1,59 @@
#!/usr/bin/env python3
"""
cookies_setup.py - Einmal-Vorlauf ("Pre-App") zum Speichern der DSGVO-/Cookie-
Zustimmung, damit der Bot sie nicht bei jedem Lauf neu
wegklicken muss.
Was passiert:
Es oeffnet sich ein sichtbares Chrome-Fenster. Du klickst die Cookie-/DSGVO-
Zustimmung an und drueckst dann hier im Terminal ENTER. Cookies UND
localStorage der Seite werden pro Origin in consent.json gespeichert.
Der Bot (app.py / immo_bot.py) spielt diesen Snapshot beim naechsten Start
in jeden - auch parallel geoeffneten - Browser ein. Damit bleibt das
Cookie-Banner weg. Funktioniert lokal und mit dem Remote-VPS-Container.
Benutzung:
python cookies_setup.py
python cookies_setup.py "https://www.immowelt.de/"
python cookies_setup.py "https://www.immowelt.de/" "https://www.immobilienscout24.de/"
"""
import sys
import core
# Seiten, deren Consent standardmaessig einmalig bestaetigt wird.
DEFAULT_URLS = [
"https://www.immowelt.de/",
"https://www.immobilienscout24.de/",
]
def main():
urls = sys.argv[1:] or DEFAULT_URLS
print(f"Consent-Datei: {core.CONSENT_PATH}\n")
# Sichtbares, lokales Chrome (headless=False), damit du die Banner
# sehen und anklicken kannst.
driver = core.make_driver(headless=False)
try:
for url in urls:
print(f"Oeffne {url}")
driver.get(url)
input(
" -> Im Browser die Cookie-/DSGVO-Zustimmung anklicken.\n"
" Fertig? Hier ENTER druecken zum Speichern ... "
)
core.save_consent(driver)
print()
print("Fertig. Der Bot startet ab jetzt ohne erneutes Cookie-Banner.")
print("(Zum Zuruecksetzen einfach consent.json loeschen und neu starten.)")
finally:
driver.quit()
if __name__ == "__main__":
main()

387
core.py Normal file
View File

@ -0,0 +1,387 @@
#!/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

21
create_test_address.py Normal file
View File

@ -0,0 +1,21 @@
#!/usr/bin/env python3
"""
Erstelle eine neue Wegwerf-Adresse + Token zum Testen.
"""
from mailtm import create_account
print("🆕 Erstelle neue Wegwerf-Adresse...")
address, password, token = create_account()
print()
print("=" * 60)
print(f"📧 Adresse: {address}")
print(f"🔑 Token: {token}")
print("=" * 60)
print()
print("Nächste Schritte:")
print(f" 1. Sende eine Mail an: {address}")
print(f" 2. Dann aufrufen:")
print(f" python test_extract_link.py {address} {token}")
print()

41
docker-compose.yml Normal file
View File

@ -0,0 +1,41 @@
# Immo-Bot lokal / auf einem VPS:
#
# docker compose up -d
#
# Danach:
# - Web-UI: http://localhost:5000
# - Live-Ansicht des Automations-Browsers: direkt in der Web-App (Button
# "👁️ Browser live ansehen" / oeffnet automatisch beim Consent-Browser).
# Die App bettet dazu noVNC (Port 7900) per iframe ein — NOVNC_URL unten.
# Alternativ direkt: http://localhost:7900 (Passwort: secret)
#
# Der Selenium-Container bringt Chrome + noVNC mit. Die App verbindet sich
# ueber SELENIUM_REMOTE_URL mit ihm (kein Chrome-Setup in der App noetig).
services:
selenium:
image: selenium/standalone-chrome:latest
shm_size: "2g"
environment:
# Parallelitaet: mehrere Browser-Sessions gleichzeitig zulassen.
# Sollte >= config.json run.parallel_workers sein.
SE_NODE_MAX_SESSIONS: "4"
SE_NODE_OVERRIDE_MAX_SESSIONS: "true"
ports:
- "4444:4444" # WebDriver (intern von der App genutzt)
- "7900:7900" # noVNC - Live-Ansicht (eingebettet in die Web-App)
restart: unless-stopped
app:
build: .
depends_on:
- selenium
environment:
# App spricht den Selenium-Container ueber den Service-Namen an
SELENIUM_REMOTE_URL: "http://selenium:4444"
# Basis-URL fuer das eingebettete noVNC-Live-View. Lokal zeigt der Browser
# auf den gemappten Host-Port 7900. (Im Cluster: /novnc/ ueber den Ingress.)
NOVNC_URL: "http://localhost:7900"
ports:
- "5000:5000" # Web-UI
restart: unless-stopped

411
immo_bot.py Normal file
View File

@ -0,0 +1,411 @@
#!/usr/bin/env python3
"""
Immo-Bot - simple UI, die den ganzen Ablauf automatisiert:
1. Kontaktformular auf immowelt (URL) mit Mock-Daten ausfuellen
und - falls gewuenscht - automatisch absenden.
2. Posteingang (IMAP) beobachten, die Bestaetigungs-Mail finden
und den "Formular ausfuellen"-Link extrahieren.
3. Diesen Link oeffnen und das Bewerbungsformular (Angular) ausfuellen.
Es werden die bestehenden Scripte wiederverwendet:
- autofill_kontakt.py -> Kontaktformular (React)
- autofill.py -> Bewerbungsformular (Angular)
Setup:
pip install selenium
(Chrome + chromedriver; Selenium 4 verwaltet den Driver selbst.)
Start:
python immo_bot.py
Hinweis E-Mail:
Fuer den Postfach-Zugriff wird IMAP mit App-Passwort genutzt.
- Gmail: Host imap.gmail.com (App-Passwort in Google-Konto erzeugen)
- Outlook: Host outlook.office365.com (persoenliche Outlook-Konten
erlauben teils nur OAuth statt Passwort - dann Gmail nehmen oder
die Mail an ein Gmail weiterleiten.)
"""
import os
import re
import time
import email
import imaplib
import threading
import tkinter as tk
from tkinter import ttk, scrolledtext
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
# Bestehende Scripte wiederverwenden
import autofill # Angular-Bewerbungsformular
import autofill_kontakt as ak # React-Kontaktformular
# ---------------------------------------------------------------------------
# Schritt 1: Kontaktformular ausfuellen (+ optional absenden)
# ---------------------------------------------------------------------------
def fill_contact_form(driver, log, applicant_email=None):
"""Fuellt das immowelt-Kontaktformular. Nutzt Funktionen aus autofill_kontakt."""
if applicant_email:
ak.MOCK["email"] = applicant_email # Mail = das Postfach, das wir beobachten
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, ak.MOCK["salutation"])
ak.fill_by_name(driver, "firstName", ak.MOCK["firstName"])
ak.fill_by_name(driver, "lastName", ak.MOCK["lastName"])
ak.fill_by_name(driver, "email", ak.MOCK["email"])
ak.fill_phone(driver, ak.MOCK["phone"])
ak.fill_by_name(driver, "message", ak.MOCK["message"])
log(f" -> E-Mail im Formular: {ak.MOCK['email']}")
return True
def submit_contact_form(driver, log):
"""Klickt den Absenden-Button des Kontaktformulars."""
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: E-Mail beobachten und Formular-Link extrahieren
# ---------------------------------------------------------------------------
def _html_from_message(msg):
"""Holt den text/html-Teil einer E-Mail als String."""
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 extract_form_link(html):
"""Sucht in der Mail den Link hinter 'Formular ausfuellen'."""
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:
clean = re.sub(r"<[^>]+>", "", text)
clean = re.sub(r"&nbsp;?", " ", clean).strip().lower()
if "formular" in clean and not any(x in href.lower() for x in exclude):
candidates.append(href)
# bevorzugt den echten Formular-Link
for c in candidates:
if "application/form" in c.lower():
return c
if candidates:
return candidates[0]
# Fallback: irgendein application/form-Link im HTML
m = re.search(r'https://[^"\s]*application/form[^"\s]*', html)
return m.group(0) if m else None
def watch_inbox_for_link(host, user, password, log,
sender_hint="", timeout=300, poll=15):
"""Pollt den Posteingang, bis eine passende Mail mit Formular-Link kommt."""
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:
M.select("INBOX")
# ungelesene Mails zuerst; sonst die neuesten
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
html = _html_from_message(msg)
link = extract_form_link(html)
if link:
log(f"Formular-Link gefunden in Mail von: {frm}")
return link
remaining = int(deadline - time.time())
log(f"Noch keine passende Mail. Warte {poll}s (Rest {remaining}s) ...")
time.sleep(poll)
finally:
try:
M.logout()
except Exception:
pass
log("!! Timeout - keine Mail mit Formular-Link gefunden.")
return None
# ---------------------------------------------------------------------------
# Schritt 3: Bewerbungsformular (Angular) ausfuellen
# ---------------------------------------------------------------------------
def fill_application_form(driver, log):
"""Fuellt das Angular-Formular ueber die Funktionen aus autofill.py."""
wait = WebDriverWait(driver, 10)
time.sleep(1.5) # Angular-Bootstrap
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
# ---------------------------------------------------------------------------
# UI
# ---------------------------------------------------------------------------
class ImmoBotUI:
def __init__(self, root):
self.root = root
self.driver = None
self.busy = False
self.form_link = None
root.title("Immo-Bot")
root.geometry("720x640")
pad = {"padx": 8, "pady": 4}
frm = ttk.Frame(root, padding=12)
frm.pack(fill="both", expand=True)
# --- Eingaben ---
ttk.Label(frm, text="immowelt Expose-/Kontakt-URL:").grid(
row=0, column=0, sticky="w", **pad)
self.url_var = tk.StringVar()
ttk.Entry(frm, textvariable=self.url_var, width=70).grid(
row=1, column=0, columnspan=3, sticky="we", **pad)
ttk.Label(frm, text="Deine E-Mail (Formular + Postfach):").grid(
row=2, column=0, sticky="w", **pad)
self.email_var = tk.StringVar(value="abutepeli@outlook.de")
ttk.Entry(frm, textvariable=self.email_var, width=40).grid(
row=3, column=0, sticky="we", **pad)
ttk.Label(frm, text="IMAP-Host:").grid(row=2, column=1, sticky="w", **pad)
self.host_var = tk.StringVar(value="outlook.office365.com")
ttk.Entry(frm, textvariable=self.host_var, width=28).grid(
row=3, column=1, sticky="we", **pad)
ttk.Label(frm, text="E-Mail App-Passwort:").grid(
row=4, column=0, sticky="w", **pad)
self.pw_var = tk.StringVar()
ttk.Entry(frm, textvariable=self.pw_var, width=40, show="*").grid(
row=5, column=0, sticky="we", **pad)
ttk.Label(frm, text="Absender-Filter (optional):").grid(
row=4, column=1, sticky="w", **pad)
self.sender_var = tk.StringVar(value="muenchner-wohnen")
ttk.Entry(frm, textvariable=self.sender_var, width=28).grid(
row=5, column=1, sticky="we", **pad)
self.submit_var = tk.BooleanVar(value=False)
ttk.Checkbutton(
frm, text="Kontaktformular automatisch ABSENDEN",
variable=self.submit_var).grid(row=6, column=0, columnspan=2,
sticky="w", **pad)
# --- Buttons ---
btns = ttk.Frame(frm)
btns.grid(row=7, column=0, columnspan=3, sticky="we", pady=8)
self.b_all = ttk.Button(btns, text="ALLES automatisch (1 -> 2 -> 3)",
command=self.run_all)
self.b_all.pack(side="left", padx=4)
self.b1 = ttk.Button(btns, text="1. Kontakt ausfuellen",
command=self.run_step1)
self.b1.pack(side="left", padx=4)
self.b2 = ttk.Button(btns, text="2. Mail -> Link",
command=self.run_step2)
self.b2.pack(side="left", padx=4)
self.b3 = ttk.Button(btns, text="3. Bewerbung ausfuellen",
command=self.run_step3)
self.b3.pack(side="left", padx=4)
# --- Log ---
self.log_box = scrolledtext.ScrolledText(frm, height=18, wrap="word")
self.log_box.grid(row=8, column=0, columnspan=3, sticky="nsew", **pad)
frm.rowconfigure(8, weight=1)
frm.columnconfigure(0, weight=1)
self.log("Bereit. URL eingeben und einen Schritt starten.")
# ---- Helfer ----
def log(self, msg):
self.root.after(0, self._append, str(msg))
def _append(self, msg):
self.log_box.insert("end", msg + "\n")
self.log_box.see("end")
def _run_bg(self, target):
if self.busy:
self.log("... laeuft bereits, bitte warten.")
return
self.busy = True
def wrap():
try:
target()
except Exception as e:
self.log(f"!! Fehler: {e}")
finally:
self.busy = False
threading.Thread(target=wrap, daemon=True).start()
def ensure_driver(self):
if self.driver is None:
remote = os.environ.get("SELENIUM_REMOTE_URL")
opts = webdriver.ChromeOptions()
if remote:
# z.B. SELENIUM_REMOTE_URL=http://VPS-IP:4444
# Rendering live sichtbar ueber noVNC: http://VPS-IP:7900 (PW: secret)
self.log(f"Verbinde mit Remote-Selenium: {remote}")
self.driver = webdriver.Remote(command_executor=remote, options=opts)
else:
self.log("Starte lokales Chrome ...")
self.driver = webdriver.Chrome(options=opts)
return self.driver
# ---- Aktionen ----
def run_step1(self):
self._run_bg(self._step1)
def _step1(self):
url = self.url_var.get().strip()
if not url:
self.log("!! Bitte URL eingeben.")
return
d = self.ensure_driver()
self.log(f"Oeffne {url}")
d.get(url)
time.sleep(2.0)
if fill_contact_form(d, self.log, self.email_var.get().strip()):
if self.submit_var.get():
submit_contact_form(d, self.log)
else:
self.log("Nicht abgesendet (Haken 'automatisch ABSENDEN' nicht gesetzt).")
def run_step2(self):
self._run_bg(self._step2)
def _step2(self):
link = watch_inbox_for_link(
self.host_var.get().strip(),
self.email_var.get().strip(),
self.pw_var.get(),
self.log,
sender_hint=self.sender_var.get().strip(),
)
if link:
self.form_link = link
self.log(f"LINK: {link}")
def run_step3(self):
self._run_bg(self._step3)
def _step3(self):
if not self.form_link:
self.log("!! Kein Formular-Link vorhanden (erst Schritt 2 ausfuehren).")
return
d = self.ensure_driver()
self.log(f"Oeffne Bewerbungsformular: {self.form_link}")
d.get(self.form_link)
fill_application_form(d, self.log)
def run_all(self):
self._run_bg(self._all)
def _all(self):
self._step1()
if not self.submit_var.get():
self.log("Hinweis: Ohne Absenden kommt keine E-Mail. "
"Setze den Haken 'automatisch ABSENDEN' fuer den Vollablauf.")
return
self._step2()
if self.form_link:
self._step3()
self.log("== Ablauf beendet ==")
def main():
root = tk.Tk()
ImmoBotUI(root)
root.mainloop()
if __name__ == "__main__":
main()

631
index_email.html Normal file
View File

File diff suppressed because one or more lines are too long

703
index_formular.html Normal file
View File

File diff suppressed because one or more lines are too long

232
index_inserat.html Normal file
View File

File diff suppressed because one or more lines are too long

148
infra/RUNBOOK-ibo.md Normal file
View File

@ -0,0 +1,148 @@
# Runbook — ibo (Immo-Bot) containerisieren, deployen, Zitadel-SSO davorschalten
Immo-Bot = Flask-Web-App + Selenium-Chrome-Sidecar (noVNC-Live-Ansicht). Zwei
feste Umgebungen (KEINE PR-Previews):
| Umgebung | Host | TLS-Secret | Deploy |
|----------|------------------------|--------------|------------------------------------------|
| dev | `dev.ibo.momentrum.de` | `ibo-dev-tls`| **automatisch** bei jedem main-Push (CI) |
| prod | `ibo.momentrum.de` | `ibo-tls` | **manuell promotet** (Tag-Copy, s. u.) |
Beide hinter demselben Zitadel-oauth2-proxy `ibo-sso`
(`cookie-domain .ibo.momentrum.de` deckt dev **und** prod ab).
Die Muster spiegeln kanban (CI/Chart/CD) und luxly-sso/dev-sso (oauth2-proxy).
Cluster-, Zitadel- und DNS-Schritte sind **manuell** — dieses Runbook ist die
Anleitung. Reihenfolge unten einhalten.
---
## 0. Repo nach Gitea bringen
```sh
cd projekte/ibo
git init && git add -A && git commit -m "ibo: containerize + CI + infra"
git remote add origin momentrum-gitea:momentrum/ibo.git # SSH-Alias wie andere Repos
git push -u origin main # push-create legt momentrum/ibo an, falls aktiviert
```
**Gitea-Actions-Secrets** am Repo `momentrum/ibo` setzen (Settings → Actions → Secrets):
- `REGISTRY_USER`, `REGISTRY_TOKEN` — Push in die Gitea-Registry.
- `DEPLOY_TOKEN` — Gitea-Token mit **Write auf `momentrum/momentrum-deploy`**
(Write-back des dev-Image-Tags).
---
## 1. Zitadel: Redirect-URI ergaenzen (KEIN neuer Client)
`ibo-sso` nutzt — wie `luxly-sso` — den **bestehenden geteilten** Zitadel-Client
(K8s-Secret `oauth2-proxy-dev` in ns `iam`). Es ist **kein neuer Client** und
**kein neues Secret** noetig. Nur am vorhandenen Client im Zitadel-Console
(`https://auth.momentrum.de`) die Redirect-URI ergaenzen:
- Redirect-URI **hinzufuegen**: `https://sso.ibo.momentrum.de/oauth2/callback`
(EIN Login-Host `sso.ibo.momentrum.de` deckt per `cookie-domain .ibo.momentrum.de`
dev **und** prod ab. Wer prod spaeter mit eigenem Blast-Radius trennen will:
eigenen Client + eigenes SealedSecret anlegen und im `ibo-sso`-Deployment
referenzieren — die `oauth2-proxy-dev`-Secret-Refs dort ersetzen.)
---
## 2. oauth2-proxy `ibo-sso` (contabo-infra) — deployen
Manifeste liegen in `contabo-infra/platform/manifests/ibo-sso/`
(Deployment/Service, Middlewares `oauth2-ibo-auth`/`oauth2-ibo-errors`, Ingress
`sso.ibo.momentrum.de`, oauth2-Passthrough-Route). Sie referenzieren das
bestehende Secret `oauth2-proxy-dev` + die ConfigMap `oauth2-proxy-dev-templates`
— **nichts einzusealen**.
Die ArgoCD-App `platform/apps/ibo-sso.yaml` wird vom platform-App-of-Apps
automatisch aufgenommen (gleiche Ablage wie `luxly-sso.yaml`) — committen &
pushen genuegt; alternativ einmalig `kubectl apply -f platform/manifests/ibo-sso/`.
Ausserdem ist das ibo-Repo in `platform/manifests/argocd-projects/internal.yaml`
(sourceRepos) bereits ergaenzt — sonst lehnt ArgoCD die App-Source ab.
Ergebnis:
- Middlewares `iam-oauth2-ibo-auth@kubernetescrd` / `iam-oauth2-ibo-errors@kubernetescrd`
stehen bereit (die App-Ingresses referenzieren sie).
- `https://sso.ibo.momentrum.de` liefert die Login-Seite (kein Auth davor).
---
## 3. DNS
A/AAAA-Records auf die Cluster-Ingress-IP zeigen lassen:
- `ibo.momentrum.de`
- `dev.ibo.momentrum.de`
- `sso.ibo.momentrum.de`
cert-manager (`letsencrypt-prod`, HTTP-01) stellt die Zertifikate `ibo-tls`,
`ibo-dev-tls`, `ibo-sso-tls` automatisch aus, sobald DNS steht.
---
## 4. Deploy-Reihenfolge
1. Repo gepusht (Schritt 0) → CI baut das erste Image und bumpt
`apps/internal/ibo/dev.yaml` (`image.tag`).
2. `ibo-sso` laeuft (Schritt 2), DNS steht (Schritt 3).
3. CD-Apps im `momentrum-deploy` committen: `apps/internal/ibo/dev.yaml` +
`apps/internal/ibo/app.yaml`. ArgoCD legt Namespace `apps-ibo` an und synct.
4. **Erstes Prod-Promote:** den von der CI in `dev.yaml` gesetzten `image.tag`
nach `app.yaml` kopieren (siehe unten) — sonst hat prod noch `tag: latest`.
---
## 5. Prod promoten (Routine)
Prod bekommt bewusst NUR den Tag, den man freigibt:
```sh
cd momentrum-deploy
DEVTAG=$(grep -E '^ tag: ' apps/internal/ibo/dev.yaml | awk '{print $2}')
sed -i "s|^ tag: .*| tag: ${DEVTAG}|" apps/internal/ibo/app.yaml
git commit -am "ibo: promote ${DEVTAG} -> prod" && git push
```
ArgoCD synct prod auf den neuen Tag. Rollback = `git revert` dieses Commits.
---
## 6. Browser-Live-Ansicht (noVNC)
Der Automations-Chrome laeuft im `selenium`-Sidecar; sein Bild wird per noVNC
gestreamt und in der Web-App eingebettet (Button „👁️ Browser live ansehen",
oeffnet auch automatisch beim Consent-Browser). Routing:
- App-Container → Chrome via `SELENIUM_REMOTE_URL=http://localhost:4444`.
- Client-Browser → `https://<host>/novnc/…` → Ingress-Pfad `/novnc`
→ stripPrefix-Middleware → Sidecar-Port 7900 (noVNC/websockify).
- Steuerbar durch die SSO-Cookie (gleicher Host, gleiche Middlewares).
Der Browser laeuft **serverseitig** im Container — es wird NICHT der lokale
Browser des Nutzers ferngesteuert (das ginge von einer gehosteten App aus nicht).
---
## 7. Break-glass (Notzugang)
Faellt Zitadel aus: die SSO-Middlewares temporaer aus `ingress.ssoMiddlewares`
im jeweiligen CD-App-YAML entfernen (die App selbst hat keine eigene Auth →
sie ist dann OFFEN). Nur kurz, danach sofort zuruecksetzen. Nie ungeschuetzt
lassen.
---
## 8. Verifikationscheckliste
- [ ] `docker compose up -d` lokal → `http://localhost:5000` UI, „Browser live
ansehen" zeigt den Chrome im iframe (localhost:7900).
- [ ] `curl -s localhost:5000/healthz` → `{"ok":true}`.
- [ ] main-Push → CI gruen (version→test→build-push→helm-validate→deploy-dev),
`dev.yaml`-Tag gebumpt, ArgoCD rollt dev.
- [ ] `https://dev.ibo.momentrum.de` → Redirect zu `auth.momentrum.de`, nach
Login landet man in der App.
- [ ] In der App eingebettetes `/novnc/` zeigt den Live-Browser.
- [ ] `https://ibo.momentrum.de` nach Promote ebenso hinter Zitadel.
- [ ] Ohne SSO-Session → Login-Redirect (kein rohes 401).

6
infra/k8s/base/ibo/Chart.yaml Normal file
View File

@ -0,0 +1,6 @@
apiVersion: v2
name: ibo
description: Immo-Bot — Flask-Web-App mit Selenium-Chrome-Sidecar (noVNC-Live-Ansicht).
type: application
version: 0.1.0
appVersion: "1.0.0"

26
infra/k8s/base/ibo/templates/_helpers.tpl Normal file
View File

@ -0,0 +1,26 @@
{{- define "ibo.name" -}}
ibo
{{- end -}}
{{- define "ibo.labels" -}}
app.kubernetes.io/name: {{ include "ibo.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
app.kubernetes.io/managed-by: {{ .Release.Service }}
{{- end -}}
{{- define "ibo.selectorLabels" -}}
app.kubernetes.io/name: {{ include "ibo.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
{{- end -}}
{{- /*
Middleware-Kette fuer die Ingress-Annotation: erst die SSO-Middlewares aus dem
CD-Repo (errors VOR auth), dann die same-namespace stripPrefix fuer /novnc.
Traefik-Ingress-Refs sind "<namespace>-<name>@kubernetescrd".
*/ -}}
{{- define "ibo.middlewares" -}}
{{- $mw := list -}}
{{- if .Values.ingress.ssoMiddlewares -}}{{- $mw = append $mw .Values.ingress.ssoMiddlewares -}}{{- end -}}
{{- $mw = append $mw (printf "%s-%s-novnc@kubernetescrd" .Release.Namespace (include "ibo.name" .)) -}}
{{- join "," $mw -}}
{{- end -}}

81
infra/k8s/base/ibo/templates/deployment.yaml Normal file
View File

@ -0,0 +1,81 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: {{ include "ibo.name" . }}
labels:
{{- include "ibo.labels" . | nindent 4 }}
spec:
replicas: {{ .Values.replicaCount }}
# In-Memory-State ⇒ nie zwei gleichzeitig.
strategy:
type: Recreate
selector:
matchLabels:
{{- include "ibo.selectorLabels" . | nindent 6 }}
template:
metadata:
labels:
{{- include "ibo.selectorLabels" . | nindent 8 }}
spec:
containers:
# --- Flask-Web-App ---------------------------------------------------
- name: app
image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
ports:
- name: http
containerPort: {{ .Values.service.appTargetPort }}
env:
# Chrome laeuft im Sidecar -> selber Pod, localhost.
- name: SELENIUM_REMOTE_URL
value: "http://localhost:4444"
# Eingebettete noVNC-Live-Ansicht laeuft ueber den Ingress-Pfad.
- name: NOVNC_URL
value: {{ .Values.app.novncUrl | quote }}
{{- with .Values.app.env }}
{{- toYaml . | nindent 12 }}
{{- end }}
readinessProbe:
httpGet:
path: /healthz
port: http
initialDelaySeconds: 5
periodSeconds: 10
livenessProbe:
httpGet:
path: /healthz
port: http
initialDelaySeconds: 20
periodSeconds: 20
resources:
{{- toYaml .Values.resources.app | nindent 12 }}
# --- Selenium-Chrome + noVNC (Sidecar) -------------------------------
- name: selenium
image: {{ .Values.selenium.image }}
env:
- name: SE_NODE_MAX_SESSIONS
value: {{ .Values.selenium.maxSessions | quote }}
- name: SE_NODE_OVERRIDE_MAX_SESSIONS
value: "true"
ports:
- name: webdriver
containerPort: 4444
- name: novnc
containerPort: 7900
readinessProbe:
httpGet:
path: /status
port: webdriver
initialDelaySeconds: 10
periodSeconds: 10
volumeMounts:
- name: dshm
mountPath: /dev/shm
resources:
{{- toYaml .Values.resources.selenium | nindent 12 }}
volumes:
# Chrome braucht ausreichend /dev/shm, sonst Tab-Crashes.
- name: dshm
emptyDir:
medium: Memory
sizeLimit: {{ .Values.selenium.shmSize }}

44
infra/k8s/base/ibo/templates/ingress.yaml Normal file
View File

@ -0,0 +1,44 @@
{{- if .Values.ingress.enabled }}
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: {{ include "ibo.name" . }}
labels:
{{- include "ibo.labels" . | nindent 4 }}
annotations:
{{- with .Values.ingress.annotations }}
{{- toYaml . | nindent 4 }}
{{- end }}
# SSO (oauth2-proxy ibo) + stripPrefix /novnc. Gilt fuer alle Router dieses
# Ingress; auf dem /-Router ist stripPrefix ein No-Op (Pfad matcht nicht).
traefik.ingress.kubernetes.io/router.middlewares: {{ include "ibo.middlewares" . }}
spec:
ingressClassName: {{ .Values.ingress.className }}
{{- if .Values.ingress.tls.enabled }}
tls:
- hosts:
- {{ .Values.ingress.host | quote }}
secretName: {{ .Values.ingress.tls.secretName }}
{{- end }}
rules:
- host: {{ .Values.ingress.host | quote }}
http:
paths:
# noVNC-Live-Ansicht (Selenium-Sidecar, Port 7900) — spezifischer Pfad
# zuerst, gewinnt in Traefik gegen das / darunter.
- path: {{ .Values.ingress.novncPath }}
pathType: Prefix
backend:
service:
name: {{ include "ibo.name" . }}
port:
number: {{ .Values.service.novncPort }}
# Web-App (Flask, Port 5000)
- path: /
pathType: Prefix
backend:
service:
name: {{ include "ibo.name" . }}
port:
number: {{ .Values.service.appPort }}
{{- end }}

16
infra/k8s/base/ibo/templates/middleware.yaml Normal file
View File

@ -0,0 +1,16 @@
{{- if .Values.ingress.enabled }}
# Strippt den /novnc-Prefix, bevor der Request an den Selenium-noVNC (7900) geht:
# der Ingress-Pfad /novnc/vnc.html -> selenium sieht /vnc.html, /novnc/websockify
# -> /websockify. Same-namespace, wird in _helpers.tpl an die Middleware-Kette
# der Ingress-Annotation angehaengt.
apiVersion: traefik.io/v1alpha1
kind: Middleware
metadata:
name: {{ include "ibo.name" . }}-novnc
labels:
{{- include "ibo.labels" . | nindent 4 }}
spec:
stripPrefix:
prefixes:
- {{ .Values.ingress.novncPath | quote }}
{{- end }}

17
infra/k8s/base/ibo/templates/service.yaml Normal file
View File

@ -0,0 +1,17 @@
apiVersion: v1
kind: Service
metadata:
name: {{ include "ibo.name" . }}
labels:
{{- include "ibo.labels" . | nindent 4 }}
spec:
type: ClusterIP
selector:
{{- include "ibo.selectorLabels" . | nindent 4 }}
ports:
- name: http
port: {{ .Values.service.appPort }}
targetPort: http
- name: novnc
port: {{ .Values.service.novncPort }}
targetPort: novnc

57
infra/k8s/base/ibo/values.yaml Normal file
View File

@ -0,0 +1,57 @@
# Defaults fuer das ibo-Chart.
# Overlays (dev / prod) im CD-Repo momentrum-deploy patchen image.tag, ingress.host,
# ingress.tls.secretName und ingress.ssoMiddlewares. Das Chart selbst ist
# umgebungs-agnostisch.
image:
repository: momentrum-k3s-01.tailcc1bcf.ts.net:3000/momentrum/ibo-web
tag: latest
pullPolicy: IfNotPresent
# IMPORTANT: 1 lassen. Die App haelt In-Memory-State (Batch-Status, SSE-Streams,
# offener Consent-Browser) — mehrere Replicas wuerden ihn nicht teilen.
replicaCount: 1
service:
# Nach aussen (ClusterIP) sichtbare Ports; targetPorts sind die Container-Ports.
appPort: 80
appTargetPort: 5000
novncPort: 7900
# Selenium-Chrome-Sidecar (Chrome + noVNC). Laeuft im selben Pod; die App spricht
# ihn ueber localhost:4444 an, noVNC lauscht auf 7900.
selenium:
image: selenium/standalone-chrome:latest
maxSessions: 4 # >= config.json run.parallel_workers
shmSize: 2Gi # /dev/shm (emptyDir Memory) — Chrome braucht das
app:
# Basis-URL des eingebetteten noVNC-Live-Views (Ingress-Pfad, siehe stripPrefix).
novncUrl: /novnc/
# Zusaetzliche Env-Vars fuer den App-Container, z.B. [{name: IBO_MODE, value: prod}]
env: []
ingress:
enabled: true
className: traefik
host: ibo.momentrum.de
novncPath: /novnc
tls:
enabled: true
secretName: ibo-tls
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
# Zitadel-SSO davor (oauth2-proxy "ibo-sso" in contabo-infra). Reihenfolge:
# errors VOR auth, damit 401 zum Login-Redirect wird. Die stripPrefix-Middleware
# fuer /novnc haengt das Chart selbst hinten an (siehe _helpers.tpl).
# Vom CD-Repo gesetzt, z.B.:
# "iam-oauth2-ibo-errors@kubernetescrd,iam-oauth2-ibo-auth@kubernetescrd"
ssoMiddlewares: ""
resources:
app:
requests: { cpu: 50m, memory: 128Mi }
limits: { cpu: "1", memory: 512Mi }
selenium:
requests: { cpu: 100m, memory: 512Mi }
limits: { cpu: "2", memory: 2Gi }

169
mailtm.py Normal file
View File

@ -0,0 +1,169 @@
#!/usr/bin/env python3
"""
mail.tm - Wegwerf-E-Mail komplett per API, ohne Einrichtung/Konto.
Nur Python-Bordmittel (urllib). Ablauf:
addr, pw, token = create_account() # zufaellige Adresse anlegen
... Adresse ins Kontaktformular eintragen ...
link = wait_for_message(token, extractor, log) # auf Mail warten
API-Doku: https://docs.mail.tm
"""
import json
import time
import random
import string
import urllib.request
import urllib.error
BASE = "https://api.mail.tm"
def _req(method, path, data=None, token=None):
url = BASE + path
headers = {"Content-Type": "application/json", "Accept": "application/json"}
if token:
headers["Authorization"] = "Bearer " + token
body = json.dumps(data).encode() if data is not None else None
req = urllib.request.Request(url, data=body, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=20) as resp:
raw = resp.read().decode("utf-8", errors="replace")
return json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
detail = e.read().decode("utf-8", errors="replace")
raise RuntimeError(f"mail.tm {method} {path} -> HTTP {e.code}: {detail}")
def _rand(n):
return "".join(random.choice(string.ascii_lowercase + string.digits)
for _ in range(n))
def _members(data):
"""Liste aus der Antwort holen - egal ob reine Liste oder Hydra-Format."""
if isinstance(data, list):
return data
if isinstance(data, dict):
return data.get("hydra:member", data.get("member", []))
return []
def get_domain():
"""Erste aktive mail.tm-Domain holen."""
data = _req("GET", "/domains?page=1")
domains = _members(data)
for d in domains:
if d.get("isActive", True):
return d["domain"]
raise RuntimeError("Keine aktive mail.tm-Domain verfuegbar.")
def create_account():
"""Legt eine zufaellige Wegwerf-Adresse an. Gibt (address, password, token)."""
domain = get_domain()
address = f"{_rand(10)}@{domain}"
password = _rand(12)
# Account anlegen (kann bei Rate-Limit 429 liefern -> kurz warten & retry)
for attempt in range(3):
try:
_req("POST", "/accounts", {"address": address, "password": password})
break
except RuntimeError as e:
if "429" in str(e) and attempt < 2:
time.sleep(2)
continue
raise
token = get_token(address, password)
return address, password, token
def get_token(address, password):
data = _req("POST", "/token", {"address": address, "password": password})
return data["token"]
def list_messages(token):
data = _req("GET", "/messages?page=1", token=token)
return _members(data)
def get_message_html(token, message_id):
"""Volltext einer Nachricht als HTML (oder Text-Fallback)."""
data = _req("GET", f"/messages/{message_id}", token=token)
html = data.get("html")
if isinstance(html, list):
return "\n".join(html)
if isinstance(html, str):
return html
return data.get("text", "")
def wait_for_message(token, extractor, log,
sender_hint="", timeout=300, poll=10, should_cancel=None):
"""
Pollt das Postfach, bis eine Mail kommt, aus der `extractor(html)` einen
Link liefert. `extractor` ist eine Funktion html->str|None.
`should_cancel` (optional): Funktion ()->bool; liefert sie True, wird das
Warten sofort abgebrochen (z.B. Item geloescht/Batch gestoppt).
"""
def _cancelled_sleep(seconds):
"""Schlaeft in kleinen Schritten und bricht sofort ab, wenn should_cancel
True liefert. Gibt True zurueck, wenn abgebrochen wurde."""
slept = 0.0
while slept < seconds:
if should_cancel and should_cancel():
return True
time.sleep(0.5)
slept += 0.5
return False
deadline = time.time() + timeout
seen = set()
while time.time() < deadline:
if should_cancel and should_cancel():
log("Warten abgebrochen.")
return None
try:
msgs = list_messages(token)
except RuntimeError as e:
log(f"mail.tm Fehler: {e}")
if _cancelled_sleep(poll):
log("Warten abgebrochen.")
return None
continue
for m in msgs:
mid = m.get("id")
if not mid or mid in seen:
continue
seen.add(mid)
frm = ""
if isinstance(m.get("from"), dict):
frm = m["from"].get("address", "")
if sender_hint and sender_hint.lower() not in frm.lower():
continue
html = get_message_html(token, mid)
link = extractor(html)
if link:
log(f"Mail von {frm}: Link gefunden.")
return link
remaining = int(deadline - time.time())
log(f"Noch keine passende Mail. Warte {poll}s (Rest {remaining}s) ...")
if _cancelled_sleep(poll):
log("Warten abgebrochen.")
return None
log("!! Timeout - keine Mail mit Link gefunden.")
return None
if __name__ == "__main__":
# Kleiner Selbsttest: Adresse anlegen und Posteingang einmal lesen.
a, p, t = create_account()
print("Wegwerf-Adresse:", a)
print("Passwort:", p)
print("Nachrichten aktuell:", len(list_messages(t)))
print("Schick eine Mail an obige Adresse und starte erneut zum Testen.")

2
requirements.txt Normal file
View File

@ -0,0 +1,2 @@
flask>=3.0
selenium>=4.20

70
test_extract_link.py Normal file
View File

@ -0,0 +1,70 @@
#!/usr/bin/env python3
"""
Test-Script: Mail abrufen und Link extrahieren.
Verwendung:
python test_extract_link.py eg5gyqjqp5@web-library.net TOKEN_HIER
oder: Token aus app.py Step 1 Tabelle kopieren
"""
import sys
import json
from mailtm import list_messages, get_message_html
from core import extract_form_link
def test_extract(email, token):
print(f"📧 Abrufen von Mails fuer: {email}")
print(f"🔑 Token: {token[:20]}...")
print()
try:
msgs = list_messages(token)
print(f"📬 Mails gefunden: {len(msgs)}\n")
if not msgs:
print("❌ Keine Mails im Postfach!")
return
for i, m in enumerate(msgs, 1):
msg_id = m.get("id", "?")
sender = m.get("from", {})
if isinstance(sender, dict):
sender = sender.get("address", "?")
print(f"--- Mail {i} ---")
print(f"Von: {sender}")
print(f"Betreff: {m.get('subject', '(kein Betreff)')}")
print(f"ID: {msg_id}")
html = get_message_html(token, msg_id)
link = extract_form_link(html)
if link:
print(f"✅ LINK GEFUNDEN:")
print(f" {link}")
else:
print(f"❌ Kein Link extrahierbar")
# Debug: erste 500 Zeichen zeigen
preview = html[:500].replace("\n", " ")
print(f" Preview: {preview}...")
print()
except Exception as e:
print(f"❌ Fehler: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
if len(sys.argv) < 3:
print("Verwendung:")
print(" python test_extract_link.py EMAIL TOKEN")
print()
print("Beispiel:")
print(" python test_extract_link.py eg5gyqjqp5@web-library.net abc123xyz...")
sys.exit(1)
email = sys.argv[1]
token = sys.argv[2]
test_extract(email, token)