import subprocess import os from typing import Optional from app.config import settings from app.config_reader import config_reader class EmailSender: def __init__(self): self.sendmail_path = settings.php_sendmail_path def send( self, to: str, subject: str, body: str, from_addr: Optional[str] = None, ) -> bool: if from_addr is None: from_addr = config_reader.get_mail_from() message = self._build_message(to, subject, body, from_addr) try: if os.name == "nt": return self._send_debug(to, subject, body, from_addr) proc = subprocess.run( [self.sendmail_path, "-t", "-i"], input=message, capture_output=True, text=True, timeout=30, ) return proc.returncode == 0 except Exception: return self._send_debug(to, subject, body, from_addr) def _send_debug(self, to: str, subject: str, body: str, from_addr: str) -> bool: log_dir = os.path.join(os.path.dirname(__file__), "..", "logs") os.makedirs(log_dir, exist_ok=True) log_path = os.path.join(log_dir, "emails.log") with open(log_path, "a", encoding="utf-8") as f: f.write(f"=== EMAIL ===\nFrom: {from_addr}\nTo: {to}\nSubject: {subject}\n\n{body}\n\n") return True def _build_message(self, to: str, subject: str, body: str, from_addr: str) -> str: lines = [ f"From: {from_addr}", f"To: {to}", f"Subject: {subject}", "MIME-Version: 1.0", "Content-Type: text/plain; charset=UTF-8", "Content-Transfer-Encoding: 8bit", "", body, ] return "\r\n".join(lines) def send_ticket_escalation( self, to: str, subject: str, user_info: str, contact: str, inquiry: str, history: str, ) -> bool: body = ( f"Поступило новое обращение через Виртуального помощника AegisOne Engineering\n\n" f"--- Данные клиента ---\n{user_info}\n" f"--- Контактные данные ---\n{contact}\n" f"--- Суть обращения ---\n{inquiry}\n" f"--- История переписки ---\n{history}\n" f"---\n" f"Дата и время: {__import__('datetime').datetime.now().strftime('%d.%m.%Y %H:%M')}" ) return self.send(to, subject, body) email_sender = EmailSender()