Update app.py

app.py

@@ -2,861 +2,1031 @@
 # -*- coding: utf-8 -*-
 
 """
-Gradio UI (robust gegen kurzzeitigen Browser-Verbindungsverlust):
-  - Startet einen Hintergrund-Job und speichert Status/Progress/Ergebnisse per job_id.
-  - UI pollt den Status (Button „Aktualisieren“), kein Streaming/Yield mehr.
-
-Run:
-  pip install gradio requests
-  python app.py
+Gradio App – robuste Async-Ausführung via Status-Endpoints
+- Rein asynchron mit Status-Endpoints
+- Läuft serverseitig weiter, auch wenn UI/Verbindung weg ist
+- Globale HTTP-Session mit Keep-Alive
+- Polling mit Backoff + Retry-After
+- Logging in Konsole + logs/<job_id>.log
+- FIX: exclude_hash richtig laden/berücksichtigen (paginierte Wholix-Suche)
+- FIX: pro Lead harte Fehlerisolierung -> kein Gradio-Absturz
 """
 
 import json
-import math
 import os
 import random
 import re
 import time
-import threading
 import uuid
+import threading
 from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Set
+
+import logging
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
 
 import gradio as gr
 import requests
+from requests.adapters import HTTPAdapter
+
+# ============================== CONFIG ====================================
 
 LEAD_BASE_URL   = "https://wholixleadgenbackend.ngrok.io"
 WHOLIX_BASE_URL = "https://api.wholix.ai"
 
-# ============================== HTTP helper ===============================
+MAX_LEADS       = 100          # Sicherheitskappe
+MAX_WORKERS     = 16           # gleichzeitige Pipelines am Server
+JOB_TTL_SEC     = 60 * 60     # wie lange fertige Jobs im Speicher bleiben (1h)
+
+# ============================== LOGGING ====================================
+
+LOG_DIR = Path("logs")
+LOG_DIR.mkdir(parents=True, exist_ok=True)
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)s | %(threadName)s | %(message)s",
+)
+APP_LOG = logging.getLogger("app")
+
+def get_job_logger(job_id: str) -> logging.Logger:
+    lg = logging.getLogger(f"job.{job_id}")
+    if not any(isinstance(h, RotatingFileHandler) for h in lg.handlers):
+        fh = RotatingFileHandler(
+            LOG_DIR / f"{job_id}.log",
+            maxBytes=512_000,
+            backupCount=2,
+            encoding="utf-8",
+        )
+        fh.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
+        lg.addHandler(fh)
+        lg.setLevel(logging.INFO)
+        lg.propagate = True
+    return lg
+
+# ========================== GLOBAL HTTP SESSION ============================
+
+GLOBAL_SES = requests.Session()
+GLOBAL_SES.headers.update({"Accept": "application/json, text/plain;q=0.9, */*;q=0.8"})
+adapter = HTTPAdapter(pool_connections=50, pool_maxsize=50, max_retries=0)
+GLOBAL_SES.mount("https://", adapter)
+GLOBAL_SES.mount("http://", adapter)
+
+try:
+    from requests_toolbelt.adapters.socket_options import TCPKeepAliveAdapter
+    ka = TCPKeepAliveAdapter(idle=30, interval=30, count=3)
+    GLOBAL_SES.mount("https://", ka)
+    GLOBAL_SES.mount("http://", ka)
+except Exception:
+    pass
+
+# ================================ HTTP =====================================
 
 class HTTPError(Exception):
-    def __init__(self, status: int, url: str, body: Any):
-        body_str = body if isinstance(body, str) else json.dumps(body, ensure_ascii=False)
-        super().__init__(f"HTTP {status} on {url}: {body_str[:500]}")
+    def __init__(self, status: int, url: str, body: Any = None, headers: Dict[str, str] = None):
+        msg = f"HTTP {status} on {url}"
+        if body is not None:
+            try:
+                bs = body if isinstance(body, str) else json.dumps(body, ensure_ascii=False)
+                msg += f": {bs[:500]}"
+            except Exception:
+                pass
+        super().__init__(msg)
         self.status = status
         self.url = url
         self.body = body
+        self.headers = headers or {}
 
 def _is_json_ct(ct: str) -> bool:
     return bool(ct and ("application/json" in ct or ct.endswith("+json")))
 
+def _safe_json(text: str):
+    try:
+        return json.loads(text)
+    except Exception:
+        repaired = text.replace("NaN", "null").replace("Infinity", "null").replace("-Infinity", "null")
+        return json.loads(repaired)
+
 def req(
     url: str,
     method: str = "GET",
     headers: Optional[Dict[str, str]] = None,
     json_body: Any = None,
     data: Any = None,
-    timeout: float = 60.0,
-    retries: int = 2,
-    retry_on: Tuple[int, ...] = (408, 425, 429, 500, 502, 503, 504, 520, 522, 524),
+    timeout: Tuple[float, float] = (5.0, 15.0),
     session: Optional[requests.Session] = None,
-):
-    s = session or requests.Session()
-    attempt = 0
-    while True:
-        try:
-            r = s.request(
-                method=method.upper(),
-                url=url,
-                headers=headers,
-                json=json_body,
-                data=data,
-                timeout=timeout,
-            )
-            ct = r.headers.get("content-type", "")
-            text = r.text or ""
-            if not r.ok:
+) -> Any:
+    s = session or GLOBAL_SES
+    r = s.request(
+        method=method.upper(),
+        url=url,
+        headers=headers,
+        json=json_body,
+        data=data,
+        timeout=timeout,
+    )
+    ct = r.headers.get("content-type", "")
+    text = r.text or ""
+    if not r.ok:
+        body = None
+        if text:
+            try:
+                body = r.json() if _is_json_ct(ct) or (not ct and text.strip().startswith(("{", "["))) else text
+            except Exception:
                 body = text
-                try:
-                    if _is_json_ct(ct):
-                        body = r.json()
-                except Exception:
-                    pass
-                raise HTTPError(r.status_code, url, body)
-            if text == "":
-                return None
-            if _is_json_ct(ct) or (not ct and text.strip().startswith(("{", "["))):
-                try:
-                    return r.json()
-                except Exception:
-                    repaired = text.replace("NaN", "null").replace("Infinity", "null").replace("-Infinity", "null")
-                    return json.loads(repaired)
-            return text
-        except (requests.Timeout, requests.ConnectionError, HTTPError) as e:
-            code = e.status if isinstance(e, HTTPError) else 0
-            should = attempt < retries and (code in retry_on or code == 0)
-            if not should:
-                raise
-            attempt += 1
-            base = min(2.5, 0.4 * (2 ** attempt))  # full-jitter
-            delay = random.random() * base
-            time.sleep(delay)
-
-# ============================= Curl parsing ===============================
-
-CURL_DATA_RE = re.compile(
-    r"""--data(?:-raw)?\s+(?P<quote>['"])(?P<body>.*?)(?P=quote)""",
-    re.DOTALL
-)
-HDR_XTOKEN_RE = re.compile(r"""-H\s+(?P<q>['"])X-Token-Id:\s*(?P<val>[^'"]+)(?P=q)""", re.IGNORECASE)
-
-def parse_curl(curl_text: str) -> Tuple[str, Dict[str, Any]]:
-    """
-    Extract X-Token-Id and JSON body from a curl command.
-    """
-    token_id = ""
-    body_str = ""
-
-    m = HDR_XTOKEN_RE.search(curl_text)
-    if m:
-        token_id = m.group("val").strip()
-
-    if not token_id:
-        hdr_inline = re.search(r"X-Token-Id:\s*([a-zA-Z0-9\-\._]+)", curl_text, re.IGNORECASE)
-        if hdr_inline:
-            token_id = hdr_inline.group(1).strip()
-
-    md = CURL_DATA_RE.search(curl_text)
-    if md:
-        body_str = md.group("body").strip()
-    else:
-        md2 = re.search(r"-d\s+(?P<q>['\"])(?P<body>.*?)(?P=q)", curl_text, re.DOTALL)
-        if md2:
-            body_str = md2.group("body").strip()
+        raise HTTPError(r.status_code, url, body=body, headers=dict(r.headers))
+    if text == "":
+        return None
+    if _is_json_ct(ct) or (not ct and text.strip().startswith(("{", "["))):
+        try:
+            return r.json()
+        except Exception:
+            return _safe_json(text)
+    return text
 
-    if not body_str:
-        raise ValueError("Konnte den JSON Body aus dem curl nicht finden (erwarte --data-raw '...').")
+# ========================= Async Start & Poll ==============================
 
+def _looks_like_ngrok_html(body: Any) -> bool:
     try:
-        payload = json.loads(body_str)
-    except json.JSONDecodeError:
-        candidate = body_str.replace("\r\n", "\n").replace("\r", "\n")
-        payload = json.loads(candidate)
-
-    if not token_id:
-        env_token = os.getenv("X_TOKEN_ID", "").strip()
-        if env_token:
-            token_id = env_token
-
-    if not token_id:
-        raise ValueError("Konnte keinen X-Token-Id Header im curl (oder env X_TOKEN_ID) finden.")
-
-    return token_id, payload
-
-# ============================ Wholix helpers ==============================
+        s = body if isinstance(body, str) else json.dumps(body, ensure_ascii=False)
+    except Exception:
+        s = str(body)
+    s_lower = (s or "").lower()
+    return ("cdn.ngrok.com" in s_lower) or ("ngrok" in s_lower and "<html" in s_lower)
 
