Added onboarding progress tracking & landing page

This commit is contained in:
ajaysi
2025-10-02 13:20:15 +05:30
parent e57d2577f8
commit 510b79bbf8
135 changed files with 25917 additions and 5768 deletions

View File

@@ -35,14 +35,14 @@ class StepData:
class OnboardingProgress:
"""Manages onboarding progress with persistence and validation."""
def __init__(self):
def __init__(self, progress_file: Optional[str] = None):
self.steps = self._initialize_steps()
self.current_step = 1
self.started_at = datetime.now().isoformat()
self.last_updated = datetime.now().isoformat()
self.is_completed = False
self.completed_at = None
self.progress_file = ".onboarding_progress.json"
self.progress_file = progress_file or ".onboarding_progress.json"
# Load existing progress if available
self.load_progress()
@@ -297,9 +297,11 @@ class APIKeyManager:
"mistral": None,
"tavily": None,
"serper": None,
"metaphor": None,
"metaphor": None, # legacy mapping for Exa, kept for backward compatibility
"exa": None,
"firecrawl": None,
"stability": None
"stability": None,
"copilotkit": None,
}
self.load_api_keys()
@@ -370,9 +372,9 @@ class APIKeyManager:
}
},
"Deep Search": {
"METAPHOR_API_KEY": {
"EXA_API_KEY": {
"url": "https://dashboard.exa.ai/login",
"description": "Enables advanced web search capabilities",
"description": "Exa (formerly Metaphor) for advanced web search",
"setup_steps": [
"Visit the Exa AI dashboard",
"Sign up for a free account",
@@ -402,6 +404,17 @@ class APIKeyManager:
"Generate your API key"
]
}
},
"UI": {
"COPILOTKIT_API_KEY": {
"url": "https://copilotkit.ai",
"description": "CopilotKit public API key for in-app assistant",
"setup_steps": [
"Sign up or log in to CopilotKit",
"Navigate to API Keys",
"Generate a public API key (ck_pub_...)"
]
}
}
}
@@ -443,9 +456,11 @@ class APIKeyManager:
"MISTRAL_API_KEY": "mistral",
"TAVILY_API_KEY": "tavily",
"SERPER_API_KEY": "serper",
"METAPHOR_API_KEY": "metaphor",
"METAPHOR_API_KEY": "metaphor", # legacy
"EXA_API_KEY": "exa",
"FIRECRAWL_API_KEY": "firecrawl",
"STABILITY_API_KEY": "stability"
"STABILITY_API_KEY": "stability",
"COPILOTKIT_API_KEY": "copilotkit",
}
for env_var, provider in env_mapping.items():
@@ -485,9 +500,11 @@ class APIKeyManager:
"mistral": "MISTRAL_API_KEY",
"tavily": "TAVILY_API_KEY",
"serper": "SERPER_API_KEY",
"metaphor": "METAPHOR_API_KEY",
"metaphor": "METAPHOR_API_KEY", # legacy
"exa": "EXA_API_KEY",
"firecrawl": "FIRECRAWL_API_KEY",
"stability": "STABILITY_API_KEY"
"stability": "STABILITY_API_KEY",
"copilotkit": "COPILOTKIT_API_KEY",
}
env_var = env_mapping.get(provider)
@@ -529,6 +546,7 @@ class APIKeyManager:
# Global instance for the application
_onboarding_progress = None
_user_onboarding_progress_cache: Dict[str, OnboardingProgress] = {}
def get_onboarding_progress() -> OnboardingProgress:
"""Get the global onboarding progress instance."""
@@ -536,6 +554,17 @@ def get_onboarding_progress() -> OnboardingProgress:
get_onboarding_progress._instance = OnboardingProgress()
return get_onboarding_progress._instance
def get_onboarding_progress_for_user(user_id: str) -> OnboardingProgress:
"""Get or create a per-user onboarding progress instance persisted to a user-specific file."""
global _user_onboarding_progress_cache
safe_user_id = ''.join([c if c.isalnum() or c in ('-', '_') else '_' for c in str(user_id)])
if safe_user_id in _user_onboarding_progress_cache:
return _user_onboarding_progress_cache[safe_user_id]
progress_file = f".onboarding_progress_{safe_user_id}.json"
instance = OnboardingProgress(progress_file=progress_file)
_user_onboarding_progress_cache[safe_user_id] = instance
return instance
def get_api_key_manager() -> APIKeyManager:
"""Get the global API key manager instance."""
if not hasattr(get_api_key_manager, '_instance'):

View File

