"""
dashboard.py -- Lead Enrichment Dashboard
Starten mit: streamlit run dashboard.py --server.port=8500
"""
import sqlite3
import pandas as pd
import streamlit as st
from streamlit_autorefresh import st_autorefresh
from pathlib import Path
DB_PATH = str(Path(__file__).parent / "email_agent.db")
st.set_page_config(page_title="Lead Enrichment", layout="wide", page_icon="L")
# -- Globales CSS --------------------------------------------------------------
st.markdown("""
""", unsafe_allow_html=True)
# -- Auto-Refresh --------------------------------------------------------------
auto_refresh = st.sidebar.checkbox("Auto-Refresh (30s)", value=True)
# Wird weiter unten aktiviert -- nur wenn kein Kontakt offen
# -- Daten laden ---------------------------------------------------------------
@st.cache_data(ttl=28)
def load_contacts(campaign_id: int) -> pd.DataFrame:
conn = sqlite3.connect(DB_PATH)
df = pd.read_sql_query(
"SELECT * FROM cold_contacts WHERE campaign_id = ? ORDER BY CAST(source_id AS INTEGER)",
conn, params=(campaign_id,)
)
conn.close()
return df
@st.cache_data(ttl=28)
def load_campaigns() -> list:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
rows = conn.execute("SELECT * FROM outreach_campaigns ORDER BY id DESC").fetchall()
conn.close()
return [dict(r) for r in rows]
def val(row, key):
v = row.get(key)
return str(v) if (v is not None and str(v).strip() not in ('', 'None', 'nan')) else ''
# -- Sidebar -------------------------------------------------------------------
campaigns = load_campaigns()
campaign_labels = {c["id"]: c["name"] for c in campaigns}
selected_id = st.sidebar.selectbox(
"Kampagne",
options=list(campaign_labels.keys()),
format_func=lambda x: campaign_labels[x]
)
st.sidebar.divider()
ALL_STATUSES = ["raw", "enriched", "needs_manual", "low_data_lead", "active", "replied", "converted"]
status_filter = st.sidebar.multiselect(
"Status", options=ALL_STATUSES,
default=["raw", "enriched", "needs_manual", "low_data_lead"]
)
if st.sidebar.button("Neu laden"):
st.cache_data.clear()
st.rerun()
# -- Daten ---------------------------------------------------------------------
df_all = load_contacts(selected_id)
def count(s):
return len(df_all[df_all["status"] == s])
df = df_all[df_all["status"].isin(status_filter)].copy() if status_filter else df_all.copy()
# -- Kompakte KPI-Leiste -------------------------------------------------------
enriched_n = count("enriched")
raw_n = count("raw")
manual_n = count("needs_manual")
low_n = count("low_data_lead")
sent_n = count("active") + count("replied") + count("converted")
total_n = len(df_all)
pct = f"{enriched_n/max(total_n,1)*100:.0f}%"
st.markdown(f"""
Gesamt: {total_n}
|
Enriched: {enriched_n} ({pct})
|
Raw: {raw_n}
|
Manuell: {manual_n}
|
Low Data: {low_n}
|
Gesendet: {sent_n}
{" ⟳ Auto-Refresh aktiv" if auto_refresh else ""}
""", unsafe_allow_html=True)
# -- Kontakt-Navigation (Session State) ----------------------------------------
DISPLAY_COLS = [
"source_id", "company", "email", "website", "status",
"company_description", "pain_point", "company_size", "headquarters",
"founding_year", "technology_stack", "key_products", "linkedin_url",
"industry", "language", "icp_recommended_product"
]
ENRICHMENT_COLS = {
"company_description", "pain_point", "company_size", "headquarters",
"founding_year", "technology_stack", "key_products", "linkedin_url",
"industry", "language", "icp_recommended_product"
}
existing_cols = [c for c in DISPLAY_COLS if c in df.columns]
df_display = df[existing_cols].copy()
contact_ids = df_all["source_id"].tolist()
if "contact_idx" not in st.session_state:
st.session_state.contact_idx = 0
# Sidebar Kontakt-Auswahl
contact_options_list = ["-- keiner --"] + [
f"[{r['source_id']}] {r['company']} ({r['status']})"
for _, r in df_display.iterrows()
]
contact_ids_list = ["-- keiner --"] + df_display["source_id"].tolist()
sidebar_sel = st.sidebar.selectbox(
"Kontakt Detail", options=range(len(contact_ids_list)),
format_func=lambda i: contact_options_list[i]
)
# Auto-Refresh: nur aktiv wenn kein Kontakt offen
contact_selected = sidebar_sel > 0
if auto_refresh and not contact_selected:
st_autorefresh(interval=30000, key="autorefresh")
st.sidebar.caption("Refresh aktiv (30s)")
elif auto_refresh and contact_selected:
st.sidebar.caption("Refresh pausiert (Kontakt offen)")
# -- Tabs ----------------------------------------------------------------------
tab1, tab2, tab3, tab4 = st.tabs([f"Tabelle ({len(df_display)})", "Detail-Ansicht", "Lead Finder", "Enrichment"])
# ---- TAB 1: TABELLE ----------------------------------------------------------
with tab1:
def style_df(df):
styled = pd.DataFrame("", index=df.index, columns=df.columns)
for col in df.columns:
if col in ENRICHMENT_COLS:
styled[col] = df[col].apply(
lambda v: "background-color: #E2EFDA" if (pd.notna(v) and str(v).strip() not in ('', 'None', 'nan'))
else "background-color: #FFE0E0"
)
return styled
st.dataframe(
df_display.style.apply(style_df, axis=None),
use_container_width=True, height=560
)
st.caption("Gruen = Enrichment vorhanden | Rot = leer | Kein Hintergrund = Original-Felder")
# ---- TAB 2: DETAIL-ANSICHT ---------------------------------------------------
with tab2:
selected_source_id = contact_ids_list[sidebar_sel] if sidebar_sel > 0 else "-- keiner --"
if selected_source_id == "-- keiner --":
st.info("Waehle links in der Sidebar einen Kontakt aus.")
else:
contact_row = df_all[df_all["source_id"] == selected_source_id]
if contact_row.empty:
st.warning("Kontakt nicht gefunden.")
else:
c = contact_row.iloc[0]
# Navigation
all_ids = df_display["source_id"].tolist()
cur_idx = all_ids.index(selected_source_id) if selected_source_id in all_ids else 0
total_shown = len(all_ids)
nav1, nav2, nav3 = st.columns([1, 1, 8])
with nav1:
if st.button("←", key="prev") and cur_idx > 0:
prev_id = all_ids[cur_idx - 1]
st.session_state["_nav_id"] = prev_id
with nav2:
if st.button("→", key="next") and cur_idx < total_shown - 1:
next_id = all_ids[cur_idx + 1]
st.session_state["_nav_id"] = next_id
nav3.markdown(f"{cur_idx+1} / {total_shown}
", unsafe_allow_html=True)
# Ueberschreibe Kontakt wenn navigiert
if "_nav_id" in st.session_state:
nav_row = df_all[df_all["source_id"] == st.session_state["_nav_id"]]
if not nav_row.empty:
c = nav_row.iloc[0]
# Status Badge
status = val(c, "status")
badge_class = f"badge-{status}" if status in ["enriched","raw","needs_manual","low_data_lead","active"] else "badge-raw"
# Info-Header
website = val(c, "website") or ""
email = val(c, "email") or ""
hq = val(c, "headquarters") or ""
company = val(c, "company") or ""
src_id = val(c, "source_id") or ""
web_link = f'{website.replace("https://","").replace("http://","")}' if website else "—"
hq_str = f"| {hq}" if hq else ""
st.markdown(f"""
""", unsafe_allow_html=True)
# Basis-Grid
industry = val(c, "industry") or "—"
size = val(c, "company_size") or "—"
founded = val(c, "founding_year") or "—"
language = val(c, "language") or "—"
linkedin = val(c, "linkedin_url") or ""
tech = val(c, "technology_stack") or ""
role = val(c, "role") or "—"
li_link = f'LinkedIn' if linkedin else "—"
tech_badges = " ".join(
f'{t.strip()}'
for t in tech.replace("[","").replace("]","").replace('"','').split(",")
if t.strip() and t.strip() not in ('','None')
) if tech else "—"
st.markdown(f"""
LinkedIn: {li_link} | Rolle: {role} | Tech: {tech_badges}
""", unsafe_allow_html=True)
# 4-Karten Layout
desc = val(c, "company_description") or ""
products = val(c, "key_products") or ""
pain = val(c, "pain_point") or ""
icp_prod = val(c, "icp_recommended_product") or ""
icp_val = val(c, "icp") or ""
icp_conf = val(c, "icp_confidence") or ""
def card(title, content, empty_text="Keine Daten"):
body = f'{content}
' if content else f'{empty_text}
'
return f'{title}
{body}'
col_a, col_b = st.columns(2)
with col_a:
st.markdown(card("Ueber das Unternehmen", desc), unsafe_allow_html=True)
st.markdown(card("Pain Point", pain), unsafe_allow_html=True)
with col_b:
st.markdown(card("Produkte & Loesungen", products), unsafe_allow_html=True)
icp_content = f"{icp_prod}
ICP: {icp_val} | Konfidenz: {icp_conf}" if (icp_prod or icp_val) else ""
st.markdown(card("ICP / Empfehlung", icp_content), unsafe_allow_html=True)
# Notizen
notes = val(c, "notes") or ""
our_offer = val(c, "our_offer") or ""
enr_src = val(c, "enrichment_source") or ""
enr_date = val(c, "enrichment_date") or ""
if enr_date and len(enr_date) > 10:
enr_date = enr_date[:10]
st.markdown(f"""
Quelle: {enr_src or '—'} | Datum: {enr_date or '—'}
""", unsafe_allow_html=True)
if notes:
with st.expander("Notizen"):
st.write(notes)
if our_offer:
with st.expander("Unser Angebot"):
st.write(our_offer)
# ---- TAB 3: LEAD FINDER ------------------------------------------------------
with tab3:
import json, subprocess
from pathlib import Path
clients_dir = Path(__file__).parent / 'clients'
st.markdown("### Lead Finder")
# Client-Configs laden
client_files = list(clients_dir.glob('*.json')) if clients_dir.exists() else []
clients_data = {}
for cf in client_files:
try:
with open(cf, encoding='utf-8') as f:
cfg = json.load(f)
clients_data[cf.stem] = cfg
except Exception:
pass
if not clients_data:
st.warning("Keine Client-Configs gefunden.")
else:
cfg_col, filter_col = st.columns([1, 2])
with cfg_col:
st.markdown("**Mandant & Lauf**")
client_options = {k: v['name'] for k, v in clients_data.items()}
selected_client = st.selectbox(
"Mandant",
options=list(client_options.keys()),
format_func=lambda x: client_options[x]
)
cfg = clients_data[selected_client]
bis_ende = st.checkbox("Bis keine mehr gefunden", value=False)
limit = 9999 if bis_ende else int(st.number_input(
"Max. neue Kontakte", min_value=10, max_value=500, value=50, step=10
))
st.markdown(f"""
Kampagne: {cfg.get('campaign_id','—')} |
ICP: {cfg.get('icp','—')}
Absender: {cfg.get('sender_name','—')} | {cfg.get('company_name','—')}
""", unsafe_allow_html=True)
with filter_col:
st.markdown("**Suchfilter**")
f1, f2 = st.columns(2)
with f1:
regionen_input = st.text_input(
"Region / Ort",
value="Mallorca" if cfg.get('icp') == 'LP' else "",
placeholder="z.B. Mallorca, Bayern, Madrid"
)
regionen = [r.strip() for r in regionen_input.split(',') if r.strip()]
sprache = st.selectbox(
"Suchsprache",
options=["Deutsch", "Spanisch", "Beides"],
index=2 if cfg.get('icp') == 'LP' else 0
)
with f2:
alle_branchen = cfg.get('target_industries', [])
branchen_sel = st.multiselect(
"Branchen aus Liste (leer = alle)",
options=alle_branchen,
default=[]
)
eigene_branchen_input = st.text_input(
"Eigene Branchen (kommagetrennt)",
placeholder="z.B. Schwimmbad, Spa, Altenheim"
)
eigene_branchen = [b.strip() for b in eigene_branchen_input.split(',') if b.strip()]
zusatz = st.text_input(
"Zusaetzlicher Suchbegriff (optional)",
placeholder="z.B. email kontakt, correo contacto"
)
# Queries live aufbauen und anzeigen
branchen = (branchen_sel if branchen_sel else alle_branchen) + eigene_branchen
orte = regionen if regionen else [""]
preview_queries = []
for branche in branchen:
for ort in orte:
ort_str = f" {ort}" if ort else ""
z_str = f" {zusatz}" if zusatz else ""
if sprache in ("Deutsch", "Beides"):
preview_queries.append(f"{branche}{ort_str} email kontakt{z_str}")
if sprache in ("Spanisch", "Beides"):
preview_queries.append(f"{branche}{ort_str} email contacto correo{z_str}")
with st.expander(f"Vorschau: {len(preview_queries)} Suchqueries", expanded=False):
for q in preview_queries[:30]:
st.markdown(f"- `{q}`")
if len(preview_queries) > 30:
st.caption(f"... und {len(preview_queries)-30} weitere")
st.divider()
start_btn = st.button("Lead Finder starten", type="primary")
if start_btn:
if not preview_queries:
st.error("Keine Suchqueries generiert. Bitte Branche oder Region eingeben.")
else:
filter_file = "/tmp/lf_active_filter.json"
with open(filter_file, 'w', encoding='utf-8') as fh:
json.dump({"queries": preview_queries, "limit": limit}, fh, ensure_ascii=False)
st.info(f"Suche laeuft: **{cfg['name']}** | {len(preview_queries)} Queries | max. {limit if limit < 9999 else 'unbegrenzt'} Kontakte")
log_placeholder = st.empty()
script = (
f"cd /app && "
f"python3 lead_finder.py --client {selected_client} --filter-file {filter_file} 2>&1"
)
try:
result = subprocess.run(
['bash', '-c', script],
capture_output=True, text=True, timeout=600
)
output = result.stdout + result.stderr
log_placeholder.code(output[-4000:] if len(output) > 4000 else output)
st.cache_data.clear()
st.success("Fertig!")
except subprocess.TimeoutExpired:
st.warning("Timeout nach 10 Minuten.")
except Exception as e:
st.error(f"Fehler: {e}")
# ---- TAB 4: ENRICHMENT -------------------------------------------------------
with tab4:
import subprocess as _sp
st.markdown("### Enrichment")
st.markdown("Analysiert alle **raw** Kontakte: Website scrapen, Branche, Schmerzpunkt und Firmenbeschreibung via KI ermitteln.")
raw_count = count("raw")
manual_count = count("needs_manual")
col_e1, col_e2 = st.columns(2)
with col_e1:
st.markdown(f"""
Raw (bereit): {raw_count}
Manuell noetig: {manual_count}
Kampagne: {selected_id}
""", unsafe_allow_html=True)
st.markdown("")
enrich_btn = st.button("Enrichment starten", type="primary", disabled=(raw_count == 0 and manual_count == 0))
icp_btn = st.button("ICP-Klassifikation starten", disabled=(enriched_n == 0))
with col_e2:
st.markdown("**Ablauf:**")
st.markdown("""
1. **Enrichment** — scrapet Websites, KI analysiert Branche & Schmerzpunkt
2. **ICP-Klassifikation** — ordnet jeden Kontakt LP (Lueftung) oder AG (KI) zu
3. Danach: Kontakte erscheinen mit allen Infos in der Tabelle
""")
if enrich_btn:
st.info(f"Enrichment laeuft fuer Kampagne {selected_id} ({raw_count + manual_count} Kontakte)...")
log_e = st.empty()
try:
res = _sp.run(
['bash', '-c', f'cd /app && python3 multi_source_enricher.py --campaign {selected_id} 2>&1'],
capture_output=True, text=True, timeout=1800
)
output = res.stdout + res.stderr
log_e.code(output[-4000:] if len(output) > 4000 else output)
st.cache_data.clear()
st.success("Enrichment abgeschlossen!")
except _sp.TimeoutExpired:
st.warning("Timeout nach 30 Minuten. Laeuft evtl. noch im Hintergrund.")
except Exception as e:
st.error(f"Fehler: {e}")
if icp_btn:
st.info(f"ICP-Klassifikation laeuft fuer Kampagne {selected_id}...")
log_icp = st.empty()
try:
res = _sp.run(
['bash', '-c', f'cd /app && python3 icp_classifier.py --campaign {selected_id} 2>&1'],
capture_output=True, text=True, timeout=600
)
output = res.stdout + res.stderr
log_icp.code(output[-4000:] if len(output) > 4000 else output)
st.cache_data.clear()
st.success("ICP-Klassifikation abgeschlossen!")
except Exception as e:
st.error(f"Fehler: {e}")