"""Author RAG — Settings Service. Business logic for author account settings: password, profile, widget config, personality, notifications, embed token, and token usage. """ from datetime import datetime, timezone import bcrypt from fastapi import HTTPException from sqlalchemy import update as sql_update from sqlalchemy.ext.asyncio import AsyncSession from app.core.access.token_crypto import create_subscription_token from app.models.user import User from app.repositories.access_repo import AccessRepository from app.schemas.admin import ( NotificationUpdate, PasswordChangeRequest, PersonalityUpdate, ProfileUpdate, WidgetConfigUpdate, ) from app.services.token_budget import tokens_remaining, usage_summary_or_empty class SettingsService: """Orchestrates author settings and account management.""" def __init__(self, db: AsyncSession) -> None: self._db = db self._access = AccessRepository(db) async def change_password(self, user: User, body: PasswordChangeRequest) -> dict: """Validate current password and update to a new bcrypt hash.""" try: valid = bcrypt.checkpw( body.current_password.encode(), user.password_hash.encode(), ) except Exception: valid = False if not valid: raise HTTPException(400, "Current password is incorrect") new_hash = bcrypt.hashpw(body.new_password.encode(), bcrypt.gensalt(12)).decode() await self._db.execute( sql_update(User).where(User.id == user.id).values(password_hash=new_hash) ) await self._db.commit() return {"message": "Password updated"} @staticmethod def get_widget_config(user: User) -> dict: """Return current widget configuration.""" return { "bot_name": user.bot_name, "welcome_message": user.welcome_message, "theme": user.widget_theme, "position": user.widget_position, "auto_open_delay": user.widget_auto_open_delay, "is_active": user.chatbot_is_active, } async def update_widget_config(self, user: User, body: WidgetConfigUpdate) -> dict: """Update widget configuration fields that were explicitly set.""" values = {} field_map = { "bot_name": "bot_name", "welcome_message": "welcome_message", "theme": "widget_theme", "position": "widget_position", "auto_open_delay": "widget_auto_open_delay", "is_active": "chatbot_is_active", } for field_name, col in field_map.items(): val = getattr(body, field_name, None) if val is not None: values[col] = val if values: await self._db.execute( sql_update(User).where(User.id == user.id).values(**values) ) await self._db.commit() return {"message": "Widget config updated"} @staticmethod def get_profile(user: User) -> dict: """Return current author profile.""" return { "full_name": user.full_name or "", "email": user.email, "website": user.website_url or "", "bio": user.bio or "", "timezone": user.timezone, } async def update_profile(self, user: User, body: ProfileUpdate) -> dict: """Update author profile fields.""" values = {} if body.full_name is not None: values["full_name"] = body.full_name if body.website is not None: values["website_url"] = body.website if body.bio is not None: values["bio"] = body.bio if body.timezone is not None: values["timezone"] = body.timezone if values: await self._db.execute( sql_update(User).where(User.id == user.id).values(**values) ) await self._db.commit() return {"message": "Profile updated"} @staticmethod def get_personality(user: User) -> dict: """Return bot personality settings.""" return { "response_style": user.response_style, "fallback_message": user.fallback_message, "out_of_scope_message": user.out_of_scope_message, "welcome_message": user.welcome_message, } async def update_personality(self, user: User, body: PersonalityUpdate) -> dict: """Update bot personality settings.""" values = {} if body.response_style is not None: values["response_style"] = body.response_style if body.fallback_message is not None: values["fallback_message"] = body.fallback_message[:500] if body.out_of_scope_message is not None: values["out_of_scope_message"] = body.out_of_scope_message[:500] if values: await self._db.execute( sql_update(User).where(User.id == user.id).values(**values) ) await self._db.commit() return {"message": "Personality settings updated"} @staticmethod def get_notifications(user: User) -> dict: """Return notification preferences.""" return { "weekly_digest": user.notify_weekly_digest, "token_alerts": user.notify_token_alerts, "new_conversation": user.notify_new_conversation, "subscription_expiry": user.notify_subscription_expiry, } async def update_notifications(self, user: User, body: NotificationUpdate) -> dict: """Update notification preferences.""" mapping = { "weekly_digest": "notify_weekly_digest", "token_alerts": "notify_token_alerts", "new_conversation": "notify_new_conversation", "subscription_expiry": "notify_subscription_expiry", } values = {} for field_name, col in mapping.items(): val = getattr(body, field_name, None) if val is not None: values[col] = val if values: await self._db.execute( sql_update(User).where(User.id == user.id).values(**values) ) await self._db.commit() return {"message": "Notification preferences updated"} async def get_embed_token(self, user: User) -> dict: """Return the active subscription token for embedding the widget.""" now = datetime.now(timezone.utc) now_naive = now.replace(tzinfo=None) access = await self._access.get_active_for_author(user.id) if not access: return { "active": False, "token": None, "message": ( "No active subscription. Contact your administrator " "to activate your chatbot." ), } token = create_subscription_token( author_id=user.id, grant_id=access.id, granted_at=access.granted_at, expires_at=access.expires_at, ) exp = access.expires_at if exp and exp.tzinfo is not None: exp = exp.replace(tzinfo=None) days_remaining = max(0, (exp - now_naive).days) if exp else 0 return { "active": True, "token": token, "grant_id": access.id, "plan": access.plan, "expires_at": access.expires_at.isoformat(), "days_remaining": days_remaining, "tokens_remaining": tokens_remaining(access), } async def get_token_usage(self, author_id: str) -> dict: """Return token budget and consumption for the active subscription.""" access = await self._access.get_active_for_author(author_id) return usage_summary_or_empty(access)