Spaces:

metricspace
/

leadgen

Sleeping

App Files Files Community

MichaelWelsch commited on Oct 7, 2025

Commit

b7828ba

verified ·

1 Parent(s): 06f02f3

Update app.py

Browse files

Files changed (1) hide show

app.py +763 -59

app.py CHANGED Viewed

@@ -1,70 +1,774 @@
 import gradio as gr
-from huggingface_hub import InferenceClient
-def respond(
-    message,
-    history: list[dict[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-    hf_token: gr.OAuthToken,
 ):
     """
-    For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
     """
-    client = InferenceClient(token=hf_token.token, model="openai/gpt-oss-20b")
-    messages = [{"role": "system", "content": system_message}]
-    messages.extend(history)
-    messages.append({"role": "user", "content": message})
-    response = ""
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        choices = message.choices
-        token = ""
-        if len(choices) and choices[0].delta.content:
-            token = choices[0].delta.content
-        response += token
-        yield response
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-chatbot = gr.ChatInterface(
-    respond,
-    type="messages",
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
-with gr.Blocks() as demo:
-    with gr.Sidebar():
-        gr.LoginButton()
-    chatbot.render()
 if __name__ == "__main__":
     demo.launch()

+import json
+import math
+import os
+import random
+import re
+import time
+import threading
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any, Dict, List, Optional, Tuple
 import gradio as gr
+import requests
+LEAD_BASE_URL   = "https://wholixleadgenbackend.ngrok.io"
+WHOLIX_BASE_URL = "https://api.wholix.ai"
+# ============================== HTTP helper ===============================
+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]}")
+        self.status = status
+        self.url = url
+        self.body = body
+def _is_json_ct(ct: str) -> bool:
+    return bool(ct and ("application/json" in ct or ct.endswith("+json")))
+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),
+    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:
+                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)
+HDR_AUTH_RE  = re.compile(r"""-H\s+(?P<q>['"])Authorization:\s*Bearer\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()
+    # Some users might pass token in env; if absent, try to look for it inline as 'X-Token-Id: ...'
+    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:
+        # fallback for -d '...'
+        md2 = re.search(r"-d\s+(?P<q>['\"])(?P<body>.*?)(?P=q)", curl_text, re.DOTALL)
+        if md2:
+            body_str = md2.group("body").strip()
+    if not body_str:
+        raise ValueError("Konnte den JSON Body aus dem curl nicht finden (erwarte --data-raw '...').")
+    # The curl example uses single quotes around valid JSON → parse directly
+    try:
+        payload = json.loads(body_str)
+    except json.JSONDecodeError as e:
+        # attempt to unescape smart single quotes or stray CRs
+        candidate = body_str.replace("\r\n", "\n").replace("\r", "\n")
+        payload = json.loads(candidate)
+    if not token_id:
+        # allow token in the payload via "X-Token-Id" or env var
+        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 ==============================
+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 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()
+    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:
+                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
+    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"]
+    t0 = time.time()
+    max_wait = 10 * 60
+    attempt = 0
+    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 key in d and str(d[key]).strip() != "":
+        return d[key]
+    k = next((k for k in d.keys() if 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):
+        try:
+            r = json.loads(r)
+        except Exception:
+            r = {"Text": r}
+    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 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
+    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:
+                        return v
+        return ""
+    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()
+    if not body:
+        body = "Hallo,\n\nkurzer Kontaktaufbau – gerne Austausch, ob wir unterstützen können.\n\nBeste Grüße"
+    to = ""
+    if isinstance(email_obj, dict):
+        to = str(email_obj.get("to") or "").strip()
+    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]:
+    if not items:
+        raise ValueError("items ist erforderlich und muss mind. 1 Lead enthalten.")
+    body = {**variables, "items": items, "item_index": 0}
+    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
+    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)
+    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: JS-kompatible Wholix-Mapping & Store (1:1) ========
+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:
+    """
+    1:1 wie JS filterWholixContactFields:
+    - 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):
+    """
+    1:1 wie JS normalizeWholixDropdown:
+    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) for x in (val.get("keys") or []) if str(x).strip()]
+        vs = [str(x) 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) 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
+def map_to_wholix_record(lead: dict, draft: dict, tag_text: str = "AI") -> dict:
+    """
+    JS mapToWholixContactsPayload → Python
+    """
+    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 = str(p.get("email") or "").strip()
+    if not email:
+        e = ValueError("E-Mail-Adresse fehlt – Wholix benötigt 'email' als Pflichtfeld.")
+        e.name = "ValidationError"
+        raise e
+    depts_raw = p.get("departments")
+    if isinstance(depts_raw, list):
+        departments = ", ".join([str(x) for x in depts_raw if str(x).strip()]) or None
+    else:
+        departments = (str(depts_raw) if depts_raw is not None else None)
+    company_url = (
+        str(c.get("url") or c.get("website") or c.get("domain") or "").strip()
+        or str(ctx.get("url") or "").strip()
+        or None
+    )
+    mail = m.get("email") if isinstance(m, dict) else {}
+    mail_body    = (mail.get("body")    or "").strip() if isinstance(mail, dict) else ""
+    mail_subject = (mail.get("subject") or "").strip() if isinstance(mail, dict) else ""
+    followup1    = m.get("followup1")
+    followup2    = m.get("followup2")
+    street_parts = [c.get("street_name"), c.get("street_number")]
+    street = " ".join([str(x).strip() for x in street_parts if str(x or "").strip()]) or ""
+    street = street or None
+    payload = {
+        "firstname":    p.get("first_name") or None,
+        "lastname":     p.get("last_name")  or None,
+        "email":        email,
+        "adress":       street,                             # (sic)
+        "city":         c.get("city") or None,
+        "postcode":     c.get("zip_code") or None,
+        "phonenumber":  p.get("phone") or None,
+        "job_title":    p.get("job_title") or None,
+        "departments":  departments,
+        "linkedin_url": p.get("linkedin_url") or None,
+        "company_name": c.get("name") or None,
+        "company_url":  company_url,
+        "message_mail":          mail_body or None,
+        "message_mail_subject":  mail_subject or None,
+        "message_followup1":     followup1 or None,
+        "message_followup2":     followup2 or None,
+        "exclude_hash": (lead.get("exclude_hash") or None),
+        "status_field": { "keys": ["Kontakt aufgenommen"], "values": ["Kontakt aufgenommen"] },
+        "tags":         { "keys": [str(tag_text).strip("[]") or "AI"], "values": [str(tag_text).strip("[]") or "AI"] },
+    }
+    normalized = filter_wholix_contact_fields(payload)
+    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_store_contact(token: str, record: dict, module: str = "Contacts") -> dict:
+    """
+    JS wholixStoreContact exakt nachgebildet (mit Fallbacks)
+    """
+    email = str((record or {}).get("email") or "").strip()
+    if not email:
+        e = ValueError("Wholix: 'email' ist Pflichtfeld und darf nicht leer sein.")
+        e.name = "ValidationError"
+        raise e
+    normalized = filter_wholix_contact_fields({ **record, "email": email })
+    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)
+    url = f"{WHOLIX_BASE_URL}/api/v1/table-object-data/store-objects"
+    headers = {"Authorization": f"Bearer {token}"}
+    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
+        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 = ""
+        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
+    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)
+# ======== END DROP-IN =====================================================
+# ====================== Background-Jobs (robust UI) =======================
+LEAD_COUNTS = [1, 2, 3, 5, 10, 20, 50, 100, 200]
+# In-Memory Job Store
+EXEC = ThreadPoolExecutor(max_workers=16)
+JOBS: Dict[str, Dict[str, Any]] = {}  # job_id -> state dict
+def _job_init(job_id: str):
+    JOBS[job_id] = {
+        "log": [],          # List[str]
+        "progress": 0.0,    # 0..1
+        "rows": [],         # Ergebnisse (List[Dict])
+        "done": False,
+        "error": None,      # str | None
+        "lock": threading.RLock(),
+    }
+def _job_emit(job_id: str, msg: str = "", progress: Optional[float] = None, rows: Optional[List[Dict[str, Any]]] = None):
+    st = JOBS.get(job_id)
+    if not st:
+        return
+    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
+def _job_finish(job_id: str, error: Optional[str] = None):
+    st = JOBS.get(job_id)
+    if not st:
+        return
+    with st["lock"]:
+        st["error"] = error
+        st["done"]  = True
+        if error:
+            st["log"].append(f"❌ {error}")
+        else:
+            st["log"].append("✅ Alles erledigt.")
+def run_pipeline_bg(job_id: str, curl_text: str, n_leads: int):
+    """
+    1:1 wie vorherige run_pipeline(), aber ohne yield – stattdessen Status ins JOBS-Store schreiben.
+    """
+    results: List[Dict[str, Any]] = []
+    def log(msg: str):
+        # kleine Bequemlichkeit, auto-emit mit aktuellem Fortschritt
+        st = JOBS.get(job_id, {})
+        prog = st.get("progress", 0.0)
+        _job_emit(job_id, msg=msg, progress=prog, rows=results)
+    # parse curl
+    try:
+        token_id, payload = parse_curl(curl_text)
+    except Exception as e:
+        _job_finish(job_id, error=f"Parse-Fehler: {e}")
+        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.")
+        return
+    filters      = payload.get("filters") or {}
+    icp_text     = payload.get("Produkt_und_Dienstleistungsbeschreibung") or payload.get("icp_text") 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 ""
+    tag_text     = payload.get("Wholic_tag") or payload.get("Wholix_tag") or "AI"
+    total_steps = max(1, n_leads) * 4 + 2  # login + excludes + (lead + email + store)*N
+    step = 0
+    # 1) Login
+    step += 1
+    _job_emit(job_id, msg="→ Logge bei Wholix ein …", progress=step/total_steps, rows=results)
+    try:
+        wh_token = wholix_login(wh_email, wh_pass)
+    except Exception as e:
+        _job_finish(job_id, error=f"Wholix-Login: {e}")
+        return
+    # 2) Excludes
+    step += 1
+    _job_emit(job_id, msg="→ Lade Exclude-Hashes …", progress=step/total_steps, rows=results)
+    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 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
+        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,
+        }
+        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
+        # 5) Store
+        step += 1
+        _job_emit(job_id, msg="   → Speichere Kontakt + Nachricht in Wholix …", progress=step/total_steps, rows=results)
+        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
+        if lead.get("exclude_hash"):
+            excludes.append(lead["exclude_hash"])
+        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)
+        # 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)
+    _job_finish(job_id, error=None)
+# ================================ Gradio 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.")
+        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)
+        with gr.Row():
+            count = gr.Dropdown(choices=[str(x) for x in LEAD_COUNTS], value="1", label="Anzahl Leads")
+        with gr.Row():
+            start_btn = gr.Button("🚀 Start", variant="primary")
+            poll_btn  = gr.Button("🔄 Aktualisieren")
+        with gr.Row():
+            job_id_tb = gr.Textbox(label="Job-ID", interactive=False)
+        with gr.Row():
+            status = gr.Textbox(label="Status / Log", lines=18, interactive=False, elem_classes=["logbox"])
+        with gr.Row():
+            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"],
+                label="Ergebnisse",
+                interactive=False,
+                wrap=True,
+                row_count=(0, "dynamic"),
+                col_count=(5, "fixed"),
+            )
+        # ----- Handlers -----
+        def start_job(curl_text: str, n: str):
+            try:
+                n_int = int(n)
+            except Exception:
+                n_int = 1
+            job_id = str(uuid.uuid4())
+            _job_init(job_id)
+            _job_emit(job_id, msg=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))
+                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, []
+            with st["lock"]:
+                log = "\n".join(st["log"][-500:])
+                prog = int(round(st["progress"] * 100))
+                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]
+        )
+    return demo
 if __name__ == "__main__":
     demo.launch()