""" Persona API endpoints for ALwrity. Handles writing persona generation, management, and platform-specific adaptations. """ from fastapi import HTTPException, Depends from pydantic import BaseModel, Field from typing import Dict, Any, List, Optional from datetime import datetime from loguru import logger from sqlalchemy.orm import Session from services.persona_analysis_service import PersonaAnalysisService from services.database import get_db class PersonaGenerationRequest(BaseModel): """Request model for persona generation.""" onboarding_session_id: Optional[int] = Field(None, description="Specific onboarding session ID to use") force_regenerate: bool = Field(False, description="Force regeneration even if persona exists") class PersonaResponse(BaseModel): """Response model for persona data.""" persona_id: int persona_name: str archetype: str core_belief: str confidence_score: float platforms: List[str] created_at: str class PlatformPersonaResponse(BaseModel): """Response model for platform-specific persona.""" platform_type: str sentence_metrics: Dict[str, Any] lexical_features: Dict[str, Any] content_format_rules: Dict[str, Any] engagement_patterns: Dict[str, Any] platform_best_practices: Dict[str, Any] class PersonaGenerationResponse(BaseModel): """Response model for persona generation result.""" success: bool persona_id: Optional[int] = None message: str confidence_score: Optional[float] = None data_sufficiency: Optional[float] = None platforms_generated: List[str] = [] class LinkedInPersonaValidationRequest(BaseModel): """Request model for LinkedIn persona validation.""" persona_data: Dict[str, Any] class LinkedInPersonaValidationResponse(BaseModel): """Response model for LinkedIn persona validation.""" is_valid: bool quality_score: float completeness_score: float professional_context_score: float linkedin_optimization_score: float missing_fields: List[str] incomplete_fields: List[str] recommendations: List[str] quality_issues: List[str] strengths: List[str] validation_details: Dict[str, Any] # Dependency to get persona service def get_persona_service() -> PersonaAnalysisService: """Get the persona analysis service instance.""" return PersonaAnalysisService() async def generate_persona(user_id: int, request: PersonaGenerationRequest): """Generate a new writing persona from onboarding data.""" try: logger.info(f"Generating persona for user {user_id}") persona_service = get_persona_service() # Check if persona already exists and force_regenerate is False if not request.force_regenerate: existing_personas = persona_service.get_user_personas(user_id) if existing_personas: return PersonaGenerationResponse( success=False, message="Persona already exists. Use force_regenerate=true to create a new one.", persona_id=existing_personas[0]["id"] ) # Generate new persona result = persona_service.generate_persona_from_onboarding( user_id=user_id, onboarding_session_id=request.onboarding_session_id ) if "error" in result: return PersonaGenerationResponse( success=False, message=result["error"] ) return PersonaGenerationResponse( success=True, persona_id=result["persona_id"], message="Persona generated successfully", confidence_score=result["analysis_metadata"]["confidence_score"], data_sufficiency=result["analysis_metadata"].get("data_sufficiency", 0.0), platforms_generated=list(result["platform_personas"].keys()) ) except Exception as e: logger.error(f"Error generating persona: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to generate persona: {str(e)}") async def get_user_personas(user_id: str): """Get all personas for a user using PersonaData.""" try: from services.persona_data_service import PersonaDataService persona_service = PersonaDataService() all_personas = persona_service.get_all_platform_personas(user_id) return { "personas": all_personas, "total_count": len(all_personas), "platforms": list(all_personas.keys()) } except Exception as e: logger.error(f"Error getting user personas: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to get personas: {str(e)}") async def get_persona_details(user_id: str, persona_id: int): """Get detailed information about a specific persona using PersonaData.""" try: from services.persona_data_service import PersonaDataService persona_service = PersonaDataService() persona_data = persona_service.get_user_persona_data(user_id) if not persona_data: raise HTTPException(status_code=404, detail="Persona not found") # Return the complete persona data with all platforms return { "persona_id": persona_data.get('id'), "core_persona": persona_data.get('core_persona', {}), "platform_personas": persona_data.get('platform_personas', {}), "quality_metrics": persona_data.get('quality_metrics', {}), "selected_platforms": persona_data.get('selected_platforms', []), "created_at": persona_data.get('created_at'), "updated_at": persona_data.get('updated_at') } except HTTPException: raise except Exception as e: logger.error(f"Error getting persona details: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to get persona details: {str(e)}") async def get_platform_persona(user_id: str, platform: str): """Get persona adaptation for a specific platform using PersonaData.""" try: from services.persona_data_service import PersonaDataService persona_service = PersonaDataService() platform_persona = persona_service.get_platform_persona(user_id, platform) if not platform_persona: raise HTTPException(status_code=404, detail=f"No persona found for platform {platform}") return platform_persona except HTTPException: raise except Exception as e: logger.error(f"Error getting platform persona: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to get platform persona: {str(e)}") async def get_persona_summary(user_id: str): """Get persona summary for a user using PersonaData.""" try: from services.persona_data_service import PersonaDataService persona_service = PersonaDataService() summary = persona_service.get_persona_summary(user_id) return summary except Exception as e: logger.error(f"Error getting persona summary: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to get persona summary: {str(e)}") async def update_persona(user_id: str, persona_id: int, update_data: Dict[str, Any]): """Update an existing persona using PersonaData.""" try: from services.persona_data_service import PersonaDataService from models.onboarding import PersonaData persona_service = PersonaDataService() # For PersonaData, we update the core_persona field if 'core_persona' in update_data: # Get current persona data persona_data = persona_service.get_user_persona_data(user_id) if not persona_data: raise HTTPException(status_code=404, detail="Persona not found") # Update core persona with new data persona_service.db.query(PersonaData).filter( PersonaData.id == persona_data.get('id') ).update({ 'core_persona': update_data['core_persona'], 'updated_at': datetime.utcnow() }) persona_service.db.commit() persona_service.db.close() return { "message": "Persona updated successfully", "persona_id": persona_data.get('id'), "updated_at": datetime.utcnow().isoformat() } else: raise HTTPException(status_code=400, detail="core_persona field is required for updates") except HTTPException: raise except Exception as e: logger.error(f"Error updating persona: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to update persona: {str(e)}") async def delete_persona(user_id: str, persona_id: int): """Delete a persona using PersonaData (not recommended, personas are generated during onboarding).""" try: from services.persona_data_service import PersonaDataService from models.onboarding import PersonaData persona_service = PersonaDataService() # Get persona data persona_data = persona_service.get_user_persona_data(user_id) if not persona_data: raise HTTPException(status_code=404, detail="Persona not found") # For PersonaData, we mark it as deleted by setting a flag # Note: In production, you might want to add a deleted_at field or similar # For now, we'll just return a warning that deletion is not recommended logger.warning(f"Delete persona requested for user {user_id}. PersonaData deletion is not recommended.") return { "message": "Persona deletion requested. Note: Personas are generated during onboarding and deletion is not recommended.", "persona_id": persona_data.get('id'), "alternative": "Consider re-running onboarding to regenerate persona if needed." } except HTTPException: raise except Exception as e: logger.error(f"Error deleting persona: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to delete persona: {str(e)}") async def update_platform_persona(user_id: str, platform: str, update_data: Dict[str, Any]): """Update platform-specific persona fields using PersonaData.""" try: from services.persona_data_service import PersonaDataService persona_service = PersonaDataService() # Update platform-specific persona data success = persona_service.update_platform_persona(user_id, platform, update_data) if not success: raise HTTPException(status_code=404, detail=f"No platform persona found for platform {platform}") return { "message": "Platform persona updated successfully", "platform": platform, "user_id": user_id, "updated_at": datetime.utcnow().isoformat() } except HTTPException: raise except Exception as e: logger.error(f"Error updating platform persona: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to update platform persona: {str(e)}") async def generate_platform_persona(user_id: str, platform: str, db_session): """ Generate a platform-specific persona from core persona and save it. Args: user_id: User ID from auth platform: Platform name (facebook, linkedin, etc.) db_session: Database session from FastAPI dependency injection Returns: Generated platform persona with validation results """ try: logger.info(f"Generating {platform} persona for user {user_id}") # Import services from services.persona_data_service import PersonaDataService from services.onboarding.database_service import OnboardingDatabaseService persona_data_service = PersonaDataService(db_session=db_session) onboarding_service = OnboardingDatabaseService(db=db_session) # Get core persona data persona_data = persona_data_service.get_user_persona_data(user_id) if not persona_data: raise HTTPException(status_code=404, detail="Core persona not found") core_persona = persona_data.get('core_persona', {}) if not core_persona: raise HTTPException(status_code=404, detail="Core persona data is empty") # Get onboarding data for context onboarding_session = onboarding_service.get_session_by_user(user_id) if not onboarding_session: raise HTTPException(status_code=404, detail="Onboarding session not found") # Get website analysis for context website_analysis = onboarding_service.get_website_analysis(user_id) research_prefs = onboarding_service.get_research_preferences(user_id) onboarding_data = { "website_url": website_analysis.get('website_url', '') if website_analysis else '', "writing_style": website_analysis.get('writing_style', {}) if website_analysis else {}, "content_characteristics": website_analysis.get('content_characteristics', {}) if website_analysis else {}, "target_audience": website_analysis.get('target_audience', '') if website_analysis else '', "research_preferences": research_prefs or {} } # Generate platform persona based on platform generated_persona = None platform_service = None if platform.lower() == 'facebook': from services.persona.facebook.facebook_persona_service import FacebookPersonaService platform_service = FacebookPersonaService() generated_persona = platform_service.generate_facebook_persona( core_persona, onboarding_data ) elif platform.lower() == 'linkedin': from services.persona.linkedin.linkedin_persona_service import LinkedInPersonaService platform_service = LinkedInPersonaService() generated_persona = platform_service.generate_linkedin_persona( core_persona, onboarding_data ) else: raise HTTPException(status_code=400, detail=f"Unsupported platform: {platform}") # Check for errors in generation if "error" in generated_persona: raise HTTPException(status_code=500, detail=generated_persona["error"]) # Save the generated platform persona to database success = persona_data_service.save_platform_persona(user_id, platform, generated_persona) if not success: raise HTTPException(status_code=500, detail=f"Failed to save {platform} persona") logger.info(f"✅ Successfully generated and saved {platform} persona for user {user_id}") return { "success": True, "platform": platform, "persona": generated_persona, "validation_results": generated_persona.get("validation_results", {}), "quality_score": generated_persona.get("validation_results", {}).get("quality_score", 0) } except HTTPException: raise except Exception as e: logger.error(f"Error generating {platform} persona: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to generate {platform} persona: {str(e)}") async def check_facebook_persona(user_id: str, db: Session): """Check if Facebook persona exists for user.""" try: from services.persona_data_service import PersonaDataService persona_data_service = PersonaDataService(db_session=db) persona_data = persona_data_service.get_user_persona_data(user_id) if not persona_data: return { "has_persona": False, "has_core_persona": False, "message": "No persona data found", "onboarding_completed": False } platform_personas = persona_data.get('platform_personas', {}) facebook_persona = platform_personas.get('facebook') if platform_personas else None # Check if core persona exists has_core_persona = bool(persona_data.get('core_persona')) # Assume onboarding is completed if persona data exists onboarding_completed = True return { "has_persona": bool(facebook_persona), "has_core_persona": has_core_persona, "persona": facebook_persona, "onboarding_completed": onboarding_completed } except Exception as e: logger.error(f"Error checking Facebook persona for user {user_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) async def validate_persona_generation_readiness(user_id: int): """Check if user has sufficient onboarding data for persona generation.""" try: persona_service = get_persona_service() # Get onboarding data onboarding_data = persona_service._collect_onboarding_data(user_id) if not onboarding_data: return { "ready": False, "message": "No onboarding data found. Please complete onboarding first.", "missing_steps": ["All onboarding steps"], "data_sufficiency": 0.0 } data_sufficiency = persona_service._calculate_data_sufficiency(onboarding_data) missing_steps = [] if not onboarding_data.get("website_analysis"): missing_steps.append("Website Analysis (Step 2)") if not onboarding_data.get("research_preferences"): missing_steps.append("Research Preferences (Step 3)") ready = data_sufficiency >= 50.0 # Require at least 50% data sufficiency return { "ready": ready, "message": "Ready for persona generation" if ready else "Insufficient data for reliable persona generation", "missing_steps": missing_steps, "data_sufficiency": data_sufficiency, "recommendations": [ "Complete website analysis for better style detection", "Provide research preferences for content type optimization" ] if not ready else [] } except Exception as e: logger.error(f"Error validating persona generation readiness: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to validate readiness: {str(e)}") async def generate_persona_preview(user_id: int): """Generate a preview of what the persona would look like without saving.""" try: persona_service = get_persona_service() # Get onboarding data onboarding_data = persona_service._collect_onboarding_data(user_id) if not onboarding_data: raise HTTPException(status_code=400, detail="No onboarding data available") # Generate core persona (without saving) core_persona = persona_service._generate_core_persona(onboarding_data) if "error" in core_persona: raise HTTPException(status_code=400, detail=core_persona["error"]) # Generate sample platform adaptation (just one for preview) sample_platform = "linkedin" platform_preview = persona_service._generate_single_platform_persona( core_persona, sample_platform, onboarding_data ) return { "preview": { "identity": core_persona.get("identity", {}), "linguistic_fingerprint": core_persona.get("linguistic_fingerprint", {}), "tonal_range": core_persona.get("tonal_range", {}), "sample_platform": { "platform": sample_platform, "adaptation": platform_preview } }, "confidence_score": core_persona.get("confidence_score", 0.0), "data_sufficiency": persona_service._calculate_data_sufficiency(onboarding_data) } except HTTPException: raise except Exception as e: logger.error(f"Error generating persona preview: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to generate preview: {str(e)}") async def get_supported_platforms(): """Get list of supported platforms for persona generation.""" return { "platforms": [ { "id": "twitter", "name": "Twitter/X", "description": "Microblogging platform optimized for short, engaging content", "character_limit": 280, "optimal_length": "120-150 characters" }, { "id": "linkedin", "name": "LinkedIn", "description": "Professional networking platform for thought leadership content", "character_limit": 3000, "optimal_length": "150-300 words" }, { "id": "instagram", "name": "Instagram", "description": "Visual-first platform with engaging captions", "character_limit": 2200, "optimal_length": "125-150 words" }, { "id": "facebook", "name": "Facebook", "description": "Social networking platform for community engagement", "character_limit": 63206, "optimal_length": "40-80 words" }, { "id": "blog", "name": "Blog Posts", "description": "Long-form content optimized for SEO and engagement", "word_count": "800-2000 words", "seo_optimized": True }, { "id": "medium", "name": "Medium", "description": "Publishing platform for storytelling and thought leadership", "word_count": "1000-3000 words", "storytelling_focus": True }, { "id": "substack", "name": "Substack", "description": "Newsletter platform for building subscriber relationships", "format": "email newsletter", "subscription_focus": True } ] } class LinkedInOptimizationRequest(BaseModel): """Request model for LinkedIn algorithm optimization.""" persona_data: Dict[str, Any] class LinkedInOptimizationResponse(BaseModel): """Response model for LinkedIn algorithm optimization.""" optimized_persona: Dict[str, Any] optimization_applied: bool optimization_details: Dict[str, Any] async def validate_linkedin_persona( request: LinkedInPersonaValidationRequest, persona_service: PersonaAnalysisService = Depends(get_persona_service) ): """ Validate LinkedIn persona data for completeness and quality. This endpoint provides comprehensive validation of LinkedIn persona data, including core fields, LinkedIn-specific optimizations, professional context, and content quality assessments. """ try: logger.info("Validating LinkedIn persona data") # Get LinkedIn persona service from services.persona.linkedin.linkedin_persona_service import LinkedInPersonaService linkedin_service = LinkedInPersonaService() # Validate the persona data validation_results = linkedin_service.validate_linkedin_persona(request.persona_data) logger.info(f"LinkedIn persona validation completed: Quality Score: {validation_results['quality_score']:.1f}%") return LinkedInPersonaValidationResponse(**validation_results) except Exception as e: logger.error(f"Error validating LinkedIn persona: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to validate LinkedIn persona: {str(e)}" ) async def optimize_linkedin_persona( request: LinkedInOptimizationRequest, persona_service: PersonaAnalysisService = Depends(get_persona_service) ): """ Optimize LinkedIn persona data for maximum algorithm performance. This endpoint applies comprehensive LinkedIn algorithm optimization to persona data, including content quality optimization, multimedia strategy, engagement optimization, timing optimization, and professional context optimization. """ try: logger.info("Optimizing LinkedIn persona for algorithm performance") # Get LinkedIn persona service from services.persona.linkedin.linkedin_persona_service import LinkedInPersonaService linkedin_service = LinkedInPersonaService() # Apply algorithm optimization optimized_persona = linkedin_service.optimize_for_linkedin_algorithm(request.persona_data) # Extract optimization details optimization_details = optimized_persona.get("algorithm_optimization", {}) logger.info("✅ LinkedIn persona algorithm optimization completed successfully") return LinkedInOptimizationResponse( optimized_persona=optimized_persona, optimization_applied=True, optimization_details={ "optimization_categories": list(optimization_details.keys()), "total_optimization_strategies": sum(len(strategies) if isinstance(strategies, list) else 1 for category in optimization_details.values() for strategies in category.values() if isinstance(category, dict)), "optimization_timestamp": datetime.utcnow().isoformat() } ) except Exception as e: logger.error(f"Error optimizing LinkedIn persona: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to optimize LinkedIn persona: {str(e)}" ) class FacebookPersonaValidationRequest(BaseModel): """Request model for Facebook persona validation.""" persona_data: Dict[str, Any] class FacebookPersonaValidationResponse(BaseModel): """Response model for Facebook persona validation.""" is_valid: bool quality_score: float completeness_score: float facebook_optimization_score: float engagement_strategy_score: float content_format_score: float audience_targeting_score: float community_building_score: float missing_fields: List[str] incomplete_fields: List[str] recommendations: List[str] quality_issues: List[str] strengths: List[str] validation_details: Dict[str, Any] class FacebookOptimizationRequest(BaseModel): """Request model for Facebook algorithm optimization.""" persona_data: Dict[str, Any] class FacebookOptimizationResponse(BaseModel): """Response model for Facebook algorithm optimization.""" optimized_persona: Dict[str, Any] optimization_applied: bool optimization_details: Dict[str, Any] async def validate_facebook_persona( request: FacebookPersonaValidationRequest, persona_service: PersonaAnalysisService = Depends(get_persona_service) ): """ Validate Facebook persona data for completeness and quality. This endpoint provides comprehensive validation of Facebook persona data, including core fields, Facebook-specific optimizations, engagement strategies, content formats, audience targeting, and community building assessments. """ try: logger.info("Validating Facebook persona data") # Get Facebook persona service from services.persona.facebook.facebook_persona_service import FacebookPersonaService facebook_service = FacebookPersonaService() # Validate the persona data validation_results = facebook_service.validate_facebook_persona(request.persona_data) logger.info(f"Facebook persona validation completed: Quality Score: {validation_results['quality_score']:.1f}%") return FacebookPersonaValidationResponse(**validation_results) except Exception as e: logger.error(f"Error validating Facebook persona: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to validate Facebook persona: {str(e)}" ) async def optimize_facebook_persona( request: FacebookOptimizationRequest, persona_service: PersonaAnalysisService = Depends(get_persona_service) ): """ Optimize Facebook persona data for maximum algorithm performance. This endpoint applies comprehensive Facebook algorithm optimization to persona data, including engagement optimization, content quality optimization, timing optimization, audience targeting optimization, and community building strategies. """ try: logger.info("Optimizing Facebook persona for algorithm performance") # Get Facebook persona service from services.persona.facebook.facebook_persona_service import FacebookPersonaService facebook_service = FacebookPersonaService() # Apply algorithm optimization optimized_persona = facebook_service.optimize_for_facebook_algorithm(request.persona_data) # Extract optimization details optimization_details = optimized_persona.get("algorithm_optimization", {}) logger.info("✅ Facebook persona algorithm optimization completed successfully") # Use the optimization metadata from the service optimization_metadata = optimized_persona.get("optimization_metadata", {}) return FacebookOptimizationResponse( optimized_persona=optimized_persona, optimization_applied=True, optimization_details={ "optimization_categories": optimization_metadata.get("optimization_categories", []), "total_optimization_strategies": optimization_metadata.get("total_optimization_strategies", 0), "optimization_timestamp": optimization_metadata.get("optimization_timestamp", datetime.utcnow().isoformat()) } ) except Exception as e: logger.error(f"Error optimizing Facebook persona: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to optimize Facebook persona: {str(e)}" )