Coverage for src/cstlcore/newsletter/services.py: 22%

1from datetime import datetime, timedelta 

2 

3import jwt 

4from loguru import logger 

5 

6from cstlcore.emails.services import send_email 

7from cstlcore.settings import settings 

8 

9 

10def generate_unsubscribe_token(email: str) -> str: 

11 payload = { 

12 "email": email, 

13 "exp": datetime.now() + timedelta(days=30), # Token valid for 30 days 

14 } 

15 return jwt.encode(payload, settings.newsletter.secret_key, algorithm="HS256") 

16 

17 

18def verify_unsubscribe_token(token: str) -> str | None: 

19 try: 

20 payload = jwt.decode( 

21 token, settings.newsletter.secret_key, algorithms=["HS256"] 

22 ) 

23 return payload["email"] 

24 except jwt.ExpiredSignatureError: 

25 return None 

26 except jwt.InvalidTokenError: 

27 return None 

28 

29 

30def send_email_newsletter( 

31 recipients: list[str], 

32 title: str, 

33 content: str, 

34 unsubscribe_base_link: str | None = None, 

35) -> dict[str, list[str]]: 

36 results = {"sent": [], "failed": []} 

37 

38 for recipient_email in recipients: 

39 # Generate user-specific unsubscribe link if base link provided 

40 unsubscribe_link = None 

41 if unsubscribe_base_link: 

42 # You would typically generate a unique token for each user 

43 token = generate_unsubscribe_token(recipient_email) 

44 unsubscribe_link = ( 

45 f"{unsubscribe_base_link}?email={recipient_email}&token={token}" 

46 ) 

47 

48 subject = f"{title} - Constellations Newsletter" 

49 

50 # Build the email body 

51 body = content + "\n\n" 

52 

53 # Add unsubscribe link if provided 

54 if unsubscribe_link: 

55 body += f""" 

56 If you no longer wish to receive these newsletters, you can unsubscribe here: 

57 {unsubscribe_link} 

58 

59 --- 

60 """ 

61 

62 body += "This is an automated email from Constellations." 

63 

64 if send_email(recipient_email, body, subject): 

65 results["sent"].append(recipient_email) 

66 else: 

67 results["failed"].append(recipient_email) 

68 

69 logger.info("\nNewsletter sending summary:") 

70 logger.info(f"Successfully sent: {len(results['sent'])}") 

71 logger.info(f"Failed to send: {len(results['failed'])}") 

72 

73 if results["failed"]: 

74 logger.error(f"Failed recipients: {', '.join(results['failed'])}") 

75 

76 return results