Files
gurixandClaude 1bb117fd98 Implement Phase 4: AI-Powered Feedback Analysis (User Story 2)
Implement automatic AI analysis of feedback submissions using Claude API,
including language detection, categorization, summarization, and translation
to product owner's preferred language.

Tasks Completed (T065-T092):
- T065-T069: Unit tests for AI analyzer (5 tests)
- T070: Integration test for full AI analysis workflow
- T071: Created AIAnalyzer abstract base class interface
- T072: Added AnalysisResult dataclass to feedback model
- T073: Implemented ClaudeAnalyzer with Anthropic SDK
- T074: Integrated Claude API with 45s timeout
- T075: Designed single-call analysis prompt
- T076: Language detection implementation
- T077: Category extraction with validation
- T078: Summary generation (1-2 sentences)
- T079: Translation extraction
- T080: Timeout handling for Claude API
- T081: API error handling with proper exceptions
- T082: Analysis storage to analysis.md file
- T083: Formatted markdown output for analysis
- T084: Background analysis trigger on submission
- T085: Non-blocking async analysis via threading
- T086: Status update to 'analyzing' before analysis
- T087: Status update to 'analyzed' on success
- T088: Status update to 'analysis_failed' on error
- T089: Metadata update with category and language
- T090: Environment configuration for ANTHROPIC_API_KEY
- T091: Verification that original content.txt preserved (FR-016)
- T092: Verification that images not analyzed via OCR (FR-021)

Features:
- Abstract AIAnalyzer interface for multiple AI providers
- ClaudeAnalyzer implementation using Anthropic API
- Background threading for non-blocking analysis
- Flask app context management in background threads
- Comprehensive error handling and status tracking
- Original content preservation (FR-016 compliance)
- Image storage without OCR (FR-021 compliance)

Testing:
- 5 unit tests for AI analyzer components
- 2 integration tests for full analysis workflow
- All 46 tests passing (1 skipped)
- Mock-based testing to avoid API calls

Files Changed:
- app/models/feedback.py: Added AnalysisResult dataclass
- app/routes/submission.py: Background analysis integration
- app/services/ai_analyzer.py: NEW - AI analysis service
- app/services/feedback_storage.py: Analysis storage methods
- tests/unit/test_ai_analyzer.py: NEW - Unit tests (5 tests)
- tests/integration/test_ai_analysis_flow.py: NEW - Integration tests (2 tests)
- tests/integration/test_feedback_submission_flow.py: Threading mock added

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-16 21:14:10 +02:00

276 lines
8.2 KiB
Python