-def wholix_login(email: str, password: str) -> str:
-    url = f"{WHOLIX_BASE_URL}/api/v1/auth/login"
-    res = req(url, method="POST", json_body={"email": email, "password": password}, timeout=30)
-    token = (res or {}).get("token")
-    if not token:
-        raise RuntimeError("Wholix-Login fehlgeschlagen: kein token in der Response.")
-    return token
+def start_async_job(url: str, body: dict, headers: dict, session: Optional[requests.Session] = None) -> str:
+    """
+    Startet einen asynchronen Backend-Job und gibt die job_id zurück.
+    Neu: interne Retries (Backoff + Jitter) bei 404/408/425/429/5xx und ngrok-HTML.
+    Macht vorher einen Warmup-Ping auf die Base-URL.
+    """
+    s = session or GLOBAL_SES
 
-def wholix_fetch_excludes(token: str, module: str = "Contacts", per_page: int = 500, max_pages: int = 100) -> List[str]:
-    url = f"{WHOLIX_BASE_URL}/api/v1/table-object-data/fetch-paginated-results"
-    headers = {"Authorization": f"Bearer {token}"}
-    exclude_hashes: List[str] = []
-    seen = set()
-    last_page = math.inf
-    session = requests.Session()
+    # --- Warmup: ngrok/Server "aufwecken" (ignoriert Fehler) ---
+    try:
+        base = re.split(r"(?<=://[^/]+)", url)[0]  # "https://host.tld"
+        s.get(base, timeout=(3.0, 3.0))
+    except Exception:
+        pass
 
-    for page in range(1, max_pages + 1):
-        if page > last_page:
-            break
-        payload = {"module": module, "action": "search", "page": page, "per_page": per_page}
-        res = req(url, method="POST", headers=headers, json_body=payload, timeout=60, session=session) or {}
-        data_block = res.get("data") or {}
-        rows = data_block.get("data") or data_block.get("results") or []
-        last_page = data_block.get("last_page", page)
-        if not rows:
-            break
-        for row in rows:
-            h = str(row.get("exclude_hash") or row.get("excludeHash") or row.get("exclude_id") or "").strip()
-            if not h or h in seen:
+    # --- POST mit Retry ---
+    attempts = 0
+    max_attempts = 6
+    delay = 0.8
+    while True:
+        attempts += 1
+        try:
+            res = req(
+                url=url,
+                method="POST",
+                headers={**headers, "Prefer": "respond-async"},
+                json_body=body,
+                timeout=(5.0, 15.0),
+                session=s,
+            )
+            job_id = (res or {}).get("job_id")
+            ok = (res or {}).get("ok", False)
+            if not ok or not job_id:
+                raise RuntimeError(f"Backend hat keine job_id geliefert (res={res!r})")
+            return job_id
+
+        except HTTPError as e:
+            transient_codes = {404, 408, 425, 429, 500, 502, 503, 504, 520, 522, 524}
+            is_transient = (e.status in transient_codes) or _looks_like_ngrok_html(e.body)
+            if attempts < max_attempts and is_transient:
+                # Retry-After respektieren
+                ra = 0.0
+                try:
+                    if isinstance(e.body, dict):
+                        ra = float(e.body.get("retry_after", 0) or 0)
+                except Exception:
+                    ra = 0.0
+                if not ra:
+                    try:
+                        ra_hdr = (e.headers or {}).get("Retry-After", "")
+                        ra = float(ra_hdr) if ra_hdr else 0.0
+                    except Exception:
+                        ra = 0.0
+                sleep_for = ra or (delay + random.uniform(0, 0.5 * delay))
+                logging.warning(f"start_async_job: transient {e.status}, retry {attempts}/{max_attempts} in {sleep_for:.1f}s …")
+                time.sleep(sleep_for)
+                delay = min(8.0, delay * 1.8)
                 continue
-            seen.add(h)
-            exclude_hashes.append(h)
-    return exclude_hashes
-
-# ======================= Lead + Email generation =========================
-
-def _pick_single(payload: Any) -> Dict[str, Any]:
-    def unwrap(x):
-        if isinstance(x, dict) and x.get("ok") and any(k in x for k in ("result", "data", "content")):
-            return x.get("result") or x.get("data") or x.get("content")
-        return x
-    p = unwrap(payload)
-    if isinstance(p, dict) and "items" in p and isinstance(p["items"], list) and p["items"]:
-        return p["items"][0]
-    if isinstance(p, list) and p:
-        return p[0]
-    if isinstance(p, dict):
-        return p
-    raise RuntimeError("Unerwartete Lead-Payload-Form.")
-
-def lead_suggest(token_id: str, filters: Dict[str, Any], icp_text: str, exclude_ids: List[str]) -> Dict[str, Any]:
-    start_url = f"{LEAD_BASE_URL}/lead/suggest?async=1"
-    headers = {"X-Token-Id": token_id, "Prefer": "respond-async"}
-    start_body = {"filters": filters, "icp_text": icp_text or "", "exclude_ids": exclude_ids or []}
-
-    try:
-        start = req(start_url, method="POST", headers=headers, json_body=start_body, timeout=60)
-    except HTTPError:
-        start = None
+            # nicht transient oder Versuche aufgebraucht
+            raise
 
-    if not start or not start.get("ok") or not start.get("job_id"):
-        sync = req(f"{LEAD_BASE_URL}/lead/suggest", method="POST", headers={"X-Token-Id": token_id}, json_body=start_body, timeout=180)
-        return _pick_single(sync)
 
-    job_id = start["job_id"]
+def poll_status(
+    url: str,
+    headers: dict,
+    max_wait_sec: int = 60 * 20,
+    min_delay: float = 1.0,
+    max_delay: float = 30.0,
+    session: Optional[requests.Session] = None,
+    on_tick=None,
+) -> dict:
     t0 = time.time()
-    max_wait = 10 * 60
     attempt = 0
+    delay = min_delay
     while True:
-        st = req(f"{LEAD_BASE_URL}/lead/suggest/status/{job_id}?t={int(time.time()*1000)}",
-                 method="GET", headers={"X-Token-Id": token_id}, timeout=60)
-        if st and st.get("status") == "done" and st.get("ok"):
-            return _pick_single(st)
-        if time.time() - t0 > max_wait:
-            raise TimeoutError("Timeout beim Warten auf Lead-Ergebnis.")
-        attempt += 1
-        time.sleep(min(2.4, 0.6 * attempt))
-
-def _ci_get(d: Dict[str, Any], key: str) -> Any:
-    if not isinstance(d, dict):
-        return None
-    if key in d and str(d[key]).strip() != "":
-        return d[key]
-    k = next((k for k in d.keys() if isinstance(k, str) and k.lower() == key.lower() and str(d[k]).strip() != ""), None)
-    return d.get(k) if k else None
-
-def _normalize_draft_result(raw: Any) -> Dict[str, Any]:
-    r = raw
-    if isinstance(r, str):
+        if time.time() - t0 > max_wait_sec:
+            raise TimeoutError("Timeout beim Warten auf Ergebnis.")
+
         try:
-            r = json.loads(r)
-        except Exception:
-            r = {"Text": r}
+            st = req(url, method="GET", headers=headers, timeout=(5.0, 6.0), session=session)
+        except HTTPError as e:
+            if e.status in (408, 425, 429, 500, 502, 503, 504, 520, 522, 524):
+                ra = 0.0
+                if isinstance(e.body, dict):
+                    try:
+                        ra = float(e.body.get("retry_after", 0))
+                    except Exception:
+                        ra = 0.0
+                if not ra:
+                    try:
+                        ra_hdr = (e.headers or {}).get("Retry-After", "")
+                        ra = float(ra_hdr) if ra_hdr else 0.0
+                    except Exception:
+                        ra = 0.0
+                attempt += 1
+                delay = min(max_delay, max(min_delay, 0.7 * (2 ** attempt)))
+                time.sleep(ra or (delay + random.uniform(0, 0.4 * delay)))
+                continue
+            raise
 
-    if isinstance(r, dict) and r.get("ok") and any(k in r for k in ("result", "content", "data")):
-        r = r.get("result") or r.get("content") or r.get("data") or r
+        if callable(on_tick):
+            try:
+                on_tick(st)
+            except Exception:
+                pass
 
-    if isinstance(r, dict) and "message" in r and isinstance(r["message"], dict):
-        r = r["message"]
-    elif isinstance(r, dict) and "results" in r and isinstance(r["results"], list) and r["results"]:
-        cand = r["results"][0]
-        if isinstance(cand, dict) and "message" in cand and isinstance(cand["message"], dict):
-            r = cand["message"]
-        else:
-            r = cand
+        if st and st.get("ok") and st.get("status") == "done":
+            return st
 
-    email_obj = r.get("email") if isinstance(r, dict) else {}
-    messages_obj = r.get("messages") if isinstance(r, dict) else {}
-
-    def pick(*keys):
-        for src in (r, email_obj, messages_obj):
-            if isinstance(src, dict):
-                for k in keys:
-                    v = _ci_get(src, k)
-                    if v is not None and str(v).strip() != "":
-                        return v
-        return ""
+        attempt = min(attempt + 1, 20)
+        delay = min(max_delay, max(min_delay, delay * 1.6))
+        time.sleep(delay)
 
