"""Author RAG Chatbot SaaS — Email Celery Tasks. All email sending is done here as background tasks — never blocks API. """ from datetime import datetime, timedelta, timezone import structlog from app.tasks.celery_app import celery_app logger = structlog.get_logger(__name__) @celery_app.task(name="app.tasks.email_task.send_weekly_digests") def send_weekly_digests(): """Send weekly digest emails to all authors who opted in.""" import asyncio asyncio.run(_run_weekly_digests()) async def _run_weekly_digests(): from sqlalchemy import select from app.dependencies import _get_session_factory from app.models.user import User from app.repositories.analytics_repo import AnalyticsRepository from app.repositories.access_repo import AccessRepository from app.services.email_service import EmailService from app.services.token_budget import percent_used email = EmailService() since = datetime.now(timezone.utc) - timedelta(days=7) async with _get_session_factory()() as db: result = await db.execute( select(User).where( User.role == "author", User.is_active == True, User.notify_weekly_digest == True, ) ) authors = result.scalars().all() analytics = AnalyticsRepository(db) access_repo = AccessRepository(db) sent = 0 for author in authors: try: stats = await analytics.weekly_digest_stats(author.id, since) access = await access_repo.get_active_for_author(author.id) token_pct = int(round(percent_used(access), 0)) if access else 0 email.send_weekly_digest( to=author.email, name=author.full_name or author.email, chat_count=stats["chat_count"], top_book=stats["top_book"], token_pct=token_pct, top_country=stats["top_country"], ) sent += 1 except Exception as e: logger.warning("Weekly digest email failed", author_id=author.id, error=str(e)) logger.info("Weekly digests sent", count=sent) @celery_app.task(name="app.tasks.email_task.send_token_warning") def send_token_warning(author_id: str, pct_used: int): """Send token budget warning email to an author.""" import asyncio asyncio.run(_send_token_warning(author_id, pct_used)) async def _send_token_warning(author_id: str, pct_used: int): from app.dependencies import _get_session_factory from app.repositories.access_repo import AccessRepository from app.repositories.user_repo import UserRepository from app.services.email_service import EmailService from app.services.token_budget import total_budget, tokens_remaining email = EmailService() async with _get_session_factory()() as db: user = await UserRepository(db).get_by_id(author_id) access = await AccessRepository(db).get_active_for_author(author_id) if user and user.notify_token_alerts and access: budget = total_budget(access) used = access.tokens_used or 0 remaining = tokens_remaining(access) days_left = 30 if access.expires_at: delta = (access.expires_at - datetime.now(timezone.utc)).days days_left = max(delta, 1) daily = used / max(30 - days_left, 1) if used else 0 forecast = "soon" if daily > 0 and remaining / daily < 7 else f"in ~{days_left} days" email.send_token_budget_warning( to=user.email, name=user.full_name or user.email, pct_used=pct_used, tokens_used=used, tokens_total=budget, forecast_date=forecast, ) @celery_app.task(name="app.tasks.email_task.send_publish_help_request") def send_publish_help_request(payload: dict): """Send publish-help request to support and optional author confirmation.""" from app.services.email_service import EmailService email = EmailService() try: email.send_publish_help_request(**payload) author_email = payload.get("author_email") if author_email: email.send_publish_help_confirmation( to=author_email, name=payload.get("author_name", author_email), book_title=payload.get("book_title", "your book"), platform_names=payload.get("platform_names", []), ) logger.info("Publish help emails sent", author_email=author_email) except Exception as exc: logger.error("Publish help email failed", error=str(exc)) raise @celery_app.task(name="app.tasks.email_task.send_contact_request") def send_contact_request(payload: dict): """Send general contact message to support.""" from app.services.email_service import EmailService email = EmailService() try: email.send_contact_request(**payload) logger.info("Contact email sent", author_email=payload.get("author_email")) except Exception as exc: logger.error("Contact email failed", error=str(exc)) raise