@@ -71,9 +71,15 @@ class StyleDetectionLogic:
social_media = content.get('social_media', {})
content_structure = content.get('content_structure', {})
# Construct the enhanced analysis prompt
prompt = f"""Analyze the following website content for comprehensive writing style, tone, and characteristics.
This is a detailed analysis for content personalization and AI-powered content generation.
# Construct the enhanced analysis prompt (strict JSON, minified, stable keys)
prompt = f"""Analyze the following website content for comprehensive writing style, tone, and characteristics for personalization and AI generation.
RULES:
- Return ONE single-line MINIFIED JSON object only. No markdown, code fences, comments, or prose.
- Use EXACTLY the keys and ordering from the schema below. Do not add extra top-level keys.
- For unknown/unavailable fields use empty string "" or empty array [] and explain in meta.uncertainty.
- Keep text concise; avoid repeating input text.
- Assume token budget; consider only first 5000 chars of main_content and first 10 headings.
WEBSITE INFORMATION:
- Domain: {domain_info.get('domain_name', 'Unknown')}
@@ -91,10 +97,10 @@ class StyleDetectionLogic:
- Has Call-to-Action: {content_structure.get('has_call_to_action', False)}
CONTENT TO ANALYZE:
Title: {title}
Description: {description}
Main Content: {main_content[:5000]} # Enhanced content length
Key Headings: {headings[:10]} # First 10 headings for context
- Title: {title}
- Description: {description}
- Main Content (truncated): {main_content[:5000]}
- Key Headings (first 10): {headings[:10]}
ANALYSIS REQUIREMENTS:
1. Analyze the writing style, tone, and voice characteristics
@@ -106,68 +112,38 @@ class StyleDetectionLogic:
7. Consider the website type and industry context
8. Analyze social media presence impact on content style
IMPORTANT: Respond ONLY with a JSON object in the following format. Do not include any additional text, explanations, or markdown formatting:
REQUIRED JSON SCHEMA (stable key order):
{{
"writing_style": {{
"tone": "detailed tone description with context",
"voice": "active/passive with explanation",
"complexity": "simple/moderate/complex with reasoning",
"engagement_level": "low/medium/high with justification",
"brand_personality": "detailed brand personality analysis",
"formality_level": "casual/semi-formal/formal/professional",
"emotional_appeal": "rational/emotional/mixed with examples"
}},
"content_characteristics": {{
"sentence_structure": "detailed analysis of sentence patterns",
"vocabulary_level": "basic/intermediate/advanced with examples",
"paragraph_organization": "detailed structure analysis",
"content_flow": "detailed flow analysis",
"readability_score": "estimated readability level",
"content_density": "high/medium/low with reasoning",
"visual_elements_usage": "analysis of how visual elements complement text"
}},
"target_audience": {{
"demographics": ["detailed demographic analysis"],
"expertise_level": "beginner/intermediate/advanced with reasoning",
"industry_focus": "detailed industry analysis",
"geographic_focus": "detailed geographic analysis",
"psychographic_profile": "detailed psychographic analysis",
"pain_points": ["identified audience pain points"],
"motivations": ["identified audience motivations"]
}},
"content_type": {{
"primary_type": "detailed content type analysis",
"secondary_types": ["list of secondary content types"],
"purpose": "detailed content purpose analysis",
"call_to_action": "detailed CTA analysis",
"conversion_focus": "high/medium/low with reasoning",
"educational_value": "high/medium/low with reasoning"
}},
"brand_analysis": {{
"brand_voice": "detailed brand voice analysis",
"brand_values": ["identified brand values"],
"brand_positioning": "detailed positioning analysis",
"competitive_differentiation": "detailed differentiation analysis",
"trust_signals": ["identified trust elements"],
"authority_indicators": ["identified authority elements"]
}},
"content_strategy_insights": {{
"strengths": ["content strengths"],
"weaknesses": ["content weaknesses"],
"opportunities": ["content opportunities"],
"threats": ["content threats"],
"recommended_improvements": ["specific improvement suggestions"],
"content_gaps": ["identified content gaps"]
}},
"recommended_settings": {{
"writing_tone": "recommended tone for AI generation",
"target_audience": "recommended audience focus",
"content_type": "recommended content type",
"creativity_level": "low/medium/high with reasoning",
"geographic_location": "recommended geographic focus",
"industry_context": "recommended industry approach",
"brand_alignment": "recommended brand alignment strategy"
}}
"writing_style": {{
"tone": "", "voice": "", "complexity": "", "engagement_level": "",
"brand_personality": "", "formality_level": "", "emotional_appeal": ""
}},
"content_characteristics": {{
"sentence_structure": "", "vocabulary_level": "", "paragraph_organization": "",
"content_flow": "", "readability_score": "", "content_density": "",
"visual_elements_usage": ""
}},
"target_audience": {{
"demographics": [], "expertise_level": "", "industry_focus": "", "geographic_focus": "",
"psychographic_profile": "", "pain_points": [], "motivations": []
}},
"content_type": {{
"primary_type": "", "secondary_types": [], "purpose": "", "call_to_action": "",
"conversion_focus": "", "educational_value": ""
}},
"brand_analysis": {{
"brand_voice": "", "brand_values": [], "brand_positioning": "", "competitive_differentiation": "",
"trust_signals": [], "authority_indicators": []
}},
"content_strategy_insights": {{
"strengths": [], "weaknesses": [], "opportunities": [], "threats": [],
"recommended_improvements": [], "content_gaps": []
}},
"recommended_settings": {{
"writing_tone": "", "target_audience": "", "content_type": "", "creativity_level": "",
"geographic_location": "", "industry_context": "", "brand_alignment": ""
}},
"meta": {{"schema_version": "1.1", "confidence": 0.0, "notes": "", "uncertainty": {{"fields": []}}}}
}}
"""
@@ -290,22 +266,25 @@ class StyleDetectionLogic:
main_content = content.get("main_content", "")
prompt = f"""Analyze the following content for recurring writing patterns and style characteristics.
Focus on identifying patterns in sentence structure, vocabulary usage, and writing techniques.
Content: {main_content[:3000]}
IMPORTANT: Respond ONLY with a JSON object in the following format:
prompt = f"""Analyze the content for recurring writing patterns and style characteristics.
RULES:
- Return ONE single-line MINIFIED JSON object only. No markdown, code fences, comments, or prose.
- Use EXACTLY the keys and ordering from the schema below. No extra top-level keys.
- If uncertain, set empty values and list field names in meta.uncertainty.fields.
- Keep responses concise and avoid quoting long input spans.
Content (truncated to 3000 chars): {main_content[:3000]}
REQUIRED JSON SCHEMA (stable key order):
{{
"patterns": {{
"sentence_length": "short/medium/long",
"vocabulary_patterns": ["list of patterns"],
"rhetorical_devices": ["list of devices used"],
"paragraph_structure": "description",
"transition_phrases": ["list of common transitions"]
}},
"style_consistency": "high/medium/low",
"unique_elements": ["list of unique style elements"]
"patterns": {{
"sentence_length": "", "vocabulary_patterns": [], "rhetorical_devices": [],
"paragraph_structure": "", "transition_phrases": []
}},
"style_consistency": "",
"unique_elements": [],
"meta": {{"schema_version": "1.1", "confidence": 0.0, "notes": "", "uncertainty": {{"fields": []}}}}
}}
"""
@@ -352,7 +331,7 @@ class StyleDetectionLogic:
brand_analysis = analysis_results.get('brand_analysis', {})
content_strategy_insights = analysis_results.get('content_strategy_insights', {})
prompt = f"""Based on the following comprehensive style analysis, generate detailed content creation guidelines for AI-powered content generation.
prompt = f"""Generate actionable content creation guidelines based on the style analysis.
ANALYSIS DATA:
Writing Style: {writing_style}
@@ -362,85 +341,31 @@ class StyleDetectionLogic:
Content Strategy Insights: {content_strategy_insights}
REQUIREMENTS:
1. Create actionable guidelines for AI content generation
2. Provide specific recommendations for maintaining brand voice
3. Include strategies for audience engagement
4. Address content gaps and opportunities
5. Consider competitive positioning
6. Provide technical writing recommendations
7. Include SEO and conversion optimization tips
8. Address content structure and formatting
- Return ONE single-line MINIFIED JSON object only. No markdown, code fences, comments, or prose.
- Use EXACTLY the keys and ordering from the schema below. No extra top-level keys.
- Provide concise, implementation-ready bullets with an example for key items (e.g., tone and CTA examples).
- Include negative guidance (what to avoid) tied to brand constraints where applicable.
- If uncertain, set empty values and list field names in meta.uncertainty.fields.
IMPORTANT: Respond ONLY with a JSON object in the following format:
IMPORTANT: REQUIRED JSON SCHEMA (stable key order):
{{
"guidelines": {{
"tone_recommendations": [
"specific tone guidelines with examples",
"brand voice consistency tips",
"emotional appeal strategies"
],
"structure_guidelines": [
"content structure recommendations",
"formatting best practices",
"organization strategies"
],
"vocabulary_suggestions": [
"specific vocabulary recommendations",
"industry terminology guidance",
"language complexity advice"
],
"engagement_tips": [
"audience engagement strategies",
"interaction techniques",
"conversion optimization tips"
],
"audience_considerations": [
"specific audience targeting advice",
"pain point addressing strategies",
"motivation-based content tips"
],
"brand_alignment": [
"brand voice consistency guidelines",
"brand value integration tips",
"competitive differentiation strategies"
],
"seo_optimization": [
"keyword integration strategies",
"content optimization tips",
"search visibility recommendations"
],
"conversion_optimization": [
"call-to-action strategies",
"conversion funnel optimization",
"lead generation techniques"
]
}},
"best_practices": [
"comprehensive best practices list",
"industry-specific recommendations",
"quality assurance guidelines"
],
"avoid_elements": [
"elements to avoid with explanations",
"common pitfalls to prevent",
"brand-inappropriate content types"
],
"content_strategy": "comprehensive content strategy recommendation with specific action items",
"ai_generation_tips": [
"specific tips for AI content generation",
"prompt optimization strategies",
"quality control measures"
],
"competitive_advantages": [
"identified competitive advantages",
"differentiation strategies",
"market positioning recommendations"
],
"content_calendar_suggestions": [
"content frequency recommendations",
"topic planning strategies",
"seasonal content opportunities"
]
"guidelines": {{
"tone_recommendations": [],
"structure_guidelines": [],
"vocabulary_suggestions": [],
"engagement_tips": [],
"audience_considerations": [],
"brand_alignment": [],
"seo_optimization": [],
"conversion_optimization": []
}},
"best_practices": [],
"avoid_elements": [],
"content_strategy": "",
"ai_generation_tips": [],
"competitive_advantages": [],
"content_calendar_suggestions": [],
"meta": {{"schema_version": "1.1", "confidence": 0.0, "notes": "", "uncertainty": {{"fields": []}}}}
}}
"""

View File

View File

@@ -0,0 +1,5 @@
"""
Wix integration modular services package.
"""

View File

@@ -0,0 +1,82 @@
from typing import Any, Dict, Optional, Tuple
import requests
from loguru import logger
import base64
import hashlib
import secrets
class WixAuthService:
def __init__(self, client_id: Optional[str], redirect_uri: str, base_url: str):
self.client_id = client_id
self.redirect_uri = redirect_uri
self.base_url = base_url
def generate_authorization_url(self, state: Optional[str] = None) -> Tuple[str, str]:
if not self.client_id:
raise ValueError("Wix client ID not configured")
code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode('utf-8').rstrip('=')
code_challenge = base64.urlsafe_b64encode(
hashlib.sha256(code_verifier.encode('utf-8')).digest()
).decode('utf-8').rstrip('=')
oauth_url = 'https://www.wix.com/oauth/authorize'
from urllib.parse import urlencode
params = {
'client_id': self.client_id,
'redirect_uri': self.redirect_uri,
'response_type': 'code',
'scope': 'BLOG.CREATE-DRAFT,BLOG.PUBLISH,MEDIA.MANAGE',
'code_challenge': code_challenge,
'code_challenge_method': 'S256'
}
if state:
params['state'] = state
return f"{oauth_url}?{urlencode(params)}", code_verifier
def exchange_code_for_tokens(self, code: str, code_verifier: str) -> Dict[str, Any]:
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
data = {
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': self.redirect_uri,
'client_id': self.client_id,
'code_verifier': code_verifier,
}
token_url = f'{self.base_url}/oauth2/token'
response = requests.post(token_url, headers=headers, data=data)
response.raise_for_status()
return response.json()
def refresh_access_token(self, refresh_token: str) -> Dict[str, Any]:
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
data = {
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
'client_id': self.client_id,
}
token_url = f'{self.base_url}/oauth2/token'
response = requests.post(token_url, headers=headers, data=data)
response.raise_for_status()
return response.json()
def get_site_info(self, access_token: str) -> Dict[str, Any]:
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
response = requests.get(f"{self.base_url}/sites/v1/site", headers=headers)
response.raise_for_status()
return response.json()
def get_current_member(self, access_token: str, client_id: Optional[str]) -> Dict[str, Any]:
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
if client_id:
headers['wix-client-id'] = client_id
response = requests.get(f"{self.base_url}/members/v1/members/my", headers=headers)
response.raise_for_status()
return response.json()

View File

@@ -0,0 +1,60 @@
from typing import Any, Dict, List, Optional
import requests
from loguru import logger
class WixBlogService:
def __init__(self, base_url: str, client_id: Optional[str]):
self.base_url = base_url
self.client_id = client_id
def headers(self, access_token: str, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
h: Dict[str, str] = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json',
}
if self.client_id:
h['wix-client-id'] = self.client_id
if extra:
h.update(extra)
return h
def create_draft_post(self, access_token: str, payload: Dict[str, Any], extra_headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
response = requests.post(f"{self.base_url}/blog/v3/draft-posts", headers=self.headers(access_token, extra_headers), json=payload)
response.raise_for_status()
return response.json()
def publish_draft(self, access_token: str, draft_post_id: str, extra_headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
response = requests.post(f"{self.base_url}/blog/v3/draft-posts/{draft_post_id}/publish", headers=self.headers(access_token, extra_headers))
response.raise_for_status()
return response.json()
def list_categories(self, access_token: str, extra_headers: Optional[Dict[str, str]] = None) -> List[Dict[str, Any]]:
response = requests.get(f"{self.base_url}/blog/v3/categories", headers=self.headers(access_token, extra_headers))
response.raise_for_status()
return response.json().get('categories', [])
def create_category(self, access_token: str, label: str, description: Optional[str] = None, language: Optional[str] = None, extra_headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
payload: Dict[str, Any] = {'category': {'label': label}, 'fieldsets': ['URL']}
if description:
payload['category']['description'] = description
if language:
payload['category']['language'] = language
response = requests.post(f"{self.base_url}/blog/v3/categories", headers=self.headers(access_token, extra_headers), json=payload)
response.raise_for_status()
return response.json()
def list_tags(self, access_token: str, extra_headers: Optional[Dict[str, str]] = None) -> List[Dict[str, Any]]:
response = requests.get(f"{self.base_url}/blog/v3/tags", headers=self.headers(access_token, extra_headers))
response.raise_for_status()
return response.json().get('tags', [])
def create_tag(self, access_token: str, label: str, language: Optional[str] = None, extra_headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
payload: Dict[str, Any] = {'label': label, 'fieldsets': ['URL']}
if language:
payload['language'] = language
response = requests.post(f"{self.base_url}/blog/v3/tags", headers=self.headers(access_token, extra_headers), json=payload)
response.raise_for_status()
return response.json()

View File

@@ -0,0 +1,59 @@
from typing import Any, Dict, List
def convert_content_to_ricos(content: str, images: List[str] = None) -> Dict[str, Any]:
"""
Convert simple markdown-like text into minimal valid Ricos JSON.
"""
paragraphs = content.split('\n\n')
nodes = []
import uuid
for paragraph in paragraphs:
text = paragraph.strip()
if not text:
continue
node_id = str(uuid.uuid4())
text_node_id = str(uuid.uuid4())
if text.startswith('#'):
level = len(text) - len(text.lstrip('#'))
heading_text = text.lstrip('# ').strip()
nodes.append({
'id': node_id,
'type': 'HEADING',
'nodes': [{
'id': text_node_id,
'type': 'TEXT',
'textData': {
'text': heading_text,
'decorations': []
}
}],
'headingData': { 'level': min(level, 6) }
})
else:
nodes.append({
'id': node_id,
'type': 'PARAGRAPH',
'nodes': [{
'id': text_node_id,
'type': 'TEXT',
'textData': {
'text': text,
'decorations': []
}
}],
'paragraphData': {}
})
return {
'nodes': nodes,
'metadata': { 'version': 1, 'id': str(uuid.uuid4()) },
'documentStyle': {
'paragraph': { 'decorations': [], 'nodeStyle': {}, 'lineHeight': '1.5' }
}
}

View File

@@ -0,0 +1,23 @@
from typing import Any, Dict
import requests
class WixMediaService:
def __init__(self, base_url: str):
self.base_url = base_url
def import_image(self, access_token: str, image_url: str, display_name: str) -> Dict[str, Any]:
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json',
}
payload = {
'url': image_url,
'mediaType': 'IMAGE',
'displayName': display_name,
}
response = requests.post(f"{self.base_url}/media/v1/files/import", headers=headers, json=payload)
response.raise_for_status()
return response.json()

View File

@@ -0,0 +1,109 @@
from typing import Any, Dict, Optional
import jwt
import json
def normalize_token_string(access_token: Any) -> Optional[str]:
try:
if isinstance(access_token, str):
return access_token
if isinstance(access_token, dict):
token_str = access_token.get('access_token') or access_token.get('value')
if token_str:
return token_str
at = access_token.get('accessToken')
if isinstance(at, dict):
return at.get('value')
if isinstance(at, str):
return at
return None
except Exception:
return None
def extract_member_id_from_access_token(access_token: Any) -> Optional[str]:
try:
token_str: Optional[str] = None
if isinstance(access_token, str):
token_str = access_token
elif isinstance(access_token, dict):
token_str = access_token.get('access_token') or access_token.get('value')
if not token_str:
at = access_token.get('accessToken')
if isinstance(at, dict):
token_str = at.get('value')
elif isinstance(at, str):
token_str = at
if not token_str:
return None
if token_str.startswith('OauthNG.JWS.'):
jwt_part = token_str[12:]
data = jwt.decode(jwt_part, options={"verify_signature": False, "verify_aud": False})
else:
data = jwt.decode(token_str, options={"verify_signature": False, "verify_aud": False})
data_payload = data.get('data')
if isinstance(data_payload, str):
try:
data_payload = json.loads(data_payload)
except Exception:
pass
if isinstance(data_payload, dict):
instance = data_payload.get('instance', {})
if isinstance(instance, dict):
site_member_id = instance.get('siteMemberId')
if isinstance(site_member_id, str) and site_member_id:
return site_member_id
for key in ['memberId', 'sub', 'authorizedSubject', 'id', 'siteMemberId']:
val = data_payload.get(key)
if isinstance(val, str) and val:
return val
member = data_payload.get('member') or {}
if isinstance(member, dict):
val = member.get('id')
if isinstance(val, str) and val:
return val
for key in ['memberId', 'sub', 'authorizedSubject']:
val = data.get(key)
if isinstance(val, str) and val:
return val
member = data.get('member') or {}
if isinstance(member, dict):
val = member.get('id')
if isinstance(val, str) and val:
return val
return None
except Exception:
return None
def decode_wix_token(access_token: str) -> Dict[str, Any]:
token_str = str(access_token)
if token_str.startswith('OauthNG.JWS.'):
jwt_part = token_str[12:]
return jwt.decode(jwt_part, options={"verify_signature": False, "verify_aud": False})
return jwt.decode(token_str, options={"verify_signature": False, "verify_aud": False})
def extract_meta_from_token(access_token: str) -> Dict[str, Optional[str]]:
try:
payload = decode_wix_token(access_token)
data_payload = payload.get('data', {})
if isinstance(data_payload, str):
try:
data_payload = json.loads(data_payload)
except Exception:
pass
instance = (data_payload or {}).get('instance', {})
return {
'siteMemberId': instance.get('siteMemberId'),
'metaSiteId': instance.get('metaSiteId'),
'permissions': instance.get('permissions'),
}
except Exception:
return {'siteMemberId': None, 'metaSiteId': None, 'permissions': None}

View File

@@ -31,8 +31,12 @@ def llm_text_gen(prompt: str, system_prompt: Optional[str] = None, json_struct:
logger.info("[llm_text_gen] Starting text generation")
logger.debug(f"[llm_text_gen] Prompt length: {len(prompt)} characters")
# Initialize API key manager
# Initialize API key manager and reload keys from .env file
api_key_manager = APIKeyManager()
api_key_manager.load_api_keys() # Force reload from .env file
# Debug: Log loaded API keys
logger.debug(f"[llm_text_gen] Loaded API keys: {api_key_manager.get_all_keys()}")
# Set default values for LLM parameters
gpt_provider = "google" # Default to Google Gemini

View File

@@ -0,0 +1,251 @@
"""
Progressive Setup Service
Handles progressive backend initialization based on user onboarding progress.
"""
import os
import json
from typing import Dict, Any, Optional, List
from datetime import datetime
from loguru import logger
from sqlalchemy.orm import Session
from sqlalchemy import text
from services.user_workspace_manager import UserWorkspaceManager
from services.api_key_manager import get_onboarding_progress_for_user
class ProgressiveSetupService:
"""Manages progressive backend setup based on user progress."""
def __init__(self, db_session: Session):
self.db = db_session
self.workspace_manager = UserWorkspaceManager(db_session)
def initialize_user_environment(self, user_id: str) -> Dict[str, Any]:
"""Initialize user environment based on their onboarding progress."""
try:
logger.info(f"Initializing environment for user {user_id}")
# Get user's onboarding progress
progress = get_onboarding_progress_for_user(user_id)
current_step = progress.current_step
# Create or get user workspace
workspace = self.workspace_manager.get_user_workspace(user_id)
if not workspace:
workspace = self.workspace_manager.create_user_workspace(user_id)
# Set up features progressively
setup_status = self.workspace_manager.setup_progressive_features(user_id, current_step)
# Initialize user-specific services
services_status = self._initialize_user_services(user_id, current_step)
return {
"user_id": user_id,
"onboarding_step": current_step,
"workspace": workspace,
"setup_status": setup_status,
"services": services_status,
"initialized_at": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error initializing user environment: {e}")
raise
def _initialize_user_services(self, user_id: str, step: int) -> Dict[str, Any]:
"""Initialize user-specific services based on onboarding step."""
services = {
"ai_services": {"enabled": False, "services": []},
"content_services": {"enabled": False, "services": []},
"research_services": {"enabled": False, "services": []},
"integration_services": {"enabled": False, "services": []}
}
try:
# Step 1: AI Services
if step >= 1:
services["ai_services"]["enabled"] = True
services["ai_services"]["services"] = ["gemini", "exa", "copilotkit"]
self._setup_user_ai_services(user_id)
# Step 2: Content Services
if step >= 2:
services["content_services"]["enabled"] = True
services["content_services"]["services"] = ["content_analysis", "style_detection"]
self._setup_user_content_services(user_id)
# Step 3: Research Services
if step >= 3:
services["research_services"]["enabled"] = True
services["research_services"]["services"] = ["web_research", "fact_checking"]
self._setup_user_research_services(user_id)
# Step 5: Integration Services
if step >= 5:
services["integration_services"]["enabled"] = True
services["integration_services"]["services"] = ["wix", "linkedin", "wordpress"]
self._setup_user_integration_services(user_id)
return services
except Exception as e:
logger.error(f"Error initializing user services: {e}")
return services
def _setup_user_ai_services(self, user_id: str):
"""Set up AI services for the user."""
# Create user-specific AI service configuration
user_config = {
"gemini": {
"enabled": True,
"model": "gemini-pro",
"max_tokens": 4000,
"temperature": 0.7
},
"exa": {
"enabled": True,
"search_depth": "standard",
"max_results": 10
},
"copilotkit": {
"enabled": True,
"assistant_type": "content",
"context_window": 8000
}
}
# Store in user workspace
self.workspace_manager.update_user_config(user_id, {
"ai_services": user_config
})
def _setup_user_content_services(self, user_id: str):
"""Set up content services for the user."""
# Create content analysis configuration
content_config = {
"style_analysis": {
"enabled": True,
"analysis_depth": "comprehensive"
},
"content_generation": {
"enabled": True,
"templates": ["blog", "social", "email"]
},
"quality_checking": {
"enabled": True,
"checks": ["grammar", "tone", "readability"]
}
}
self.workspace_manager.update_user_config(user_id, {
"content_services": content_config
})
def _setup_user_research_services(self, user_id: str):
"""Set up research services for the user."""
# Create research configuration
research_config = {
"web_research": {
"enabled": True,
"sources": ["exa", "serper"],
"max_results": 20
},
"fact_checking": {
"enabled": True,
"verification_level": "standard"
},
"content_validation": {
"enabled": True,
"checks": ["accuracy", "relevance", "freshness"]
}
}
self.workspace_manager.update_user_config(user_id, {
"research_services": research_config
})
def _setup_user_integration_services(self, user_id: str):
"""Set up integration services for the user."""
# Create integration configuration
integration_config = {
"wix": {
"enabled": False,
"connected": False,
"auto_publish": False
},
"linkedin": {
"enabled": False,
"connected": False,
"auto_schedule": False
},
"wordpress": {
"enabled": False,
"connected": False,
"auto_publish": False
}
}
self.workspace_manager.update_user_config(user_id, {
"integration_services": integration_config
})
def get_user_environment_status(self, user_id: str) -> Dict[str, Any]:
"""Get current user environment status."""
try:
workspace = self.workspace_manager.get_user_workspace(user_id)
if not workspace:
return {"error": "User workspace not found"}
progress = get_onboarding_progress_for_user(user_id)
return {
"user_id": user_id,
"onboarding_step": progress.current_step,
"workspace_exists": True,
"workspace_path": workspace["workspace_path"],
"config": workspace["config"],
"last_updated": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error getting user environment status: {e}")
return {"error": str(e)}
def upgrade_user_environment(self, user_id: str, new_step: int) -> Dict[str, Any]:
"""Upgrade user environment when they progress in onboarding."""
try:
logger.info(f"Upgrading environment for user {user_id} to step {new_step}")
# Get current status
current_status = self.get_user_environment_status(user_id)
current_step = current_status.get("onboarding_step", 1)
if new_step <= current_step:
return {"message": "No upgrade needed", "current_step": current_step}
# Set up new features
setup_status = self.workspace_manager.setup_progressive_features(user_id, new_step)
services_status = self._initialize_user_services(user_id, new_step)
return {
"user_id": user_id,
"upgraded_from_step": current_step,
"upgraded_to_step": new_step,
"new_features": setup_status["features_enabled"],
"services": services_status,
"upgraded_at": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error upgrading user environment: {e}")
raise
def cleanup_user_environment(self, user_id: str) -> bool:
"""Clean up user environment (for account deletion)."""
try:
return self.workspace_manager.cleanup_user_workspace(user_id)
except Exception as e:
logger.error(f"Error cleaning up user environment: {e}")
return False

View File

@@ -6,6 +6,7 @@ replacing mock research with real-time industry information.
Available Services:
- GoogleSearchService: Real-time industry research using Google Custom Search API
- ExaService: Competitor discovery and analysis using Exa API
- Source ranking and credibility assessment
- Content extraction and insight generation
@@ -14,8 +15,10 @@ Version: 1.0
Last Updated: January 2025
"""
from services.research.google_search_service import GoogleSearchService
from .google_search_service import GoogleSearchService
from .exa_service import ExaService
__all__ = [
"GoogleSearchService"
"GoogleSearchService",
"ExaService"
]

