"""AI-powered feedback analysis service""" from abc import ABC, abstractmethod import anthropic import re import time from app.models.feedback import AnalysisResult class AIAnalyzer(ABC): """Abstract base class for AI-powered feedback analyzers Subclasses must implement the analyze_feedback method to provide categorization, summarization, and translation capabilities. """ @abstractmethod def analyze_feedback(self, feedback_text, target_language, product_id): """Analyze feedback using AI Args: feedback_text: The feedback text to analyze target_language: Language code for translation (e.g., 'en', 'de') product_id: Product ID for context Returns: AnalysisResult: Analysis results including category, summary, and translation """ pass class ClaudeAnalyzer(AIAnalyzer): """Claude AI-based feedback analyzer using Anthropic API Uses Claude to analyze feedback and extract: - Category (bug, feature_request, question, complaint, praise, other) - Original language detection - Summary (1-2 sentences) - Translation to target language """ # Valid feedback categories VALID_CATEGORIES = ['bug', 'feature_request', 'question', 'complaint', 'praise', 'other'] def __init__(self, api_key): """Initialize Claude analyzer Args: api_key: Anthropic API key """ self.api_key = api_key self.client = anthropic.Anthropic(api_key=api_key) def analyze_feedback(self, feedback_text, target_language, product_id): """Analyze feedback using Claude API Args: feedback_text: The feedback text to analyze target_language: Language code for translation (e.g., 'en', 'de') product_id: Product ID for context Returns: AnalysisResult: Analysis results Raises: Exception: If API call fails or timeout occurs """ # Design prompt for Claude API (T075) prompt = self._build_analysis_prompt(feedback_text, target_language) # Retry logic for transient failures (T081) max_retries = 3 retry_delay = 1 # seconds for attempt in range(max_retries): try: # Call Claude API with timeout (T074, T080) response = self._call_claude_api(prompt, timeout=45) # Extract analysis components from response raw_analysis = response.content[0].text # Extract category (T077) category = self._extract_category(raw_analysis) # Detect original language (T076) original_language = self._extract_language(raw_analysis) # Extract summary (T078) summary = self._extract_summary(raw_analysis) # Extract translation (T079) translation = self._extract_translation(raw_analysis) return AnalysisResult( category=category, original_language=original_language, summary=summary, translation=translation, raw_analysis=raw_analysis ) except anthropic.APITimeoutError as e: # Handle API timeouts (T080) - don't retry timeouts raise Exception(f"Claude API timeout after 45s: {str(e)}") except anthropic.APIError as e: # Handle API errors with retry logic (T081) # Check if this is a transient error (rate limit, server error) error_type = type(e).__name__ is_retryable = any(x in error_type.lower() for x in ['ratelimit', 'server', 'unavailable', 'overloaded']) if is_retryable and attempt < max_retries - 1: # Wait before retrying with exponential backoff wait_time = retry_delay * (2 ** attempt) time.sleep(wait_time) continue else: # Not retryable or max retries exceeded raise Exception(f"Claude API error: {str(e)}") except Exception as e: # General error handling - don't retry raise Exception(f"Analysis failed: {str(e)}") def _build_analysis_prompt(self, feedback_text, target_language): """Build the analysis prompt for Claude Prompt design (T075): Single call to categorize, summarize, and translate """ return f"""Analyze the following user feedback and provide a structured analysis. User Feedback: {feedback_text} Please provide your analysis in the following format: # Feedback Analysis **Category**: [Choose ONE: bug, feature_request, question, complaint, praise, other] **Original Language**: [Detect the language code, e.g., en, de, fr, es] **Summary**: [Provide a concise 1-2 sentence summary of the feedback] **Translation**: [Translate the feedback to {target_language}. If already in {target_language}, write "(same as original)"] Important: - Be accurate in language detection - Choose the most appropriate category - Keep the summary brief but informative - Translate naturally and accurately""" def _call_claude_api(self, prompt, timeout=45): """Call Claude API with proper configuration Args: prompt: The prompt to send to Claude timeout: Timeout in seconds (default 45s per T080) Returns: API response object Raises: anthropic.APITimeoutError: If request times out anthropic.APIError: If API returns an error """ return self.client.messages.create( model="claude-haiku-4-5-20251001", max_tokens=1000, timeout=timeout, messages=[ { "role": "user", "content": prompt } ] ) def _extract_category(self, analysis_text): """Extract category from analysis text (T077) Args: analysis_text: Raw analysis markdown Returns: str: Category (defaults to 'other' if not found or invalid) """ # Look for pattern: **Category**: bug match = re.search(r'\*\*Category\*\*:\s*(\w+)', analysis_text, re.IGNORECASE) if match: category = match.group(1).lower() # Validate category if category in self.VALID_CATEGORIES: return category # Default to 'other' if not found or invalid return 'other' def _extract_language(self, analysis_text): """Extract detected language from analysis text (T076) Args: analysis_text: Raw analysis markdown Returns: str: Language code (defaults to 'unknown' if not found) """ # Look for pattern: **Original Language**: en match = re.search(r'\*\*Original Language\*\*:\s*(\w+)', analysis_text, re.IGNORECASE) if match: return match.group(1).lower() # Default to 'unknown' return 'unknown' def _extract_summary(self, analysis_text): """Extract summary from analysis text (T078) Args: analysis_text: Raw analysis markdown Returns: str: Summary text (defaults to empty string if not found) """ # Look for pattern: **Summary**: [text] match = re.search(r'\*\*Summary\*\*:\s*(.+?)(?=\n\*\*|\n\n|$)', analysis_text, re.IGNORECASE | re.DOTALL) if match: return match.group(1).strip() # Default to empty string return '' def _extract_translation(self, analysis_text): """Extract translation from analysis text (T079) Args: analysis_text: Raw analysis markdown Returns: str: Translated text (defaults to empty string if not found) """ # Look for pattern: **Translation**: [text] match = re.search(r'\*\*Translation\*\*:\s*(.+?)(?=\n\*\*|\n\n|$)', analysis_text, re.IGNORECASE | re.DOTALL) if match: return match.group(1).strip() # Default to empty string return ''