-    subject = (pick("Betreff", "subject", "email_subject") or "Kurzer Austausch").strip()
-    body    = (pick("Text", "text", "Body", "body", "email_body", "content") or "").strip()
-    fu1     = (pick("FollowUp1", "followup1", "LinkedIn", "linkedin", "li") or "").strip()
-    fu2     = (pick("FollowUp2", "followup2", "Facebook", "facebook", "fb") or "").strip()
+# ============================= Services ====================================
 
-    if not body:
-        body = "Hallo,\n\nkurzer Kontaktaufbau – gerne Austausch, ob wir unterstützen können.\n\nBeste Grüße"
+def unwrap_result(payload: Any) -> Any:
+    p = payload or {}
+    if isinstance(p, dict) and p.get("ok") and any(k in p for k in ("result", "data", "content")):
+        p = p.get("result") or p.get("data") or p.get("content")
+    return p
 
-    to = ""
-    if isinstance(email_obj, dict):
-        to = str(email_obj.get("to") or "").strip()
+def suggest_single_lead(token_id: str, filters: dict, icp_text: str, exclude_ids: List[str]) -> dict:
+    res_job_id = start_async_job(
+        url=f"{LEAD_BASE_URL}/lead/suggest?async=1",
+        body={"filters": filters, "icp_text": icp_text or "", "exclude_ids": exclude_ids or []},
+        headers={"X-Token-Id": token_id},
+    )
+    st = poll_status(
+        url=f"{LEAD_BASE_URL}/lead/suggest/status/{res_job_id}?t={int(time.time()*1000)}",
+        headers={"X-Token-Id": token_id},
+        max_wait_sec=60 * 20,
+        session=GLOBAL_SES,
+    )
+    res = unwrap_result(st)
+    if isinstance(res, dict) and isinstance(res.get("items"), list) and res["items"]:
+        return res["items"][0]
+    if isinstance(res, list) and res:
+        return res[0]
+    return res if isinstance(res, dict) else {"result": res}
+
+def _flatten_text(val: Any) -> str:
+    """
+    Newlines/Tabs/Mehrfach-Whitespace sauber zu einem Space zusammenziehen.
+    Keine Feldraten, nur Plain-String-Verarbeitung.
+    """
+    s = "" if val is None else str(val)
+    s = s.replace("\r\n", "\n").replace("\r", "\n")
+    s = re.sub(r"\s+", " ", s).strip()
+    return s
 
-    return {
-        "email": {"to": to, "subject": subject, "body": body},
-        "followup1": fu1,
-        "followup2": fu2,
-    }
 
-def email_generate(token_id: str, variables: Dict[str, Any], items: List[Dict[str, Any]]) -> Dict[str, Any]:
+def normalize_draft(raw: Any) -> Dict[str, Any]:
+    r = unwrap_result(raw) or raw or {}
+    if isinstance(r, dict) and "message" in r and isinstance(r["message"], dict):
+        r = r["message"]
+    email_obj = r.get("email") if isinstance(r, dict) else {}
+    def pick(obj, *keys):
+        if not isinstance(obj, dict): return ""
+        for k in keys:
+            for kk, vv in obj.items():
+                if isinstance(kk, str) and kk.lower() == k.lower():
+                    if isinstance(vv, str) and vv.strip(): return vv.strip()
+                    if vv: return vv
+        return ""
+    subject = pick(email_obj, "subject", "email_subject", "Betreff") or pick(r, "subject", "email_subject", "Betreff") 
+    body    = pick(email_obj, "body", "text", "content") or pick(r, "body", "text", "content") 
+    fu1     = pick(r, "followup1", "FollowUp1", "LinkedIn")
+    fu2     = pick(r, "followup2", "FollowUp2", "Facebook")
+    to      = pick(email_obj, "to")
+    return {"email": {"to": to, "subject": subject, "body": body}, "followup1": fu1, "followup2": fu2}
+
+def email_generate_async(token_id: str, variables: dict, items: List[dict]) -> dict:
+    """
+    Startet /email/generate (async) und gibt nur ECHTE Endpoint-Daten zurück.
+    - Übergibt globale Variablen 1:1 (inkl. Signatur) auf Root-Ebene.
+    - Keine Feldumbenennung, keine Heuristik.
+    Rückgabe:
+      {
+        "email": {"subject": <Betreff>, "body": <Text>},
+        "followup1": <FollowUp1>,
+        "followup2": <FollowUp2>,
+        "raw": <unwrap_result(...)>
+      }
+    """
     if not items:
-        raise ValueError("items ist erforderlich und muss mind. 1 Lead enthalten.")
-    body = {**variables, "items": items, "item_index": 0}
+        raise ValueError("items fehlt (mindestens 1 Lead erforderlich).")
+
+    # Nur die Keys weiterreichen, die dein Backend tatsächlich kennt/erwartet.
+    # Wir filtern NICHT um – wir vertrauen der übergebenen payload (keine Fantasie-Felder).
+    root_allowed = {
+        "Produkt_und_Dienstleistungsbeschreibung",
+        "CTA",
+        "Signatur",
+        "Checkliste_Landingpage",
+        "homepage_url",   # optional, falls im Frontend genutzt
+        "tags",           # wird serverseitig ignoriert, aber ok
+        "Touch_Point",    # optional – stört den Backend-Kontrakt nicht
+    }
 
-    start = None
-    try:
-        start = req(f"{LEAD_BASE_URL}/email/generate?async=1",
-                    method="POST",
-                    headers={"X-Token-Id": token_id, "Prefer": "respond-async"},
-                    json_body=body, timeout=60)
-    except HTTPError:
-        pass
+    safe_vars = {}
+    for k, v in (variables or {}).items():
+        # 1:1 durchlassen für die bekannten Felder
+        if k in root_allowed:
+            safe_vars[k] = v
 
-    if not start or not start.get("ok") or not start.get("job_id"):
-        res = req(f"{LEAD_BASE_URL}/email/generate",
-                  method="POST",
-                  headers={"X-Token-Id": token_id},
-                  json_body=body, timeout=180)
-        return _normalize_draft_result(res)
+    # Payload für den echten Endpoint: Root-Variablen + Items
+    start_body = {**safe_vars, "items": items, "item_index": 0}
 
-    job_id = start["job_id"]
-    t0 = time.time()
-    max_wait = 10 * 60
-    attempt = 0
-    while True:
-        st = req(f"{LEAD_BASE_URL}/email/generate/status/{job_id}?t={int(time.time()*1000)}",
-                 method="GET", headers={"X-Token-Id": token_id}, timeout=60)
-        if st and st.get("ok") and st.get("status") == "done":
-            return _normalize_draft_result(st)
-        if time.time() - t0 > max_wait:
-            raise TimeoutError("Timeout bei der E-Mail-Generierung.")
-        attempt += 1
-        time.sleep(min(2.4, 0.6 * attempt))
-
-# ======== DROP-IN: Wholix-Mapping & Store (FIXED) ========================
-
-ALLOWED_FIELDS = {
-    "firstname",
-    "lastname",
-    "email",                 # Pflichtfeld
-    "adress",                # (sic) exakt so geschrieben
-    "city",
-    "postcode",
-    "phonenumber",
-    "job_title",
-    "departments",           # Text, KEINE Liste
-    "linkedin_url",
-    "company_name",
-    "company_url",
-    "message_mail",
-    "message_mail_subject",
-    "message_followup1",
-    "message_followup2",
-    "exclude_hash",
-    "status_field",          # Multi-Select: { keys:[], values:[] }
-    "tags",                  # Multi-Select: { keys:[], values:[] }
-}
-
-def filter_wholix_contact_fields(obj: dict) -> dict:
-    """
-    - nur erlaubte Felder
-    - email immer getrimmt
-    - Strings getrimmt; leere Werte raus
-    """
-    out = {}
-    for k, v in (obj or {}).items():
-        if k not in ALLOWED_FIELDS:
-            continue
-        if k == "email":
-            out["email"] = str(v or "").strip()
-            continue
-        if v is None:
-            continue
-        s = v.strip() if isinstance(v, str) else v
-        if isinstance(s, str) and s == "":
-            continue
-        out[k] = s
-    return out
-
-def normalize_wholix_dropdown(val):
-    """
-    akzeptiert {keys,values}, Array oder String
-    → normalisiert zu {keys:[...], values:[...]} oder None
-    """
-    if isinstance(val, dict) and ("keys" in val or "values" in val):
-        ks = [str(x).strip() for x in (val.get("keys") or []) if str(x).strip()]
-        vs = [str(x).strip() for x in (val.get("values") or []) if str(x).strip()]
-        if not vs and ks:
-            vs = ks[:]
-        return {"keys": ks, "values": vs} if (ks or vs) else None
-    if isinstance(val, list):
-        ks = [str(x).strip() for x in val if str(x).strip()]
-        return {"keys": ks, "values": ks} if ks else None
-    if isinstance(val, str) and val.strip():
-        s = val.strip()
-        return {"keys": [s], "values": [s]}
-    return None
-
-# ---------- helpers for robust mapping ----------
-
-def _first_non_empty(*vals):
-    for v in vals:
-        if isinstance(v, str) and v.strip():
-            return v.strip()
-        if v not in (None, "", [], {}):
-            return v
-    return None
-
-def _from_ci(d: dict, *keys, default=None):
-    if not isinstance(d, dict):
-        return default
-    for k in keys:
-        if k in d and str(d[k]).strip() != "":
-            return d[k]
-        for dk in d.keys():
-            if isinstance(dk, str) and dk.lower() == k.lower() and str(d[dk]).strip() != "":
-                return d[dk]
-    return default
-
-def _join_nonempty(parts, sep=" "):
-    return sep.join([str(x).strip() for x in parts if str(x or "").strip()])
-
-def _parse_maybe_json_list(value):
-    """
-    Accept list, JSON-string list, or bracketed string → return list[str]
-    """
-    if value is None:
-        return []
-    if isinstance(value, list):
-        return [str(x).strip() for x in value if str(x).strip()]
-    s = str(value).strip()
-    if not s:
-        return []
-    # try JSON
-    if (s.startswith("[") and s.endswith("]")) or (s.startswith("(") and s.endswith(")")):
-        try:
-            arr = json.loads(s.replace("(", "[").replace(")", "]"))
-            if isinstance(arr, list):
-                return [str(x).strip() for x in arr if str(x).strip()]
-        except Exception:
-            # crude fallback
-            s2 = s.strip("[]()")
-            parts = [p.strip().strip("'").strip('"') for p in s2.split(",")]
-            return [p for p in parts if p]
-    # plain string, maybe delimited
-    if "," in s:
-        return [p.strip() for p in s.split(",") if p.strip()]
-    return [s]
-
-def _normalize_tag_items(tag_text):
-    """
-    Accepts: "AI", "[AI]", "AI, Sales", '["AI","Sales"]' → returns list[str]
-    """
-    if isinstance(tag_text, list):
-        return [str(x).strip() for x in tag_text if str(x).strip()]
-    if tag_text is None:
-        return []
-    s = str(tag_text).strip()
-    if not s:
-        return []
+    # Debug-Log: zeigen, welche Keys wir wirklich senden (inkl. Signatur)
     try:
-        if s.startswith("[") and s.endswith("]"):
-            arr = json.loads(s)
-            if isinstance(arr, list):
-                return [str(x).strip() for x in arr if str(x).strip()]
+        sent_keys = sorted(list(start_body.keys()))
+        sig_preview = str(start_body.get("Signatur") or "")[:120]
+        APP_LOG.info(f"/email/generate body keys: {sent_keys}")
+        APP_LOG.info(f"/email/generate Signatur (preview): {sig_preview}")
     except Exception:
         pass
-    for sep in [",", "|", ";"]:
-        if sep in s:
-            return [p.strip() for p in s.split(sep) if p.strip()]
-    return [s.strip("[]")]
-
-# ---------- the fixed mapper ----------
-
-def map_to_wholix_record(lead: dict, draft: dict, tag_text: str = "AI") -> dict:
-    """
-    FIXED:
-      - nutzt jetzt 'draft' für message_mail/subject/followups
-      - bereinigt departments (keine ["..."] Reste)
-      - breite Fallbacks für job_title / linkedin_url / adress / city / postcode / company_url / exclude_hash
-      - Tags: nur Benutzer-Tags, KEINE Departments mehr
-    """
-    p   = (lead or {}).get("person")   or {}
-    c   = (lead or {}).get("company")  or {}
-    m   = (lead or {}).get("messages") or {}
-    ctx = (lead or {}).get("context")  or {}
-
-    # --- Email (REQUIRED) ---
-    email = str(_first_non_empty(p.get("email"), _from_ci(p, "mail", "email_address")) or "").strip()
-    if not email:
-        e = ValueError("E-Mail-Adresse fehlt – Wholix benötigt 'email' als Pflichtfeld.")
-        e.name = "ValidationError"
-        raise e
-
-    # --- Departments (TEXT) — clean up list-like strings ---
-    depts_list = _parse_maybe_json_list(p.get("departments"))
-    departments_txt = ", ".join(depts_list) if depts_list else None
-
-    # --- Company URL with wide fallbacks ---
-    company_url = _first_non_empty(
-        c.get("url"), c.get("website"), c.get("domain"),
-        c.get("homepage_url"), c.get("website_url"), c.get("url_normalized"),
-        ctx.get("url"), (lead or {}).get("homepage_url"),
-    )
-
-    # --- Message from generated draft ---
-    draft = draft or {}
-    draft_email  = draft.get("email") if isinstance(draft, dict) else {}
-    msg_subject  = _first_non_empty(
-        _from_ci(draft_email, "subject", "email_subject"),
-        _from_ci(draft, "subject", "email_subject", "Betreff"),
-        _from_ci(m, "message_mail_subject"),
-    )
-    msg_body     = _first_non_empty(
-        _from_ci(draft_email, "body", "text", "content"),
-        _from_ci(draft, "body", "Text", "content", "email_body"),
-        _from_ci(m, "message_mail"),
-    )
-    followup1    = _first_non_empty(
-        _from_ci(draft, "followup1", "FollowUp1", "LinkedIn", "linkedin", "li"),
-        _from_ci(m, "followup1", "message_followup1"),
-    )
-    followup2    = _first_non_empty(
-        _from_ci(draft, "followup2", "FollowUp2", "Facebook", "facebook", "fb"),
-        _from_ci(m, "followup2", "message_followup2"),
-    )
 
