import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import List, Dict, Optional
import os
from datetime import datetime
class EmailNotifier:
"""Email notification system"""
def __init__(self, config: Dict):
self.smtp_server = config.get('smtp_server', 'localhost')
self.smtp_port = int(config.get('smtp_port', 587))
self.username = config.get('username', '')
self.password = config.get('password', '')
self.from_email = config.get('from_email', self.username)
self.security = config.get('security', 'starttls')
# Handle to_emails - can be string or list
to_emails = config.get('to_emails', [])
if isinstance(to_emails, str):
self.to_emails = [to_emails]
else:
self.to_emails = to_emails
def send_notification(self, scraper_name: str, new_results: List[Dict]) -> bool:
"""Send email notification for new results"""
if not new_results:
return True
if not self.to_emails:
print("Keine Empfänger-Emails konfiguriert")
return False
try:
# Create message
msg = MIMEMultipart()
msg['From'] = self.from_email
msg['To'] = ', '.join(self.to_emails)
msg['Subject'] = f"Neue Wohnungen gefunden: {len(new_results)} neue Ergebnisse für {scraper_name}"
# Create email body
body = self._create_email_body(scraper_name, new_results)
msg.attach(MIMEText(body, 'html'))
# Send email with security settings
if self.security in ['ssl', 'ssl/tls']:
server = smtplib.SMTP_SSL(self.smtp_server, self.smtp_port)
else:
server = smtplib.SMTP(self.smtp_server, self.smtp_port)
# Apply security settings for non-SSL
if self.security in ['tls', 'starttls']:
server.starttls()
# Login if credentials provided
if self.username and self.password:
server.login(self.username, self.password)
server.send_message(msg)
server.quit()
print(f"Email-Benachrichtigung gesendet an {len(self.to_emails)} Empfänger")
return True
except Exception as e:
print(f"Fehler beim Senden der Email: {e}")
return False
def _create_email_body(self, scraper_name: str, new_results: List[Dict]) -> str:
"""Create HTML email body"""
html = f"""
🏠 Neue Wohnungen gefunden - {scraper_name}
Zeitpunkt: {datetime.now().strftime('%d.%m.%Y %H:%M')}
Anzahl neuer Ergebnisse: {len(new_results)}
Neue Wohnungen:
| PLZ |
Adresse |
Link |
"""
for result in new_results:
plz = result.get('plz', 'N/A')
address = result.get('address', 'N/A')
link = result.get('link', '#')
# Handle empty or None links
if not link or link == '#' or link == 'None':
link_display = 'Kein Link'
link_href = '#'
else:
link_display = 'Details'
link_href = link
html += f"""
| {plz} |
{address} |
{link_display} |
"""
html += """
Diese Nachricht wurde automatisch vom Flat Scraper gesendet.
"""
return html
class ConsoleNotifier:
"""Console notification for testing"""
def send_notification(self, scraper_name: str, new_results: List[Dict]) -> bool:
"""Print notification to console"""
if not new_results:
return True
print(f"\n{'='*50}")
print(f"🏠 NEUE WOHNUNGEN GEFUNDEN: {scraper_name}")
print(f"Zeitpunkt: {datetime.now().strftime('%d.%m.%Y %H:%M')}")
print(f"Anzahl: {len(new_results)}")
print(f"{'='*50}")
for result in new_results:
plz = result.get('plz', 'N/A')
address = result.get('address', 'N/A')
link = result.get('link', '#')
print(f"📍 PLZ {plz}: {address}")
if link != '#':
print(f" 🔗 {link}")
print()
return True