"""
lead_finder.py -- Findet neue Leads via DuckDuckGo + Website-Scraping.
Liest Suchqueries aus dem aktiven Client-Config-File und sucht nach
Unternehmen in der Zielregion. Extrahiert Email, Website, Firmenname
und speichert als 'raw' Kontakte in der DB.
Verwendung:
python lead_finder.py --client lueftungsprofi --limit 50
python lead_finder.py --client agent_solutions --limit 30
python lead_finder.py --list-clients
"""
import os
import re
import json
import time
import logging
import argparse
import sqlite3
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
from dotenv import load_dotenv
load_dotenv(dotenv_path=Path(__file__).parent / '.env', override=True)
try:
import requests
from bs4 import BeautifulSoup
HAS_SCRAPING = True
except ImportError:
HAS_SCRAPING = False
try:
from ddgs import DDGS
HAS_DDGS = True
except ImportError:
try:
from duckduckgo_search import DDGS
HAS_DDGS = True
except ImportError:
HAS_DDGS = False
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
CLIENTS_DIR = Path(__file__).parent / 'clients'
DB_PATH = Path(__file__).parent / 'email_agent.db'
FREEMAIL = {
'gmail.com', 'yahoo.com', 'outlook.com', 'hotmail.com', 'web.de',
'gmx.de', 'gmx.net', 'icloud.com', 'me.com', 't-online.de',
'freenet.de', 'aol.com', 'mail.com', 'yahoo.es', 'hotmail.es',
}
HEADERS = {
'User-Agent': (
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
'AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/120.0.0.0 Safari/537.36'
),
'Accept-Language': 'de-DE,de;q=0.9,es;q=0.8,en;q=0.7',
}
EMAIL_REGEX = re.compile(r'[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}')
SKIP_DOMAINS = {
'google.com', 'google.es', 'facebook.com', 'instagram.com',
'twitter.com', 'linkedin.com', 'youtube.com', 'tripadvisor.com',
'booking.com', 'airbnb.com', 'wikipedia.org', 'yelp.com',
'trustpilot.com', 'amazon.com', 'amazon.es',
}
def load_client(client_id: str) -> dict:
path = CLIENTS_DIR / f'{client_id}.json'
if not path.exists():
raise FileNotFoundError(f"Client-Config nicht gefunden: {path}")
with open(path, encoding='utf-8') as f:
return json.load(f)
def list_clients() -> list:
return [p.stem for p in CLIENTS_DIR.glob('*.json')]
def scrape_for_contacts(url: str) -> dict:
result = {'email': None, 'phone': None, 'company': None, 'website': url}
if not HAS_SCRAPING:
return result
try:
resp = requests.get(url, timeout=8, headers=HEADERS,
allow_redirects=True, verify=False)
if resp.status_code != 200:
return result
soup = BeautifulSoup(resp.content, 'html.parser')
company = None
if soup.title and soup.title.string:
company = soup.title.string.strip().split('|')[0].split('-')[0].strip()[:80]
if not company:
h1 = soup.find('h1')
if h1:
company = h1.get_text(strip=True)[:80]
result['company'] = company
emails_found = []
for a in soup.find_all('a', href=True):
href = a['href']
if href.startswith('mailto:'):
email = href[7:].split('?')[0].strip().lower()
if email and '@' in email:
emails_found.append(email)
if not emails_found:
emails_found = [e.lower() for e in EMAIL_REGEX.findall(soup.get_text())]
for email in emails_found:
domain = email.split('@')[1] if '@' in email else ''
if domain and domain not in FREEMAIL and len(email) < 100:
result['email'] = email
break
phone_pattern = re.compile(
r'(\+34[\s\-]?[6-9]\d{2}[\s\-]?\d{3}[\s\-]?\d{3}'
r'|\+49[\s\-]?\d{3,5}[\s\-]?\d{4,10}'
r'|0\d{2,5}[\s\-]\d{4,10})'
)
phone_match = phone_pattern.search(soup.get_text())
if phone_match:
result['phone'] = phone_match.group(0).strip()
result['website'] = str(resp.url)
except Exception as e:
logger.debug(f"Scraping fehlgeschlagen ({url}): {e}")
return result
def search_leads(query: str, max_results: int = 8) -> list:
if not HAS_DDGS:
logger.warning("duckduckgo_search nicht installiert.")
return []
try:
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=max_results))
urls = []
for r in results:
href = r.get('href', '')
if not href or not href.startswith('http'):
continue
domain = urlparse(href).netloc.replace('www.', '')
if any(skip in domain for skip in SKIP_DOMAINS):
continue
urls.append(href)
return urls
except Exception as e:
logger.debug(f"DDG-Suche fehlgeschlagen ({query!r}): {e}")
return []
def save_contact(campaign_id: int, data: dict, offset: int) -> bool:
if not data.get('email'):
return False # Ohne Email kein Outreach moeglich
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
if data.get('email'):
c.execute("SELECT id FROM cold_contacts WHERE email = ?", (data['email'],))
if c.fetchone():
conn.close()
return False
if data.get('website'):
try:
domain = urlparse(data['website']).netloc.replace('www.', '')
c.execute(
"SELECT id FROM cold_contacts WHERE website LIKE ? AND campaign_id = ?",
(f'%{domain}%', campaign_id)
)
if c.fetchone():
conn.close()
return False
except Exception:
pass
c.execute(
"SELECT MAX(CAST(source_id AS INTEGER)) FROM cold_contacts WHERE campaign_id = ?",
(campaign_id,)
)
row = c.fetchone()
next_source_id = (row[0] or 0) + 1 + offset
c.execute("""
INSERT INTO cold_contacts
(campaign_id, source_id, first_name, last_name, company, email, website, phone, status,
enrichment_source, created_at, updated_at)
VALUES (?, ?, '', '', ?, ?, ?, ?, 'raw', 'lead_finder', datetime('now'), datetime('now'))
""", (
campaign_id,
str(next_source_id),
data.get('company') or '',
data.get('email') or '',
data.get('website') or '',
data.get('phone') or '',
))
conn.commit()
conn.close()
return True
def run_finder(client_id: str, limit: int = 50, filter_file: str = None) -> dict:
config = load_client(client_id)
campaign_id = config.get('campaign_id')
if not campaign_id:
logger.error(f"Kein campaign_id in Config '{client_id}' definiert.")
return {'found': 0, 'saved': 0}
# Filter-File vom Dashboard hat Vorrang vor Config-Queries
if filter_file:
try:
with open(filter_file, encoding='utf-8') as f:
params = json.load(f)
queries = params.get('queries', [])
limit = params.get('limit', limit)
logger.info(f"Filter-File geladen: {len(queries)} Queries, Limit {limit}")
except Exception as e:
logger.error(f"Filter-File Fehler: {e}")
queries = config.get('search_queries', [])
else:
queries = config.get('search_queries', [])
logger.info(f"Lead Finder: {config['name']} | Kampagne {campaign_id} | {len(queries)} Queries | Limit {limit}")
found_total = 0
saved_total = 0
offset = 0
for query in queries:
if saved_total >= limit:
break
logger.info(f"Suche: {query!r}")
urls = search_leads(query, max_results=10)
logger.info(f" {len(urls)} URLs gefunden")
for url in urls:
if saved_total >= limit:
break
domain = urlparse(url).netloc.replace('www.', '')
data = scrape_for_contacts(url)
found_total += 1
saved = save_contact(campaign_id, data, offset)
if saved:
saved_total += 1
offset += 1
logger.info(
f" [+] {data.get('company', domain)} | "
f"{data.get('email', 'keine Email')} | gesamt: {saved_total}"
)
else:
logger.debug(f" [-] Duplikat/kein Kontakt: {domain}")
time.sleep(1.5)
time.sleep(3)
logger.info(f"\nFertig: {found_total} geprueft, {saved_total} neu gespeichert.")
return {'found': found_total, 'saved': saved_total}
def main():
parser = argparse.ArgumentParser(description='Lead Finder -- findet neue Leads via Web-Suche.')
parser.add_argument('--client', type=str, help='Client-ID (z.B. lueftungsprofi)')
parser.add_argument('--limit', type=int, default=50, help='Max. neue Kontakte (default: 50)')
parser.add_argument('--filter-file', type=str, help='JSON-Datei mit Queries vom Dashboard')
parser.add_argument('--list-clients', action='store_true', help='Verfuegbare Clients anzeigen')
args = parser.parse_args()
if args.list_clients:
clients = list_clients()
print("\nVerfuegbare Clients:")
for c in clients:
cfg = load_client(c)
print(f" {c:20s} -> {cfg['name']} (Kampagne {cfg.get('campaign_id', '?')})")
return
if not args.client:
parser.print_help()
return
run_finder(args.client, args.limit, filter_file=args.filter_file)
if __name__ == '__main__':
main()