-    # --- Address / City / Postcode fallbacks ---
-    street = _first_non_empty(
-        _join_nonempty([c.get("street_name"), c.get("street_number")]),
-        c.get("address"), c.get("address1"), c.get("address_line1"),
-        c.get("street"), c.get("street_address"),
+    # Async-Start & Poll
+    res_job_id = start_async_job(
+        url=f"{LEAD_BASE_URL}/email/generate?async=1",
+        body=start_body,
+        headers={"X-Token-Id": token_id},
     )
-    city = _first_non_empty(c.get("city"), c.get("town"), c.get("locality"))
-    postcode = _first_non_empty(c.get("zip_code"), c.get("postal_code"), c.get("postcode"), c.get("zip"))
-
-    # --- Job title with fallbacks ---
-    job_title = _first_non_empty(
-        p.get("job_title"),
-        p.get("job_title_de_DE"),
-        p.get("title"),
-        p.get("position"),
-        _from_ci(p, "role"),
+    st = poll_status(
+        url=f"{LEAD_BASE_URL}/email/generate/status/{res_job_id}?t={int(time.time()*1000)}",
+        headers={"X-Token-Id": token_id},
+        max_wait_sec=60 * 20,
+        session=GLOBAL_SES,
     )
 
-    # --- LinkedIn URL with fallbacks ---
-    linkedin_url = _first_non_empty(
-        p.get("linkedin_url"),
-        p.get("linkedin"),
-        p.get("linkedin_profile"),
-        p.get("linkedinUrl"),
-        p.get("li"),
-        p.get("li_url"),
-    )
+    # Ergebnis 1:1 auswerten – nur echte Felder
+    raw = unwrap_result(st) or {}
+    results = raw.get("results") if isinstance(raw, dict) else None
+    first = (results[0] if isinstance(results, list) and results else {}) or {}
+    msg = first.get("message") if isinstance(first, dict) else {}
 
-    # --- exclude_hash fallbacks ---
-    exclude_hash = _first_non_empty(
-        lead.get("exclude_hash"),
-        c.get("exclude_hash"),
-        p.get("exclude_hash"),
-        lead.get("combined_id"),
-    )
+    subj = str((msg or {}).get("Betreff") or "")
+    text = str((msg or {}).get("Text") or "")
+    fu1  = str((msg or {}).get("FollowUp1") or "")
+    fu2  = str((msg or {}).get("FollowUp2") or "")
 
-    # --- Tags: ONLY what user provided (no departments mirroring) ---
-    tag_items = _normalize_tag_items(tag_text)
-    tags_dropdown = {"keys": tag_items, "values": tag_items} if tag_items else None
-
-    payload = {
-        # Person
-        "firstname":    p.get("first_name") or None,
-        "lastname":     p.get("last_name")  or None,
-        "email":        email,
-        "adress":       street or None,           # (sic)
-        "city":         city or None,
-        "postcode":     postcode or None,
-        "phonenumber":  p.get("phone") or None,
-        "job_title":    job_title or None,
-        "departments":  departments_txt,
-        "linkedin_url": linkedin_url or None,
-
-        # Company
-        "company_name": _first_non_empty(c.get("name"), c.get("company_name")),
-        "company_url":  company_url or None,
-
-        # Message
-        "message_mail":          msg_body or None,
-        "message_mail_subject":  msg_subject or None,
-        "message_followup1":     followup1 or None,
-        "message_followup2":     followup2 or None,
-
-        # Other
-        "exclude_hash": exclude_hash or None,
-
-        # Dropdowns
-        "status_field": { "keys": ["Kontakt aufgenommen"], "values": ["Kontakt aufgenommen"] },
-        "tags":         tags_dropdown,
+    return {
+        "email": {"subject": subj, "body": text},
+        "followup1": fu1,
+        "followup2": fu2,
+        "raw": raw,
     }
 
-    normalized = filter_wholix_contact_fields(payload)
-
-    # Normalize dropdowns
-    if "status_field" in normalized:
-        fixed = normalize_wholix_dropdown(normalized["status_field"])
-        if fixed: normalized["status_field"] = fixed
-        else:     normalized.pop("status_field", None)
 
-    if "tags" in normalized:
-        fixed = normalize_wholix_dropdown(normalized["tags"])
-        if fixed: normalized["tags"] = fixed
-        else:     normalized.pop("tags", None)
-
-    return normalized
+def wholix_login(email: str, password: str) -> str:
+    res = req(f"{WHOLIX_BASE_URL}/api/v1/auth/login", method="POST",
+              json_body={"email": email, "password": password}, timeout=(5.0, 15.0))
+    token = (res or {}).get("token")
+    if not token:
+        raise RuntimeError("Wholix-Login fehlgeschlagen.")
+    return token
 
 def wholix_store_contact(token: str, record: dict, module: str = "Contacts") -> dict:
     """
-    Wholix-Store mit Dropdown-Fallbacks:
-      1) Normales {keys,values}
-      2) Legacy {value}
-      3) Ohne problematische Felder
+    Sendet NUR erlaubte Felder an Wholix und saniert problematische Werte:
+      - URLs (linkedin_url/company_url): Schema ergänzen, invalid droppen
+      - departments: immer Plain-Text
+      - Multi-Select (status_field/tags): nur korrektes {keys,values}
+      - Leere Strings -> None und weglassen
     """
-    email = str((record or {}).get("email") or "").strip()
+    import urllib.parse as _urlparse
+
+    if not isinstance(record, dict):
+        raise ValueError("Wholix: record muss ein dict sein.")
+
+    email = str((record.get("email") or "")).strip()
     if not email:
-        e = ValueError("Wholix: 'email' ist Pflichtfeld und darf nicht leer sein.")
-        e.name = "ValidationError"
-        raise e
+        raise ValueError("Wholix: 'email' ist Pflichtfeld.")
+
+    ALLOWED = {
+        "firstname",
+        "lastname",
+        "email",                 # Pflichtfeld
+        "adress",                # (sic) genau so
+        "city",
+        "postcode",
+        "phonenumber",
+        "job_title",
+        "departments",           # Text
+        "linkedin_url",
+        "company_name",
+        "company_url",
+        "message_mail",
+        "message_mail_subject",
+        "message_followup1",
+        "message_followup2",
+        "exclude_hash",
+        "status_field",          # Multi-Select: {keys:[], values:[]}
+        "tags",                  # Multi-Select: {keys:[], values:[]}
+    }
 
-    normalized = filter_wholix_contact_fields({ **record, "email": email })
+    def _clean_str(v):
+        if v is None:
+            return None
+        s = str(v).strip()
+        return s if s else None
 
-    if "status_field" in normalized:
-        fixed = normalize_wholix_dropdown(normalized["status_field"])
-        if fixed: normalized["status_field"] = fixed
-        else:     normalized.pop("status_field", None)
+    def _coerce_departments(v):
+        if v is None:
+            return None
+        if isinstance(v, (list, tuple, set)):
+            v = ", ".join(str(x).strip() for x in v if str(x).strip())
+        else:
+            v = str(v).strip()
+            # String wie "[Marketing]" -> "Marketing"
+            if v.startswith("[") and v.endswith("]"):
+                v = v[1:-1].strip().strip("'\"")
+        return _clean_str(v)
+
+    def _normalize_url(u):
+        u = _clean_str(u)
+        if not u:
+            return None
+        if not re.match(r"^[a-zA-Z][a-zA-Z0-9+\-.]*://", u):
+            u = "https://" + u
+        try:
+            pr = _urlparse.urlparse(u)
+            if not pr.scheme or not pr.netloc:
+                return None
+            return u
+        except Exception:
+            return None
+
+    out = {}
+    for k in ALLOWED:
+        if k not in record:
+            continue
+        val = record[k]
+
+        if k == "departments":
+            val = _coerce_departments(val)
+        elif k in ("linkedin_url", "company_url"):
+            val = _normalize_url(val)
+        elif k in ("status_field", "tags"):
+            # nur korrektes Schema durchlassen
+            if not (isinstance(val, dict) and "keys" in val and "values" in val):
+                val = None
+        else:
+            val = _clean_str(val)
+
+        if val is not None:
+            out[k] = val
 
-    if "tags" in normalized:
-        fixed = normalize_wholix_dropdown(normalized["tags"])
-        if fixed: normalized["tags"] = fixed
-        else:     normalized.pop("tags", None)
+    # Pflichtfeld sicher
+    out["email"] = _clean_str(email)
 
     url = f"{WHOLIX_BASE_URL}/api/v1/table-object-data/store-objects"
     headers = {"Authorization": f"Bearer {token}"}
+    body = {"module": module, "action": "store", "data": [out]}
+    return req(url, method="POST", headers=headers, json_body=body, timeout=(5.0, 30.0))
 
-    body = {"module": module, "action": "store", "data": [normalized]}
-    try:
-        return req(url, method="POST", headers=headers, json_body=body, timeout=60)
-    except Exception as e1:
-        from requests import HTTPError as _ReqHTTPError
-        status = getattr(e1, "status", None) or (e1.response.status_code if isinstance(e1, _ReqHTTPError) and e1.response else None)
-        msg = ""
-        try:
-            msg = json.dumps(getattr(e1, "body", "") or "").lower()
-        except Exception:
-            pass
-        if status != 422 or not any(k in (msg or "") for k in ("status_field", "tags")):
-            raise
 
-    try:
-        legacy = dict(normalized)
-        def to_value(v):
-            if isinstance(v, dict) and "keys" in v and v["keys"]:
-                return v["keys"][0]
-            if isinstance(v, dict) and "values" in v and v["values"]:
-                return v["values"][0]
-            if isinstance(v, list) and v:
-                return v[0]
-            if isinstance(v, str) and v.strip():
-                return v.strip()
-            return None
+# ======= NEW: Wholix-Excludes paginiert laden (wie in deinem JS) ===========
+
+def wholix_fetch_excludes(token: str,
+                          module_name: str = "Contacts",
+                          per_page: int = 500,
+                          max_pages: int = 100,
+                          dedupe: bool = True) -> List[Dict[str, str]]:
+    path = f"{WHOLIX_BASE_URL}/api/v1/table-object-data/fetch-paginated-results"
+    headers = {"Authorization": f"Bearer {token}"}
+    out: List[Dict[str, str]] = []
+    seen: Set[str] = set()
 
-        if "status_field" in legacy and legacy["status_field"]:
-            legacy["status_field"] = {"value": to_value(legacy["status_field"])}
-        if "tags" in legacy and legacy["tags"]:
-            legacy["tags"] = {"value": to_value(legacy["tags"])}
-
-        body2 = {"module": module, "action": "store", "data": [legacy]}
-        return req(url, method="POST", headers=headers, json_body=body2, timeout=60)
-    except Exception as e2:
-        from requests import HTTPError as _ReqHTTPError
-        status2 = getattr(e2, "status", None) or (e2.response.status_code if isinstance(e2, _ReqHTTPError) and e2.response else None)
-        msg2 = ""
+    last_page = 10**9
+    for page in range(1, max_pages + 1):
+        if page > last_page:
+            break
+        payload = {"module": module_name, "action": "search", "page": page, "per_page": per_page}
         try:
-            msg2 = json.dumps(getattr(e2, "body", "") or "").lower()
-        except Exception:
-            pass
-        if status2 != 422 or not any(k in (msg2 or "") for k in ("status_field", "tags")):
-            raise
+            res = req(path, method="POST", headers=headers, json_body=payload, timeout=(5.0, 30.0))
+        except Exception as e:
+            APP_LOG.warning(f"Wholix-Excludes Page {page} Fehler: {e}")
+            break
 
-    stripped = dict(normalized)
-    stripped.pop("status_field", None)
-    stripped.pop("tags", None)
-    body3 = {"module": module, "action": "store", "data": [stripped]}
-    return req(url, method="POST", headers=headers, json_body=body3, timeout=60)
+        data_block = (res or {}).get("data") or {}
+        rows = data_block.get("data") or data_block.get("results") or []
+        if isinstance(data_block.get("last_page"), (int, float)):
+            last_page = int(data_block.get("last_page"))
+        else:
+            last_page = page
+
+        if not isinstance(rows, list) or not rows:
+            break
 
-# ====================== Background-Jobs (robust UI) =======================
+        for row in rows:
+            ex = (row.get("exclude_hash") or row.get("excludeHash") or row.get("exclude_id") or "").strip()
+            if not ex:
+                continue
+            if dedupe and ex in seen:
+                continue
+            if dedupe:
+                seen.add(ex)
+            cname = (row.get("company_name") or row.get("companyName") or "").strip()
+            out.append({"exclude_hash": ex, "company_name": cname})
+    return out
 
-LEAD_COUNTS = [1, 2, 3, 5, 10, 20, 50, 100, 200]
+# ========================== Job Management =================================
 
-# In-Memory Job Store
-EXEC = ThreadPoolExecutor(max_workers=16)
-JOBS: Dict[str, Dict[str, Any]] = {}  # job_id -> state dict
+EXEC = ThreadPoolExecutor(max_workers=MAX_WORKERS)
+JOBS: Dict[str, Dict[str, Any]] = {}
 
 def _job_init(job_id: str):
     JOBS[job_id] = {
-        "log": [],          # List[str]
-        "progress": 0.0,    # 0..1
-        "rows": [],         # Ergebnisse (List[Dict])
+        "log": [],
+        "progress": 0,
+        "rows": [],
         "done": False,
-        "error": None,      # str | None
+        "error": None,
         "lock": threading.RLock(),
+        "created_at": time.time(),
+        "finished_at": None,
     }
 
-def _job_emit(job_id: str, msg: str = "", progress: Optional[float] = None, rows: Optional[List[Dict[str, Any]]] = None):
+def _job_emit(job_id: str, msg: str = None, progress: Optional[int] = None, rows_append: Optional[dict] = None):
     st = JOBS.get(job_id)
-    if not st:
-        return
+    if not st: return
+    lg = get_job_logger(job_id)
     with st["lock"]:
         if msg:
             st["log"].append(msg)
-            st["log"] = st["log"][-1000:]  # cap
-        if progress is not None:
-            st["progress"] = max(0.0, min(1.0, progress))
-        if rows is not None:
-            st["rows"] = rows[-1000:]      # cap
+            st["log"] = st["log"][-1000:]
+            APP_LOG.info(f"[{job_id}] {msg}")
+            lg.info(msg)
+        if isinstance(progress, int):
+            st["progress"] = max(0, min(100, progress))
+        if rows_append:
+            st["rows"].append(rows_append)
+            st["rows"] = st["rows"][-1000:]
+            APP_LOG.info(f"[{job_id}] row: {rows_append}")
+            lg.info(f"row: {rows_append}")
 
 def _job_finish(job_id: str, error: Optional[str] = None):
     st = JOBS.get(job_id)
-    if not st:
-        return
+    if not st: return
+    lg = get_job_logger(job_id)
     with st["lock"]:
         st["error"] = error
-        st["done"]  = True
-        if error:
-            st["log"].append(f"❌ {error}")
+        st["done"] = True
+        st["finished_at"] = time.time()
+        final = "✅ Fertig." if not error else f"❌ {error}"
+        st["log"].append(final)
+        APP_LOG.info(f"[{job_id}] {final}")
+        (lg.error if error else lg.info)(final)
+
+def _gc_jobs():
+    now = time.time()
+    for jid, st in list(JOBS.items()):
+        if st.get("done") and st.get("finished_at") and now - st["finished_at"] > JOB_TTL_SEC:
+            JOBS.pop(jid, None)
+
+# ========================== Pipeline (Background) ===========================
+
+LEAD_COUNTS = [1, 2, 3, 4, 5, 10, 15, 20, 40, 80, 100, 200, 300, 400, 500, 1000]
+
+CURL_DATA_RE = re.compile(r"""--data(?:-raw)?\s+(?P<q>['"])(?P<body>.*?)(?P=q)""", re.DOTALL)
+HDR_XTOKEN_RE = re.compile(r"""-H\s+(?P<q>['"])X-Token-Id:\s*(?P<val>[^'"]+)(?P=q)""", re.IGNORECASE)
+
+def _find_data_quote_start(s: str) -> Tuple[int, Optional[str]]:
+    m = re.search(r"--data(?:-raw)?\s+(['\"])", s)
+    if not m:
+        return -1, None
+    return m.end(1), m.group(1)
+
+def _scan_quoted_payload(s: str, i: int, q: str) -> Tuple[str, int]:
+    out = []
+    n = len(s)
+    while i < n:
+        ch = s[i]
+        if q == "'":
+            if i + 4 < n and s[i:i+5] == "'\"'\"'":
+                out.append("'"); i += 5; continue
+            if ch == "'":
+                return "".join(out), i + 1
+            out.append(ch); i += 1
         else:
-            st["log"].append("✅ Alles erledigt.")
+            if ch == "\\" and i + 1 < n:
+                out.append(s[i+1]); i += 2; continue
+            if ch == '"':
+                return "".join(out), i + 1
+            out.append(ch); i += 1
+    return "".join(out), i
 
-def run_pipeline_bg(job_id: str, curl_text: str, n_leads: int):
+def parse_curl(curl_text: str) -> Tuple[str, Dict[str, Any]]:
     """
-    Background-Pipeline:
-      1) Wholix-Login
-      2) Excludes laden
-      3..N) Lead holen → Nachricht generieren → in Wholix speichern
+    Extrahiert X-Token-Id und JSON aus einem curl mit --data-raw '...'.
+    Keine Feldumbenennungen, keine Heuristik – JSON wird 1:1 übernommen.
     """
-    results: List[Dict[str, Any]] = []
+    import shlex
+
+    if not isinstance(curl_text, str) or not curl_text.strip():
+        raise ValueError("Leerer curl-Text.")
+
+    # 1) Token aus Header holen (robust für einfache/doppelte Quotes)
+    token_id = ""
+    # -H 'X-Token-Id: abc'   oder   -H "X-Token-Id: abc"
+    m = re.search(r"""-H\s+(["'])X-Token-Id:\s*([^"']+)\1""", curl_text, re.IGNORECASE)
+    if m:
+        token_id = m.group(2).strip()
+
+    # fallback: unquoted header
+    if not token_id:
+        m2 = re.search(r"""X-Token-Id:\s*([A-Za-z0-9\-\._]+)""", curl_text, re.IGNORECASE)
+        if m2:
+            token_id = m2.group(1).strip()
+
+    if not token_id:
+        env_token = os.getenv("X_TOKEN_ID", "").strip()
+        if env_token:
+            token_id = env_token
+    if not token_id:
+        raise ValueError("Konnte keinen X-Token-Id Header im curl (oder env X_TOKEN_ID) finden.")
 
-    def log(msg: str):
-        st = JOBS.get(job_id, {})
-        prog = st.get("progress", 0.0)
-        _job_emit(job_id, msg=msg, progress=prog, rows=results)
+    # 2) JSON-Body nach --data / --data-raw extrahieren (einfach/doppelt-quoted)
+    qmatch = re.search(r"""--data(?:-raw)?\s+(['"])""", curl_text)
+    if not qmatch:
+        raise ValueError("Konnte den JSON Body nicht finden (erwarte --data-raw '...').")
+    quote = qmatch.group(1)
+    start = qmatch.end(1)
+
+    # Payload bis zum passenden schließenden Quote lesen (beachtet Escapes)
+    out = []
+    i = start
+    n = len(curl_text)
+    if quote == "'":
+        # Bash-Rule: in single quotes sind nur Sequenzen '\"'\"' als eingebettetes einzelnes '
+        while i < n:
+            if i + 4 < n and curl_text[i:i+5] == "'\"'\"'":
+                out.append("'")
+                i += 5
+                continue
+            ch = curl_text[i]
+            if ch == "'":
+                break
+            out.append(ch)
+            i += 1
+    else:
+        # Doppel-Quotes: Backslashes beachten
+        while i < n:
+            ch = curl_text[i]
+            if ch == "\\" and i + 1 < n:
+                out.append(curl_text[i+1])
+                i += 2
+                continue
+            if ch == '"':
+                break
+            out.append(ch)
+            i += 1
 
-    # parse curl
+    body_str = "".join(out).strip()
+    # 3) JSON laden (ohne Felder zu „erdenken“)
+    try:
+        payload = json.loads(body_str)
+    except json.JSONDecodeError:
+        # CRLF -> LF und nochmal versuchen
+        payload = json.loads(body_str.replace("\r\n", "\n").replace("\r", "\n"))
+
+    return token_id, payload
+
+
+def run_pipeline_bg(job_id: str, curl_text: str, n_leads_ui: int):
+    lg = get_job_logger(job_id)
+    APP_LOG.info(f"[{job_id}] Background job gestartet (n={n_leads_ui})")
+    lg.info(f"Background job gestartet (n={n_leads_ui})")
+
+    _gc_jobs()
+    _job_emit(job_id, "Job gestartet.")
     try:
         token_id, payload = parse_curl(curl_text)
-    except Exception as e:
-        _job_finish(job_id, error=f"Parse-Fehler: {e}")
+    except Exception:
+        logging.exception(f"[{job_id}] Parse-Fehler")
+        _job_finish(job_id, "Parse-Fehler: siehe Logs (Stacktrace).")
         return
 
     wh_email = payload.get("wholix_email") or payload.get("Wholix_email") or ""
     wh_pass  = payload.get("wholix_passwort") or payload.get("wholix_password") or ""
     if not wh_email or not wh_pass:
-        _job_finish(job_id, error="In der JSON-Payload fehlen wholix_email / wholix_passwort.")
+        _job_finish(job_id, "In der JSON-Payload fehlen wholix_email / wholix_passwort.")
         return
 
+    # Globale Felder exakt wie im Curl (1:1)
     filters      = payload.get("filters") or {}
-    icp_text     = payload.get("Produkt_und_Dienstleistungsbeschreibung") or payload.get("icp_text") or ""
+    icp_text     = payload.get("Produkt_und_Dienstleistungsbeschreibung") or ""
     checklist    = payload.get("Checkliste_Landingpage") or ""
     signature    = payload.get("Signatur") or ""
     cta          = payload.get("CTA") or ""
     homepage_url = payload.get("icp_homepage_url") or ""
-    # Tags: in beliebigen Formen erlauben (AI | [AI] | "AI, Sales" | ["AI","Sales"])
     raw_tag      = payload.get("Wholic_tag") or payload.get("Wholix_tag") or "AI"
-    tag_text     = raw_tag  # Mapper parst das robust
 
-    # optional limit aus Payload
+    # Lead-Anzahl
     try:
-        n_leads = int(payload.get("limit", n_leads))
+        n_leads = int(n_leads_ui)
     except Exception:
-        pass
-    n_leads = max(1, n_leads)
+        n_leads = 1
+    n_leads = max(1, min(n_leads, MAX_LEADS))
 
-    total_steps = max(1, n_leads) * 4 + 2  # login + excludes + (lead + email + store)*N
+    # Fortschritt: 1 (Login) + pro Lead 4 Schritte
+    total_steps = (n_leads * 4) + 1
     step = 0
 
-    # 1) Login
+    # 1) Wholix Login
     step += 1
-    _job_emit(job_id, msg="→ Logge bei Wholix ein …", progress=step/total_steps, rows=results)
+    _job_emit(job_id, "→ Wholix Login …", progress=int(step / total_steps * 100))
     try:
         wh_token = wholix_login(wh_email, wh_pass)
-    except Exception as e:
-        _job_finish(job_id, error=f"Wholix-Login: {e}")
+    except Exception:
+        logging.exception(f"[{job_id}] Wholix-Login fehlgeschlagen")
+        _job_finish(job_id, "Wholix-Login fehlgeschlagen (Details in Logs).")
         return
 
-    # 2) Excludes
-    step += 1
-    _job_emit(job_id, msg="→ Lade Exclude-Hashes …", progress=step/total_steps, rows=results)
+    # ======= EXCLUDES LADEN / MERGEN =======================================
+    initial_excludes = set()
     try:
-        excludes = wholix_fetch_excludes(wh_token)
-        _job_emit(job_id, msg=f"   {len(excludes)} Exclude-Hashes gefunden.")
-    except Exception as e:
-        _job_emit(job_id, msg=f"⚠️  Excludes konnten nicht geladen werden: {e} — fahre ohne fort.")
-        excludes = []
+        for x in (payload.get("exclude_ids") or []):
+            if x: initial_excludes.add(str(x).strip())
+    except Exception:
+        pass
 
-    for i in range(1, n_leads + 1):
-        # 3) Lead
-        step += 1
-        _job_emit(job_id, msg=f"→ [{i}/{n_leads}] Fordere Lead an …", progress=step/total_steps, rows=results)
-        try:
-            lead = lead_suggest(token_id, filters, icp_text, excludes)
-        except Exception as e:
-            _job_emit(job_id, msg=f"❌ Lead-Fehler: {e}", progress=step/total_steps, rows=results)
-            continue
+    _job_emit(job_id, f"→ Wholix-Excludes laden … (Seed: {len(initial_excludes)})")
+    try:
+        wh_excludes = wholix_fetch_excludes(wh_token)  # paginiert
+        for row in wh_excludes:
+            ex = (row.get("exclude_hash") or "").strip()
+            if ex:
+                initial_excludes.add(ex)
+        _job_emit(job_id, f"   {len(wh_excludes)} Excludes aus Wholix, gesamt {len(initial_excludes)}")
+    except Exception:
+        logging.exception(f"[{job_id}] Wholix-Excludes laden fehlgeschlagen")
+        _job_emit(job_id, "   Excludes laden übersprungen (Fehler).")
+
+    excludes: List[str] = list(initial_excludes)
+    excludes_set: Set[str] = set(initial_excludes)
+
+    # ---- Helper zum sicheren Auslesen echter Felder (ohne Raten) -----------
+    def pick(d: dict, *keys):
+        for k in keys:
+            if isinstance(d, dict) and k in d and d[k] not in (None, "", [], {}):
+                return d[k]
+        return None
+
+    def norm_tags(raw) -> dict:
+        """
+        Multi-Select für Wholix: {keys:[…],values:[…]}
+        - String "[AI]" -> "AI"
+        - String "AI"   -> "AI"
+        - Liste ["AI","X"] -> entsprechend erweitern
+        """
+        if raw is None:
+            return {"keys": [], "values": []}
+        vals: List[str] = []
+        if isinstance(raw, str):
+            s = raw.strip()
+            # wenn JSON-Array als String übergeben wurde
+            if (s.startswith("[") and s.endswith("]")):
+                try:
+                    arr = json.loads(s)
+                    if isinstance(arr, list):
+                        vals = [str(x).strip() for x in arr if str(x).strip()]
+                    else:
+                        vals = [s.strip("[] ").strip().strip("'\"")]
+                except Exception:
+                    vals = [s.strip("[] ").strip().strip("'\"")]
+            else:
+                vals = [s]
+        elif isinstance(raw, (list, tuple, set)):
+            vals = [str(x).strip() for x in raw if str(x).strip()]
+        else:
+            vals = [str(raw).strip()]
+
+        # leere filtern
+        vals = [v for v in vals if v]
+        return {"keys": vals, "values": vals}
+
+    def norm_departments(v) -> str:
+        if v is None:
+            return None
+        if isinstance(v, (list, tuple, set)):
+            return ", ".join(str(x).strip() for x in v if str(x).strip()) or None
+        s = str(v).strip()
+        if s.startswith("[") and s.endswith("]"):
+            s = s[1:-1].strip().strip("'\"")
+        return s or None
 
-        person = (lead or {}).get("person") or {}
-        company = (lead or {}).get("company") or {}
-        _job_emit(job_id, msg=f"   Lead: {person.get('first_name','?')} {person.get('last_name','?')} @ {company.get('name') or company.get('company_name','?')}")
-
-        # 4) Email draft
-        step += 1
-        _job_emit(job_id, msg="   → Generiere Nachricht …", progress=step/total_steps, rows=results)
-        items = [{"combined_id": lead.get("combined_id"), "company": company, "person": person}]
-        variables = {
-            "Touch_Point": "LinkedIn DM",
-            **({"homepage_url": homepage_url} if homepage_url else {}),
-            "Produkt_und_Dienstleistungsbeschreibung": icp_text,
-            "Checkliste_Landingpage": checklist,
-            "CTA": cta,
-            "Signatur": signature,
-        }
+    def norm_url(u: Any) -> Optional[str]:
+        u = "" if u is None else str(u).strip()
+        if not u:
+            return None
+        if not re.match(r"^[a-zA-Z][a-zA-Z0-9+\-.]*://", u):
+            u = "https://" + u
         try:
-            draft = email_generate(token_id, variables, items)
-        except Exception as e:
-            _job_emit(job_id, msg=f"❌ Email-Generate-Fehler: {e}", progress=step/total_steps, rows=results)
-            continue
+            from urllib.parse import urlparse
+            pr = urlparse(u)
+            if pr.scheme and pr.netloc:
+                return u
+        except Exception:
+            pass
+        return None
 
-        # 5) Store (MAPPER FIXED)
-        step += 1
-        _job_emit(job_id, msg="   → Speichere Kontakt + Nachricht in Wholix …", progress=step/total_steps, rows=results)
+    for i in range(1, n_leads + 1):
         try:
-            record = map_to_wholix_record(lead, draft, tag_text=tag_text)
-            store_res = wholix_store_contact(wh_token, record)
-        except Exception as e:
-            _job_emit(job_id, msg=f"❌ Wholix-Store-Fehler: {e}", progress=step/total_steps, rows=results)
-            continue
+            # 2) Lead holen
+            step += 1
+            _job_emit(job_id, f"→ [{i}/{n_leads}] Lead vorschlagen … (excludes={len(excludes_set)})",
+                      progress=int(step / total_steps * 100))
+            try:
+                lead = suggest_single_lead(token_id, filters, icp_text, excludes)
+            except Exception:
+                logging.exception(f"[{job_id}] Lead-Fehler")
+                _job_emit(job_id, "❌ Lead-Fehler: siehe Logs")
+                continue
 
-        if lead.get("exclude_hash"):
-            excludes.append(lead["exclude_hash"])
+            if not isinstance(lead, dict):
+                _job_emit(job_id, "❌ Ungültige Lead-Struktur")
+                continue
 
-        result_row = {
-            "person": f"{person.get('first_name','')} {person.get('last_name','')}".strip(),
-            "email": (draft.get("email") or {}).get("to") or person.get("email") or "",
-            "company": company.get("name") or company.get("company_name") or "",
-            "subject": (draft.get("email") or {}).get("subject") or "",
-            "stored_ok": bool(store_res),
-        }
-        results.append(result_row)
+            person  = (lead.get("person")  or {})
+            company = (lead.get("company") or {})
+            combined_id = str(lead.get("combined_id") or "").strip()
 
-        # 6) small completion step for this iteration
-        step += 1
-        _job_emit(job_id, msg="   ✓ Fertig für diesen Lead.", progress=step/total_steps, rows=results)
+            if combined_id and combined_id in excludes_set:
+                _job_emit(job_id, f"   ⚠️ Lead übersprungen (bereits excluded): {combined_id}")
+                continue
 
-    _job_finish(job_id, error=None)
+            _job_emit(job_id, f"   Lead: {person.get('first_name','?')} {person.get('last_name','?')} @ {company.get('name') or company.get('company_name','?')}")
+
+            # 3) Nachricht generieren (echte Endpoint-Daten)
+            step += 1
+            _job_emit(job_id, "   → Nachricht generieren …", progress=int(step / total_steps * 100))
+            items = [{"combined_id": combined_id, "company": company, "person": person}]
+            variables = {
+                "Produkt_und_Dienstleistungsbeschreibung": icp_text,
+                "CTA": cta,
+                "Signatur": signature,
+                "Checkliste_Landingpage": checklist,
+                "homepage_url": homepage_url,   # optional
+                "tags": raw_tag,                # kommt als "[AI]" → später normiert
+                "Touch_Point": "LinkedIn DM",   # optional
+            }
+            try:
+                draft = email_generate_async(token_id, variables, items)
+            except Exception:
+                logging.exception(f"[{job_id}] Email-Generate-Fehler")
+                _job_emit(job_id, "❌ Email-Generate-Fehler: siehe Logs")
+                continue
 
-# ================================ Gradio UI ===============================
+            subj_raw = (draft.get("email") or {}).get("subject", "")
+            body_raw = (draft.get("email") or {}).get("body", "")
+            subj_flat = _flatten_text(subj_raw)
+            body_flat = _flatten_text(body_raw)
+
+            _job_emit(job_id, f"   ✉️ Subject: {subj_flat}")
+            _job_emit(job_id, f"   ✉️ Message: {body_flat}")
+
+            # 4) Speichern in Wholix
+            step += 1
+            _job_emit(job_id, "   → Speichere in Wholix …", progress=int(step / total_steps * 100))
+
+            email_to = str(person.get("email") or "").strip()
+            if not email_to:
+                _job_emit(job_id, "   ⚠️ Keine E-Mail vorhanden – Speichern übersprungen.")
+                stored_ok = False
+            else:
+                try:
+                    # --- Felder befüllen NUR wenn real vorhanden -----------------
+                    firstname   = pick(person, "first_name")
+                    lastname    = pick(person, "last_name")
+                    job_title   = pick(person, "job_title", "title", "position")
+                    departments = norm_departments(pick(person, "departments"))
+                    linkedin    = pick(person, "linkedin_url")
+
+                    # Kontakt-Adressdaten: erst person, dann company (falls vorhanden)
+                    phonenumber = pick(person, "phone", "phonenumber") or pick(company, "phone", "company_phone", "phonenumber")
+                    adress      = pick(person, "adress", "address", "street") or pick(company, "adress", "address", "street")
+                    city        = pick(person, "city") or pick(company, "city")
+                    postcode    = pick(person, "postcode", "postal_code", "zip") or pick(company, "postcode", "postal_code", "zip")
+
+                    company_name = pick(company, "name", "company_name")
+                    company_url  = norm_url(pick(company, "url", "website_url", "website", "domain"))
+
+                    # Tags korrekt als Multi-Select (Dropdown-Wert ohne [])
+                    tags_ms = norm_tags(raw_tag)
+
+                    record = {
+                        "firstname":  firstname,
+                        "lastname":   lastname,
+                        "email":      email_to,
+                        "phonenumber": phonenumber,
+                        "adress":     adress,
+                        "city":       city,
+                        "postcode":   postcode,
+
+                        "job_title":  job_title,
+                        "departments": departments,
+                        "linkedin_url": linkedin,
+
+                        "company_name": company_name,
+                        "company_url":  company_url,
+
+                        "message_mail_subject": subj_raw or None,
+                        "message_mail":         body_raw or None,
+                        "message_followup1":    draft.get("followup1") or None,
+                        "message_followup2":    draft.get("followup2") or None,
+
+                        "exclude_hash": combined_id or None,
+
+                        "status_field": {"keys": ["Kontakt aufgenommen"], "values": ["Kontakt aufgenommen"]},
+                        "tags":         tags_ms,
+                    }
+
+                    # Log: kurze Vorschau der wichtigsten Felder
+                    try:
+                        APP_LOG.info(f"Store-> firstname={firstname} lastname={lastname} email={email_to} job_title={job_title} dept={departments} phone={phonenumber} city={city} postcode={postcode} company_url={company_url}")
+                        APP_LOG.info(f"Store-> tags={tags_ms}")
+                    except Exception:
+                        pass
+
+                    store_res = wholix_store_contact(wh_token, record)
+                    stored_ok = bool(store_res)
+                except Exception:
+                    logging.exception(f"[{job_id}] Wholix-Store-Fehler")
+                    _job_emit(job_id, "❌ Wholix-Store-Fehler: siehe Logs")
+                    stored_ok = False
+
+            if combined_id:
+                excludes_set.add(combined_id)
+                excludes.append(combined_id)
+
+            _job_emit(job_id, rows_append={
+                "person":  f"{person.get('first_name','')} {person.get('last_name','')}".strip(),
+                "email":   email_to,
+                "company": company.get("name") or company.get("company_name") or "",
+                "subject": subj_flat,
+                "message": body_flat,   # Body bleibt Body
+                "stored_ok": stored_ok,
+            })
+
+            # 5) Abschluss
+            step += 1
+            _job_emit(job_id, "   ✓ Lead abgeschlossen.", progress=int(min(99, step / total_steps * 100)))
+
+        except Exception:
+            logging.exception(f"[{job_id}] Unerwarteter Fehler im Lead-Durchlauf")
+            _job_emit(job_id, "❌ Unerwarteter Fehler – Details in Logs.")
+
+    _job_emit(job_id, progress=100)
+    _job_finish(job_id, None)
+
+
+# ================================ UI =======================================
 
 def build_ui():
     with gr.Blocks(theme=gr.themes.Soft(), css="""
       .logbox textarea { font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, "Liberation Mono", monospace; font-size: 12.5px; line-height: 1.35; }
     """) as demo:
-        gr.Markdown("## Wholix Lead → Message → Store (robust, Background-Job + Polling)")
-        gr.Markdown("Füge unten deinen **gesamten `curl`** ein (inkl. `X-Token-Id` Header und JSON `--data-raw`), wähle die Lead-Anzahl und klicke **Start**. "
-                    "Die Verarbeitung läuft als Hintergrund-Job weiter – selbst wenn der Browser kurz offline ist. "
-                    "Mit „Aktualisieren“ holst du den aktuellen Status ab.")
+        gr.Markdown("## Wholix Lead → Message → Store (rein asynchron, robust für lange Jobs)")
+        gr.Markdown(
+            "Füge deinen kompletten **`curl`** (mit `X-Token-Id` und JSON `--data-raw`) ein, wähle die Anzahl Leads und klicke **Start**. "
+            "Die Verarbeitung läuft serverseitig weiter – selbst wenn der Browser/Tab schließt. "
+            "Mit **Aktualisieren** holst du den aktuellen Status ab."
+        )
 
         with gr.Row():
-            curl_in = gr.Textbox(label="curl Befehl", placeholder="curl -sS -N -X POST 'https://.../stream' -H 'X-Token-Id: ...' --data-raw '{...}'", lines=12)
+            curl_in = gr.Textbox(
+                label="curl Befehl",
+                placeholder="curl -sS -N -X POST 'https://.../stream' -H 'X-Token-Id: ...' --data-raw '{...}'",
+                lines=12
+            )
         with gr.Row():
             count = gr.Dropdown(choices=[str(x) for x in LEAD_COUNTS], value="1", label="Anzahl Leads")
 
@@ -865,7 +1035,7 @@ def build_ui():
             poll_btn  = gr.Button("🔄 Aktualisieren")
 
         with gr.Row():
-            job_id_tb = gr.Textbox(label="Job-ID", interactive=False)
+            job_id_tb = gr.Textbox(label="Job-ID", interactive=True)
 
         with gr.Row():
             status = gr.Textbox(label="Status / Log", lines=18, interactive=False, elem_classes=["logbox"])
@@ -873,61 +1043,53 @@ def build_ui():
             progress = gr.Slider(label="Progress", minimum=0, maximum=100, value=0, interactive=False)
         with gr.Row():
             out = gr.Dataframe(
-                headers=["person", "email", "company", "subject", "stored_ok"],
+                headers=["person", "email", "company", "subject", "message", "stored_ok"],
                 label="Ergebnisse",
                 interactive=False,
                 wrap=True,
                 row_count=(0, "dynamic"),
-                col_count=(5, "fixed"),
+                col_count=(6, "fixed"),
             )
 
-        # ----- Handlers -----
-
         def start_job(curl_text: str, n: str):
             try:
                 n_int = int(n)
             except Exception:
                 n_int = 1
+            n_int = max(1, min(n_int, MAX_LEADS))
+
             job_id = str(uuid.uuid4())
             _job_init(job_id)
-            _job_emit(job_id, msg=f"Job gestartet: {job_id}")
+            _job_emit(job_id, f"Job gestartet: {job_id}")
 
-            # Start background work
             EXEC.submit(run_pipeline_bg, job_id, curl_text, n_int)
 
-            # Return initial state
             st = JOBS[job_id]
             with st["lock"]:
                 log = "\n".join(st["log"])
-                prog = int(round(st["progress"] * 100))
+                prog = int(st["progress"])
                 rows = st["rows"]
             return log, prog, rows, job_id
 
         def poll_job(job_id: str):
             st = JOBS.get(job_id)
             if not st:
-                return "Unbekannte Job-ID.", 0, []
+                return "Unbekannte oder abgelaufene Job-ID.", 0, []
             with st["lock"]:
                 log = "\n".join(st["log"][-500:])
-                prog = int(round(st["progress"] * 100))
+                prog = int(st["progress"])
                 rows = st["rows"]
             return log, prog, rows
 
-        start_btn.click(
-            start_job,
-            inputs=[curl_in, count],
-            outputs=[status, progress, out, job_id_tb]
-        )
-
-        poll_btn.click(
-            poll_job,
-            inputs=[job_id_tb],
-            outputs=[status, progress, out]
-        )
+        start_btn.click(start_job, inputs=[curl_in, count],
+                        outputs=[status, progress, out, job_id_tb])
+        poll_btn.click(poll_job, inputs=[job_id_tb],
+                       outputs=[status, progress, out])
 
     return demo
 
+# ============================== MAIN =======================================
+
 if __name__ == "__main__":
     app = build_ui()
-    # Du kannst server_name="0.0.0.0" setzen, wenn du LAN-Zugriff brauchst
-    app.launch(share=True, debug=True)
+    app.launch(server_name="0.0.0.0", debug=True, share=True)