kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add brand analysis columns to onboarding database and migration scripts

Browse Source
This commit is contained in:
ajaysi
2025-10-11 17:05:42 +05:30
parent b1ebe1034e
commit 1df12a64a2
25 changed files with 2415 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
backend/api/onboarding_utils/onboarding_summary_service.py
View File

@@ -9,6 +9,7 @@ from loguru import logger
from services.api_key_manager import get_api_key_manager
from services.database import get_db
from services.onboarding_database_service import OnboardingDatabaseService
from services.website_analysis_service import WebsiteAnalysisService
from services.research_preferences_service import ResearchPreferencesService
from services.persona_analysis_service import PersonaAnalysisService
@@ -23,14 +24,10 @@ class OnboardingSummaryService:
Args:
user_id: Clerk user ID from authenticated request
"""
# Convert Clerk user ID to integer for database compatibility
try:
self.user_id_int = int(user_id.replace('user_', '').replace('-', '')[:8], 16) % 2147483647
except:
self.user_id_int = hash(user_id) % 2147483647
self.user_id = user_id # Store Clerk user ID (string)
self.db_service = OnboardingDatabaseService()
self.user_id = user_id # Store original Clerk ID for logging
self.session_id = self.user_id_int # Use user ID as session ID for backwards compatibility
logger.info(f"OnboardingSummaryService initialized for user {user_id} (database mode)")
async def get_onboarding_summary(self) -> Dict[str, Any]:
"""Get comprehensive onboarding summary for FinalStep."""
@@ -69,40 +66,75 @@ class OnboardingSummaryService:
raise HTTPException(status_code=500, detail="Internal server error")
def _get_api_keys(self) -> Dict[str, Any]:
"""Get configured API keys."""
api_manager = get_api_key_manager()
return api_manager.get_all_keys()
def _get_website_analysis(self) -> Optional[Dict[str, Any]]:
"""Get website analysis data."""
"""Get configured API keys from database."""
try:
db = next(get_db())
website_service = WebsiteAnalysisService(db)
return website_service.get_analysis_by_session(self.session_id)
api_keys = self.db_service.get_api_keys(self.user_id, db)
logger.info(f"Retrieved {len(api_keys)} API keys from database for user {self.user_id}")
return api_keys
except Exception as e:
logger.warning(f"Could not get website analysis: {str(e)}")
logger.error(f"Error getting API keys from database: {e}")
return {}
def _get_website_analysis(self) -> Optional[Dict[str, Any]]:
"""Get website analysis data from database (Step 2)."""
try:
db = next(get_db())
website_data = self.db_service.get_website_analysis(self.user_id, db)
if website_data:
logger.info(f"Retrieved website analysis from database for user {self.user_id}")
else:
logger.warning(f"No website analysis found in database for user {self.user_id}")
return website_data
except Exception as e:
logger.error(f"Error getting website analysis from database: {e}")
return None
def _get_research_preferences(self) -> Optional[Dict[str, Any]]:
"""Get research preferences data."""
"""Get research preferences data from database (Step 3)."""
try:
db = next(get_db())
research_service = ResearchPreferencesService(db)
return research_service.get_research_preferences(self.session_id)
research_data = self.db_service.get_research_preferences(self.user_id, db)
if research_data:
logger.info(f"Retrieved research preferences from database for user {self.user_id}")
else:
logger.warning(f"No research preferences found in database for user {self.user_id}")
return research_data
except Exception as e:
logger.warning(f"Could not get research preferences: {str(e)}")
logger.error(f"Error getting research preferences from database: {e}")
return None
def _get_personalization_settings(self, research_preferences: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""Get personalization settings from research preferences."""
if not research_preferences:
"""Get personalization settings from Step 4 (Persona) database."""
try:
# Try to get from Step 4 (Persona) in database
db = next(get_db())
persona_data = self.db_service.get_persona_data(self.user_id, db)
if persona_data:
logger.info(f"Retrieved persona data from database for user {self.user_id}")
# Extract personalization settings from persona data
if 'corePersona' in persona_data:
core_persona = persona_data.get('corePersona', {})
return {
'writing_style': core_persona.get('linguistic_fingerprint', {}).get('tone', 'Professional'),
'tone': core_persona.get('tonal_range', {}).get('primary_tone', 'Formal'),
'brand_voice': core_persona.get('identity', {}).get('voice', 'Trustworthy and Expert')
}
# Fallback to research preferences if persona data not available
if research_preferences:
logger.info(f"Using research preferences as fallback for personalization")
return {
'writing_style': research_preferences.get('writing_style', {}).get('tone', 'Professional'),
'tone': research_preferences.get('writing_style', {}).get('voice', 'Formal'),
'brand_voice': research_preferences.get('writing_style', {}).get('complexity', 'Trustworthy and Expert')
}
return None
except Exception as e:
logger.error(f"Error getting personalization settings from database: {e}")
return None
return {
'writing_style': research_preferences.get('writing_style', {}).get('tone', 'Professional'),
'tone': research_preferences.get('writing_style', {}).get('voice', 'Formal'),
'brand_voice': research_preferences.get('writing_style', {}).get('complexity', 'Trustworthy and Expert')
}
def _check_persona_readiness(self, website_analysis: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""Check if persona can be generated."""