Spaces:
Running
Running
File size: 2,326 Bytes
3c2b4ae |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
import logging
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from core.settings import cnf
from database.connections import get_mongo_client
logger = logging.getLogger(__name__)
def _send_email(to_email: str, subject: str, body: str) -> bool:
"""Low-level Gmail SMTP email sender."""
try:
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
EMAIL_USER = cnf.EMAIL_USER
EMAIL_PASS = cnf.EMAIL_PASS
if not all([to_email, subject, body, EMAIL_USER, EMAIL_PASS]):
logger.error("Missing required email parameters")
return False
msg = MIMEMultipart()
msg["From"] = EMAIL_USER
msg["To"] = to_email
msg["Subject"] = subject
msg.attach(MIMEText(body, "plain"))
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
server.starttls()
server.login(EMAIL_USER, EMAIL_PASS)
server.sendmail(EMAIL_USER, to_email, msg.as_string())
logger.info(f"Email sent to {to_email}")
return True
except Exception as e:
logger.error(f"Failed to send email: {e}")
return False
def all_tasks_completed_notification(user_email: str, product_name: str) -> bool:
"""
Send success email if the user has completed image, script, and video tasks.
"""
db = get_mongo_client()
if db is None:
logger.error("MongoDB not available for notifications")
return False
has_image = db["results"].count_documents({"created_by": user_email, "type": {"$in": ["generation", "variation"]}}) > 0
has_script = db["script_generator"].count_documents({"created_by": user_email}) > 0
has_video = db["video_analyses"].count_documents({"created_by": user_email}) > 0
if has_image and has_script and has_video:
body = f"""Hey {user_email},
🎉 Congratulations!
You have successfully:
- Generated Images
- Generated Scripts
- Analyzed Videos
Your work for **{product_name}** has been completed successfully.
You can now review all assets in the AI Library.
Best regards,
The Creative AdGenesis Team
"""
return _send_email(user_email, "All Tasks Completed Successfully ", body)
logger.info(f"Tasks not fully completed for {user_email}")
return False
|