import json
import sqlite3
import html
from pathlib import Path
import pandas as pd
import streamlit as st
from v2_schema import init_schema, sync_projects
from prospect_store_v2 import init as init_prospects
DB_PATH = Path(__file__).parent / "email_agent.db"
st.set_page_config(page_title="Lead Engine 2.0", layout="wide", page_icon="L")
# Kompakter Seitenkopf: Lead Engine klar am oberen Seitenrand verankern.
st.markdown("""
<style>
.block-container {padding-top: 1.35rem !important;}
h1 {margin-top: 0 !important; margin-bottom: .15rem !important;}
div[data-testid="stCaptionContainer"] {margin-bottom: .45rem;}
div[data-testid="stButton"] button p {text-align:left !important; width:100%;}
div[data-testid="stButton"] button {justify-content:flex-start !important; text-align:left !important;}
div[data-testid="stButton"] button > div {justify-content:flex-start !important; width:100% !important;}
div[data-testid="stButton"] button p {text-align:left !important; width:100% !important; margin-left:0 !important;}
/* Kompakter Produkt-/Zielmarktkatalog */
div[data-testid="stHorizontalBlock"]:has(button[kind="secondary"]) {align-items:center;}
div[data-testid="stMetric"] {padding-top:0 !important; padding-bottom:0 !important;}
div[data-testid="stMetric"] label {margin-bottom:0 !important;}
div[data-testid="stMetricValue"] {line-height:1.05 !important;}
div[data-testid="stVerticalBlock"] > div:has(button[kind="secondary"]) {margin-top:0 !important; margin-bottom:0 !important;}
hr {margin-top:.35rem !important; margin-bottom:.35rem !important;}
</style>
""", unsafe_allow_html=True)
# Bei einem echten UI-Update einmalig an den Seitenanfang springen.
LEAD_ENGINE_VERSION = "2.2.0"
LEAD_ENGINE_UI_VERSION = LEAD_ENGINE_VERSION
try:
import streamlit.components.v1 as components
if st.session_state.get("ui_version_seen") != LEAD_ENGINE_UI_VERSION:
components.html("<script>window.parent.scrollTo({top:0,left:0,behavior:'instant'});</script>", height=0)
st.session_state["ui_version_seen"] = LEAD_ENGINE_UI_VERSION
except Exception:
pass
init_schema(DB_PATH)
init_prospects(DB_PATH)
sync_projects(DB_PATH)
head_l, head_r = st.columns([8, 1.35], vertical_alignment="center")
with head_l:
st.title(f"Lead Engine 2.0 · v{LEAD_ENGINE_VERSION}")
with head_r:
if st.button("Aktualisieren", key="refresh_lead_engine", use_container_width=True):
st.rerun()
st.caption("Agent Solutions · projektgetriebene Lead-Recherche, Bewertung und Human-Gate")
def company_scale_summary(lead):
import re
parts = []
texts = []
for field in ('company_size','need_signals','fit_rationale','sources_json'):
val = lead.get(field) if hasattr(lead, 'get') else None
if val:
texts.append(str(val))
text = " ".join(texts)
low = text.lower()
def clean_num(raw):
digits = re.sub(r"[^0-9]", "", raw or "")
return int(digits) if digits else None
# 1) Mitarbeiterzahl -> vereinbarte Größenklasse.
people_patterns = [
r"(?i)(?:über|mehr als|ca\.?|rund|etwa)?\s*([0-9][0-9\. ,]*)\+?\s*(?:mitarbeitende|mitarbeiter|beschäftigte|employees|people|experts|professionals|staff)",
r"(?i)(?:mitarbeitende|mitarbeiter|beschäftigte|employees|people|experts|professionals|staff)\s*[:\-]?\s*(?:über|mehr als|ca\.?|rund|etwa)?\s*([0-9][0-9\. ,]*)",
]
employee_found = False
for pat in people_patterns:
m = re.search(pat, text)
if m:
n = clean_num(m.group(1))
if n:
employee_found = True
if n <= 50: parts.append('/ bis 50')
elif n <= 250: parts.append('/ 51–250')
elif n <= 1000: parts.append('/ 251–1.000')
else: parts.append('/ über 1.000')
break
# 2) Umsatz, falls vorhanden.
revenue_patterns = [
r"(?i)(?:umsatz|revenue|turnover|sales)\D{0,25}(?:EUR|€)?\s*([0-9]+(?:[\.,][0-9]+)?)\s*(?:mio\.?|million|m\b)",
r"(?i)(?:EUR|€)\s*([0-9]+(?:[\.,][0-9]+)?)\s*(?:mio\.?|million|m\b)\D{0,25}(?:umsatz|revenue|turnover|sales)",
r"(?i)([0-9]+(?:[\.,][0-9]+)?)\s*(?:mio\.?|million|m\b)\s*(?:EUR|€)?\D{0,25}(?:umsatz|revenue|turnover|sales)",
]
for pat in revenue_patterns:
m = re.search(pat, text)
if m:
val = m.group(1).replace('.', ',')
parts.append(f'{val} Mio. € Umsatz')
break
# Milliarden-Umsatz/Volumen separat.
if not any('Umsatz' in x for x in parts):
m = re.search(r"(?i)(?:umsatz|revenue|turnover|sales|revenue handled)\D{0,25}(?:EUR|€)?\s*([0-9]+(?:[\.,][0-9]+)?)\s*(?:billion|mrd\.?|bn\b|b\b)", text)
if m:
parts.append(f"{m.group(1).replace('.', ',')} Mrd. € Umsatz")
# 3) Ein weiterer aussagekräftiger Betriebsindikator.
metrics = [
(r"(?i)\b([0-9]{1,4})\+?\s*(?:hotels?|häuser|haeuser|establishments?)\b", 'Hotels', 5000),
(r"(?i)\b([0-9]{1,5})\+?\s*(?:rooms?|zimmer)\b", 'Zimmer', 50000),
(r"(?i)\b([0-9]{1,4})\+?\s*(?:offices?|büros|bueros)\b", 'Büros', 5000),
(r"(?i)\b([0-9]{1,5})\+?\s*(?:stores?|filialen|branches|standorte)\b", 'Standorte', 50000),
(r"(?i)\b([0-9]{1,7})\+?\s*(?:properties|immobilien|objekte)\b", 'Objekte', 9999999),
(r"(?i)\b([0-9]{1,6})\+?\s*(?:vehicles|fahrzeuge|cars|autos)\b", 'Fahrzeuge', 999999),
]
metric_added = False
for pat, label, max_n in metrics:
m = re.search(pat, text)
if m:
n = int(m.group(1))
if 0 < n <= max_n:
shown = f"{n:,}".replace(',', '.')
plus = '+' if '+' in m.group(0) else ''
parts.append(f'{shown}{plus} {label}')
metric_added = True
break
# Ausgeschriebene kleine Zahlen wie "vier Hotels" ebenfalls erkennen.
if not metric_added:
word_nums = {'ein':1,'eine':1,'einen':1,'zwei':2,'drei':3,'vier':4,'fünf':5,'fuenf':5,'sechs':6,'sieben':7,'acht':8,'neun':9,'zehn':10,'elf':11,'zwölf':12,'zwoelf':12}
m = re.search(r"(?i)\b(ein|eine|einen|zwei|drei|vier|fünf|fuenf|sechs|sieben|acht|neun|zehn|elf|zwölf|zwoelf)\s+(?:[^.;,]{0,20}\s)?(?:hotels?|häuser|haeuser)\b", text)
if m:
parts.append(f"{word_nums[m.group(1).lower()]} Hotels")
# 4) Kunden-/Gäste-/Reservierungsvolumen, falls noch Platz.
volume_patterns = [
(r"(?i)([0-9]+(?:[\.,][0-9]+)?)\s*(?:mio\.?|million)\s*(?:clients?|kunden|gäste|gaeste)", 'Mio. Kunden/Jahr'),
(r"(?i)([0-9]+(?:[\.,][0-9]+)?)\s*(?:mio\.?|million)\s*(?:reservations?|reservierungen)", 'Mio. Reservierungen/Jahr'),
(r"(?i)([0-9]+(?:[\.,][0-9]+)?)\s*(?:mio\.?|million)\s*(?:roomnights?|zimmernächte|zimmernaechte)", 'Mio. Zimmernächte/Jahr'),
]
if len(parts) < 3:
for pat, label in volume_patterns:
m = re.search(pat, text)
if m:
parts.append(f"{m.group(1).replace('.', ',')} {label}")
break
# Wenn keine Mitarbeiterzahl vorhanden war, alternativen Größenindikator zuerst zeigen.
if not parts:
return '/ ?'
return ' · '.join(parts[:3])
def mallorca_region(lead):
text = " ".join(str((lead.get(k) if hasattr(lead,'get') else '') or '') for k in ('location','company_size','fit_rationale','sources_json')).lower()
# Inselweit, wenn klar mehrere Regionen/Orte genannt werden.
islandwide_markers = ['multiple mallorca', 'mallorca-wide', 'across mallorca', 'throughout the island', 'island-wide', 'inselweit']
if any(x in text for x in islandwide_markers):
return 'Inselweit'
regions = [
('Ostmallorca', ['cala millor','sa coma','s\'illot','s’illot','porto cristo','manacor','cala ratjada','capdepera','arta','artà','canyamel','cala bona','calas de mallorca','porto colom','portocolom','cala d’or','cala dor']),
('Nordmallorca', ['alcudia','alcúdia','port d\'alcudia','puerto de alcudia','pollensa','pollença','port de pollenca','port de pollença','can picafort','playa de muro','muro']),
('Südmallorca', ['llucmajor','el arenal','s\'arenal','s’arenal','colonia de sant jordi','ses salines','campos','santanyi','santanyí']),
('Südwestmallorca', ['calvia','calvià','palmanova','magaluf','santa ponsa','portals nous','puerto portals','andratx','port d\'andratx','puerto de andratx','paguera','peguera']),
('Westmallorca', ['soller','sóller','port de soller','port de sóller','deia','deià','valldemossa','banyalbufar','esporles']),
('Palma', ['palma de mallorca','palma, mallorca','0700','0701','0702','0703','0704','0705','0706','0707','0708','0709']),
('Zentralmallorca', ['inca','binissalem','sineu','santa maria del cami','santa maria del camí','alaro','alaró','marratxi','marratxí']),
]
hits=[]
for region, markers in regions:
if any(m in text for m in markers):
hits.append(region)
if len(set(hits)) >= 2:
return 'Inselweit'
if hits:
return hits[0]
if 'mallorca' in text:
return 'Mallorca'
return ''
def query(sql: str, params=()) -> pd.DataFrame:
with sqlite3.connect(DB_PATH) as conn:
return pd.read_sql_query(sql, conn, params=params)
projects = query("SELECT * FROM product_catalog ORDER BY potential_score DESC, title")
prospects = query("SELECT * FROM prospects_v2 ORDER BY created_at DESC")
companies = query("SELECT * FROM companies_v2 ORDER BY canonical_name")
company_matches = query("""
SELECT m.*, pp.title AS primary_project_title
FROM company_project_matches_v2 m
JOIN company_project_matches_v2 pm ON pm.company_id=m.company_id AND pm.is_primary=1
JOIN product_catalog pp ON pp.project_id=pm.project_id
""")
drafts = query("SELECT * FROM email_drafts_v2 ORDER BY updated_at DESC")
reviews = query("SELECT * FROM email_evaluations_v2 ORDER BY updated_at DESC")
matches = query("""
SELECT m.*, c.company, c.email, c.website, p.title AS project_title
FROM lead_project_matches m
LEFT JOIN cold_contacts c ON c.id=m.contact_id
LEFT JOIN product_catalog p ON p.project_id=m.project_id
ORDER BY m.lead_score DESC
""")
approved_count = int((prospects["human_gate"] == "approved").sum()) if not prospects.empty else 0
if "main_view" not in st.session_state:
st.session_state["main_view"] = "projects"
nav1, nav2, nav3, nav4 = st.columns(4)
if nav1.button(f"Kundenprojekte {len(projects)}", key="nav_projects", use_container_width=True):
st.session_state["main_view"] = "projects"
st.rerun()
if nav2.button(f"Kundendatenbank {len(companies)}", key="nav_database", use_container_width=True):
st.session_state["main_view"] = "database"
st.rerun()
if nav3.button(f"Projektzuordnungen {len(company_matches)}", key="nav_matches", use_container_width=True):
st.session_state["main_view"] = "scoring"
st.rerun()
if nav4.button(f"Freigaben {approved_count}", key="nav_approvals", use_container_width=True):
st.session_state["main_view"] = "scoring"
st.rerun()
view = st.session_state["main_view"]
if view == "projects":
st.markdown("## Aktueller Produkt- und Zielmarktkatalog")
st.caption(f"{len(projects)} Kundenprojekte · jedes Projekt mit eigenem Lead-Pool, eigener Bewertung und eigener Ansprache")
st.divider()
if projects.empty:
st.info("Noch keine Kundenprojekte vorhanden.")
else:
for _, project in projects.iterrows():
project_id = project["project_id"]
lead_count = int((prospects["project_id"] == project_id).sum()) if not prospects.empty else 0
cols = st.columns([5, 1.2, 1.2, 1.6])
cols[0].markdown(f"**{project['title']}** \
{project['status']} · Potenzial {project['potential_score']} · Konfidenz {project['confidence']}")
cols[1].metric("Leads", lead_count)
cols[2].metric("Potenzial", project["potential_score"])
if cols[3].button("Leads ansehen", key=f"show_{project_id}", use_container_width=True):
current = st.session_state.get("selected_project_id")
st.session_state["selected_project_id"] = None if current == project_id else project_id
st.rerun()
if st.session_state.get("selected_project_id") == project_id:
st.markdown(f"### Leads für: {project['title']}")
project_leads = prospects[prospects["project_id"] == project_id].copy()
if project_leads.empty:
st.info("Für dieses Projekt sind noch keine Leads vorhanden.")
else:
for _, lead in project_leads.iterrows():
lead_id = int(lead["id"])
link = company_matches[company_matches["prospect_id"] == lead_id]
is_primary = bool(int(link.iloc[0]["is_primary"])) if not link.empty else True
primary_title = str(link.iloc[0]["primary_project_title"]) if not link.empty else project['title']
company_id = int(lead.get("company_id")) if pd.notna(lead.get("company_id")) else None
master_rows = companies[companies["id"] == company_id] if company_id is not None else companies.iloc[0:0]
master = master_rows.iloc[0] if not master_rows.empty else lead
company_name = master.get("canonical_name") if hasattr(master, "get") and master.get("canonical_name") else lead['company']
if not is_primary:
alt_cols = st.columns([5, 4])
alt_cols[0].markdown(f"**{company_name}**")
alt_cols[1].markdown(f"<div style='white-space:nowrap;padding-top:0.15rem'>Hinweis: vollständig beschrieben unter <b>{primary_title}</b></div>", unsafe_allow_html=True)
continue
row_cols = st.columns([5, 2.4, 1.6])
if row_cols[0].button(str(company_name), key=f"lead_{project_id}_{lead_id}", use_container_width=True):
current = st.session_state.get("selected_lead_id")
st.session_state["selected_lead_id"] = None if current == lead_id else lead_id
st.rerun()
row_cols[1].markdown(f"<div style='text-align:left;padding-top:0.55rem;font-weight:700;white-space:nowrap;overflow:hidden;text-overflow:ellipsis'>{company_scale_summary(master)} · <span style='font-weight:500'>{mallorca_region(master)}</span></div>", unsafe_allow_html=True)
if st.session_state.get("selected_lead_id") == lead_id:
with st.container(border=True):
head_cols = st.columns([5, 2.4, 1.6])
head_cols[0].markdown(f"#### {company_name}")
head_cols[1].markdown(f"<div style='text-align:left;padding-top:0.25rem;font-weight:700;white-space:nowrap;overflow:hidden;text-overflow:ellipsis'>{company_scale_summary(master)} · <span style='font-weight:500'>{mallorca_region(master)}</span></div>", unsafe_allow_html=True)
st.markdown(f"**Branche:** {master.get('industry') or lead.get('industry') or '—'}")
st.markdown(f"**Website:** {master.get('website') or lead.get('website') or '—'}")
st.markdown(f"**Kontakt:** {master.get('contact_page') or master.get('email') or master.get('phone') or lead.get('contact_page') or lead.get('email') or lead.get('phone') or '—'}")
st.markdown(f"**Primärprojekt:** {project['title']}")
st.markdown(f"**Warum interessant:** {lead.get('fit_rationale') or '—'}")
signals = lead.get('need_signals')
if signals:
try:
items = json.loads(signals)
if items:
st.markdown("**Bedarfssignale:**")
for item in items:
st.markdown(f"- {item}")
except Exception:
st.markdown(f"**Bedarfssignale:** {signals}")
sources = lead.get('sources_json')
if sources:
try:
srcs = json.loads(sources)
if srcs:
st.markdown("**Quellen:**")
for src in srcs:
src_label = src.get('claim') or src.get('source_type') or src.get('url')
st.markdown(f"- [{src_label}]({src.get('url')})")
except Exception:
pass
st.markdown("### Individuelle Erstkontakt-E-Mail")
lead_draft = drafts[(drafts["prospect_id"] == lead_id) & (drafts["status"] != "blocked_alternate_project")] if not drafts.empty else drafts
if lead_draft.empty:
st.info("Der individuelle E-Mail-Entwurf wird von Hermes vorbereitet.")
else:
d = lead_draft.iloc[0]
st.markdown(f"**Betreff:** {d['subject']}")
st.text_area("E-Mail-Entwurf", value=d['body'], height=320, key=f"draft_{lead_id}")
st.caption("Entwurf zur Prüfung · kein automatischer Versand")
lead_review = reviews[reviews["prospect_id"] == lead_id] if not reviews.empty else reviews
if not lead_review.empty:
r = lead_review.iloc[0]
st.markdown("### Kritische Empfängerbewertung")
st.metric("Gesamtscore", f"{r['overall_score']:.1f}/10" if r['overall_score'] is not None else "—")
st.markdown(f"**Urteil:** {r['verdict'] or '—'}")
st.markdown(f"**Stärkster Punkt:** {r['strongest_point'] or '—'}")
st.markdown(f"**Größtes Risiko:** {r['biggest_risk'] or '—'}")
st.markdown(f"**Verbesserung:** {r['improvement'] or '—'}")
st.markdown(f"**Würde ich antworten?** {r['would_reply'] or '—'} – {r['reply_reason'] or ''}")
st.divider()
elif view == "database":
st.subheader("Kundendatenbank")
st.caption(f"{len(companies)} eindeutige Firmen · {len(company_matches)} Projektzuordnungen · zentrale Firmenstammdaten")
if companies.empty:
st.info("Noch keine Firmen in der Kundendatenbank vorhanden.")
else:
project_names = dict(zip(projects["project_id"], projects["title"]))
db_rows = []
for _, company in companies.iterrows():
cid = int(company["id"])
cm = company_matches[company_matches["company_id"] == cid].copy()
primary = cm[cm["is_primary"] == 1]
primary_title = primary.iloc[0]["primary_project_title"] if not primary.empty else "—"
other_titles = []
for _, mr in cm[cm["is_primary"] != 1].iterrows():
title = project_names.get(mr["project_id"], mr["project_id"])
if title not in other_titles:
other_titles.append(title)
db_rows.append({
"Firma": company.get("canonical_name") or "—",
"Branche": company.get("industry") or "—",
"Region": mallorca_region(company) or company.get("location") or "—",
"Größe": company_scale_summary(company),
"Website": company.get("website") or "—",
"Kontakt": company.get("email") or company.get("phone") or company.get("contact_page") or "—",
"Primärprojekt": primary_title,
"Weitere passende Projekte": " · ".join(other_titles) if other_titles else "—",
"Projektmatches": len(cm),
})
customer_db = pd.DataFrame(db_rows).sort_values("Firma", key=lambda x: x.str.lower())
search_term = st.text_input("Firma suchen", placeholder="Name, Branche, Region oder Projekt …", key="customer_db_search")
if search_term:
mask = customer_db.astype(str).apply(lambda col: col.str.contains(search_term, case=False, na=False)).any(axis=1)
customer_db = customer_db[mask]
st.caption(f"Angezeigt: {len(customer_db)} Firmen")
# Vollbreiten-Tabelle ohne horizontales Scrollen: feste Spaltenanteile,
# lange Inhalte umbrechen innerhalb der Zelle statt die Tabelle zu verbreitern.
widths = {
"Firma": 14, "Branche": 13, "Region": 8, "Größe": 11,
"Website": 7, "Kontakt": 10, "Primärprojekt": 14,
"Weitere passende Projekte": 20, "Projektmatches": 3,
}
headers = {"Projektmatches": "Anz."}
cols_order = list(widths.keys())
table_html = [
"<div class='customer-db-wrap'><table class='customer-db-table'><colgroup>",
*[f"<col style='width:{widths[c]}%'>" for c in cols_order],
"</colgroup><thead><tr>",
*[f"<th title='{html.escape(c)}'>{html.escape(headers.get(c,c))}</th>" for c in cols_order],
"</tr></thead><tbody>",
]
for _, row in customer_db.iterrows():
table_html.append("<tr>")
for c in cols_order:
raw = str(row.get(c, "—"))
if c == "Website" and raw not in ("", "—", "nan"):
href = html.escape(raw, quote=True)
val = f"<a href='{href}' target='_blank' rel='noopener'>Website ↗</a>"
elif c == "Kontakt" and "@" in raw and raw not in ("", "—", "nan"):
safe = html.escape(raw)
val = f"<a href='mailto:{html.escape(raw, quote=True)}'>{safe}</a>"
else:
val = html.escape(raw)
table_html.append(f"<td>{val}</td>")
table_html.append("</tr>")
table_html.append("</tbody></table></div>")
st.markdown("""
<style>
.customer-db-wrap {width:100%; overflow-x:hidden;}
.customer-db-table {width:100%; table-layout:fixed; border-collapse:collapse; font-size:11.5px;}
.customer-db-table th,.customer-db-table td,.customer-db-table a {border-bottom:1px solid rgba(128,128,128,.28); padding:6px 7px; text-align:left; vertical-align:top; white-space:normal !important; overflow-wrap:anywhere !important; word-break:break-word !important; hyphens:auto;}
.customer-db-table th {font-weight:700; position:sticky; top:0; background:#0e1117; color:#fff !important; z-index:1;}
.customer-db-table td:last-child,.customer-db-table th:last-child {text-align:center; white-space:normal; overflow-wrap:anywhere; word-break:break-word;}
.customer-db-table a {white-space:nowrap;}
</style>
""", unsafe_allow_html=True)
st.markdown("".join(table_html), unsafe_allow_html=True)
elif view == "scoring":
st.subheader("Bewertung und Human-Gate")
if matches.empty:
st.info("Noch keine neuen Lead-Scores vorhanden. Bewertungen entstehen erst aus neu recherchierten potenziellen Kunden.")
else:
score_cols = [
"company", "project_title", "lead_score", "industry_fit",
"problem_pressure", "company_fit", "economic_value",
"reachability", "evidence_quality", "sale_probability",
"human_gate", "rationale"
]
st.dataframe(matches[score_cols], use_container_width=True, hide_index=True)
st.divider()
st.caption("Lead Engine 2.0 · projektbezogene Leads bleiben strikt nach Kundenprojekt getrennt.")