View File

@@ -0,0 +1,270 @@
"""
AI Prompts for Competitor Analysis
This module contains prompts for analyzing competitor data from Exa API
to generate actionable insights for content strategy and competitive positioning.
"""
COMPETITOR_ANALYSIS_PROMPT = """
You are a competitive intelligence analyst specializing in content strategy and market positioning.
**TASK**: Analyze competitor data to provide actionable insights for content strategy and competitive positioning.
**COMPETITOR DATA**:
{competitor_context}
**USER'S WEBSITE**: {user_url}
**INDUSTRY CONTEXT**: {industry_context}
**ANALYSIS REQUIREMENTS**:
1. **Market Position Analysis**
- Identify the competitive landscape structure
- Determine market leaders vs. challengers
- Assess market saturation and opportunities
2. **Content Strategy Insights**
- Analyze competitor content themes and topics
- Identify content gaps and opportunities
- Suggest unique content angles for differentiation
3. **Competitive Advantages**
- Highlight what makes each competitor unique
- Identify areas where the user can differentiate
- Suggest positioning strategies
4. **SEO and Marketing Insights**
- Analyze competitor positioning and messaging
- Identify keyword and content opportunities
- Suggest marketing strategies
**OUTPUT FORMAT** (JSON):
{{
"market_analysis": {{
"competitive_landscape": "Description of market structure",
"market_leaders": ["List of top 3 competitors"],
"market_opportunities": ["List of 3-5 opportunities"],
"saturation_level": "high/medium/low"
}},
"content_strategy": {{
"common_themes": ["List of common content themes"],
"content_gaps": ["List of 5 content opportunities"],
"unique_angles": ["List of 3 unique content angles"],
"content_frequency_insights": "Analysis of publishing patterns"
}},
"competitive_positioning": {{
"differentiation_opportunities": ["List of 5 ways to differentiate"],
"unique_value_propositions": ["List of 3 unique positioning ideas"],
"target_audience_insights": "Analysis of competitor audience targeting"
}},
"seo_opportunities": {{
"keyword_gaps": ["List of 5 keyword opportunities"],
"content_topics": ["List of 5 high-value content topics"],
"marketing_channels": ["List of competitor marketing strategies"]
}},
"actionable_recommendations": [
"List of 5 specific, actionable recommendations"
],
"risk_assessment": {{
"competitive_threats": ["List of 3 main threats"],
"market_barriers": ["List of 2-3 barriers to entry"],
"success_factors": ["List of 3 key success factors"]
}}
}}
**INSTRUCTIONS**:
- Be specific and actionable in your recommendations
- Focus on opportunities for differentiation
- Consider the user's industry context
- Prioritize recommendations by impact and feasibility
- Use data from the competitor analysis to support insights
- Keep recommendations practical and implementable
**QUALITY STANDARDS**:
- Each recommendation should be specific and actionable
- Insights should be based on actual competitor data
- Focus on differentiation and competitive advantage
- Consider both short-term and long-term strategies
- Ensure recommendations are relevant to the user's industry
"""
CONTENT_GAP_ANALYSIS_PROMPT = """
You are a content strategist analyzing competitor content to identify gaps and opportunities.
**TASK**: Analyze competitor content patterns to identify content gaps and opportunities.
**COMPETITOR CONTENT DATA**:
{competitor_context}
**USER'S INDUSTRY**: {industry_context}
**TARGET AUDIENCE**: {target_audience}
**ANALYSIS FOCUS**:
1. **Content Topic Analysis**
- Identify most common content topics across competitors
- Find underserved or missing topics
- Analyze content depth and quality patterns
2. **Content Format Opportunities**
- Identify popular content formats among competitors
- Find format gaps and opportunities
- Suggest innovative content approaches
3. **Audience Targeting Gaps**
- Analyze competitor audience targeting
- Identify underserved audience segments
- Suggest audience expansion opportunities
4. **SEO Content Opportunities**
- Identify high-value keywords competitors are missing
- Find long-tail keyword opportunities
- Suggest content clusters for SEO
**OUTPUT FORMAT** (JSON):
{{
"content_gaps": [
{{
"topic": "Specific content topic",
"opportunity_level": "high/medium/low",
"reasoning": "Why this is an opportunity",
"content_angle": "Unique angle for this topic",
"estimated_difficulty": "easy/medium/hard"
}}
],
"format_opportunities": [
{{
"format": "Content format type",
"gap_reason": "Why competitors aren't using this",
"potential_impact": "Expected impact level",
"implementation_tips": "How to implement"
}}
],
"audience_gaps": [
{{
"audience_segment": "Underserved audience",
"opportunity_size": "large/medium/small",
"content_needs": "What content this audience needs",
"engagement_strategy": "How to engage this audience"
}}
],
"seo_opportunities": [
{{
"keyword_theme": "Keyword cluster theme",
"search_volume": "estimated_high/medium/low",
"competition_level": "low/medium/high",
"content_ideas": ["3-5 content ideas for this theme"]
}}
],
"priority_recommendations": [
"Top 5 prioritized content opportunities with implementation order"
]
}}
"""
COMPETITIVE_INTELLIGENCE_PROMPT = """
You are a competitive intelligence expert providing strategic insights for market positioning.
**TASK**: Generate comprehensive competitive intelligence insights for strategic decision-making.
**COMPETITOR INTELLIGENCE DATA**:
{competitor_context}
**BUSINESS CONTEXT**:
- User Website: {user_url}
- Industry: {industry_context}
- Business Model: {business_model}
- Target Market: {target_market}
**INTELLIGENCE AREAS**:
1. **Competitive Landscape Mapping**
- Market positioning analysis
- Competitive strength assessment
- Market share estimation
2. **Strategic Positioning Opportunities**
- Blue ocean opportunities
- Differentiation strategies
- Competitive moats
3. **Threat Assessment**
- Competitive threats
- Market disruption risks
- Barrier to entry analysis
4. **Growth Strategy Insights**
- Market expansion opportunities
- Partnership possibilities
- Acquisition targets
**OUTPUT FORMAT** (JSON):
{{
"competitive_landscape": {{
"market_structure": "Description of market structure",
"key_players": [
{{
"name": "Competitor name",
"position": "market_leader/challenger/niche",
"strengths": ["List of key strengths"],
"weaknesses": ["List of key weaknesses"],
"market_share": "estimated_percentage"
}}
],
"market_dynamics": "Analysis of market trends and forces"
}},
"positioning_opportunities": {{
"blue_ocean_opportunities": ["List of uncontested market spaces"],
"differentiation_strategies": ["List of positioning strategies"],
"competitive_advantages": ["List of potential advantages to build"]
}},
"threat_analysis": {{
"immediate_threats": ["List of current competitive threats"],
"future_risks": ["List of potential future risks"],
"market_barriers": ["List of barriers to success"]
}},
"strategic_recommendations": {{
"short_term_actions": ["List of 3-5 immediate actions"],
"medium_term_strategy": ["List of 3-5 strategic initiatives"],
"long_term_vision": ["List of 2-3 long-term strategic goals"]
}},
"success_metrics": {{
"kpis_to_track": ["List of key performance indicators"],
"competitive_benchmarks": ["List of metrics to benchmark against"],
"success_thresholds": ["List of success criteria"]
}}
}}
"""
# Utility function to format prompts with data
def format_competitor_analysis_prompt(competitor_context: str, user_url: str, industry_context: str = None) -> str:
"""Format the competitor analysis prompt with actual data."""
return COMPETITOR_ANALYSIS_PROMPT.format(
competitor_context=competitor_context,
user_url=user_url,
industry_context=industry_context or "Not specified"
)
def format_content_gap_prompt(competitor_context: str, industry_context: str = None, target_audience: str = None) -> str:
"""Format the content gap analysis prompt with actual data."""
return CONTENT_GAP_ANALYSIS_PROMPT.format(
competitor_context=competitor_context,
industry_context=industry_context or "Not specified",
target_audience=target_audience or "Not specified"
)
def format_competitive_intelligence_prompt(
competitor_context: str,
user_url: str,
industry_context: str = None,
business_model: str = None,
target_market: str = None
) -> str:
"""Format the competitive intelligence prompt with actual data."""
return COMPETITIVE_INTELLIGENCE_PROMPT.format(
competitor_context=competitor_context,
user_url=user_url,
industry_context=industry_context or "Not specified",
business_model=business_model or "Not specified",
target_market=target_market or "Not specified"
)

View File

