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>
276 lines
8.2 KiB
Python
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
|