"""Author RAG Chatbot SaaS — Analytics Hourly Aggregation Task. Runs every hour via Celery beat. Delegates aggregation logic to core/analytics/aggregator.py. RULE: This task is idempotent — safe to re-run for the same date. RULE: Failures here MUST NOT affect live chat — analytics is non-critical path. """ import structlog from app.tasks.celery_app import celery_app logger = structlog.get_logger(__name__) @celery_app.task( name="app.tasks.analytics_task.aggregate_hourly", max_retries=3, default_retry_delay=60, acks_late=True, ) def aggregate_hourly(): """Roll up yesterday's + today's raw events into analytics_daily rows. Delegates to core/analytics/aggregator.run_daily_rollup(). Runs for today (partial day) and yesterday (full day) to catch any missed events. """ from datetime import date, timedelta from app.services.analytics_core.aggregator import run_daily_rollup try: today = date.today() yesterday = today - timedelta(days=1) # Always process yesterday (ensures complete daily record) yesterday_result = run_daily_rollup(yesterday) logger.info("Yesterday rollup complete", **yesterday_result, date=str(yesterday)) # Also process today (partial day — updated each hour) today_result = run_daily_rollup(today) logger.info("Today rollup complete", **today_result, date=str(today)) except Exception as exc: logger.error("Analytics aggregation failed", error=str(exc)) raise