import smtplib from datetime import datetime from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.utils import formataddr from typing import Sequence from sqlalchemy.orm import Session from app.models.system import SendLog from app.services.config_store import get_config def get_smtp_settings(db: Session) -> dict[str, str]: return { "enabled": get_config(db, "smtp.enabled", "false"), "host": get_config(db, "smtp.host", ""), "port": get_config(db, "smtp.port", "587"), "username": get_config(db, "smtp.username", ""), "password": get_config(db, "smtp.password", ""), "from_email": get_config(db, "smtp.from_email", ""), "from_name": get_config(db, "smtp.from_name", ""), "reply_to": get_config(db, "smtp.reply_to", ""), "use_tls": get_config(db, "smtp.use_tls", "true"), "recipients": get_config(db, "smtp.recipients", ""), # Zeitplan "schedule_enabled": get_config(db, "smtp.schedule_enabled", "false"), "schedule_frequency": get_config(db, "smtp.schedule_frequency", "monthly"), "schedule_time": get_config(db, "smtp.schedule_time", "08:00"), "schedule_weekdays": get_config(db, "smtp.schedule_weekdays", "0"), "schedule_dom": get_config(db, "smtp.schedule_dom", "1"), "last_scheduled_date": get_config(db, "smtp.last_scheduled_date", ""), # Generierungs-Vorgaben für den geplanten Versand "gen_period": get_config(db, "smtp.gen_period", "last_month"), "gen_days": get_config(db, "smtp.gen_days", "30"), "gen_article_type": get_config(db, "smtp.gen_article_type", "new"), "gen_category": get_config(db, "smtp.gen_category", ""), } def schedule_is_due(settings: dict[str, str], now: datetime) -> bool: """Prüft, ob laut Zeitplan jetzt ein Versand fällig ist (mit Tages-Dedupe).""" if settings["enabled"].lower() != "true": return False if (settings.get("schedule_enabled") or "false").lower() != "true": return False freq = (settings.get("schedule_frequency") or "off").lower() if freq == "off": return False schedule_time = settings.get("schedule_time") or "08:00" if now.strftime("%H:%M") < schedule_time: return False today = now.strftime("%Y-%m-%d") if settings.get("last_scheduled_date") == today: return False if freq == "daily": return True if freq == "weekly": weekdays = {d.strip() for d in (settings.get("schedule_weekdays") or "").split(",") if d.strip() != ""} return str(now.weekday()) in weekdays if freq == "monthly": try: dom = int(settings.get("schedule_dom") or "1") except ValueError: dom = 1 return now.day == max(1, min(dom, 28)) return False def send_newsletter( db: Session, subject: str, html_content: str, text_content: str, recipients: Sequence[str], scheduled: bool = False, ) -> tuple[str, str]: settings = get_smtp_settings(db) if settings["enabled"].lower() != "true": detail = "SMTP ist deaktiviert. Bitte in der SMTP-Konfiguration aktivieren." _log(db, subject, recipients, "skipped", detail, scheduled) return "skipped", detail if not settings["host"] or not settings["from_email"] or not recipients: detail = "SMTP unvollständig konfiguriert (Host, Absender oder Empfänger fehlen)." _log(db, subject, recipients, "failed", detail, scheduled) return "failed", detail msg = MIMEMultipart("alternative") msg.attach(MIMEText(text_content, "plain", "utf-8")) msg.attach(MIMEText(html_content, "html", "utf-8")) msg["Subject"] = subject if settings.get("from_name"): msg["From"] = formataddr((settings["from_name"], settings["from_email"])) else: msg["From"] = settings["from_email"] msg["To"] = ", ".join(recipients) if settings.get("reply_to"): msg["Reply-To"] = settings["reply_to"] try: port = int(settings["port"]) with smtplib.SMTP(settings["host"], port, timeout=20) as server: if settings["use_tls"].lower() == "true": server.starttls() if settings["username"]: server.login(settings["username"], settings["password"]) server.sendmail(settings["from_email"], list(recipients), msg.as_string()) detail = f"Versand erfolgreich an {len(recipients)} Empfänger." _log(db, subject, recipients, "sent", detail, scheduled) return "sent", detail except Exception as exc: detail = f"Versandfehler: {exc}" _log(db, subject, recipients, "failed", detail, scheduled) return "failed", detail def mark_scheduled_sent(db: Session, now: datetime) -> None: from app.services.config_store import set_config set_config(db, "smtp.last_scheduled_date", now.strftime("%Y-%m-%d")) def _log(db: Session, subject: str, recipients: Sequence[str], status: str, detail: str, scheduled: bool) -> None: db.add( SendLog( subject=subject, recipients=", ".join(recipients) if recipients else "-", status=status, detail=detail, scheduled=scheduled, ) ) db.commit()