Coverage for modules\email\smtp_service.py: 15%

1""" 

2SMTP Service - Send emails like Outlook/Thunderbird 

3""" 

4import smtplib 

5from email.mime.multipart import MIMEMultipart 

6from email.mime.text import MIMEText 

7from email.mime.base import MIMEBase 

8from email import encoders 

9from typing import List, Optional, Tuple 

10import base64 

11 

12 

13class SMTPService: 

14 """Service for sending email via SMTP.""" 

15 

16 def __init__(self, host: str, port: int, username: str, password: str, 

17 use_ssl: bool = True, from_email: str = None, from_name: str = ''): 

18 self.host = host 

19 self.port = port 

20 self.username = username 

21 self.password = password 

22 self.use_ssl = use_ssl # True for 465, False for other ports 

23 self.use_tls = not use_ssl and port == 587 # TLS for port 587 

24 self.from_email = from_email or username 

25 self.from_name = from_name 

26 self.connection: Optional[smtplib.SMTP] = None 

27 

28 def connect(self) -> Tuple[bool, str]: 

29 """Establish SMTP connection. Returns (success, error_message).""" 

30 try: 

31 if self.use_ssl: 

32 self.connection = smtplib.SMTP_SSL(self.host, self.port) 

33 else: 

34 self.connection = smtplib.SMTP(self.host, self.port) 

35 

36 # Start TLS if needed 

37 if self.use_tls: 

38 self.connection.starttls() 

39 

40 # Login 

41 self.connection.login(self.username, self.password) 

42 return True, "" 

43 

44 except smtplib.SMTPAuthenticationError as e: 

45 error_msg = f"SMTP authentication failed: {str(e)}" 

46 print(f"SMTP connection failed: {e}") 

47 return False, error_msg 

48 except Exception as e: 

49 error_msg = f"SMTP connection failed: {str(e)}" 

50 print(f"SMTP connection failed: {e}") 

51 return False, error_msg 

52 

53 def disconnect(self): 

54 """Close SMTP connection.""" 

55 if self.connection: 

56 try: 

57 self.connection.quit() 

58 except: 

59 pass 

60 self.connection = None 

61 

62 def send_email(self, to_addresses: List[str], subject: str, body: str, 

63 cc_addresses: Optional[List[str]] = None, 

64 bcc_addresses: Optional[List[str]] = None, 

65 is_html: bool = False, 

66 attachments: Optional[List[dict]] = None) -> bool: 

67 """ 

68 Send an email. 

69  

70 Args: 

71 to_addresses: List of recipient emails 

72 subject: Email subject 

73 body: Email body 

74 cc_addresses: CC recipients 

75 bcc_addresses: BCC recipients 

76 is_html: Body is HTML format 

77 attachments: List of dicts with {filename, content, content_type} 

78 """ 

79 if not self.connection: 

80 success, error_msg = self.connect() 

81 if not success: 

82 return False, error_msg 

83 

84 try: 

85 # Create message 

86 msg = MIMEMultipart('alternative') 

87 msg['Subject'] = subject 

88 msg['From'] = self._format_address(self.from_email, self.from_name) 

89 msg['To'] = ', '.join(to_addresses) 

90 

91 if cc_addresses: 

92 msg['Cc'] = ', '.join(cc_addresses) 

93 

94 # Attach body 

95 content_type = 'html' if is_html else 'plain' 

96 msg.attach(MIMEText(body, content_type, 'utf-8')) 

97 

98 # Attach files 

99 if attachments: 

100 for attachment in attachments: 

101 part = MIMEBase('application', 'octet-stream') 

102 part.set_payload(attachment['content']) 

103 encoders.encode_base64(part) 

104 part.add_header( 

105 'Content-Disposition', 

106 f'attachment; filename="{attachment["filename"]}"' 

107 ) 

108 msg.attach(part) 

109 

110 # Build recipient list 

111 all_recipients = to_addresses.copy() 

112 if cc_addresses: 

113 all_recipients.extend(cc_addresses) 

114 if bcc_addresses: 

115 all_recipients.extend(bcc_addresses) 

116 

117 # Send 

118 self.connection.sendmail(self.from_email, all_recipients, msg.as_string()) 

119 return True, "" 

120 

121 except smtplib.SMTPAuthenticationError as e: 

122 error_msg = f"SMTP authentication failed: {str(e)}" 

123 print(f"Error sending email: {e}") 

124 return False, error_msg 

125 except smtplib.SMTPSenderRefused as e: 

126 # SMTPSenderRefused: (code, response, sender) 

127 # e.args is (code, response) 

128 error_info = e.args if e.args else () 

129 if len(error_info) >= 2: 

130 resp = error_info[1] 

131 if isinstance(resp, bytes): 

132 resp = resp.decode() 

133 error_msg = f"SMTP sender refused: {resp}" 

134 else: 

135 error_msg = f"SMTP sender refused: {str(e)}" 

136 print(f"Error sending email: {e}") 

137 return False, error_msg 

138 except smtplib.SMTPRecipientsRefused as e: 

139 error_msg = f"SMTP recipient refused: {str(e)}" 

140 print(f"Error sending email: {e}") 

141 return False, error_msg 

142 except smtplib.SMTPException as e: 

143 error_msg = f"SMTP error: {str(e)}" 

144 print(f"Error sending email: {e}") 

145 return False, error_msg 

146 except Exception as e: 

147 error_msg = f"Failed to send email: {str(e)}" 

148 print(f"Error sending email: {e}") 

149 return False, error_msg 

150 

151 def test_connection(self) -> tuple: 

152 """ 

153 Test the SMTP connection. 

154 Returns (success, message). 

155 """ 

156 try: 

157 if not self.connection: 

158 success, error_msg = self.connect() 

159 if not success: 

160 return False, error_msg 

161 

162 return True, "Connection successful" 

163 

164 except Exception as e: 

165 return False, str(e) 

166 

167 def _format_address(self, email: str, name: str = '') -> str: 

168 """Format email address with optional name.""" 

169 if name: 

170 return f"{name} <{email}>" 

171 return email 

172 

173 def __enter__(self): 

174 self.connect() 

175 return self 

176 

177 def __exit__(self, exc_type, exc_val, exc_tb): 

178 self.disconnect() 

179 

180 

181# Provider presets 

182IMAP_PRESETS = { 

183 'gmail': { 

184 'imap_host': 'imap.gmail.com', 

185 'imap_port': 993, 

186 'smtp_host': 'smtp.gmail.com', 

187 'smtp_port': 587, 

188 'use_ssl': True, 

189 }, 

190 'outlook': { 

191 'imap_host': 'outlook.office365.com', 

192 'imap_port': 993, 

193 'smtp_host': 'smtp.office365.com', 

194 'smtp_port': 587, 

195 'use_ssl': True, 

196 }, 

197 'yahoo': { 

198 'imap_host': 'imap.mail.yahoo.com', 

199 'imap_port': 993, 

200 'smtp_host': 'smtp.mail.yahoo.com', 

201 'smtp_port': 587, 

202 'use_ssl': True, 

203 }, 

204} 

205 

206 

207def get_provider_preset(provider: str) -> dict: 

208 """Get IMAP/SMTP preset for a provider.""" 

209 return IMAP_PRESETS.get(provider, { 

210 'imap_host': '', 

211 'imap_port': 993, 

212 'smtp_host': '', 

213 'smtp_port': 587, 

214 'use_ssl': True, 

215 })