@@ -0,0 +1,769 @@
"""
Exa API Service for ALwrity
This service provides competitor discovery and analysis using the Exa API,
which uses neural search to find semantically similar websites and content.
Key Features:
- Competitor discovery using neural search
- Content analysis and summarization
- Competitive intelligence gathering
- Cost-effective API usage with caching
- Integration with onboarding Step 3
Dependencies:
- aiohttp (for async HTTP requests)
- os (for environment variables)
- logging (for debugging)
Author: ALwrity Team
Version: 1.0
Last Updated: January 2025
"""
import os
import json
import asyncio
from typing import Dict, List, Optional, Any, Union
from datetime import datetime, timedelta
from loguru import logger
from urllib.parse import urlparse
from exa_py import Exa
class ExaService:
"""
Service for competitor discovery and analysis using the Exa API.
This service provides neural search capabilities to find semantically similar
websites and analyze their content for competitive intelligence.
"""
def __init__(self):
"""Initialize the Exa Service with API credentials."""
self.api_key = os.getenv("EXA_API_KEY")
if not self.api_key:
raise ValueError("Exa API key not configured. Please set EXA_API_KEY environment variable.")
else:
self.exa = Exa(api_key=self.api_key)
self.enabled = True
logger.info("Exa Service initialized successfully")
async def discover_competitors(
self,
user_url: str,
num_results: int = 10,
include_domains: Optional[List[str]] = None,
exclude_domains: Optional[List[str]] = None,
industry_context: Optional[str] = None,
website_analysis_data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Discover competitors for a given website using Exa's neural search.
Args:
user_url: The website URL to find competitors for
num_results: Number of competitor results to return (max 100)
include_domains: List of domains to include in search
exclude_domains: List of domains to exclude from search
industry_context: Industry context for better competitor discovery
Returns:
Dictionary containing competitor analysis results
"""
try:
if not self.enabled:
raise ValueError("Exa Service is not enabled - API key missing")
logger.info(f"Starting competitor discovery for: {user_url}")
# Extract user domain for exclusion
user_domain = urlparse(user_url).netloc
exclude_domains_list = exclude_domains or []
exclude_domains_list.append(user_domain)
logger.info(f"Excluding domains: {exclude_domains_list}")
# Extract insights from website analysis for better targeting
include_text_queries = []
summary_query = f"Business model, target audience, content strategy{f' in {industry_context}' if industry_context else ''}"
if website_analysis_data:
analysis = website_analysis_data.get('analysis', {})
# Extract key business terms from the analysis
if 'target_audience' in analysis:
audience = analysis['target_audience']
if isinstance(audience, dict) and 'primary_audience' in audience:
primary_audience = audience['primary_audience']
if len(primary_audience.split()) <= 5: # Exa limit
include_text_queries.append(primary_audience)
# Use industry context from analysis
if 'industry' in analysis and analysis['industry']:
industry = analysis['industry']
if len(industry.split()) <= 5:
include_text_queries.append(industry)
# Enhance summary query with analysis insights
if 'content_type' in analysis:
content_type = analysis['content_type']
summary_query += f", {content_type} content strategy"
logger.info(f"Enhanced targeting with analysis data: {include_text_queries}")
# Use the Exa SDK to find similar links with content and context
search_result = self.exa.find_similar_and_contents(
url=user_url,
num_results=min(num_results, 10), # Exa API limit
include_domains=include_domains,
exclude_domains=exclude_domains_list,
include_text=include_text_queries if include_text_queries else None,
text=True,
highlights={
"numSentences": 2,
"highlightsPerUrl": 3,
"query": "Unique value proposition, competitive advantages, market position"
},
summary={
"query": summary_query
}
)
# TODO: Add context generation once SDK supports it
# For now, we'll generate a basic context from the results
context_result = None
# Log the raw Exa API response summary (avoiding verbose markdown content)
logger.info(f"📊 Exa API response for {user_url}:")
logger.info(f" ├─ Request ID: {getattr(search_result, 'request_id', 'N/A')}")
logger.info(f" ├─ Results count: {len(getattr(search_result, 'results', []))}")
logger.info(f" └─ Cost: ${getattr(getattr(search_result, 'cost_dollars', None), 'total', 0)}")
# Note: Full raw response contains verbose markdown content - logging only summary
# To see full response, set EXA_DEBUG=true in environment
# Extract results from search
results = getattr(search_result, 'results', [])
# Log summary of results
logger.info(f" - Found {len(results)} competitors")
# Process and structure the results
competitors = self._process_competitor_results(search_result, user_url)
logger.info(f"Successfully discovered {len(competitors)} competitors for {user_url}")
return {
"success": True,
"user_url": user_url,
"competitors": competitors,
"total_competitors": len(competitors),
"analysis_timestamp": datetime.utcnow().isoformat(),
"industry_context": industry_context,
"api_cost": getattr(getattr(search_result, 'cost_dollars', None), 'total', 0) if hasattr(search_result, 'cost_dollars') and getattr(search_result, 'cost_dollars', None) else 0,
"request_id": getattr(search_result, 'request_id', None) if hasattr(search_result, 'request_id') else None
}
except asyncio.TimeoutError:
logger.error("Exa API request timed out")
return {
"success": False,
"error": "Request timed out",
"details": "The competitor discovery request took too long to complete"
}
except Exception as e:
logger.error(f"Error in competitor discovery: {str(e)}")
return {
"success": False,
"error": str(e),
"details": "An unexpected error occurred during competitor discovery"
}
def _process_competitor_results(self, search_result, user_url: str) -> List[Dict[str, Any]]:
"""
Process and structure the Exa SDK response into competitor data.
Args:
search_result: Response from Exa SDK
user_url: Original user URL for reference
Returns:
List of processed competitor data
"""
competitors = []
user_domain = urlparse(user_url).netloc
# Extract results from the SDK response
results = getattr(search_result, 'results', [])
for result in results:
try:
# Extract basic information from the result object
competitor_url = getattr(result, 'url', '')
competitor_domain = urlparse(competitor_url).netloc
# Skip if it's the same domain as the user
if competitor_domain == user_domain:
continue
# Extract content insights
summary = getattr(result, 'summary', '')
highlights = getattr(result, 'highlights', [])
highlight_scores = getattr(result, 'highlight_scores', [])
# Calculate competitive relevance score
relevance_score = self._calculate_relevance_score(result, user_url)
competitor_data = {
"url": competitor_url,
"domain": competitor_domain,
"title": getattr(result, 'title', ''),
"published_date": getattr(result, 'published_date', None),
"author": getattr(result, 'author', None),
"favicon": getattr(result, 'favicon', None),
"image": getattr(result, 'image', None),
"summary": summary,
"highlights": highlights,
"highlight_scores": highlight_scores,
"relevance_score": relevance_score,
"competitive_insights": self._extract_competitive_insights(summary, highlights),
"content_analysis": self._analyze_content_quality(result)
}
competitors.append(competitor_data)
except Exception as e:
logger.warning(f"Error processing competitor result: {str(e)}")
continue
# Sort by relevance score (highest first)
competitors.sort(key=lambda x: x["relevance_score"], reverse=True)
return competitors
def _calculate_relevance_score(self, result, user_url: str) -> float:
"""
Calculate a relevance score for competitor ranking.
Args:
result: Competitor result from Exa SDK
user_url: Original user URL
Returns:
Relevance score between 0 and 1
"""
score = 0.0
# Base score from highlight scores
highlight_scores = getattr(result, 'highlight_scores', [])
if highlight_scores:
score += sum(highlight_scores) / len(highlight_scores) * 0.4
# Score from summary quality
summary = getattr(result, 'summary', '')
if summary and len(summary) > 100:
score += 0.3
# Score from title relevance
title = getattr(result, 'title', '').lower()
if any(keyword in title for keyword in ["business", "company", "service", "solution", "platform"]):
score += 0.2
# Score from URL structure similarity
competitor_url = getattr(result, 'url', '')
if self._url_structure_similarity(user_url, competitor_url) > 0.5:
score += 0.1
return min(score, 1.0)
def _url_structure_similarity(self, url1: str, url2: str) -> float:
"""
Calculate URL structure similarity.
Args:
url1: First URL
url2: Second URL
Returns:
Similarity score between 0 and 1
"""
try:
parsed1 = urlparse(url1)
parsed2 = urlparse(url2)
# Compare path structure
path1_parts = [part for part in parsed1.path.split('/') if part]
path2_parts = [part for part in parsed2.path.split('/') if part]
if not path1_parts or not path2_parts:
return 0.0
# Calculate similarity based on path length and structure
max_parts = max(len(path1_parts), len(path2_parts))
common_parts = sum(1 for p1, p2 in zip(path1_parts, path2_parts) if p1 == p2)
return common_parts / max_parts
except Exception:
return 0.0
def _extract_competitive_insights(self, summary: str, highlights: List[str]) -> Dict[str, Any]:
"""
Extract competitive insights from summary and highlights.
Args:
summary: Content summary
highlights: Content highlights
Returns:
Dictionary of competitive insights
"""
insights = {
"business_model": "",
"target_audience": "",
"value_proposition": "",
"competitive_advantages": [],
"content_strategy": ""
}
# Combine summary and highlights for analysis
content = f"{summary} {' '.join(highlights)}".lower()
# Extract business model indicators
business_models = ["saas", "platform", "service", "product", "consulting", "agency", "marketplace"]
for model in business_models:
if model in content:
insights["business_model"] = model.title()
break
# Extract target audience indicators
audiences = ["enterprise", "small business", "startups", "developers", "marketers", "consumers"]
for audience in audiences:
if audience in content:
insights["target_audience"] = audience.title()
break
# Extract value proposition from highlights
if highlights:
insights["value_proposition"] = highlights[0][:100] + "..." if len(highlights[0]) > 100 else highlights[0]
return insights
def _analyze_content_quality(self, result) -> Dict[str, Any]:
"""
Analyze the content quality of a competitor.
Args:
result: Competitor result from Exa SDK
Returns:
Dictionary of content quality metrics
"""
quality_metrics = {
"content_depth": "medium",
"technical_sophistication": "medium",
"content_freshness": "unknown",
"engagement_potential": "medium"
}
# Analyze content depth from summary length
summary = getattr(result, 'summary', '')
if len(summary) > 300:
quality_metrics["content_depth"] = "high"
elif len(summary) < 100:
quality_metrics["content_depth"] = "low"
# Analyze technical sophistication
technical_keywords = ["api", "integration", "automation", "analytics", "data", "platform"]
highlights = getattr(result, 'highlights', [])
content_text = f"{summary} {' '.join(highlights)}".lower()
technical_count = sum(1 for keyword in technical_keywords if keyword in content_text)
if technical_count >= 3:
quality_metrics["technical_sophistication"] = "high"
elif technical_count == 0:
quality_metrics["technical_sophistication"] = "low"
return quality_metrics
async def discover_social_media_accounts(self, user_url: str) -> Dict[str, Any]:
"""
Discover social media accounts for a given website using Exa's answer API.
Args:
user_url: The website URL to find social media accounts for
Returns:
Dictionary containing social media discovery results
"""
try:
if not self.enabled:
raise ValueError("Exa Service is not enabled - API key missing")
logger.info(f"Starting social media discovery for: {user_url}")
# Extract domain from URL for better targeting
domain = urlparse(user_url).netloc.replace('www.', '')
# Use Exa's answer API to find social media accounts
result = self.exa.answer(
f"Find all social media accounts of the url: {domain}. Return a JSON object with facebook, twitter, instagram, linkedin, youtube, and tiktok fields containing the URLs or empty strings if not found.",
model="exa-pro",
text=True
)
# Log the raw Exa API response for debugging
logger.info(f"Raw Exa social media response for {user_url}:")
logger.info(f" - Request ID: {getattr(result, 'request_id', 'N/A')}")
logger.info(f" └─ Cost: ${getattr(getattr(result, 'cost_dollars', None), 'total', 0)}")
# Note: Full raw response contains verbose content - logging only summary
# To see full response, set EXA_DEBUG=true in environment
# Extract social media data
answer_text = getattr(result, 'answer', '')
citations = getattr(result, 'citations', [])
# Convert AnswerResult objects to dictionaries for JSON serialization
citations_dicts = []
for citation in citations:
if hasattr(citation, '__dict__'):
# Convert object to dictionary
citation_dict = {
'id': getattr(citation, 'id', ''),
'title': getattr(citation, 'title', ''),
'url': getattr(citation, 'url', ''),
'text': getattr(citation, 'text', ''),
'snippet': getattr(citation, 'snippet', ''),
'published_date': getattr(citation, 'published_date', None),
'author': getattr(citation, 'author', None),
'image': getattr(citation, 'image', None),
'favicon': getattr(citation, 'favicon', None)
}
citations_dicts.append(citation_dict)
else:
# If it's already a dict, use as is
citations_dicts.append(citation)
logger.info(f" - Raw answer text: {answer_text}")
logger.info(f" - Citations count: {len(citations_dicts)}")
# Parse the response from the answer (could be JSON or markdown format)
try:
import json
import re
if answer_text.strip().startswith('{'):
# Direct JSON format
answer_data = json.loads(answer_text.strip())
else:
# Parse markdown format with URLs
answer_data = {
"facebook": "",
"twitter": "",
"instagram": "",
"linkedin": "",
"youtube": "",
"tiktok": ""
}
# Extract URLs using regex patterns
facebook_match = re.search(r'Facebook.*?\[([^\]]+)\]', answer_text)
if facebook_match:
answer_data["facebook"] = facebook_match.group(1)
twitter_match = re.search(r'Twitter.*?\[([^\]]+)\]', answer_text)
if twitter_match:
answer_data["twitter"] = twitter_match.group(1)
instagram_match = re.search(r'Instagram.*?\[([^\]]+)\]', answer_text)
if instagram_match:
answer_data["instagram"] = instagram_match.group(1)
linkedin_match = re.search(r'LinkedIn.*?\[([^\]]+)\]', answer_text)
if linkedin_match:
answer_data["linkedin"] = linkedin_match.group(1)
youtube_match = re.search(r'YouTube.*?\[([^\]]+)\]', answer_text)
if youtube_match:
answer_data["youtube"] = youtube_match.group(1)
tiktok_match = re.search(r'TikTok.*?\[([^\]]+)\]', answer_text)
if tiktok_match:
answer_data["tiktok"] = tiktok_match.group(1)
except (json.JSONDecodeError, AttributeError, KeyError):
# If parsing fails, create empty structure
answer_data = {
"facebook": "",
"twitter": "",
"instagram": "",
"linkedin": "",
"youtube": "",
"tiktok": ""
}
logger.info(f" - Parsed social media accounts:")
for platform, url in answer_data.items():
if url:
logger.info(f" {platform}: {url}")
return {
"success": True,
"user_url": user_url,
"social_media_accounts": answer_data,
"citations": citations_dicts,
"analysis_timestamp": datetime.utcnow().isoformat(),
"api_cost": getattr(getattr(result, 'cost_dollars', None), 'total', 0) if hasattr(result, 'cost_dollars') and getattr(result, 'cost_dollars', None) else 0,
"request_id": getattr(result, 'request_id', None) if hasattr(result, 'request_id') else None
}
except Exception as e:
logger.error(f"Error in social media discovery: {str(e)}")
return {
"success": False,
"error": str(e),
"details": "An unexpected error occurred during social media discovery"
}
def _generate_basic_context(self, results: List[Any], user_url: str) -> str:
"""
Generate a basic context string from competitor results for LLM consumption.
Args:
results: List of competitor results from Exa API
user_url: Original user URL for reference
Returns:
Formatted context string
"""
context_parts = [
f"Competitive Analysis for: {user_url}",
f"Found {len(results)} similar websites/competitors:",
""
]
for i, result in enumerate(results[:5], 1): # Limit to top 5 for context
url = getattr(result, 'url', 'Unknown URL')
title = getattr(result, 'title', 'Unknown Title')
summary = getattr(result, 'summary', 'No summary available')
context_parts.extend([
f"{i}. {title}",
f" URL: {url}",
f" Summary: {summary[:200]}{'...' if len(summary) > 200 else ''}",
""
])
context_parts.append("Key insights:")
context_parts.append("- These competitors offer similar services or content")
context_parts.append("- Analyze their content strategy and positioning")
context_parts.append("- Identify opportunities for differentiation")
return "\n".join(context_parts)
async def analyze_competitor_content(
self,
competitor_url: str,
analysis_depth: str = "standard"
) -> Dict[str, Any]:
"""
Perform deeper analysis of a specific competitor.
Args:
competitor_url: URL of the competitor to analyze
analysis_depth: Depth of analysis ("quick", "standard", "deep")
Returns:
Dictionary containing detailed competitor analysis
"""
try:
logger.info(f"Starting detailed analysis for competitor: {competitor_url}")
# Get similar content from this competitor
similar_results = await self.discover_competitors(
competitor_url,
num_results=10,
include_domains=[urlparse(competitor_url).netloc]
)
if not similar_results["success"]:
return similar_results
# Analyze content patterns
content_patterns = self._analyze_content_patterns(similar_results["competitors"])
# Generate competitive insights
competitive_insights = self._generate_competitive_insights(
competitor_url,
similar_results["competitors"],
content_patterns
)
return {
"success": True,
"competitor_url": competitor_url,
"content_patterns": content_patterns,
"competitive_insights": competitive_insights,
"analysis_timestamp": datetime.utcnow().isoformat(),
"analysis_depth": analysis_depth
}
except Exception as e:
logger.error(f"Error in competitor content analysis: {str(e)}")
return {
"success": False,
"error": str(e),
"details": "An unexpected error occurred during competitor analysis"
}
def _analyze_content_patterns(self, competitors: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Analyze content patterns across competitors.
Args:
competitors: List of competitor data
Returns:
Dictionary of content patterns
"""
patterns = {
"common_themes": [],
"content_types": [],
"publishing_patterns": {},
"target_keywords": [],
"content_strategies": []
}
# Analyze common themes
all_summaries = [comp.get("summary", "") for comp in competitors]
# This would be enhanced with NLP analysis in a full implementation
# Analyze content types from URLs
content_types = set()
for comp in competitors:
url = comp.get("url", "")
if "/blog/" in url:
content_types.add("blog")
elif "/product/" in url or "/service/" in url:
content_types.add("product")
elif "/about/" in url:
content_types.add("about")
elif "/contact/" in url:
content_types.add("contact")
patterns["content_types"] = list(content_types)
return patterns
def _generate_competitive_insights(
self,
competitor_url: str,
competitors: List[Dict[str, Any]],
content_patterns: Dict[str, Any]
) -> Dict[str, Any]:
"""
Generate competitive insights from analysis data.
Args:
competitor_url: URL of the competitor
competitors: List of competitor data
content_patterns: Content pattern analysis
Returns:
Dictionary of competitive insights
"""
insights = {
"competitive_strengths": [],
"content_opportunities": [],
"market_positioning": "unknown",
"strategic_recommendations": []
}
# Analyze competitive strengths
for comp in competitors:
if comp.get("relevance_score", 0) > 0.7:
insights["competitive_strengths"].append({
"strength": comp.get("summary", "")[:100],
"relevance": comp.get("relevance_score", 0)
})
# Generate content opportunities
if content_patterns.get("content_types"):
insights["content_opportunities"] = [
f"Develop {content_type} content"
for content_type in content_patterns["content_types"]
]
return insights
def health_check(self) -> Dict[str, Any]:
"""
Check the health of the Exa service.
Returns:
Dictionary containing service health status
"""
try:
if not self.enabled:
return {
"status": "disabled",
"message": "Exa API key not configured",
"timestamp": datetime.utcnow().isoformat()
}
# Test with a simple request using the SDK directly
test_result = self.exa.find_similar(
url="https://example.com",
num_results=1
)
# If we get here without an exception, the API is working
return {
"status": "healthy",
"message": "Exa API is operational",
"timestamp": datetime.utcnow().isoformat(),
"test_successful": True
}
except Exception as e:
return {
"status": "error",
"message": f"Health check failed: {str(e)}",
"timestamp": datetime.utcnow().isoformat()
}
def get_cost_estimate(self, num_results: int, include_content: bool = True) -> Dict[str, Any]:
"""
Get cost estimate for Exa API usage.
Args:
num_results: Number of results requested
include_content: Whether to include content analysis
Returns:
Dictionary containing cost estimate
"""
# Exa API pricing (as of documentation)
if num_results <= 25:
search_cost = 0.005
elif num_results <= 100:
search_cost = 0.025
else:
search_cost = 1.0
content_cost = 0.0
if include_content:
# Estimate content analysis cost
content_cost = num_results * 0.001 # Rough estimate
total_cost = search_cost + content_cost
return {
"search_cost": search_cost,
"content_cost": content_cost,
"total_estimated_cost": total_cost,
"num_results": num_results,
"include_content": include_content
}

View File

@@ -1,140 +0,0 @@
"""
Test Script for 12-Step Prompt Chaining Framework
This script tests the basic functionality of the 12-step prompt chaining framework.
"""
import asyncio
import sys
import os
# Add the current directory to the Python path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from calendar_generation_datasource_framework.prompt_chaining import PromptChainOrchestrator
async def test_12_step_framework():
"""Test the 12-step prompt chaining framework."""
print("🚀 Testing 12-Step Prompt Chaining Framework")
print("=" * 50)
try:
# Initialize the orchestrator
print("📋 Initializing Prompt Chain Orchestrator...")
orchestrator = PromptChainOrchestrator()
# Test health status
print("\n🏥 Testing Health Status...")
health_status = await orchestrator.get_health_status()
print(f"✅ Health Status: {health_status}")
# Test calendar generation
print("\n🎯 Testing Calendar Generation...")
result = await orchestrator.generate_calendar(
user_id=1,
strategy_id=123,
calendar_type="monthly",
industry="technology",
business_size="sme"
)
print(f"✅ Calendar Generation Result:")
print(f" - Status: {result.get('status')}")
print(f" - Processing Time: {result.get('processing_time', 0):.2f}s")
print(f" - Quality Score: {result.get('quality_score', 0):.2f}")
print(f" - Framework Version: {result.get('framework_version')}")
# Test progress tracking
print("\n📊 Testing Progress Tracking...")
progress = await orchestrator.get_progress()
print(f"✅ Progress: {progress.get('completed_steps')}/{progress.get('total_steps')} steps completed")
print(f" - Progress Percentage: {progress.get('progress_percentage', 0):.1f}%")
print(f" - Current Phase: {progress.get('current_phase')}")
print(f" - Overall Quality Score: {progress.get('overall_quality_score', 0):.2f}")
# Test step details
print("\n🔍 Testing Step Details...")
step_details = progress.get('step_details', {})
for step_name, step_data in step_details.items():
print(f" - {step_name}: {step_data.get('status')} (Quality: {step_data.get('quality_score', 0):.2f})")
print("\n✅ All tests completed successfully!")
return True
except Exception as e:
print(f"\n❌ Test failed: {str(e)}")
import traceback
traceback.print_exc()
return False
async def test_individual_components():
"""Test individual components of the framework."""
print("\n🔧 Testing Individual Components")
print("=" * 50)
try:
from calendar_generation_datasource_framework.prompt_chaining import (
StepManager, ContextManager, ProgressTracker, ErrorHandler
)
# Test Step Manager
print("\n🎯 Testing Step Manager...")
step_manager = StepManager()
health_status = step_manager.get_health_status()
print(f"✅ Step Manager Health: {health_status}")
# Test Context Manager
print("\n📋 Testing Context Manager...")
context_manager = ContextManager()
health_status = context_manager.get_health_status()
print(f"✅ Context Manager Health: {health_status}")
# Test Progress Tracker
print("\n📊 Testing Progress Tracker...")
progress_tracker = ProgressTracker()
health_status = progress_tracker.get_health_status()
print(f"✅ Progress Tracker Health: {health_status}")
# Test Error Handler
print("\n🛡️ Testing Error Handler...")
error_handler = ErrorHandler()
health_status = error_handler.get_health_status()
print(f"✅ Error Handler Health: {health_status}")
print("\n✅ All component tests completed successfully!")
return True
except Exception as e:
print(f"\n❌ Component test failed: {str(e)}")
import traceback
traceback.print_exc()
return False
async def main():
"""Main test function."""
print("🧪 12-Step Prompt Chaining Framework Test Suite")
print("=" * 60)
# Test individual components
component_success = await test_individual_components()
# Test full framework
framework_success = await test_12_step_framework()
# Summary
print("\n📋 Test Summary")
print("=" * 30)
print(f"✅ Individual Components: {'PASSED' if component_success else 'FAILED'}")
print(f"✅ Full Framework: {'PASSED' if framework_success else 'FAILED'}")
if component_success and framework_success:
print("\n🎉 All tests passed! The 12-step framework is ready for implementation.")
else:
print("\n⚠️ Some tests failed. Please check the implementation.")
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -1,564 +0,0 @@
"""
Integration Test for 12-Step Prompt Chaining Framework
This script tests the complete integration with real AI services and database connections.
"""
import asyncio
import sys
import os
import json
from datetime import datetime
from typing import Dict, Any
# Add the current directory to Python path
sys.path.append(os.path.dirname(__file__))
# Check if we can import the real services
def check_service_availability():
"""Check which services are available."""
services_status = {
"prompt_chaining": False,
"ai_engine": False,
"keyword_researcher": False,
"competitor_analyzer": False,
"onboarding_service": False,
"ai_analytics": False,
"content_planning_db": False
}
try:
from calendar_generation_datasource_framework.prompt_chaining import PromptChainOrchestrator
services_status["prompt_chaining"] = True
print("✅ Prompt Chaining Framework available")
except ImportError as e:
print(f"❌ Prompt Chaining Framework not available: {e}")
try:
from content_gap_analyzer.ai_engine_service import AIEngineService
services_status["ai_engine"] = True
print("✅ AI Engine Service available")
except ImportError as e:
print(f"⚠️ AI Engine Service not available: {e}")
try:
from content_gap_analyzer.keyword_researcher import KeywordResearcher
services_status["keyword_researcher"] = True
print("✅ Keyword Researcher available")
except ImportError as e:
print(f"⚠️ Keyword Researcher not available: {e}")
try:
from content_gap_analyzer.competitor_analyzer import CompetitorAnalyzer
services_status["competitor_analyzer"] = True
print("✅ Competitor Analyzer available")
except ImportError as e:
print(f"⚠️ Competitor Analyzer not available: {e}")
try:
from onboarding_data_service import OnboardingDataService
services_status["onboarding_service"] = True
print("✅ Onboarding Data Service available")
except ImportError as e:
print(f"⚠️ Onboarding Data Service not available: {e}")
try:
from ai_analytics_service import AIAnalyticsService
services_status["ai_analytics"] = True
print("✅ AI Analytics Service available")
except ImportError as e:
print(f"⚠️ AI Analytics Service not available: {e}")
try:
from content_planning_db import ContentPlanningDBService
services_status["content_planning_db"] = True
print("✅ Content Planning DB Service available")
except ImportError as e:
print(f"⚠️ Content Planning DB Service not available: {e}")
return services_status
async def test_real_ai_services():
"""Test real AI services connectivity."""
print("🤖 Testing Real AI Services")
print("=" * 40)
success_count = 0
total_tests = 0
# Test AI Engine Service
try:
from content_gap_analyzer.ai_engine_service import AIEngineService
ai_engine = AIEngineService()
print("🎯 Testing AI Engine Service...")
# Test strategic insights generation
total_tests += 1
try:
result = await ai_engine.generate_strategic_insights(
strategy_data={"content_pillars": ["AI", "Technology"]},
onboarding_data={"website_analysis": {"industry": "technology"}},
industry="technology",
business_size="sme"
)
if result and isinstance(result, dict):
print(f"✅ Strategic insights generation: SUCCESS")
success_count += 1
else:
print(f"⚠️ Strategic insights generation: Empty result")
except Exception as e:
print(f"❌ Strategic insights generation: {str(e)}")
# Test content gap analysis
total_tests += 1
try:
result = await ai_engine.analyze_content_gaps(
gap_data={"content_gaps": ["Blog posts", "Video content"]},
keyword_analysis={"high_value_keywords": ["AI", "technology"]},
competitor_analysis={"insights": {"competitors": ["comp1"]}},
industry="technology"
)
if result and isinstance(result, dict):
print(f"✅ Content gap analysis: SUCCESS")
success_count += 1
else:
print(f"⚠️ Content gap analysis: Empty result")
except Exception as e:
print(f"❌ Content gap analysis: {str(e)}")
# Test audience behavior analysis
total_tests += 1
try:
result = await ai_engine.analyze_audience_behavior(
onboarding_data={"website_analysis": {"target_audience": ["developers"]}},
strategy_data={"target_audience": {"demographics": {"age": "25-35"}}},
industry="technology",
business_size="sme"
)
if result and isinstance(result, dict):
print(f"✅ Audience behavior analysis: SUCCESS")
success_count += 1
else:
print(f"⚠️ Audience behavior analysis: Empty result")
except Exception as e:
print(f"❌ Audience behavior analysis: {str(e)}")
except ImportError:
print("❌ AI Engine Service not available for testing")
# Test Keyword Researcher
try:
from content_gap_analyzer.keyword_researcher import KeywordResearcher
keyword_researcher = KeywordResearcher()
print("\n🔍 Testing Keyword Researcher...")
# Test keyword analysis
total_tests += 1
try:
result = await keyword_researcher.analyze_keywords(
target_keywords=["AI", "technology", "automation"],
industry="technology"
)
if result and isinstance(result, dict):
print(f"✅ Keyword analysis: SUCCESS")
success_count += 1
else:
print(f"⚠️ Keyword analysis: Empty result")
except Exception as e:
print(f"❌ Keyword analysis: {str(e)}")
# Test trending topics
total_tests += 1
try:
result = await keyword_researcher.get_trending_topics(
industry="technology"
)
if result and isinstance(result, list):
print(f"✅ Trending topics: SUCCESS")
success_count += 1
else:
print(f"⚠️ Trending topics: Empty result")
except Exception as e:
print(f"❌ Trending topics: {str(e)}")
except ImportError:
print("❌ Keyword Researcher not available for testing")
# Test Competitor Analyzer
try:
from content_gap_analyzer.competitor_analyzer import CompetitorAnalyzer
competitor_analyzer = CompetitorAnalyzer()
print("\n🏢 Testing Competitor Analyzer...")
# Test competitor analysis
total_tests += 1
try:
result = await competitor_analyzer.analyze_competitors(
competitor_urls=["https://example.com", "https://competitor.com"],
industry="technology"
)
if result and isinstance(result, dict):
print(f"✅ Competitor analysis: SUCCESS")
success_count += 1
else:
print(f"⚠️ Competitor analysis: Empty result")
except Exception as e:
print(f"❌ Competitor analysis: {str(e)}")
except ImportError:
print("❌ Competitor Analyzer not available for testing")
print(f"\n📊 AI Services Test Summary: {success_count}/{total_tests} tests passed")
return success_count, total_tests
async def test_data_services():
"""Test data services connectivity."""
print("\n💾 Testing Data Services")
print("=" * 40)
success_count = 0
total_tests = 0
# Test Onboarding Data Service
try:
from onboarding_data_service import OnboardingDataService
onboarding_service = OnboardingDataService()
print("👤 Testing Onboarding Data Service...")
# Test get personalized inputs
total_tests += 1
try:
result = onboarding_service.get_personalized_ai_inputs(1)
if result and isinstance(result, dict):
print(f"✅ Get personalized AI inputs: SUCCESS")
success_count += 1
else:
print(f"⚠️ Get personalized AI inputs: Empty result")
except Exception as e:
print(f"❌ Get personalized AI inputs: {str(e)}")
except ImportError:
print("❌ Onboarding Data Service not available for testing")
# Test AI Analytics Service
try:
from ai_analytics_service import AIAnalyticsService
ai_analytics = AIAnalyticsService()
print("\n🧠 Testing AI Analytics Service...")
# Test strategic intelligence generation
total_tests += 1
try:
result = await ai_analytics.generate_strategic_intelligence(1)
if result and isinstance(result, dict):
print(f"✅ Strategic intelligence generation: SUCCESS")
success_count += 1
else:
print(f"⚠️ Strategic intelligence generation: Empty result")
except Exception as e:
print(f"❌ Strategic intelligence generation: {str(e)}")
except ImportError:
print("❌ AI Analytics Service not available for testing")
# Test Content Planning DB Service
try:
from content_planning_db import ContentPlanningDBService
# Note: This would require proper database session injection
print("\n🗃️ Testing Content Planning DB Service...")
print(" Database service requires proper session injection - skipping direct test")
except ImportError:
print("❌ Content Planning DB Service not available for testing")
print(f"\n📊 Data Services Test Summary: {success_count}/{total_tests} tests passed")
return success_count, total_tests
async def test_12_step_framework_integration():
"""Test the 12-step framework with real service integration."""
print("\n🚀 Testing 12-Step Framework Integration")
print("=" * 50)
try:
from calendar_generation_datasource_framework.prompt_chaining import PromptChainOrchestrator
# Initialize orchestrator
print("📋 Initializing Prompt Chain Orchestrator...")
orchestrator = PromptChainOrchestrator()
# Check health status
health_status = await orchestrator.get_health_status()
print(f"✅ Framework Health: {health_status['status']}")
print(f"📊 Steps Configured: {health_status['steps_configured']}")
print(f"🏗️ Phases Configured: {health_status['phases_configured']}")
# Test calendar generation with real services
print("\n🎯 Testing Calendar Generation...")
try:
result = await orchestrator.generate_calendar(
user_id=1,
strategy_id=1,
calendar_type="monthly",
industry="technology",
business_size="sme"
)
print("✅ Calendar generation completed!")
print(f"📋 Result keys: {list(result.keys())}")
print(f"⏱️ Processing time: {result.get('processing_time', 0):.2f}s")
print(f"🎯 Framework version: {result.get('framework_version', 'unknown')}")
print(f"📊 Status: {result.get('status', 'unknown')}")
# Validate result structure
required_fields = [
'user_id', 'strategy_id', 'processing_time', 'generated_at',
'framework_version', 'status'
]
missing_fields = [field for field in required_fields if field not in result]
if missing_fields:
print(f"⚠️ Missing required fields: {missing_fields}")
else:
print("✅ All required fields present")
# Check for calendar content
calendar_fields = [
'daily_schedule', 'weekly_themes', 'content_recommendations',
'optimal_timing', 'performance_predictions', 'trending_topics'
]
present_fields = [field for field in calendar_fields if field in result and result[field]]
print(f"📋 Calendar content fields present: {len(present_fields)}/{len(calendar_fields)}")
return True, result
except Exception as e:
print(f"❌ Calendar generation failed: {str(e)}")
return False, None
except ImportError as e:
print(f"❌ 12-Step Framework not available: {e}")
return False, None
async def test_phase1_steps_integration():
"""Test Phase 1 steps with real service integration."""
print("\n🎯 Testing Phase 1 Steps Integration")
print("=" * 50)
try:
from calendar_generation_datasource_framework.prompt_chaining.steps.phase1_steps import (
ContentStrategyAnalysisStep,
GapAnalysisStep,
AudiencePlatformStrategyStep
)
# Test context
context = {
"user_id": 1,
"strategy_id": 1,
"calendar_type": "monthly",
"industry": "technology",
"business_size": "sme",
"user_data": {
"strategy_data": {
"content_pillars": ["AI", "Technology", "Innovation"],
"target_audience": {"demographics": {"age": "25-35", "location": "US"}},
"business_goals": ["Increase brand awareness", "Generate leads"],
"success_metrics": ["Website traffic", "Social engagement"]
},
"onboarding_data": {
"website_analysis": {"industry": "technology", "target_audience": ["developers"]},
"competitor_analysis": {"top_performers": ["competitor1", "competitor2"]},
"keyword_analysis": {"high_value_keywords": ["AI", "automation"]}
},
"gap_analysis": {
"content_gaps": ["Video content", "Interactive demos"],
"keyword_opportunities": ["machine learning", "artificial intelligence"]
},
"performance_data": {
"engagement_metrics": {"average_engagement": 0.05},
"best_performing_content": ["How-to guides", "Industry insights"]
},
"competitor_data": {
"competitor_urls": ["https://competitor1.com", "https://competitor2.com"]
}
},
"step_results": {},
"quality_scores": {},
"current_step": 0,
"phase": "initialization"
}
phase1_results = {}
# Test Step 1: Content Strategy Analysis
print("🎯 Testing Step 1: Content Strategy Analysis")
try:
step1 = ContentStrategyAnalysisStep()
result1 = await step1.run(context)
phase1_results["step_01"] = result1
print(f"✅ Step 1 Status: {result1.get('status', 'unknown')}")
print(f"📊 Step 1 Quality: {result1.get('quality_score', 0.0):.2f}")
print(f"⏱️ Step 1 Time: {result1.get('execution_time', 0.0):.2f}s")
except Exception as e:
print(f"❌ Step 1 failed: {str(e)}")
# Test Step 2: Gap Analysis & Opportunity Identification
print("\n🎯 Testing Step 2: Gap Analysis & Opportunity Identification")
try:
step2 = GapAnalysisStep()
result2 = await step2.run(context)
phase1_results["step_02"] = result2
print(f"✅ Step 2 Status: {result2.get('status', 'unknown')}")
print(f"📊 Step 2 Quality: {result2.get('quality_score', 0.0):.2f}")
print(f"⏱️ Step 2 Time: {result2.get('execution_time', 0.0):.2f}s")
except Exception as e:
print(f"❌ Step 2 failed: {str(e)}")
# Test Step 3: Audience & Platform Strategy
print("\n🎯 Testing Step 3: Audience & Platform Strategy")
try:
step3 = AudiencePlatformStrategyStep()
result3 = await step3.run(context)
phase1_results["step_03"] = result3
print(f"✅ Step 3 Status: {result3.get('status', 'unknown')}")
print(f"📊 Step 3 Quality: {result3.get('quality_score', 0.0):.2f}")
print(f"⏱️ Step 3 Time: {result3.get('execution_time', 0.0):.2f}s")
except Exception as e:
print(f"❌ Step 3 failed: {str(e)}")
# Calculate overall Phase 1 metrics
completed_steps = len([r for r in phase1_results.values() if r.get('status') == 'completed'])
total_quality = sum(r.get('quality_score', 0.0) for r in phase1_results.values())
avg_quality = total_quality / len(phase1_results) if phase1_results else 0.0
total_time = sum(r.get('execution_time', 0.0) for r in phase1_results.values())
print(f"\n📋 Phase 1 Integration Summary")
print("=" * 40)
print(f"✅ Completed Steps: {completed_steps}/3")
print(f"📊 Average Quality: {avg_quality:.2f}")
print(f"⏱️ Total Time: {total_time:.2f}s")
return completed_steps == 3, phase1_results
except ImportError as e:
print(f"❌ Phase 1 steps not available: {e}")
return False, {}
async def generate_integration_report(
services_status: Dict[str, bool],
ai_services_result: tuple,
data_services_result: tuple,
framework_result: tuple,
phase1_result: tuple
):
"""Generate comprehensive integration test report."""
print("\n📋 Integration Test Report")
print("=" * 60)
# Service availability
available_services = sum(services_status.values())
total_services = len(services_status)
print(f"🔧 Service Availability: {available_services}/{total_services}")
# AI services
ai_success, ai_total = ai_services_result
print(f"🤖 AI Services: {ai_success}/{ai_total} tests passed")
# Data services
data_success, data_total = data_services_result
print(f"💾 Data Services: {data_success}/{data_total} tests passed")
# Framework integration
framework_success, framework_data = framework_result
print(f"🚀 Framework Integration: {'SUCCESS' if framework_success else 'FAILED'}")
# Phase 1 integration
phase1_success, phase1_data = phase1_result
print(f"🎯 Phase 1 Integration: {'SUCCESS' if phase1_success else 'FAILED'}")
# Overall assessment
total_tests = ai_total + data_total + (1 if framework_success else 0) + (3 if phase1_success else 0)
total_success = ai_success + data_success + (1 if framework_success else 0) + (3 if phase1_success else len(phase1_data))
print(f"\n🎉 Overall Integration: {total_success}/{total_tests} ({total_success/total_tests*100:.1f}%)")
# Recommendations
print(f"\n📝 Recommendations:")
if available_services < total_services:
print(" • Set up missing services for full integration")
if ai_success < ai_total:
print(" • Check AI service configurations and API keys")
if data_success < data_total:
print(" • Verify database connections and service dependencies")
if not framework_success:
print(" • Debug framework integration issues")
if not phase1_success:
print(" • Review Phase 1 step implementations")
if total_success == total_tests:
print(" ✅ All systems operational - ready for production!")
# Save detailed report
report = {
"timestamp": datetime.now().isoformat(),
"service_availability": services_status,
"ai_services": {"success": ai_success, "total": ai_total},
"data_services": {"success": data_success, "total": data_total},
"framework_integration": {"success": framework_success},
"phase1_integration": {"success": phase1_success, "results": phase1_data},
"overall": {"success": total_success, "total": total_tests, "percentage": total_success/total_tests*100}
}
with open("integration_test_report.json", "w") as f:
json.dump(report, f, indent=2, default=str)
print(f"\n💾 Detailed report saved to: integration_test_report.json")
async def main():
"""Main integration test function."""
print("🧪 12-Step Framework Integration Test Suite")
print("=" * 60)
print(f"🕒 Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
# Check service availability
print("\n🔍 Checking Service Availability...")
services_status = check_service_availability()
# Test AI services
ai_services_result = await test_real_ai_services()
# Test data services
data_services_result = await test_data_services()
# Test 12-step framework integration
framework_result = await test_12_step_framework_integration()
# Test Phase 1 steps integration
phase1_result = await test_phase1_steps_integration()
# Generate comprehensive report
await generate_integration_report(
services_status,
ai_services_result,
data_services_result,
framework_result,
phase1_result
)
print(f"\n🏁 Integration test completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -1,491 +0,0 @@
"""
Real Services Integration Test for 12-Step Prompt Chaining Framework
This script tests the complete integration using real AI services and database connections.
This test should be run from the backend/services directory or with proper PYTHONPATH setup.
"""
import asyncio
import sys
import os
import json
from datetime import datetime
from typing import Dict, Any, Optional
# Add the backend directory to Python path for proper imports
backend_dir = os.path.dirname(os.path.dirname(__file__))
if backend_dir not in sys.path:
sys.path.insert(0, backend_dir)
services_dir = os.path.dirname(__file__)
if services_dir not in sys.path:
sys.path.insert(0, services_dir)
async def test_real_ai_engine_service():
"""Test real AI Engine Service with proper error handling."""
print("🤖 Testing Real AI Engine Service")
print("=" * 40)
try:
from content_gap_analyzer.ai_engine_service import AIEngineService
ai_engine = AIEngineService()
# Test strategic insights generation
print("🎯 Testing strategic insights generation...")
try:
result = await ai_engine.generate_strategic_insights(
strategy_data={
"content_pillars": ["AI", "Technology", "Innovation"],
"target_audience": {"demographics": {"age": "25-35", "industry": "technology"}},
"business_goals": ["Increase brand awareness", "Generate leads"]
},
onboarding_data={
"website_analysis": {
"industry": "technology",
"target_audience": ["developers", "tech enthusiasts"],
"content_focus": ["tutorials", "industry insights"]
}
},
industry="technology",
business_size="sme"
)
if result and isinstance(result, dict):
print(f"✅ Strategic insights generation: SUCCESS")
print(f" - Result keys: {list(result.keys())}")
if "strategic_insights" in result:
print(f" - Insights count: {len(result['strategic_insights'])}")
return True, result
else:
print(f"⚠️ Strategic insights generation: Empty result")
return False, None
except Exception as e:
print(f"❌ Strategic insights generation failed: {str(e)}")
return False, None
except ImportError as e:
print(f"❌ AI Engine Service not available: {e}")
return False, None
async def test_real_keyword_researcher():
"""Test real Keyword Researcher service."""
print("\n🔍 Testing Real Keyword Researcher")
print("=" * 40)
try:
from content_gap_analyzer.keyword_researcher import KeywordResearcher
keyword_researcher = KeywordResearcher()
# Test keyword analysis
print("🎯 Testing keyword analysis...")
try:
result = await keyword_researcher.analyze_keywords(
target_keywords=["artificial intelligence", "machine learning", "automation", "AI tools"],
industry="technology"
)
if result and isinstance(result, dict):
print(f"✅ Keyword analysis: SUCCESS")
print(f" - Result keys: {list(result.keys())}")
if "high_value_keywords" in result:
print(f" - High-value keywords: {len(result['high_value_keywords'])}")
return True, result
else:
print(f"⚠️ Keyword analysis: Empty result")
return False, None
except Exception as e:
print(f"❌ Keyword analysis failed: {str(e)}")
return False, None
except ImportError as e:
print(f"❌ Keyword Researcher not available: {e}")
return False, None
async def test_real_onboarding_service():
"""Test real Onboarding Data Service."""
print("\n👤 Testing Real Onboarding Data Service")
print("=" * 40)
try:
from onboarding_data_service import OnboardingDataService
onboarding_service = OnboardingDataService()
# Test get personalized inputs
print("🎯 Testing get personalized AI inputs...")
try:
result = onboarding_service.get_personalized_ai_inputs(1)
if result and isinstance(result, dict):
print(f"✅ Get personalized AI inputs: SUCCESS")
print(f" - Result keys: {list(result.keys())}")
if "website_analysis" in result:
print(f" - Website analysis available")
if "keyword_analysis" in result:
print(f" - Keyword analysis available")
return True, result
else:
print(f"⚠️ Get personalized AI inputs: Empty result")
return False, None
except Exception as e:
print(f"❌ Get personalized AI inputs failed: {str(e)}")
return False, None
except ImportError as e:
print(f"❌ Onboarding Data Service not available: {e}")
return False, None
async def test_real_data_processing():
"""Test real data processing modules."""
print("\n💾 Testing Real Data Processing Modules")
print("=" * 40)
try:
from calendar_generation_datasource_framework.data_processing import (
ComprehensiveUserDataProcessor,
StrategyDataProcessor,
GapAnalysisDataProcessor
)
# Test comprehensive user data processor
print("🎯 Testing ComprehensiveUserDataProcessor...")
try:
processor = ComprehensiveUserDataProcessor()
result = await processor.get_comprehensive_user_data(1, 1)
if result and isinstance(result, dict):
print(f"✅ ComprehensiveUserDataProcessor: SUCCESS")
print(f" - Result keys: {list(result.keys())}")
return True, result
else:
print(f"⚠️ ComprehensiveUserDataProcessor: Empty result")
return False, None
except Exception as e:
print(f"❌ ComprehensiveUserDataProcessor failed: {str(e)}")
return False, None
except ImportError as e:
print(f"❌ Data Processing modules not available: {e}")
return False, None
async def test_phase1_with_real_services():
"""Test Phase 1 steps with real service integration."""
print("\n🎯 Testing Phase 1 Steps with Real Services")
print("=" * 50)
try:
from calendar_generation_datasource_framework.prompt_chaining.steps.phase1_steps import (
ContentStrategyAnalysisStep,
GapAnalysisStep,
AudiencePlatformStrategyStep
)
# Get real data
real_context = {
"user_id": 1,
"strategy_id": 1,
"calendar_type": "monthly",
"industry": "technology",
"business_size": "sme",
"user_data": {
"strategy_data": {
"content_pillars": ["AI", "Technology", "Innovation", "Tutorials"],
"target_audience": {
"demographics": {"age": "25-35", "location": "US", "industry": "technology"},
"interests": ["AI", "machine learning", "programming", "tech trends"]
},
"business_goals": ["Increase brand awareness", "Generate leads", "Establish thought leadership"],
"success_metrics": ["Website traffic", "Social engagement", "Lead generation"]
},
"onboarding_data": {
"website_analysis": {
"industry": "technology",
"target_audience": ["developers", "tech enthusiasts", "AI researchers"],
"content_focus": ["tutorials", "industry insights", "product reviews"],
"competitive_landscape": ["competitor1.com", "competitor2.com"]
},
"competitor_analysis": {
"top_performers": ["OpenAI Blog", "Google AI Blog", "MIT Technology Review"],
"content_types": ["research papers", "tutorials", "industry news"]
},
"keyword_analysis": {
"high_value_keywords": ["artificial intelligence", "machine learning", "AI tools", "automation"],
"search_volume": {"artificial intelligence": 100000, "machine learning": 80000}
}
},
"gap_analysis": {
"content_gaps": ["Video tutorials", "Interactive demos", "Case studies", "Beginner guides"],
"keyword_opportunities": ["AI for beginners", "machine learning tutorial", "AI tools comparison"],
"implementation_priority": {"high": ["Video tutorials"], "medium": ["Case studies"]}
},
"performance_data": {
"engagement_metrics": {"average_engagement": 0.05, "peak_engagement_time": "9am-11am"},
"best_performing_content": ["How-to guides", "Industry insights", "Product comparisons"],
"platform_performance": {"linkedin": 0.08, "twitter": 0.03, "blog": 0.12}
},
"competitor_data": {
"competitor_urls": ["https://openai.com/blog", "https://ai.googleblog.com"],
"analysis_date": datetime.now().isoformat()
}
},
"step_results": {},
"quality_scores": {},
"current_step": 0,
"phase": "initialization"
}
phase1_results = {}
total_execution_time = 0
# Test Step 1: Content Strategy Analysis with real services
print("🎯 Testing Step 1: Content Strategy Analysis with Real Services")
try:
step1 = ContentStrategyAnalysisStep()
result1 = await step1.run(real_context)
phase1_results["step_01"] = result1
total_execution_time += result1.get('execution_time', 0.0)
print(f"✅ Step 1 Status: {result1.get('status', 'unknown')}")
print(f"📊 Step 1 Quality: {result1.get('quality_score', 0.0):.2f}")
print(f"⏱️ Step 1 Time: {result1.get('execution_time', 0.0):.2f}s")
# Check if real services were used
step_result = result1.get('result', {})
strategy_summary = step_result.get('content_strategy_summary', {})
if strategy_summary.get('content_pillars'):
print(f" ✅ Real strategy data processed: {len(strategy_summary['content_pillars'])} pillars")
except Exception as e:
print(f"❌ Step 1 failed: {str(e)}")
# Test Step 2: Gap Analysis with real services
print("\n🎯 Testing Step 2: Gap Analysis & Opportunity Identification with Real Services")
try:
step2 = GapAnalysisStep()
result2 = await step2.run(real_context)
phase1_results["step_02"] = result2
total_execution_time += result2.get('execution_time', 0.0)
print(f"✅ Step 2 Status: {result2.get('status', 'unknown')}")
print(f"📊 Step 2 Quality: {result2.get('quality_score', 0.0):.2f}")
print(f"⏱️ Step 2 Time: {result2.get('execution_time', 0.0):.2f}s")
# Check if real services were used
step_result = result2.get('result', {})
gap_analysis = step_result.get('prioritized_gaps', {})
if gap_analysis.get('content_gaps'):
print(f" ✅ Real gap data processed: {len(gap_analysis['content_gaps'])} gaps")
except Exception as e:
print(f"❌ Step 2 failed: {str(e)}")
# Test Step 3: Audience & Platform Strategy with real services
print("\n🎯 Testing Step 3: Audience & Platform Strategy with Real Services")
try:
step3 = AudiencePlatformStrategyStep()
result3 = await step3.run(real_context)
phase1_results["step_03"] = result3
total_execution_time += result3.get('execution_time', 0.0)
print(f"✅ Step 3 Status: {result3.get('status', 'unknown')}")
print(f"📊 Step 3 Quality: {result3.get('quality_score', 0.0):.2f}")
print(f"⏱️ Step 3 Time: {result3.get('execution_time', 0.0):.2f}s")
# Check if real services were used
step_result = result3.get('result', {})
audience_personas = step_result.get('audience_personas', {})
if audience_personas.get('demographics'):
print(f" ✅ Real audience data processed")
except Exception as e:
print(f"❌ Step 3 failed: {str(e)}")
# Calculate overall metrics
completed_steps = len([r for r in phase1_results.values() if r.get('status') == 'completed'])
total_quality = sum(r.get('quality_score', 0.0) for r in phase1_results.values())
avg_quality = total_quality / len(phase1_results) if phase1_results else 0.0
print(f"\n📋 Phase 1 Real Services Integration Summary")
print("=" * 50)
print(f"✅ Completed Steps: {completed_steps}/3")
print(f"📊 Average Quality: {avg_quality:.2f}")
print(f"⏱️ Total Time: {total_execution_time:.2f}s")
return completed_steps == 3, phase1_results
except ImportError as e:
print(f"❌ Phase 1 steps not available: {e}")
return False, {}
async def test_end_to_end_calendar_generation():
"""Test complete end-to-end calendar generation with real services."""
print("\n🚀 Testing End-to-End Calendar Generation with Real Services")
print("=" * 60)
try:
from calendar_generation_datasource_framework.prompt_chaining import PromptChainOrchestrator
# Initialize orchestrator
print("📋 Initializing Prompt Chain Orchestrator...")
orchestrator = PromptChainOrchestrator()
# Test full calendar generation
print("🎯 Testing complete calendar generation...")
try:
result = await orchestrator.generate_calendar(
user_id=1,
strategy_id=1,
calendar_type="monthly",
industry="technology",
business_size="sme"
)
print("✅ End-to-end calendar generation completed!")
# Analyze result quality
quality_score = result.get('quality_score', 0.0)
ai_confidence = result.get('ai_confidence', 0.0)
processing_time = result.get('processing_time', 0.0)
print(f"📊 Quality Score: {quality_score:.2f}")
print(f"🤖 AI Confidence: {ai_confidence:.2f}")
print(f"⏱️ Processing Time: {processing_time:.2f}s")
print(f"🎯 Framework Version: {result.get('framework_version', 'unknown')}")
# Check calendar content completeness
calendar_fields = [
'daily_schedule', 'weekly_themes', 'content_recommendations',
'optimal_timing', 'performance_predictions', 'trending_topics',
'content_pillars', 'platform_strategies', 'gap_analysis_insights'
]
present_fields = [field for field in calendar_fields if field in result and result[field]]
completeness_score = len(present_fields) / len(calendar_fields) * 100
print(f"📋 Content Completeness: {completeness_score:.1f}% ({len(present_fields)}/{len(calendar_fields)} fields)")
# Check step results
step_results = result.get('step_results_summary', {})
completed_steps = len([s for s in step_results.values() if s.get('status') == 'completed'])
print(f"🎯 Steps Completed: {completed_steps}/12")
return True, {
'quality_score': quality_score,
'ai_confidence': ai_confidence,
'processing_time': processing_time,
'completeness_score': completeness_score,
'completed_steps': completed_steps
}
except Exception as e:
print(f"❌ End-to-end calendar generation failed: {str(e)}")
return False, None
except ImportError as e:
print(f"❌ Prompt Chain Orchestrator not available: {e}")
return False, None
async def generate_real_services_report(test_results: Dict[str, Any]):
"""Generate comprehensive real services integration report."""
print("\n📋 Real Services Integration Report")
print("=" * 60)
# Service connectivity
services_tested = 0
services_working = 0
for test_name, (success, data) in test_results.items():
services_tested += 1
if success:
services_working += 1
print(f"{test_name}: SUCCESS")
else:
print(f"{test_name}: FAILED")
connectivity_score = services_working / services_tested * 100 if services_tested > 0 else 0
print(f"\n🔧 Service Connectivity: {services_working}/{services_tested} ({connectivity_score:.1f}%)")
# Phase 1 integration analysis
if 'phase1_real_services' in test_results:
phase1_success, phase1_data = test_results['phase1_real_services']
if phase1_success:
avg_quality = sum(r.get('quality_score', 0.0) for r in phase1_data.values()) / len(phase1_data)
total_time = sum(r.get('execution_time', 0.0) for r in phase1_data.values())
print(f"🎯 Phase 1 Quality: {avg_quality:.2f}")
print(f"⏱️ Phase 1 Time: {total_time:.2f}s")
# End-to-end analysis
if 'e2e_calendar_generation' in test_results:
e2e_success, e2e_data = test_results['e2e_calendar_generation']
if e2e_success and e2e_data:
print(f"🚀 E2E Quality: {e2e_data['quality_score']:.2f}")
print(f"🤖 E2E Confidence: {e2e_data['ai_confidence']:.2f}")
print(f"📋 E2E Completeness: {e2e_data['completeness_score']:.1f}%")
# Overall assessment
if connectivity_score >= 80:
print(f"\n🎉 EXCELLENT: Real services integration ready for production!")
elif connectivity_score >= 60:
print(f"\n✅ GOOD: Most services working, minor issues to resolve")
elif connectivity_score >= 40:
print(f"\n⚠️ FAIR: Some services working, significant improvements needed")
else:
print(f"\n❌ POOR: Major service integration issues, requires attention")
# Save detailed report
report = {
"timestamp": datetime.now().isoformat(),
"service_connectivity": {
"working": services_working,
"tested": services_tested,
"percentage": connectivity_score
},
"test_results": test_results,
"overall_status": "excellent" if connectivity_score >= 80 else "good" if connectivity_score >= 60 else "fair" if connectivity_score >= 40 else "poor"
}
with open("real_services_integration_report.json", "w") as f:
json.dump(report, f, indent=2, default=str)
print(f"\n💾 Detailed report saved to: real_services_integration_report.json")
async def main():
"""Main real services integration test function."""
print("🧪 Real Services Integration Test Suite")
print("=" * 60)
print(f"🕒 Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
test_results = {}
# Test individual real services
test_results['ai_engine'] = await test_real_ai_engine_service()
test_results['keyword_researcher'] = await test_real_keyword_researcher()
test_results['onboarding_service'] = await test_real_onboarding_service()
test_results['data_processing'] = await test_real_data_processing()
# Test Phase 1 with real services
test_results['phase1_real_services'] = await test_phase1_with_real_services()
# Test end-to-end calendar generation
test_results['e2e_calendar_generation'] = await test_end_to_end_calendar_generation()
# Generate comprehensive report
await generate_real_services_report(test_results)
print(f"\n🏁 Real services integration test completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
if __name__ == "__main__":
asyncio.run(main())

View File

@@ -0,0 +1,357 @@
"""
User Workspace Manager
Handles user-specific workspace creation, configuration, and progressive setup.
"""
import os
import json
import shutil
from pathlib import Path
from typing import Dict, Any, Optional, List
from datetime import datetime
from loguru import logger
from sqlalchemy.orm import Session
from sqlalchemy import text
class UserWorkspaceManager:
"""Manages user-specific workspaces and progressive setup."""
def __init__(self, db_session: Session):
self.db = db_session
self.base_workspace_dir = Path("lib/workspace")
self.user_workspaces_dir = self.base_workspace_dir / "users"
def create_user_workspace(self, user_id: str) -> Dict[str, Any]:
"""Create a complete user workspace with progressive setup."""
try:
logger.info(f"Creating workspace for user {user_id}")
# Create user-specific directories
user_dir = self.user_workspaces_dir / f"user_{user_id}"
user_dir.mkdir(parents=True, exist_ok=True)
# Create subdirectories
subdirs = [
"content",
"research",
"config",
"cache",
"exports",
"templates"
]
for subdir in subdirs:
(user_dir / subdir).mkdir(exist_ok=True)
# Create user-specific configuration
config = self._create_user_config(user_id)
config_file = user_dir / "config" / "user_config.json"
with open(config_file, 'w') as f:
json.dump(config, f, indent=2)
# Create user-specific database tables if needed
self._create_user_database_tables(user_id)
logger.info(f"✅ User workspace created: {user_dir}")
return {
"user_id": user_id,
"workspace_path": str(user_dir),
"config": config,
"created_at": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error creating user workspace: {e}")
raise
def _create_user_config(self, user_id: str) -> Dict[str, Any]:
"""Create user-specific configuration."""
return {
"user_id": user_id,
"created_at": datetime.now().isoformat(),
"onboarding_completed": False,
"api_keys": {
"gemini": None,
"exa": None,
"copilotkit": None
},
"preferences": {
"research_depth": "standard",
"content_types": ["blog", "social"],
"auto_research": True
},
"workspace_settings": {
"max_content_items": 1000,
"cache_duration_hours": 24,
"export_formats": ["json", "csv", "pdf"]
}
}
def _create_user_database_tables(self, user_id: str):
"""Create user-specific database tables."""
try:
# Create user-specific content tables
user_tables = [
f"user_{user_id}_content_items",
f"user_{user_id}_research_cache",
f"user_{user_id}_ai_analyses",
f"user_{user_id}_exports"
]
for table in user_tables:
create_sql = f"""
CREATE TABLE IF NOT EXISTS {table} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id VARCHAR(50) NOT NULL,
data JSON,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
self.db.execute(text(create_sql))
self.db.commit()
logger.info(f"✅ User-specific tables created for user {user_id}")
except Exception as e:
logger.error(f"Error creating user database tables: {e}")
self.db.rollback()
raise
def get_user_workspace(self, user_id: str) -> Optional[Dict[str, Any]]:
"""Get user workspace information."""
user_dir = self.user_workspaces_dir / f"user_{user_id}"
if not user_dir.exists():
return None
config_file = user_dir / "config" / "user_config.json"
if config_file.exists():
with open(config_file, 'r') as f:
config = json.load(f)
return {
"user_id": user_id,
"workspace_path": str(user_dir),
"config": config
}
return None
def update_user_config(self, user_id: str, updates: Dict[str, Any]) -> bool:
"""Update user configuration."""
try:
user_dir = self.user_workspaces_dir / f"user_{user_id}"
config_file = user_dir / "config" / "user_config.json"
if config_file.exists():
with open(config_file, 'r') as f:
config = json.load(f)
# Deep merge updates
self._deep_merge(config, updates)
with open(config_file, 'w') as f:
json.dump(config, f, indent=2)
logger.info(f"✅ User config updated for user {user_id}")
return True
return False
except Exception as e:
logger.error(f"Error updating user config: {e}")
return False
def _deep_merge(self, base: Dict, updates: Dict):
"""Deep merge two dictionaries."""
for key, value in updates.items():
if key in base and isinstance(base[key], dict) and isinstance(value, dict):
self._deep_merge(base[key], value)
else:
base[key] = value
def setup_progressive_features(self, user_id: str, onboarding_step: int) -> Dict[str, Any]:
"""Set up features progressively based on onboarding progress."""
setup_status = {
"user_id": user_id,
"step": onboarding_step,
"features_enabled": [],
"tables_created": [],
"services_initialized": []
}
try:
# Step 1: API Keys - Enable basic AI services
if onboarding_step >= 1:
self._setup_ai_services(user_id)
setup_status["features_enabled"].append("ai_services")
setup_status["services_initialized"].append("gemini")
setup_status["services_initialized"].append("exa")
setup_status["services_initialized"].append("copilotkit")
# Step 2: Website Analysis - Enable content analysis
if onboarding_step >= 2:
self._setup_content_analysis(user_id)
setup_status["features_enabled"].append("content_analysis")
setup_status["tables_created"].append(f"user_{user_id}_content_analysis")
# Step 3: Research - Enable research capabilities
if onboarding_step >= 3:
self._setup_research_services(user_id)
setup_status["features_enabled"].append("research_services")
setup_status["tables_created"].append(f"user_{user_id}_research_cache")
# Step 4: Personalization - Enable user-specific features
if onboarding_step >= 4:
self._setup_personalization(user_id)
setup_status["features_enabled"].append("personalization")
setup_status["tables_created"].append(f"user_{user_id}_preferences")
# Step 5: Integrations - Enable external integrations
if onboarding_step >= 5:
self._setup_integrations(user_id)
setup_status["features_enabled"].append("integrations")
setup_status["services_initialized"].append("wix")
setup_status["services_initialized"].append("linkedin")
# Step 6: Complete - Enable all features
if onboarding_step >= 6:
self._setup_complete_features(user_id)
setup_status["features_enabled"].append("all_features")
setup_status["tables_created"].append(f"user_{user_id}_complete_workspace")
logger.info(f"✅ Progressive setup completed for user {user_id} at step {onboarding_step}")
return setup_status
except Exception as e:
logger.error(f"Error in progressive setup: {e}")
raise
def _setup_ai_services(self, user_id: str):
"""Set up AI services for the user."""
# Create user-specific AI service configuration
user_dir = self.user_workspaces_dir / f"user_{user_id}"
ai_config = user_dir / "config" / "ai_services.json"
ai_services = {
"gemini": {"enabled": True, "model": "gemini-pro"},
"exa": {"enabled": True, "search_depth": "standard"},
"copilotkit": {"enabled": True, "assistant_type": "content"}
}
with open(ai_config, 'w') as f:
json.dump(ai_services, f, indent=2)
def _setup_content_analysis(self, user_id: str):
"""Set up content analysis capabilities."""
# Create content analysis tables
create_sql = f"""
CREATE TABLE IF NOT EXISTS user_{user_id}_content_analysis (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content_id VARCHAR(100),
analysis_type VARCHAR(50),
results JSON,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
self.db.execute(text(create_sql))
self.db.commit()
def _setup_research_services(self, user_id: str):
"""Set up research services."""
# Create research cache table
create_sql = f"""
CREATE TABLE IF NOT EXISTS user_{user_id}_research_cache (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query_hash VARCHAR(64),
research_data JSON,
expires_at DATETIME,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
self.db.execute(text(create_sql))
self.db.commit()
def _setup_personalization(self, user_id: str):
"""Set up personalization features."""
# Create user preferences table
create_sql = f"""
CREATE TABLE IF NOT EXISTS user_{user_id}_preferences (
id INTEGER PRIMARY KEY AUTOINCREMENT,
preference_type VARCHAR(50),
preference_data JSON,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
)
"""
self.db.execute(text(create_sql))
self.db.commit()
def _setup_integrations(self, user_id: str):
"""Set up external integrations."""
# Create integrations configuration
user_dir = self.user_workspaces_dir / f"user_{user_id}"
integrations_config = user_dir / "config" / "integrations.json"
integrations = {
"wix": {"enabled": False, "connected": False},
"linkedin": {"enabled": False, "connected": False},
"wordpress": {"enabled": False, "connected": False}
}
with open(integrations_config, 'w') as f:
json.dump(integrations, f, indent=2)
def _setup_complete_features(self, user_id: str):
"""Set up complete feature set."""
# Create comprehensive workspace
user_dir = self.user_workspaces_dir / f"user_{user_id}"
# Create additional directories for complete setup
complete_dirs = [
"ai_models",
"content_templates",
"export_templates",
"backup"
]
for dir_name in complete_dirs:
(user_dir / dir_name).mkdir(exist_ok=True)
# Create final configuration
final_config = {
"setup_complete": True,
"all_features_enabled": True,
"last_updated": datetime.now().isoformat()
}
self.update_user_config(user_id, final_config)
def cleanup_user_workspace(self, user_id: str) -> bool:
"""Clean up user workspace (for account deletion)."""
try:
user_dir = self.user_workspaces_dir / f"user_{user_id}"
if user_dir.exists():
shutil.rmtree(user_dir)
# Drop user-specific tables
user_tables = [
f"user_{user_id}_content_items",
f"user_{user_id}_research_cache",
f"user_{user_id}_ai_analyses",
f"user_{user_id}_exports",
f"user_{user_id}_content_analysis",
f"user_{user_id}_preferences"
]
for table in user_tables:
try:
self.db.execute(text(f"DROP TABLE IF EXISTS {table}"))
except:
pass # Table might not exist
self.db.commit()
logger.info(f"✅ User workspace cleaned up for user {user_id}")
return True
except Exception as e:
logger.error(f"Error cleaning up user workspace: {e}")
return False

View File

@@ -233,6 +233,19 @@ def validate_api_key(provider: str, api_key: str) -> Dict[str, Any]:
if len(api_key) < 10:
return {'valid': False, 'error': 'Metaphor API key seems too short'}
elif provider == "exa":
# Exa API keys are UUIDs (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)
import re
exa_uuid_regex = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE)
if not exa_uuid_regex.match(api_key):
return {'valid': False, 'error': 'Exa API key must be a valid UUID (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)'}
elif provider == "copilotkit":
if not api_key.startswith("ck_pub_"):
return {'valid': False, 'error': 'CopilotKit API key must start with "ck_pub_"'}
if len(api_key) < 20:
return {'valid': False, 'error': 'CopilotKit API key seems too short'}
elif provider == "firecrawl":
if len(api_key) < 10:
return {'valid': False, 'error': 'Firecrawl API key seems too short'}
@@ -277,21 +290,49 @@ def validate_step_data(step_number: int, data: Dict[str, Any]) -> List[str]:
"""Validate step-specific data with enhanced logic."""
errors = []
if step_number == 1: # AI LLM Providers
logger.info(f"[validate_step_data] Validating step {step_number} with data: {data}")
if step_number == 1: # AI LLM Providers - Now requires Gemini, Exa, and CopilotKit
required_providers = ['gemini', 'exa', 'copilotkit']
missing_providers = []
logger.info(f"[validate_step_data] Step 1 validation - data type: {type(data)}, data: {data}")
if not data or 'api_keys' not in data:
errors.append("At least one API key must be configured")
logger.warning(f"[validate_step_data] No data or api_keys missing. data: {data}")
errors.append("API keys configuration is required")
elif not data['api_keys']:
errors.append("At least one API key must be configured")
logger.warning(f"[validate_step_data] api_keys is empty. data: {data}")
errors.append("API keys configuration is required")
else:
# Validate each configured API key
for provider in data['api_keys']:
if provider not in ['openai', 'gemini', 'anthropic', 'mistral']:
errors.append(f"Unknown provider: {provider}")
# Check for all required providers
for provider in required_providers:
if provider not in data['api_keys'] or not data['api_keys'][provider]:
missing_providers.append(provider)
if missing_providers:
errors.append(f"Missing required API keys: {', '.join(missing_providers)}")
# Validate each configured API key format
for provider, api_key in data['api_keys'].items():
if provider in required_providers and api_key:
if provider == 'gemini' and not api_key.startswith('AIza'):
errors.append("Gemini API key must start with 'AIza'")
elif provider == 'exa':
# Exa API keys are UUIDs (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)
import re
exa_uuid_regex = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$', re.IGNORECASE)
if not exa_uuid_regex.match(api_key):
errors.append("Exa API key must be a valid UUID (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)")
elif provider == 'copilotkit' and not api_key.startswith('ck_pub_'):
errors.append("CopilotKit API key must start with 'ck_pub_'")
elif step_number == 2: # Website Analysis
if not data or 'website_url' not in data:
# Accept both 'website' and 'website_url' for backwards compatibility
website_url = data.get('website') or data.get('website_url') if data else None
if not website_url:
errors.append("Website URL is required")
elif not validate_website_url(data['website_url']):
elif not validate_website_url(website_url):
errors.append("Invalid website URL format")
elif step_number == 3: # AI Research

View File

@@ -0,0 +1,418 @@
"""
Wix Integration Service
Handles authentication, permission checking, and blog publishing to Wix websites.
"""
import os
import json
import requests
from typing import Dict, Any, Optional, List
from loguru import logger
from datetime import datetime, timedelta
import base64
from urllib.parse import urlencode, parse_qs
import jwt
import base64 as b64
from services.integrations.wix.blog import WixBlogService
from services.integrations.wix.media import WixMediaService
from services.integrations.wix.utils import extract_meta_from_token, normalize_token_string, extract_member_id_from_access_token as utils_extract_member
from services.integrations.wix.content import convert_content_to_ricos as ricos_builder
from services.integrations.wix.auth import WixAuthService
class WixService:
"""Service for interacting with Wix APIs"""
def __init__(self):
self.client_id = os.getenv('WIX_CLIENT_ID')
self.redirect_uri = os.getenv('WIX_REDIRECT_URI', 'https://littery-sonny-unscrutinisingly.ngrok-free.dev/wix/callback')
self.base_url = 'https://www.wixapis.com'
self.oauth_url = 'https://www.wix.com/oauth/authorize'
# Modular services
self.blog_service = WixBlogService(self.base_url, self.client_id)
self.media_service = WixMediaService(self.base_url)
self.auth_service = WixAuthService(self.client_id, self.redirect_uri, self.base_url)
if not self.client_id:
logger.warning("Wix client ID not configured. Set WIX_CLIENT_ID environment variable.")
def get_authorization_url(self, state: str = None) -> str:
"""
Generate Wix OAuth authorization URL for "on behalf of user" authentication
This implements the "Authenticate on behalf of a Wix User" flow as described in:
https://dev.wix.com/docs/build-apps/develop-your-app/access/authentication/authenticate-on-behalf-of-a-wix-user
Args:
state: Optional state parameter for security
Returns:
Authorization URL for user to visit
"""
url, code_verifier = self.auth_service.generate_authorization_url(state)
self._code_verifier = code_verifier
return url
def _create_redirect_session_for_auth(self, redirect_uri: str, client_id: str, code_challenge: str, state: str) -> str:
"""
Create a redirect session for Wix Headless OAuth authentication using Redirects API
Args:
redirect_uri: The redirect URI for OAuth callback
client_id: The OAuth client ID
code_challenge: The PKCE code challenge
state: The OAuth state parameter
Returns:
The redirect URL for OAuth authentication
"""
try:
# According to Wix documentation, we need to use the Redirects API
# to create a redirect session for OAuth authentication
# This is the correct approach for Wix Headless OAuth
# For now, return the direct OAuth URL as a fallback
# In production, this should call the Wix Redirects API
redirect_url = f"https://www.wix.com/oauth/authorize?client_id={client_id}&redirect_uri={redirect_uri}&response_type=code&scope=BLOG.CREATE-DRAFT,BLOG.PUBLISH,MEDIA.MANAGE&code_challenge={code_challenge}&code_challenge_method=S256&state={state}"
logger.info(f"Generated Wix Headless OAuth redirect URL: {redirect_url}")
logger.warning("Using direct OAuth URL - should implement Redirects API for production")
return redirect_url
except Exception as e:
logger.error(f"Failed to create redirect session for auth: {e}")
raise
def exchange_code_for_tokens(self, code: str, code_verifier: str = None) -> Dict[str, Any]:
"""
Exchange authorization code for access and refresh tokens using PKCE
Args:
code: Authorization code from Wix
code_verifier: PKCE code verifier (uses stored one if not provided)
Returns:
Token response with access_token, refresh_token, etc.
"""
if not self.client_id:
raise ValueError("Wix client ID not configured")
if not code_verifier:
code_verifier = getattr(self, '_code_verifier', None)
if not code_verifier:
raise ValueError("Code verifier not found. Please provide code_verifier parameter.")
try:
return self.auth_service.exchange_code_for_tokens(code, code_verifier)
except requests.RequestException as e:
logger.error(f"Failed to exchange code for tokens: {e}")
raise
def refresh_access_token(self, refresh_token: str) -> Dict[str, Any]:
"""
Refresh access token using refresh token (Wix Headless OAuth)
Args:
refresh_token: Valid refresh token
Returns:
New token response
"""
if not self.client_id:
raise ValueError("Wix client ID not configured")
try:
return self.auth_service.refresh_access_token(refresh_token)
except requests.RequestException as e:
logger.error(f"Failed to refresh access token: {e}")
raise
def get_site_info(self, access_token: str) -> Dict[str, Any]:
"""
Get information about the connected Wix site
Args:
access_token: Valid access token
Returns:
Site information
"""
token_str = normalize_token_string(access_token)
if not token_str:
raise ValueError("Invalid access token format for create_blog_post")
try:
return self.auth_service.get_site_info(token_str)
except requests.RequestException as e:
logger.error(f"Failed to get site info: {e}")
raise
def get_current_member(self, access_token: str) -> Dict[str, Any]:
"""
Get current member information (for third-party apps)
Args:
access_token: Valid access token
Returns:
Current member information
"""
token_str = normalize_token_string(access_token)
if not token_str:
raise ValueError("Invalid access token format for get_current_member")
try:
return self.auth_service.get_current_member(token_str, self.client_id)
except requests.RequestException as e:
logger.error(f"Failed to get current member: {e}")
raise
def extract_member_id_from_access_token(self, access_token: Any) -> Optional[str]:
return utils_extract_member(access_token)
def _normalize_token_string(self, access_token: Any) -> Optional[str]:
return normalize_token_string(access_token)
def check_blog_permissions(self, access_token: str) -> Dict[str, Any]:
"""
Check if the app has required blog permissions
Args:
access_token: Valid access token
Returns:
Permission status
"""
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json',
'wix-client-id': self.client_id or ''
}
try:
# Try to list blog categories to check permissions
response = requests.get(
f"{self.base_url}/blog/v1/categories",
headers=headers
)
if response.status_code == 200:
return {
'has_permissions': True,
'can_create_posts': True,
'can_publish': True
}
elif response.status_code == 403:
return {
'has_permissions': False,
'can_create_posts': False,
'can_publish': False,
'error': 'Insufficient permissions'
}
else:
response.raise_for_status()
except requests.RequestException as e:
logger.error(f"Failed to check blog permissions: {e}")
return {
'has_permissions': False,
'error': str(e)
}
def import_image_to_wix(self, access_token: str, image_url: str, display_name: str = None) -> str:
"""
Import external image to Wix Media Manager
Args:
access_token: Valid access token
image_url: URL of the image to import
display_name: Optional display name for the image
Returns:
Wix media ID
"""
try:
result = self.media_service.import_image(
access_token,
image_url,
display_name or f'Imported Image {datetime.now().strftime("%Y%m%d_%H%M%S")}'
)
return result['file']['id']
except requests.RequestException as e:
logger.error(f"Failed to import image to Wix: {e}")
raise
def convert_content_to_ricos(self, content: str, images: List[str] = None) -> Dict[str, Any]:
return ricos_builder(content, images)
def create_blog_post(self, access_token: str, title: str, content: str,
cover_image_url: str = None, category_ids: List[str] = None,
tag_ids: List[str] = None, publish: bool = True,
member_id: str = None) -> Dict[str, Any]:
"""
Create and optionally publish a blog post on Wix
Args:
access_token: Valid access token
title: Blog post title
content: Blog post content
cover_image_url: Optional cover image URL
category_ids: Optional list of category IDs
tag_ids: Optional list of tag IDs
publish: Whether to publish immediately or save as draft
member_id: Required for third-party apps - the member ID of the post author
Returns:
Created blog post information
"""
if not member_id:
raise ValueError("memberId is required for third-party apps creating blog posts")
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
# Build valid Ricos rich content (minimum: one paragraph with text)
ricos_content = self.convert_content_to_ricos(content or "This is a post from ALwrity.", None)
# Minimal payload per Wix docs: title, memberId, and richContent
blog_data = {
'draftPost': {
'title': title,
'memberId': member_id, # Required for third-party apps
'richContent': ricos_content,
'excerpt': (content or '').strip()[:200]
},
'publish': publish,
'fieldsets': ['URL'] # Simplified fieldsets
}
# Add cover image if provided
if cover_image_url:
try:
media_id = self.import_image_to_wix(access_token, cover_image_url, f'Cover: {title}')
blog_data['draftPost']['media'] = {
'wixMedia': {
'image': {'id': media_id}
},
'displayed': True,
'custom': True
}
except Exception as e:
logger.warning(f"Failed to import cover image: {e}")
# Add categories if provided
if category_ids:
blog_data['draftPost']['categoryIds'] = category_ids
# Add tags if provided
if tag_ids:
blog_data['draftPost']['tagIds'] = tag_ids
try:
# Check what permissions we have in the token
logger.info("DEBUG: Checking token permissions...")
try:
import jwt
# Extract token string manually since _normalize_access_token doesn't exist
token_str = str(access_token)
if token_str and token_str.startswith('OauthNG.JWS.'):
jwt_part = token_str[12:]
payload = jwt.decode(jwt_part, options={"verify_signature": False, "verify_aud": False})
logger.info(f"DEBUG: Full token payload: {payload}")
# Check for permissions in various possible locations
data_payload = payload.get('data', {})
if isinstance(data_payload, str):
try:
data_payload = json.loads(data_payload)
except:
pass
instance_data = data_payload.get('instance', {})
permissions = instance_data.get('permissions', '')
scopes = instance_data.get('scopes', [])
meta_site_id = instance_data.get('metaSiteId')
if isinstance(meta_site_id, str) and meta_site_id:
headers['wix-site-id'] = meta_site_id
logger.info(f"DEBUG: Added wix-site-id header: {meta_site_id}")
logger.info(f"DEBUG: Token permissions: {permissions}")
logger.info(f"DEBUG: Token scopes: {scopes}")
else:
logger.info("DEBUG: Could not decode token for permission check")
except Exception as perm_e:
logger.warning(f"DEBUG: Failed to check permissions: {perm_e}")
logger.info(f"DEBUG: Sending simplified blog data: {json.dumps(blog_data, indent=2)}")
extra_headers = {}
if 'wix-site-id' in headers:
extra_headers['wix-site-id'] = headers['wix-site-id']
result = self.blog_service.create_draft_post(access_token, blog_data, extra_headers or None)
logger.info(f"DEBUG: Create draft result: {result}")
return result
except requests.RequestException as e:
logger.error(f"Failed to create blog post: {e}")
if hasattr(e, 'response') and e.response is not None:
logger.error(f"Response body: {e.response.text}")
raise
def get_blog_categories(self, access_token: str) -> List[Dict[str, Any]]:
"""
Get available blog categories
Args:
access_token: Valid access token
Returns:
List of blog categories
"""
try:
return self.blog_service.list_categories(access_token)
except requests.RequestException as e:
logger.error(f"Failed to get blog categories: {e}")
raise
def get_blog_tags(self, access_token: str) -> List[Dict[str, Any]]:
"""
Get available blog tags
Args:
access_token: Valid access token
Returns:
List of blog tags
"""
try:
return self.blog_service.list_tags(access_token)
except requests.RequestException as e:
logger.error(f"Failed to get blog tags: {e}")
raise
def publish_draft_post(self, access_token: str, draft_post_id: str) -> Dict[str, Any]:
"""
Publish a draft post by ID.
"""
try:
result = self.blog_service.publish_draft(access_token, draft_post_id)
logger.info(f"DEBUG: Publish result: {result}")
return result
except requests.RequestException as e:
logger.error(f"Failed to publish draft post: {e}")
raise
def create_category(self, access_token: str, label: str, description: Optional[str] = None,
language: Optional[str] = None) -> Dict[str, Any]:
"""
Create a blog category.
"""
try:
return self.blog_service.create_category(access_token, label, description, language)
except requests.RequestException as e:
logger.error(f"Failed to create category: {e}")
raise
def create_tag(self, access_token: str, label: str, language: Optional[str] = None) -> Dict[str, Any]:
"""
Create a blog tag.
"""
try:
return self.blog_service.create_tag(access_token, label, language)
except requests.RequestException as e:
logger.error(f"Failed to create tag: {e}")
raise