Spaces:
Running
Running
| """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"], | |
| } | |