"""Author RAG — Admin Analytics Service. Business logic for admin dashboard analytics endpoints. Delegates all DB access to AnalyticsRepository and VisitorRepository. """ from datetime import datetime, timedelta, timezone from sqlalchemy.ext.asyncio import AsyncSession from app.repositories.analytics_repo import AnalyticsRepository from app.repositories.visitor_repo import VisitorRepository class AdminAnalyticsService: """Orchestrates admin analytics queries and response shaping.""" def __init__(self, db: AsyncSession) -> None: self._db = db self._analytics = AnalyticsRepository(db) self._visitors = VisitorRepository(db) def _since(self, days: int) -> datetime: return datetime.now(timezone.utc) - timedelta(days=days) async def daily_sessions(self, author_id: str, days: int) -> dict: """Return daily session counts for dashboard charts.""" since = self._since(days) rows = await self._analytics.daily_session_counts(author_id, since) return { "daily_sessions": [{"date": str(row.date), "count": row.count} for row in rows], "period_days": days, } async def conversion_funnel(self, author_id: str, days: int) -> dict: """Return conversion funnel data.""" since = self._since(days) counts = await self._analytics.funnel_counts(author_id, since) return { "funnel": [ {"stage": "Widget Loads", "count": counts["total_sessions"]}, {"stage": "Chats Started", "count": counts["chat_started"]}, {"stage": "Book Discussed", "count": counts["book_discussed"]}, {"stage": "Link Shown", "count": counts["link_shown"]}, {"stage": "Link Clicked", "count": counts["link_clicked"]}, ], "period_days": days, } async def intent_distribution(self, author_id: str, days: int) -> dict: """Return distribution of detected intent labels.""" since = self._since(days) rows = await self._analytics.intent_distribution(author_id, since) return { "intents": [{"intent": intent or "unknown", "count": count} for intent, count in rows], "period_days": days, } async def unanswered_questions(self, author_id: str, days: int) -> dict: """Return boundary/fallback trigger stats.""" since = self._since(days) daily_rows = await self._analytics.boundary_daily_trend(author_id, since) total_boundary, total_turns = await self._analytics.boundary_totals(author_id, since) boundary_rate = round(total_boundary / total_turns * 100, 1) if total_turns > 0 else 0.0 return { "period_days": days, "total_turns": total_turns, "unanswered_count": total_boundary, "unanswered_rate_percent": boundary_rate, "daily_trend": [{"date": str(row.date), "count": row.boundary_count} for row in daily_rows], "tip": ( "Add Q&A pairs or upload more book content to reduce unanswered rate." if boundary_rate > 20 else "Good coverage! Unanswered rate is below 20%." ), } async def book_engagement(self, author_id: str, days: int) -> dict: """Return per-book engagement breakdown.""" since = self._since(days) books = await self._analytics.ready_books_map(author_id) if not books: return {"books": [], "period_days": days} stats_rows = await self._analytics.per_book_engagement( author_id, list(books.keys()), since, ) book_stats = [] for row in stats_rows: shown = int(row.link_shown_count or 0) clicked = int(row.link_clicked_count or 0) turns = int(row.total_turns or 0) boundary = int(row.boundary_count or 0) book_stats.append({ "book_id": row.book_id, "book_title": books.get(row.book_id, "Unknown"), "unique_sessions": int(row.unique_sessions or 0), "total_turns": turns, "link_shown": shown, "link_clicks": clicked, "click_through_rate": round(clicked / shown * 100, 1) if shown > 0 else 0.0, "boundary_rate": round(boundary / turns * 100, 1) if turns > 0 else 0.0, "avg_faithfulness": round(float(row.avg_faithfulness or 0), 3), }) return {"books": book_stats, "period_days": days} async def activity_heatmap(self, author_id: str, days: int) -> dict: """Return hourly × weekday activity heatmap.""" since = self._since(days) rows = await self._analytics.activity_heatmap_rows(author_id, since) heatmap = [] max_count = 0 for row in rows: iso_weekday = (int(row.weekday) - 1) % 7 count = int(row.count) max_count = max(max_count, count) heatmap.append({"weekday": iso_weekday, "hour": int(row.hour), "count": count}) return { "heatmap": sorted(heatmap, key=lambda x: (x["weekday"], x["hour"])), "weekday_labels": ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"], "max_count": max_count, "period_days": days, } async def visitor_stats(self, author_id: str, days: int) -> dict: """Return unique visitor counts with new vs returning breakdown.""" since = self._since(days) total = await self._visitors.count_total(author_id, since) returning = await self._visitors.count_returning(author_id, since) new_visitors = total - returning returning_rate = round(returning / total * 100, 1) if total > 0 else 0.0 return { "period_days": days, "total_visitors": total, "new_visitors": new_visitors, "returning_visitors": returning, "returning_rate_pct": returning_rate, } async def geo_breakdown(self, author_id: str, days: int) -> dict: """Return geographic distribution of visitors.""" since = self._since(days) countries = await self._visitors.get_country_distribution(author_id, since) cities = await self._visitors.get_city_distribution(author_id, since) return {"period_days": days, "countries": countries, "cities": cities} async def device_breakdown(self, author_id: str, days: int) -> dict: """Return device, browser, and OS distribution.""" since = self._since(days) breakdown = await self._visitors.get_device_distribution(author_id, since) return {"period_days": days, **breakdown} async def session_stats(self, author_id: str, days: int) -> dict: """Return session-level aggregate statistics.""" since = self._since(days) stats = await self._analytics.session_aggregate_stats(author_id, since) avg_rating = stats["avg_rating"] return { "period_days": days, "total_sessions": stats["total_sessions"], "avg_turns_per_session": round(float(stats["avg_turns"] or 0), 2), "avg_rating": round(float(avg_rating), 2) if avg_rating else None, "rated_sessions": stats["rated_count"], }