"""Author RAG Chatbot SaaS — Analytics Aggregator. Runs as a Celery beat task every hour. Aggregates raw analytics_events into pre-computed analytics_daily rows. RULE: This task is idempotent — safe to re-run for the same date. RULE: Uses INSERT ... ON CONFLICT (upsert) so partial runs don't create duplicates. RULE: Failures here MUST NOT affect live chat — analytics is non-critical path. """ import asyncio import uuid from datetime import date, datetime, timedelta, timezone import structlog logger = structlog.get_logger(__name__) async def _run_daily_rollup(target_date: date) -> dict: """Aggregate analytics_events for target_date into analytics_daily. For each author that had events on target_date: - Count total chat turns - Count unique visitor fingerprints - Sum tokens used - Sum link clicks - Average session turns - Average faithfulness score Args: target_date: The calendar date to aggregate (UTC). Returns: Dict summarising rows written: {authors_processed, rows_written}. """ from sqlalchemy import func, select, text from app.dependencies import _get_session_factory from app.models.analytics import AnalyticsEvent, AnalyticsDaily async with _get_session_factory()() as db: # Fetch per-author aggregates for the target date day_start = datetime(target_date.year, target_date.month, target_date.day, tzinfo=timezone.utc) day_end = day_start + timedelta(days=1) result = await db.execute( select( AnalyticsEvent.author_id, func.count(AnalyticsEvent.id).label("total_chats"), func.count(func.distinct(AnalyticsEvent.visitor_fingerprint)).label("unique_visitors"), func.sum(AnalyticsEvent.prompt_tokens + AnalyticsEvent.completion_tokens).label("total_tokens"), # DESIGN-1 fix: was summing link_shown (button displayed) — should be # link_clicked (button actually clicked). Fixes inflated conversion funnel. func.sum(AnalyticsEvent.link_clicked.cast("int")).label("total_link_clicks"), func.avg(AnalyticsEvent.faithfulness_score).label("avg_faithfulness"), ).where( AnalyticsEvent.timestamp >= day_start, AnalyticsEvent.timestamp < day_end, ).group_by(AnalyticsEvent.author_id) ) rows = result.all() if not rows: logger.info("Aggregator: no events for date", date=target_date.isoformat()) return {"authors_processed": 0, "rows_written": 0} rows_written = 0 for row in rows: author_id = row.author_id # Upsert into analytics_daily (raw SQL for portability) await db.execute( text(""" INSERT INTO analytics_daily (id, author_id, date, total_chats, unique_visitors, total_tokens_used, total_link_clicks, avg_session_turns, avg_faithfulness_score) VALUES (:row_id, :author_id, :date, :total_chats, :unique_visitors, :total_tokens, :total_link_clicks, 0.0, :avg_faithfulness) ON CONFLICT (author_id, date) DO UPDATE SET total_chats = EXCLUDED.total_chats, unique_visitors = EXCLUDED.unique_visitors, total_tokens_used = EXCLUDED.total_tokens_used, total_link_clicks = EXCLUDED.total_link_clicks, avg_faithfulness_score = EXCLUDED.avg_faithfulness_score """), { "row_id": str(uuid.uuid4()), "author_id": author_id, "date": target_date.isoformat(), "total_chats": int(row.total_chats or 0), "unique_visitors": int(row.unique_visitors or 0), "total_tokens": int(row.total_tokens or 0), "total_link_clicks": int(row.total_link_clicks or 0), "avg_faithfulness": float(row.avg_faithfulness or 0.0), }, ) rows_written += 1 await db.commit() logger.info( "Daily rollup complete", date=target_date.isoformat(), authors=len(rows), rows_written=rows_written, ) return {"authors_processed": len(rows), "rows_written": rows_written} def run_daily_rollup(target_date: date | None = None) -> dict: """Synchronous wrapper — called by Celery analytics_task. Args: target_date: Date to aggregate. Defaults to yesterday (UTC). Returns: Aggregation result dict. """ if target_date is None: target_date = (datetime.now(timezone.utc) - timedelta(days=1)).date() logger.info("Starting daily analytics rollup", date=target_date.isoformat()) return asyncio.run(_run_daily_rollup(target_date))