"""Daily digest email scheduler — daemon thread, fires once per day at 09:00. Uses only stdlib (threading, time, smtplib, email). No APScheduler or schedule package required. Started from app.py after extensions are initialised. """ from __future__ import annotations import logging import os import smtplib import threading import time from datetime import datetime, timedelta from zoneinfo import ZoneInfo from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText logger = logging.getLogger("apex") _IST = ZoneInfo("Asia/Kolkata") _FIRE_HOUR = 9 # 09:00 IST _CHECK_INTERVAL = 60 # seconds between clock checks def _seconds_until(hour: int) -> float: """Seconds until the next occurrence of `hour:00:00`.""" now = datetime.now(tz=_IST) target = now.replace(hour=hour, minute=0, second=0, microsecond=0) if now >= target: # already past today's window — schedule for tomorrow target = (target + timedelta(days=1)).replace(hour=_FIRE_HOUR, minute=0, second=0, microsecond=0) return (target - now).total_seconds() def _send_digest_email( digest_text: str, recipients: list[str], smtp_host: str, smtp_port: int, smtp_from: str, ) -> None: """Send a plain-text + HTML digest email via the configured SMTP relay.""" today = datetime.now(tz=_IST).date() subject = f"Apex Daily Sales Digest — {today.strftime('%d %b %Y')}" html_body = f"""\

Apex Sales Digest

{today.strftime('%A, %d %B %Y')}

{digest_text.replace(chr(10), '
')}

Sent automatically by Apex — Rashi Peripherals Sales Intelligence

""" msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = smtp_from msg["To"] = ", ".join(recipients) msg.attach(MIMEText(digest_text, "plain")) msg.attach(MIMEText(html_body, "html")) try: with smtplib.SMTP(smtp_host, smtp_port, timeout=15) as server: server.sendmail(smtp_from, recipients, msg.as_string()) logger.info("[DigestScheduler] Digest sent to %d recipient(s).", len(recipients)) except Exception as exc: logger.error("[DigestScheduler] SMTP send failed: %s", exc) class DigestScheduler: """Background daemon that sends the daily sales digest at 09:00.""" def __init__( self, digest_service, # DailyDigestService instance redis_client, recipients: list[str] | None = None, ): self._digest_service = digest_service self._redis = redis_client self._recipients: list[str] = recipients or [] self._smtp_host = os.getenv("SMTP_HOST", "localhost") self._smtp_port = int(os.getenv("SMTP_PORT", "25")) self._smtp_from = os.getenv("SMTP_FROM", "apex@rptechindia.com") self._thread: threading.Thread | None = None self._stop_event = threading.Event() def add_recipient(self, email: str) -> None: if email and email not in self._recipients: self._recipients.append(email) def _fired_today_key(self) -> str: return f"digest_scheduler:fired:{datetime.now(tz=_IST).date().isoformat()}" def _already_fired_today(self) -> bool: try: return bool(self._redis.get(self._fired_today_key())) except Exception: return False def _mark_fired_today(self) -> None: try: self._redis.set(self._fired_today_key(), "1", ex=86400) except Exception as exc: logger.warning("[DigestScheduler] Could not mark fired: %s", exc) def _run_loop(self) -> None: logger.info("[DigestScheduler] Started. Will fire at %02d:00 daily.", _FIRE_HOUR) while not self._stop_event.is_set(): now = datetime.now(tz=_IST) if now.hour == _FIRE_HOUR and not self._already_fired_today(): self._fire() self._stop_event.wait(_CHECK_INTERVAL) def _fire(self) -> None: if not self._recipients: logger.info("[DigestScheduler] No recipients configured — skipping send.") self._mark_fired_today() return # Use a generic tenant key for global digest tenant_id = os.getenv("DIGEST_TENANT_ID", "default") digest = self._digest_service.get_digest(tenant_id) text = self._digest_service.format_digest_text(digest) _send_digest_email( digest_text=text, recipients=self._recipients, smtp_host=self._smtp_host, smtp_port=self._smtp_port, smtp_from=self._smtp_from, ) self._mark_fired_today() def start(self) -> None: if self._thread and self._thread.is_alive(): return self._stop_event.clear() self._thread = threading.Thread( target=self._run_loop, name="DigestScheduler", daemon=True, ) self._thread.start() def stop(self) -> None: self._stop_event.set()