Coverage for modules\email\services.py: 18%
286 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 17:01 +0800
1"""
2Email Services - Business logic for email operations
3"""
4import uuid
5from typing import List, Optional, Tuple
6from uuid import UUID
7from datetime import datetime
8import base64
9from django.db import transaction
10from django.utils import timezone
12from .models import EmailAccount, EmailFolder, EmailMessage, EmailAttachment
13from .imap_service import IMAPService
14from .smtp_service import SMTPService, IMAP_PRESETS, get_provider_preset
15from .gmail_oauth import GmailOAuthService
18class EmailAccountService:
19 """Service for managing email accounts."""
21 @staticmethod
22 def create_imap_account(user, tenant_id: UUID, data: dict) -> Tuple[EmailAccount, bool, str]:
23 """
24 Create and verify an IMAP email account.
25 Returns (account, success, message).
26 """
27 # Get provider preset or use custom settings
28 provider = data.get('provider', 'other')
29 preset = get_provider_preset(provider)
31 # Use provided values or fall back to preset
32 imap_host = data.get('imap_host') or preset.get('imap_host', '')
33 imap_port = data.get('imap_port') or preset.get('imap_port', 993)
34 smtp_host = data.get('smtp_host') or preset.get('smtp_host', '')
35 smtp_port = data.get('smtp_port') or preset.get('smtp_port', 587)
36 use_ssl = data.get('use_ssl', preset.get('use_ssl', True))
38 # Encrypt password (simple base64 for now - in production use proper encryption)
39 password = data.get('password', '')
40 encrypted_password = base64.b64encode(password.encode()).decode()
42 # Test IMAP connection
43 print(f"[EMAIL DEBUG] Testing IMAP: {imap_host}:{imap_port}")
44 try:
45 with IMAPService(
46 host=imap_host,
47 port=imap_port,
48 username=data['email_address'],
49 password=password,
50 ssl=use_ssl
51 ) as imap:
52 print(f"[EMAIL DEBUG] IMAP connection object: {imap.connection}")
53 if not imap.connection:
54 print("[EMAIL DEBUG] IMAP connection is None/False")
55 return None, False, "Failed to connect to IMAP server"
57 # Test SMTP as well
58 print(f"[EMAIL DEBUG] Testing SMTP: {smtp_host}:{smtp_port}")
59 with SMTPService(
60 host=smtp_host,
61 port=smtp_port,
62 username=data['email_address'],
63 password=password,
64 use_ssl=(smtp_port == 465),
65 from_email=data['email_address']
66 ) as smtp:
67 print(f"[EMAIL DEBUG] SMTP connection object: {smtp.connection}")
68 if not smtp.connection:
69 print("[EMAIL DEBUG] SMTP connection is None/False")
70 return None, False, "Failed to connect to SMTP server"
72 except Exception as e:
73 import traceback
74 print(f"[EMAIL DEBUG] Exception: {type(e).__name__}: {str(e)}")
75 traceback.print_exc()
76 return None, False, f"Connection failed: {str(e)}"
78 # Get User instance from user ID
79 from django.contrib.auth import get_user_model
80 User = get_user_model()
81 user_obj = User.objects.get(id=user)
83 # Create account
84 account = EmailAccount.objects.create(
85 tenant_id=tenant_id,
86 user=user_obj,
87 provider=provider,
88 connection_type='imap',
89 email_address=data['email_address'],
90 display_name=data.get('display_name', ''),
91 imap_host=imap_host,
92 imap_port=imap_port,
93 smtp_host=smtp_host,
94 smtp_port=smtp_port,
95 encrypted_password=encrypted_password,
96 use_ssl=use_ssl,
97 is_verified=True,
98 )
100 # Sync folders
101 EmailFolderService.sync_folders(account)
103 return account, True, "Account connected successfully"
105 @staticmethod
106 def create_oauth_account(user, tenant_id: UUID, credentials: dict) -> EmailAccount:
107 """Create an OAuth-based email account."""
108 account = EmailAccount.objects.create(
109 tenant_id=tenant_id,
110 user=user,
111 provider='gmail',
112 connection_type='oauth',
113 email_address=credentials.get('email'),
114 display_name=credentials.get('name', ''),
115 oauth_token=credentials.get('token'),
116 oauth_refresh_token=credentials.get('refresh_token'),
117 is_verified=True,
118 )
119 return account
121 @staticmethod
122 def decrypt_password(encrypted_password: str) -> str:
123 """Decrypt password for IMAP/SMTP auth."""
124 try:
125 return base64.b64decode(encrypted_password.encode()).decode()
126 except:
127 return ''
129 @staticmethod
130 def sync_account(account: EmailAccount) -> Tuple[bool, str]:
131 """Sync emails for an account."""
132 if account.connection_type == 'oauth':
133 return EmailAccountService._sync_oauth(account)
134 else:
135 return EmailAccountService._sync_imap(account)
137 @staticmethod
138 def _sync_imap(account: EmailAccount) -> Tuple[bool, str]:
139 """Sync emails via IMAP."""
140 password = EmailAccountService.decrypt_password(account.encrypted_password)
142 try:
143 with IMAPService(
144 host=account.imap_host,
145 port=account.imap_port,
146 username=account.email_address,
147 password=password,
148 ssl=account.use_ssl
149 ) as imap:
150 from django.db import connection
152 # Get tenant_id and account_id as hex strings (no dashes)
153 tenant_id_hex = account.tenant_id.hex if hasattr(account.tenant_id, 'hex') else str(account.tenant_id)
154 account_id_hex = account.id.hex if hasattr(account.id, 'hex') else str(account.id)
156 # Sync folders using raw SQL to avoid Django UUID ORM issues
157 folders = imap.list_folders()
159 for folder_data in folders:
160 folder_name = folder_data.get('name', 'Unknown')
161 remote_id = folder_name
162 folder_type = EmailAccountService._detect_folder_type(folder_name)
163 unread = imap.get_unread_count(folder_name)
165 # Use raw SQL to upsert folder (avoid Django ORM UUID issues)
166 with connection.cursor() as cursor:
167 # Check if folder exists
168 cursor.execute(
169 "SELECT id, unread_count FROM email_emailfolder WHERE tenant_id = %s AND account_id = %s AND remote_id = %s",
170 [tenant_id_hex, account_id_hex, remote_id]
171 )
172 row = cursor.fetchone()
174 if row:
175 # Update existing folder
176 cursor.execute(
177 "UPDATE email_emailfolder SET name = %s, folder_type = %s, unread_count = %s WHERE id = %s",
178 [folder_name, folder_type, unread, row[0]]
179 )
180 else:
181 # Insert new folder
182 now = timezone.now()
183 folder_uuid = str(uuid.uuid4()).replace('-', '')
184 cursor.execute(
185 """INSERT INTO email_emailfolder (id, created_at, updated_at, tenant_id, name, remote_id, folder_type, unread_count, account_id)
186 VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)""",
187 [folder_uuid, now, now, tenant_id_hex, folder_name, remote_id, folder_type, unread, account_id_hex]
188 )
190 # Update last sync time using raw SQL
191 with connection.cursor() as cursor:
192 cursor.execute(
193 "UPDATE email_emailaccount SET last_sync_at = %s WHERE id = %s",
194 [timezone.now(), account_id_hex]
195 )
197 return True, "Sync completed"
199 except Exception as e:
200 return False, f"Sync failed: {str(e)}"
202 @staticmethod
203 def _sync_oauth(account: EmailAccount) -> Tuple[bool, str]:
204 """Sync emails via OAuth."""
205 try:
206 gmail = GmailOAuthService()
207 gmail.creds = account.oauth_token
209 # Sync folders (Gmail uses labels)
210 labels = ['INBOX', 'SENT', 'DRAFT', 'TRASH', 'SPAM']
211 for label in labels:
212 folder, _ = EmailFolder.objects.get_or_create(
213 tenant_id=account.tenant_id,
214 account=account,
215 name=label,
216 defaults={
217 'folder_type': label.lower(),
218 }
219 )
221 # Get unread count via API
222 if gmail.connect():
223 messages = gmail.list_messages(label=label, max_results=100)
224 unread = sum(1 for m in messages if not m.get('is_read', True))
225 folder.unread_count = unread
226 folder.save()
228 account.last_sync_at = timezone.now()
229 account.save()
230 return True, "OAuth sync completed"
232 except Exception as e:
233 return False, f"OAuth sync failed: {str(e)}"
235 @staticmethod
236 def _detect_folder_type(folder_name: str) -> str:
237 """Detect folder type from name."""
238 name_lower = folder_name.lower().replace(' ', '').replace('[gmail]', '')
240 if name_lower in ['inbox']:
241 return 'inbox'
242 elif name_lower in ['sent', '[gmail]/sent', 'sent mail', 'sent messages']:
243 return 'sent'
244 elif name_lower in ['drafts', '[gmail]/drafts']:
245 return 'draft'
246 elif name_lower in ['trash', '[gmail]/trash', 'deleted']:
247 return 'trash'
248 elif name_lower in ['spam', '[gmail]/spam', 'junk']:
249 return 'spam'
250 else:
251 return 'custom'
254class EmailFolderService:
255 """Service for managing email folders."""
257 @staticmethod
258 def sync_folders(account: EmailAccount):
259 """Sync folders from email server."""
260 if account.connection_type == 'oauth':
261 # Gmail uses labels
262 labels = ['INBOX', 'SENT', 'DRAFT', 'TRASH', 'SPAM']
263 for label in labels:
264 folder_type = label.lower()
265 # Map special folder types
266 if label == 'DRAFT':
267 folder_type = 'draft'
268 elif label == 'TRASH':
269 folder_type = 'trash'
270 elif label == 'SPAM':
271 folder_type = 'spam'
272 EmailFolder.objects.get_or_create(
273 tenant_id=account.tenant_id,
274 account=account,
275 name=label,
276 defaults={'folder_type': folder_type}
277 )
278 # Create outbox folder for sent items pending sync
279 EmailFolder.objects.get_or_create(
280 tenant_id=account.tenant_id,
281 account=account,
282 name='OUTBOX',
283 defaults={'folder_type': 'outbox'}
284 )
285 else:
286 # IMAP folders
287 password = EmailAccountService.decrypt_password(account.encrypted_password)
288 try:
289 with IMAPService(
290 host=account.imap_host,
291 port=account.imap_port,
292 username=account.email_address,
293 password=password,
294 ssl=account.use_ssl
295 ) as imap:
296 folders = imap.list_folders()
297 for folder_data in folders:
298 EmailFolder.objects.get_or_create(
299 tenant_id=account.tenant_id,
300 account=account,
301 remote_id=folder_data.get('name'),
302 defaults={
303 'name': folder_data.get('name', 'Unknown'),
304 'folder_type': EmailAccountService._detect_folder_type(
305 folder_data.get('name', '')
306 ),
307 }
308 )
309 except:
310 pass
312 @staticmethod
313 def get_folder_stats(account_id: UUID, tenant_id: UUID) -> dict:
314 """Get folder statistics."""
315 folders = EmailFolder.objects.filter(
316 tenant_id=tenant_id,
317 account_id=account_id
318 )
320 return {
321 folder.name: {
322 'unread': folder.unread_count,
323 'total': folder.messages.count(),
324 }
325 for folder in folders
326 }
329class EmailMessageService:
330 """Service for managing email messages."""
332 @staticmethod
333 def fetch_messages(account: EmailAccount, folder_name: str = 'INBOX',
334 page: int = 1, per_page: int = 50) -> List[EmailMessage]:
335 """Fetch messages from server and store locally."""
336 if account.connection_type == 'oauth':
337 return EmailMessageService._fetch_oauth_messages(account, folder_name, page, per_page)
338 else:
339 return EmailMessageService._fetch_imap_messages(account, folder_name, page, per_page)
341 @staticmethod
342 def _fetch_imap_messages(account: EmailAccount, folder_name: str,
343 page: int, per_page: int) -> List[EmailMessage]:
344 """Fetch messages via IMAP."""
345 password = EmailAccountService.decrypt_password(account.encrypted_password)
347 messages = []
348 try:
349 with IMAPService(
350 host=account.imap_host,
351 port=account.imap_port,
352 username=account.email_address,
353 password=password,
354 ssl=account.use_ssl
355 ) as imap:
356 # Get tenant_id and account_id as hex strings
357 tenant_id_hex = account.tenant_id.hex if hasattr(account.tenant_id, 'hex') else str(account.tenant_id).replace('-', '')
358 account_id_hex = account.id.hex if hasattr(account.id, 'hex') else str(account.id).replace('-', '')
360 # Use raw SQL to find folder (Django ORM fails due to UUID vs integer mismatch)
361 from django.db import connection
362 with connection.cursor() as cursor:
363 cursor.execute(
364 "SELECT id FROM email_emailfolder WHERE tenant_id = %s AND account_id = %s AND name = %s LIMIT 1",
365 [tenant_id_hex, account_id_hex, folder_name]
366 )
367 row = cursor.fetchone()
368 if not row:
369 return []
370 folder_id = row[0]
372 # Get remote messages
373 remote_messages = imap.fetch_messages(folder_name, page, per_page)
375 for remote_msg in remote_messages:
376 message_id = remote_msg.get('message_id', '')
378 # Use raw SQL to check if message exists
379 with connection.cursor() as cursor:
380 cursor.execute(
381 "SELECT id FROM email_emailmessage WHERE tenant_id = %s AND account_id = %s AND message_id = %s",
382 [tenant_id_hex, account_id_hex, message_id]
383 )
384 row = cursor.fetchone()
386 if row:
387 # Update existing message read status and folder
388 is_read = 'UNSEEN' not in remote_msg.get('flags', [])
389 cursor.execute(
390 "UPDATE email_emailmessage SET is_read = %s, folder_id = %s WHERE id = %s",
391 [1 if is_read else 0, folder_id, row[0]]
392 )
393 else:
394 # Create new message with folder_id
395 msg_id = remote_msg.get('id') or message_id
396 now = timezone.now()
397 cursor.execute(
398 """INSERT INTO email_emailmessage
399 (id, created_at, updated_at, tenant_id, account_id, folder_id, message_id,
400 from_address, from_name, subject, snippet, is_read, is_starred, is_draft,
401 is_sent, sent_at, received_at)
402 VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
403 [
404 msg_id,
405 now, now,
406 tenant_id_hex, account_id_hex, folder_id,
407 message_id,
408 remote_msg.get('from_address', ''),
409 remote_msg.get('from_name', ''),
410 remote_msg.get('subject', ''),
411 remote_msg.get('snippet', ''),
412 1 if 'UNSEEN' not in remote_msg.get('flags', []) else 0,
413 1 if '\\Starred' in remote_msg.get('flags', []) else 0,
414 1 if '\\Draft' in remote_msg.get('flags', []) else 0,
415 1 if '\\Sent' in remote_msg.get('flags', []) else 0,
416 remote_msg.get('sent_at'),
417 remote_msg.get('sent_at'),
418 ]
419 )
421 # Fetch the updated messages for this folder using raw SQL
422 with connection.cursor() as cursor:
423 cursor.execute(
424 """SELECT CAST(id AS TEXT), message_id, subject, snippet, from_address, from_name,
425 to_addresses, cc_addresses, is_read, is_starred, is_draft, is_sent,
426 sent_at, received_at, folder_id, account_id
427 FROM email_emailmessage
428 WHERE tenant_id = %s AND account_id = %s AND folder_id = %s
429 ORDER BY received_at DESC
430 LIMIT %s OFFSET %s""",
431 [tenant_id_hex, account_id_hex, folder_id, per_page, (page - 1) * per_page]
432 )
433 rows = cursor.fetchall()
435 for row in rows:
436 messages.append({
437 'id': row[0],
438 'message_id': row[1],
439 'subject': row[2],
440 'snippet': row[3],
441 'from_address': row[4],
442 'from_name': row[5],
443 'to_addresses': row[6] or [],
444 'cc_addresses': row[7],
445 'is_read': bool(row[8]),
446 'is_starred': bool(row[9]),
447 'is_draft': bool(row[10]),
448 'is_sent': bool(row[11]),
449 'sent_at': row[12],
450 'received_at': row[13],
451 'folder_id': row[14],
452 'account_id': row[15],
453 })
455 except Exception as e:
456 print(f"Error fetching IMAP messages: {e}")
457 import traceback
458 traceback.print_exc()
460 return messages
462 @staticmethod
463 def _fetch_oauth_messages(account: EmailAccount, folder_name: str,
464 page: int, per_page: int) -> List[EmailMessage]:
465 """Fetch messages via OAuth."""
466 try:
467 gmail = GmailOAuthService()
468 gmail.creds = account.oauth_token
470 if not gmail.connect():
471 return []
473 remote_messages = gmail.list_messages(label=folder_name.upper(), max_results=per_page)
475 folder = EmailFolder.objects.filter(
476 tenant_id=account.tenant_id,
477 account=account,
478 name=folder_name.upper()
479 ).first()
481 messages = []
482 for remote_msg in remote_messages:
483 msg, _ = EmailMessage.objects.update_or_create(
484 tenant_id=account.tenant_id,
485 account=account,
486 message_id=remote_msg.get('message_id', ''),
487 defaults={
488 'folder': folder,
489 'from_address': remote_msg.get('from_address', ''),
490 'from_name': remote_msg.get('from_name'),
491 'subject': remote_msg.get('subject', ''),
492 'snippet': remote_msg.get('snippet', ''),
493 'is_read': remote_msg.get('is_read', True),
494 'sent_at': remote_msg.get('sent_at'),
495 'received_at': remote_msg.get('sent_at'),
496 }
497 )
498 messages.append(msg)
500 return messages
502 except Exception as e:
503 print(f"Error fetching OAuth messages: {e}")
504 return []
506 @staticmethod
507 def get_message_detail(account: EmailAccount, message_id: UUID) -> Optional[EmailMessage]:
508 """Get message with full body and attachments."""
509 try:
510 msg = EmailMessage.objects.get(
511 id=message_id,
512 tenant_id=account.tenant_id,
513 account=account
514 )
516 # If body is empty, fetch from server
517 if not msg.body_text and msg.message_id and account.connection_type == 'imap':
518 password = EmailAccountService.decrypt_password(account.encrypted_password)
520 with IMAPService(
521 host=account.imap_host,
522 port=account.imap_port,
523 username=account.email_address,
524 password=password,
525 ssl=account.use_ssl
526 ) as imap:
527 # Find message by UID or Message-ID
528 folder_name = msg.folder.name if msg.folder else 'INBOX'
529 full_msg = imap.fetch_message(msg.uid or msg.message_id)
531 if full_msg:
532 msg.body_text = full_msg.get('body_text', '')
533 msg.body_html = full_msg.get('body_html', '')
534 msg.save()
536 return msg
538 except EmailMessage.DoesNotExist:
539 return None
541 @staticmethod
542 def send_email(account: EmailAccount, to_addresses: List[dict], subject: str,
543 body: str, cc: List[dict] = None, bcc: List[dict] = None,
544 is_html: bool = False) -> Tuple[bool, str]:
545 """Send an email via SMTP or OAuth."""
546 if account.connection_type == 'oauth':
547 return EmailMessageService._send_oauth_email(account, to_addresses, subject, body, cc, bcc, is_html)
548 else:
549 return EmailMessageService._send_smtp_email(account, to_addresses, subject, body, cc, bcc, is_html)
551 @staticmethod
552 def _send_smtp_email(account: EmailAccount, to_addresses: List[dict], subject: str,
553 body: str, cc: List[dict], bcc: List[dict],
554 is_html: bool) -> Tuple[bool, str]:
555 """Send email via SMTP."""
556 password = EmailAccountService.decrypt_password(account.encrypted_password)
558 try:
559 with SMTPService(
560 host=account.smtp_host,
561 port=account.smtp_port,
562 username=account.email_address,
563 password=password,
564 use_ssl=(account.smtp_port == 465),
565 from_email=account.email_address,
566 from_name=account.display_name
567 ) as smtp:
568 to_emails = [a['email'] for a in to_addresses]
569 cc_emails = [a['email'] for a in cc] if cc else None
570 bcc_emails = [a['email'] for a in bcc] if bcc else None
572 success, error_msg = smtp.send_email(
573 to_addresses=to_emails,
574 subject=subject,
575 body=body,
576 cc_addresses=cc_emails,
577 bcc_addresses=bcc_emails,
578 is_html=is_html
579 )
581 if success:
582 # Save to sent folder
583 folder = EmailFolder.objects.filter(
584 tenant_id=account.tenant_id,
585 account=account,
586 folder_type='sent'
587 ).first()
589 EmailMessage.objects.create(
590 tenant_id=account.tenant_id,
591 account=account,
592 folder=folder,
593 message_id=f"<sent-{datetime.now().timestamp()}@{account.email_address}>",
594 from_address=account.email_address,
595 from_name=account.display_name,
596 to_addresses=to_addresses,
597 cc_addresses=cc,
598 bcc_addresses=bcc,
599 subject=subject,
600 body_text=body if not is_html else None,
601 body_html=body if is_html else None,
602 sent_at=timezone.now(),
603 received_at=timezone.now(),
604 is_sent=True,
605 )
606 return True, "Email sent successfully"
607 else:
608 return False, error_msg or "Failed to send email"
610 except Exception as e:
611 return False, f"Send failed: {str(e)}"
613 @staticmethod
614 def _send_oauth_email(account: EmailAccount, to_addresses: List[dict], subject: str,
615 body: str, cc: List[dict], bcc: List[dict],
616 is_html: bool) -> Tuple[bool, str]:
617 """Send email via Gmail OAuth."""
618 try:
619 gmail = GmailOAuthService()
620 gmail.creds = account.oauth_token
622 if not gmail.connect():
623 return False, "Failed to connect to Gmail"
625 to_email = to_addresses[0]['email'] if to_addresses else ''
627 if gmail.send_message(to_email, subject, body, is_html):
628 return True, "Email sent successfully"
629 else:
630 return False, "Failed to send email"
632 except Exception as e:
633 return False, f"OAuth send failed: {str(e)}"
635 @staticmethod
636 def mark_as_read(message: EmailMessage) -> bool:
637 """Mark message as read."""
638 message.is_read = True
639 message.save()
641 # Also mark on server
642 if message.account and message.account.connection_type == 'imap' and message.uid:
643 password = EmailAccountService.decrypt_password(message.account.encrypted_password)
644 try:
645 with IMAPService(
646 host=message.account.imap_host,
647 port=message.account.imap_port,
648 username=message.account.email_address,
649 password=password,
650 ssl=message.account.use_ssl
651 ) as imap:
652 imap.mark_as_read(message.uid)
653 except:
654 pass
656 return True
658 @staticmethod
659 def mark_as_unread(message: EmailMessage) -> bool:
660 """Mark message as unread."""
661 message.is_read = False
662 message.save()
663 return True
665 @staticmethod
666 def delete_message(message: EmailMessage) -> bool:
667 """Delete a message (move to trash)."""
668 if message.account and message.account.connection_type == 'imap' and message.uid:
669 password = EmailAccountService.decrypt_password(message.account.encrypted_password)
670 try:
671 with IMAPService(
672 host=message.account.imap_host,
673 port=message.account.imap_port,
674 username=message.account.email_address,
675 password=password,
676 ssl=message.account.use_ssl
677 ) as imap:
678 imap.delete_message(message.uid)
679 except:
680 pass
682 message.delete()
683 return True