"""Author RAG — Dashboard Service. Business logic for admin dashboard stats and session moderation. Delegates DB access to SessionRepository and BookRepository. """ from datetime import datetime, timezone from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.platform_listing import BookPlatformListing from app.repositories.book_repo import BookRepository from app.repositories.session_repo import SessionListStats, SessionRepository from app.services.platform_presence import count_missing_listings class DashboardService: """Orchestrates dashboard stats and session management.""" def __init__(self, db: AsyncSession) -> None: self._db = db self._sessions = SessionRepository(db) self._books = BookRepository(db) async def get_dashboard_stats(self, author_id: str) -> dict: """Return high-level dashboard stats for the author.""" books = await self._books.list_for_author(author_id) total_sessions = await self._sessions.count_for_author(author_id) book_ids = [b.id for b in books] gaps_books = 0 total_missing = 0 if book_ids: listings_result = await self._db.execute( select(BookPlatformListing).where( BookPlatformListing.author_id == author_id, BookPlatformListing.book_id.in_(book_ids), ) ) by_book: dict[str, list[BookPlatformListing]] = {} for row in listings_result.scalars().all(): by_book.setdefault(row.book_id, []).append(row) for listings in by_book.values(): missing = count_missing_listings(listings) if missing > 0: gaps_books += 1 total_missing += missing return { "total_books": len(books), "active_books": sum(1 for b in books if b.status == "active"), "total_sessions": total_sessions, "total_clicks": 0, "distribution_gaps": { "books_with_gaps": gaps_books, "total_missing_platforms": total_missing, }, } @staticmethod def _session_summary( s: object, stats: SessionListStats | None = None, mask_fp: bool = False, ) -> dict: """Serialize a ChatSession for API response.""" fp = s.visitor_fingerprint if mask_fp and fp: fp = fp[:8] + "..." row = { "id": s.id, "visitor_fingerprint": fp, "visitor_name": s.visitor_name, "visitor_email": getattr(s, "visitor_email", None), "ip_address": getattr(s, "ip_address", None), "country": s.country_name, "city": s.city, "device_type": s.device_type, "turn_count": s.turn_count, "rating": s.rating, "created_at": s.created_at.isoformat() if s.created_at else None, "updated_at": s.updated_at.isoformat() if getattr(s, "updated_at", None) else None, "blocked": s.blocked, "message_count": stats.message_count if stats else 0, "last_message_at": ( stats.last_message_at.isoformat() if stats and stats.last_message_at else None ), "first_user_preview": stats.first_user_preview if stats else None, } return row async def list_sessions(self, author_id: str, limit: int, offset: int) -> dict: """List reader sessions with pagination.""" total = await self._sessions.count_for_author(author_id) sessions = await self._sessions.list_for_author(author_id, limit, offset) stats_map = await self._sessions.get_list_stats_batch([s.id for s in sessions]) return { "sessions": [ self._session_summary(s, stats_map.get(s.id)) for s in sessions ], "total": total, "offset": offset, "limit": limit, } async def search_sessions( self, author_id: str, query: str, page: int, per_page: int, ) -> dict: """Full-text search across chat messages.""" if not query.strip(): return {"sessions": [], "total": 0} sessions, total = await self._sessions.search_by_message_content( author_id, query, page, per_page, ) stats_map = await self._sessions.get_list_stats_batch([s.id for s in sessions]) return { "sessions": [ self._session_summary(s, stats_map.get(s.id), mask_fp=True) for s in sessions ], "total": total, "query": query, "page": page, } async def block_session(self, author_id: str, session_id: str) -> bool: """Block a visitor session. Returns False if not found.""" session = await self._sessions.set_blocked(author_id, session_id, True) return session is not None async def unblock_session(self, author_id: str, session_id: str) -> bool: """Unblock a visitor session. Returns False if not found.""" session = await self._sessions.set_blocked(author_id, session_id, False) return session is not None async def get_transcript( self, author_id: str, session_id: str, page: int, per_page: int, ) -> dict | None: """Return paginated message transcript. None if session not found.""" session = await self._sessions.get_for_author(author_id, session_id) if not session: return None total = await self._sessions.message_count(session_id) messages = await self._sessions.list_messages(session_id, page, per_page) return { "session": { "id": session.id, "visitor_fingerprint": session.visitor_fingerprint[:8] + "...", "visitor_name": session.visitor_name, "visitor_email": session.visitor_email, "ip_address": session.ip_address, "country": session.country_name, "city": session.city, "device_type": session.device_type, "browser": session.browser, "os": session.os, "rating": session.rating, "blocked": session.blocked, "created_at": session.created_at.isoformat() if session.created_at else None, "summary": session.summary, }, "messages": [ { "id": m.id, "role": m.role, "content": m.content, "intent": m.intent, "intent_confidence": m.intent_confidence, "faithfulness_score": m.faithfulness_score, "hallucination_detected": m.hallucination_detected, "prompt_tokens": m.prompt_tokens, "completion_tokens": m.completion_tokens, "response_ms": m.response_ms, "annotation": m.annotation, "flag_type": m.flag_type, "user_feedback": m.user_feedback, "created_at": m.created_at.isoformat() if m.created_at else None, } for m in messages ], "pagination": { "page": page, "per_page": per_page, "total": total, "total_pages": (total + per_page - 1) // per_page, }, } async def annotate_message( self, author_id: str, session_id: str, message_id: str, annotation: str, ) -> dict | None: """Add admin annotation to a message. None if not found.""" msg = await self._sessions.get_message_for_author( author_id, session_id, message_id, ) if not msg: return None prefix = f"[{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M')}] " existing = msg.annotation or "" msg.annotation = (prefix + annotation + "\n" + existing).strip() await self._db.commit() return {"message": "Annotation saved", "annotation": msg.annotation} async def flag_message( self, author_id: str, session_id: str, message_id: str, flag_type: str, ) -> dict | None: """Flag a message. None if not found.""" msg = await self._sessions.get_message_for_author( author_id, session_id, message_id, ) if not msg: return None msg.flag_type = flag_type await self._db.commit() return {"message": "Flag updated", "flag_type": flag_type}