"""Feedback model"""
import os
import uuid
from datetime import datetime
from dataclasses import dataclass
import yaml
from flask import current_app
class Feedback:
"""Feedback submission model
Attributes:
feedback_id: Unique feedback identifier (UUID)
product_id: Associated product ID
submitted_at: Submission timestamp (ISO 8601)
status: Feedback status ('new', 'analyzing', 'analyzed', 'analysis_failed', 'archived')
content_preview: First 200 chars of feedback text
has_attachments: Whether feedback has file attachments
attachment_count: Number of attached files
original_language: Detected language of feedback (set during analysis)
category: Feedback category (set during analysis)
"""
VALID_STATUSES = ['new', 'in_progress', 'resolved', 'closed', 'analyzing', 'analyzed', 'analysis_failed', 'archived']
def __init__(self, feedback_id, product_id, submitted_at=None, status='new',
content_preview='', has_attachments=False, attachment_count=0,
original_language=None, category=None):
self.feedback_id = feedback_id
self.product_id = product_id
self.submitted_at = submitted_at or datetime.utcnow().isoformat()
self.status = status
self.content_preview = content_preview
self.has_attachments = has_attachments
self.attachment_count = attachment_count
self.original_language = original_language
self.category = category
def to_dict(self):
"""Convert feedback to dictionary
Returns:
dict: Feedback metadata
"""
data = {
'feedback_id': self.feedback_id,
'product_id': self.product_id,
'submitted_at': self.submitted_at,
'status': self.status,
'content_preview': self.content_preview,
'has_attachments': self.has_attachments,
'attachment_count': self.attachment_count
}
if self.original_language:
data['original_language'] = self.original_language
if self.category:
data['category'] = self.category
return data
@classmethod
def from_dict(cls, data):
"""Create feedback from dictionary
Args:
data: Dictionary with feedback data
Returns:
Feedback: Feedback instance
"""
return cls(
feedback_id=data['feedback_id'],
product_id=data['product_id'],
submitted_at=data.get('submitted_at'),
status=data.get('status', 'new'),
content_preview=data.get('content_preview', ''),
has_attachments=data.get('has_attachments', False),
attachment_count=data.get('attachment_count', 0),
original_language=data.get('original_language'),
category=data.get('category')
)
@staticmethod
def generate_id():
"""Generate unique feedback ID
Returns:
str: UUID-based feedback ID
"""
return str(uuid.uuid4())
@staticmethod
def _get_feedback_dir(product_id, feedback_id):
"""Get feedback directory path
Args:
product_id: Product ID
feedback_id: Feedback ID
Returns:
str: Path to feedback directory
"""
return os.path.join(
current_app.config['DATA_DIR'],
'products',
product_id,
'feedback',
feedback_id
)
@staticmethod
def _get_metadata_file(product_id, feedback_id):
"""Get metadata file path
Args:
product_id: Product ID
feedback_id: Feedback ID
Returns:
str: Path to metadata.yaml
"""
feedback_dir = Feedback._get_feedback_dir(product_id, feedback_id)
return os.path.join(feedback_dir, 'metadata.yaml')
@staticmethod
def _get_content_file(product_id, feedback_id):
"""Get content file path
Args:
product_id: Product ID
feedback_id: Feedback ID
Returns:
str: Path to content.txt
"""
feedback_dir = Feedback._get_feedback_dir(product_id, feedback_id)
return os.path.join(feedback_dir, 'content.txt')
@staticmethod
def _get_attachments_dir(product_id, feedback_id):
"""Get attachments directory path
Args:
product_id: Product ID
feedback_id: Feedback ID
Returns:
str: Path to attachments directory
"""
feedback_dir = Feedback._get_feedback_dir(product_id, feedback_id)
return os.path.join(feedback_dir, 'attachments')
@classmethod
def get_by_id(cls, product_id, feedback_id):
"""Load feedback by ID
Args:
product_id: Product ID
feedback_id: Feedback ID
Returns:
Feedback or None: Feedback instance if found, None otherwise
"""
metadata_file = cls._get_metadata_file(product_id, feedback_id)
if not os.path.exists(metadata_file):
return None
with open(metadata_file, 'r') as f:
data = yaml.safe_load(f)
return cls.from_dict(data)
@classmethod
def get_all_for_product(cls, product_id):
"""Get all feedback for a product
Args:
product_id: Product ID
Returns:
list: List of Feedback instances, sorted by submitted_at (newest first)
"""
feedback_list = []
feedback_base_dir = os.path.join(
current_app.config['DATA_DIR'],
'products',
product_id,
'feedback'
)
if not os.path.exists(feedback_base_dir):
return feedback_list
for feedback_id in os.listdir(feedback_base_dir):
feedback_dir = os.path.join(feedback_base_dir, feedback_id)
if not os.path.isdir(feedback_dir):
continue
feedback = cls.get_by_id(product_id, feedback_id)
if feedback:
feedback_list.append(feedback)
# Sort by submitted_at (newest first)
feedback_list.sort(key=lambda f: f.submitted_at, reverse=True)
return feedback_list
def save_metadata(self):
"""Save feedback metadata to filesystem"""
feedback_dir = self._get_feedback_dir(self.product_id, self.feedback_id)
os.makedirs(feedback_dir, exist_ok=True)
metadata_file = self._get_metadata_file(self.product_id, self.feedback_id)
with open(metadata_file, 'w') as f:
yaml.dump(self.to_dict(), f, default_flow_style=False)
def get_content(self):
"""Load feedback content text
Returns:
str or None: Feedback content if exists, None otherwise
"""
content_file = self._get_content_file(self.product_id, self.feedback_id)
if not os.path.exists(content_file):
return None
with open(content_file, 'r') as f:
return f.read()
def get_attachments(self):
"""Get list of attachment filenames
Returns:
list: List of attachment filenames
"""
attachments_dir = self._get_attachments_dir(self.product_id, self.feedback_id)
if not os.path.exists(attachments_dir):
return []
return [f for f in os.listdir(attachments_dir)
if os.path.isfile(os.path.join(attachments_dir, f))]
def validate_status(self):
"""Validate feedback status
Returns:
bool: True if status is valid, False otherwise
"""
return self.status in self.VALID_STATUSES
@dataclass
class AnalysisResult:
"""Result of AI-powered feedback analysis
Attributes:
category: Feedback category (bug, feature_request, question, complaint, praise, other)
original_language: Detected language code (e.g., 'en', 'de', 'fr')
summary: Brief summary of feedback (1-2 sentences)
translation: Feedback translated to target language
raw_analysis: Full analysis text in markdown format
"""
category: str
original_language: str
summary: str
translation: str
raw_analysis: str