"""
knowledge_retriever.py -- RAG-Zugriff auf die YouTube Research knowledge.db.
Findet relevante Video-Summaries aus der Wissensdatenbank basierend auf
Branche und Schmerzpunkt eines Kontakts. Wird in Step-2-Emails genutzt
um einen konkreten Mehrwert (Wissens-Snippet) einzubetten.
WICHTIG: Wird NUR fuer AG-Emails verwendet (KI-Automatisierung).
Fuer LP-Emails (Lueftung) gibt es eine separate statische Faktenbasis.
"""
import os
import re
import sqlite3
import logging
from typing import Optional, Dict, List
logger = logging.getLogger(__name__)
# Pfad zur YouTube Research Wissensdatenbank
DEFAULT_KNOWLEDGE_DB = os.path.join(
os.path.dirname(__file__),
'..', 'YouTube_Research', 'dist', 'knowledge.db'
)
# Maximale Wortanzahl fuer einen Knowledge-Snippet
MAX_SNIPPET_WORDS = 80
# ---------------------------------------------------------------------------
# Branchen -> Suchbegriffe Mapping
# ---------------------------------------------------------------------------
INDUSTRY_SEARCH_TERMS: Dict[str, List[str]] = {
# Handwerk / Dienstleister
'handwerk': ['telefon agent', 'missed calls', 'phone automation', 'ki telefon'],
'sanitaer': ['telefon agent', 'phone automation', 'ki agent'],
'elektro': ['telefon agent', 'phone automation'],
'installateur': ['telefon agent', 'automation'],
# Medizin / Gesundheit
'arztpraxis': ['ki praxis', 'telefon agent arzt', 'appointment automation'],
'zahnarzt': ['telefon automation', 'appointment', 'ki agent'],
'therapeut': ['ki agent', 'automation', 'email automation'],
'physiotherapie': ['telefon agent', 'ki agent'],
# Recht / Finanzen
'rechtsanwalt': ['email agent', 'ki email', 'automation kanzlei'],
'steuerberater': ['email agent', 'ki automatisierung', 'claude agent'],
'versicherung': ['email agent', 'ki automation', 'customer service ai'],
# E-Commerce / Handel
'ecommerce': ['email agent', 'ki customer service', 'automation shop'],
'online-shop': ['email automation', 'ki agent', 'customer support ai'],
'einzelhandel': ['email agent', 'ki automatisierung'],
# Marketing / Beratung
'marketing': ['claude code', 'ki agent', 'automation workflow'],
'agentur': ['claude agent', 'ki workflow', 'automation'],
'beratung': ['ki agent', 'email automation', 'claude'],
# IT
'it': ['claude code', 'ki agent', 'agent sdk'],
'software': ['claude code', 'agent', 'automation'],
# Gastronomie / Hotel
'gastronomie': ['ki agent', 'customer service', 'telefon automation'],
'restaurant': ['ki agent', 'telefon automation', 'reservation'],
'hotel': ['portier', 'concierge ai', 'ki hotel', 'guest service'],
# Immobilien
'immobilien': ['email agent', 'ki automatisierung', 'automation'],
'makler': ['email automation', 'ki agent', 'real estate ai'],
# Tourismus
'tourismus': ['ki agent', 'buchung automation', 'customer service ai'],
'reisebuero': ['email agent', 'ki agent', 'automation'],
# Logistik
'logistik': ['telefon agent', 'ki automation', 'tracking automation'],
'transport': ['telefon agent', 'ki agent'],
# Bildung
'bildung': ['ki kurs', 'ki agent', 'automation education'],
'schule': ['ki bildung', 'automation', 'ki kurs'],
# Standard-Fallback
'_default': ['ki agent', 'claude code', 'automation', 'ki automatisierung'],
}
# ---------------------------------------------------------------------------
# LP-Fakten (statisch, fuer Lueftungsprofi-Emails)
# ---------------------------------------------------------------------------
LP_FACTS: Dict[str, List[str]] = {
'hotel': [
"Gaeste bewerten Zimmerluft als drittgroessten Faktor fuer Hotelbewertungen "
"(nach Sauberkeit und Bett).",
"Dezentrale Lueftung spart gegenueber zentralen Anlagen 30-40% "
"Installationskosten -- kein Umbau der Kanalstruktur noetig.",
"Auf Mallorca liegt die relative Luftfeuchtigkeit 8 Monate im Jahr ueber 60% "
"-- das ist die kritische Grenze fuer Schimmelwachstum.",
],
'immobilienverwaltung': [
"Schimmelbeseitigung kostet im Schnitt 3.000-8.000 EUR pro Wohneinheit -- "
"paeventive Lueftung ist rein rechnerisch nach 1-2 Jahren amortisiert.",
"80% aller Schimmelfaelle in Wohngebaeuden entstehen durch unzureichende "
"Lueftung und Waermebruecken -- nicht durch Baumaengel.",
"CO2-gesteuerte Lueftung erkennt Belegung automatisch und spart Energie "
"bei Leerstand ohne manuellen Eingriff.",
],
'hausverwaltung': [
"Schimmelreklamationen gehoeren zu den haeufigstem Mieterbeschwerden -- "
"und koennen rechtliche Konsequenzen haben.",
"Moderne Luftqualitaetssensoren (CO2, Feuchte, VOC) geben Echtzeitdaten "
"fuer proaktives Gebaeudmanagement.",
],
'buero': [
"Bei CO2-Werten ueber 1000ppm sinkt die kognitive Leistung der Mitarbeiter "
"nachweislich um bis zu 15% -- haeufig unbemerkt.",
"Dezentrale Lueftungsgeraete koennen raumweise nachgeruestet werden -- "
"kein Umbau, keine Betriebsunterbrechung.",
],
'bauunternehmen': [
"Das spanische CTE (Codigo Tecnico de Edificacion) schreibt seit 2020 "
"mechanische Lueftung in Neubauten vor.",
"getAir-Systeme erfuellen die RITE-Anforderungen fuer Wohnraumlueftung -- "
"auf Mallorca gelistete Losung.",
],
'wohnungsbau': [
"Waermerueckgewinnung bis 93% macht moderne Lueftungssysteme auch bei "
"Passivhaeusern wirtschaftlich.",
"Lueftungskonzepte gehoeren zum Pflichtprogramm in spanischen Neubauvorhaben.",
],
'_default': [
"Schlechte Raumluft ist einer der unterschaetztesten Faktoren fuer "
"Produktivitaet und Gesundheit in Innenraeumen.",
"CO2, Luftfeuchtigkeit und VOC-Belastung lassen sich durch dezentrale "
"Systeme kostenguenstig und ohne Umbau loesen.",
],
}
# ---------------------------------------------------------------------------
# Knowledge Retriever
# ---------------------------------------------------------------------------
class KnowledgeRetriever:
"""
Sucht relevante Inhalte aus der YouTube Research Wissensdatenbank.
Liefert Snippets fuer die Email-Generierung.
"""
def __init__(self, knowledge_db_path: str = None):
self.db_path = knowledge_db_path or DEFAULT_KNOWLEDGE_DB
self._check_db()
def _check_db(self):
"""Prueft ob die knowledge.db erreichbar ist."""
if not os.path.exists(self.db_path):
logger.warning(
f"knowledge.db nicht gefunden: {self.db_path}\n"
"AG-Knowledge wird nicht verfuegbar sein. "
"Bitte YouTube Research gestartet lassen."
)
def get_knowledge_snippet(self, industry: str, pain_point: str = '',
icp: str = 'AG') -> Optional[str]:
"""
Hauptfunktion. Gibt einen kurzen Wissens-Snippet zurueck.
Args:
industry: Branche des Kontakts
pain_point: Beschriebenes Problem
icp: 'AG' oder 'LP'
Returns:
Snippet-String fuer Email-Einbettung, oder None wenn nichts gefunden.
"""
if icp == 'LP':
return self._get_lp_fact(industry)
else:
return self._get_ag_snippet(industry, pain_point)
def _get_lp_fact(self, industry: str) -> Optional[str]:
"""Gibt eine statische LP-Tatsache basierend auf der Branche zurueck."""
industry_norm = industry.lower().strip()
facts = None
for key in LP_FACTS:
if key == '_default':
continue
if key in industry_norm:
facts = LP_FACTS[key]
break
if not facts:
facts = LP_FACTS['_default']
# Erstes Fact zurueckgeben
return facts[0] if facts else None
def _get_ag_snippet(self, industry: str, pain_point: str) -> Optional[str]:
"""
Sucht in der YouTube Research DB nach relevanten Inhalten.
Nutzt FTS5 fuer die Suche.
"""
if not os.path.exists(self.db_path):
return None
search_terms = self._build_search_query(industry, pain_point)
try:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
for term in search_terms:
# FTS5-Suche
try:
rows = conn.execute("""
SELECT v.title, v.summary, v.channel,
(-bm25(videos_fts)) AS score
FROM videos_fts
JOIN videos v ON videos_fts.rowid = v.id
WHERE videos_fts MATCH ?
AND v.summary IS NOT NULL
AND length(v.summary) > 100
ORDER BY score DESC
LIMIT 3
""", (term,)).fetchall()
if rows:
best = rows[0]
snippet = self._extract_snippet(
summary=best['summary'],
title=best['title'],
)
conn.close()
logger.debug(
f"Knowledge-Hit fuer '{term}': {best['title'][:50]}"
)
return snippet
except sqlite3.OperationalError:
# FTS-Query-Fehler (z.B. Sonderzeichen) -> naechsten Term versuchen
continue
conn.close()
logger.debug(f"Kein Knowledge-Hit fuer Branche '{industry}'")
return None
except Exception as e:
logger.error(f"knowledge.db Zugriffsfehler: {e}")
return None
def _build_search_query(self, industry: str, pain_point: str) -> List[str]:
"""
Erstellt priorisierte Suchbegriff-Liste.
Der erste Begriff mit einem Treffer gewinnt.
"""
industry_norm = industry.lower().strip()
terms = []
# Branchenspezifische Begriffe
for key, search_list in INDUSTRY_SEARCH_TERMS.items():
if key == '_default':
continue
if key in industry_norm:
terms.extend(search_list)
break
# Pain-Point-basierte Begriffe
if pain_point:
pain_lower = pain_point.lower()
if 'email' in pain_lower or 'antwort' in pain_lower:
terms.insert(0, 'email agent')
if 'telefon' in pain_lower or 'anruf' in pain_lower:
terms.insert(0, 'telefon agent')
if 'hotel' in pain_lower or 'gaest' in pain_lower:
terms.insert(0, 'portier concierge')
# Fallback
terms.extend(INDUSTRY_SEARCH_TERMS['_default'])
# Deduplizieren (Reihenfolge beibehalten)
seen = set()
result = []
for t in terms:
if t not in seen:
seen.add(t)
result.append(t)
return result
def _extract_snippet(self, summary: str, title: str) -> str:
"""
Extrahiert einen kompakten, emailtauglichen Snippet aus einem Summary.
Maximal MAX_SNIPPET_WORDS Woerter.
"""
if not summary:
return ''
# Erste aussagekraeftige Saetze nehmen
sentences = re.split(r'(?<=[.!?])\s+', summary.strip())
snippet_parts = []
word_count = 0
for sentence in sentences:
words = len(sentence.split())
if word_count + words > MAX_SNIPPET_WORDS:
break
if len(sentence) > 20: # Kurze Fragmente ueberspringen
snippet_parts.append(sentence.strip())
word_count += words
if not snippet_parts:
# Fallback: Ersten N Woerter nehmen
words = summary.split()[:MAX_SNIPPET_WORDS]
return ' '.join(words)
return ' '.join(snippet_parts)
def get_lp_facts_for_industry(self, industry: str) -> List[str]:
"""Gibt alle LP-Fakten fuer eine Branche zurueck (fuer Outreach Generator)."""
industry_norm = industry.lower().strip()
for key, facts in LP_FACTS.items():
if key == '_default':
continue
if key in industry_norm:
return facts
return LP_FACTS['_default']