""" Daily waitlist report — sends an evening summary email via Brevo. Run as a cron job (e.g. every day at 21:00): python -m app.daily_report Requires REPORT_RECIPIENT_EMAIL and BREVO_API_KEY in .env. """ import asyncio import datetime import logging import httpx from sqlalchemy import func, select, and_ from app.config import settings from app.db import async_session from app.models import WaitlistEntry logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) async def gather_stats() -> dict: """Collect today's waitlist statistics.""" now = datetime.datetime.now(datetime.timezone.utc) today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) async with async_session() as db: # Total counts total = (await db.execute(select(func.count(WaitlistEntry.id)))).scalar() or 0 confirmed_total = (await db.execute( select(func.count(WaitlistEntry.id)).where(WaitlistEntry.confirmed == True) # noqa: E712 )).scalar() or 0 anonymized_total = (await db.execute( select(func.count(WaitlistEntry.id)).where(WaitlistEntry.anonymized_at != None) # noqa: E711 )).scalar() or 0 pending = total - confirmed_total - anonymized_total # Today's activity new_today = (await db.execute( select(func.count(WaitlistEntry.id)).where( and_( WaitlistEntry.created_at >= today_start, WaitlistEntry.anonymized_at == None, # noqa: E711 ) ) )).scalar() or 0 confirmed_today = (await db.execute( select(func.count(WaitlistEntry.id)).where( and_( WaitlistEntry.confirmed == True, # noqa: E712 WaitlistEntry.consent_given_at >= today_start, ) ) )).scalar() or 0 anonymized_today = (await db.execute( select(func.count(WaitlistEntry.id)).where( and_( WaitlistEntry.anonymized_at != None, # noqa: E711 WaitlistEntry.anonymized_at >= today_start, ) ) )).scalar() or 0 return { "date": now.strftime("%B %d, %Y"), "total": total, "confirmed_total": confirmed_total, "pending": pending, "anonymized_total": anonymized_total, "new_today": new_today, "confirmed_today": confirmed_today, "anonymized_today": anonymized_today, "conversion_rate": round(confirmed_total / total * 100, 1) if total > 0 else 0, } async def send_report() -> bool: """Gather stats and send the daily report email.""" if not settings.REPORT_RECIPIENT_EMAIL: logger.warning("REPORT_RECIPIENT_EMAIL not set — skipping daily report") return False if not settings.brevo_configured: logger.warning("Brevo not configured — skipping daily report") return False stats = await gather_stats() html = _report_html(stats) payload = { "sender": { "name": settings.BREVO_SENDER_NAME, "email": settings.BREVO_SENDER_EMAIL, }, "to": [{"email": settings.REPORT_RECIPIENT_EMAIL}], "subject": f"Waitlist report — {stats['date']}", "htmlContent": html, } headers = { "api-key": settings.BREVO_API_KEY, "Content-Type": "application/json", "Accept": "application/json", } try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.post( "https://api.brevo.com/v3/smtp/email", headers=headers, json=payload, ) resp.raise_for_status() logger.info("Daily report sent to %s", settings.REPORT_RECIPIENT_EMAIL) return True except httpx.HTTPError: logger.exception("Failed to send daily report") return False def _report_html(s: dict) -> str: """adiuvAI-branded daily report email.""" def _stat_cell(label: str, value, color: str = "#040404") -> str: return f"""\

{value}

{label}

""" def _row(label: str, value, highlight: bool = False) -> str: color = "#e5a94e" if highlight else "#040404" return f"""\ {label} {value} """ return f"""\ Waitlist Report — {s['date']}
{s['new_today']} new signups, {s['confirmed_total']} confirmed total ({s['conversion_rate']}% conversion)
 
adiuvAI

Daily Report

{s['date']}

{_stat_cell("Confirmed", s['confirmed_total'], "#040404")} {_stat_cell("Pending", s['pending'], "#8a8ea9")} {_stat_cell("Conversion", f"{s['conversion_rate']}%", "#e5a94e")}

Today's Activity

{_row("New signups", f"+{s['new_today']}", highlight=s['new_today'] > 0)} {_row("Confirmed", f"+{s['confirmed_today']}", highlight=s['confirmed_today'] > 0)} {_row("Anonymized (expired/unsub)", s['anonymized_today'])}

All Time

{_row("Total records", s['total'])} {_row("Confirmed", s['confirmed_total'], highlight=True)} {_row("Pending confirmation", s['pending'])} {_row("Anonymized", s['anonymized_total'])}
 

Automated report from adiuvai.com

""" if __name__ == "__main__": asyncio.run(send_report())