"""Feedback storage service""" import os import shutil import yaml from datetime import datetime from flask import current_app from app.models.feedback import Feedback, AnalysisResult from app.utils.file_validator import get_safe_filename class FeedbackStorageService: """Service for storing feedback to filesystem""" @staticmethod def create_feedback(product_id, content_text=None, files=None): """Create new feedback entry Args: product_id: Product ID content_text: Feedback text content (optional) files: List of uploaded files (optional) Returns: Feedback: Created feedback instance """ # Generate unique feedback ID feedback_id = Feedback.generate_id() # Create content preview (first 200 chars) content_preview = '' if content_text: content_preview = content_text[:200] # Check attachments has_attachments = bool(files and len(files) > 0) attachment_count = len(files) if files else 0 # Create feedback instance feedback = Feedback( feedback_id=feedback_id, product_id=product_id, status='new', content_preview=content_preview, has_attachments=has_attachments, attachment_count=attachment_count ) # Create directory structure feedback_dir = Feedback._get_feedback_dir(product_id, feedback_id) os.makedirs(feedback_dir, exist_ok=True) return feedback @staticmethod def save_metadata(feedback): """Save feedback metadata to YAML file Args: feedback: Feedback instance to save """ feedback.save_metadata() @staticmethod def save_content(feedback, content_text): """Save feedback content to text file Args: feedback: Feedback instance content_text: Feedback text content """ if not content_text: return content_file = Feedback._get_content_file(feedback.product_id, feedback.feedback_id) with open(content_file, 'w', encoding='utf-8') as f: f.write(content_text) @staticmethod def save_attachments(feedback, files): """Save attachment files Args: feedback: Feedback instance files: List of Werkzeug FileStorage objects Returns: list: List of saved filenames """ if not files: return [] attachments_dir = Feedback._get_attachments_dir(feedback.product_id, feedback.feedback_id) os.makedirs(attachments_dir, exist_ok=True) saved_files = [] for file in files: if not file or file.filename == '': continue # Sanitize filename safe_filename = get_safe_filename(file.filename) # Save file file_path = os.path.join(attachments_dir, safe_filename) file.save(file_path) saved_files.append(safe_filename) return saved_files @staticmethod def save_complete_feedback(product_id, content_text=None, files=None): """Create and save complete feedback submission Args: product_id: Product ID content_text: Feedback text content (optional) files: List of uploaded files (optional) Returns: Feedback: Created and saved feedback instance """ # Create feedback feedback = FeedbackStorageService.create_feedback(product_id, content_text, files) # Save content if content_text: FeedbackStorageService.save_content(feedback, content_text) # Save attachments if files: FeedbackStorageService.save_attachments(feedback, files) # Save metadata FeedbackStorageService.save_metadata(feedback) return feedback @staticmethod def update_feedback_status(feedback, new_status): """Update feedback status Args: feedback: Feedback instance new_status: New status value Returns: bool: True if updated successfully, False otherwise """ if new_status not in Feedback.VALID_STATUSES: return False feedback.status = new_status feedback.save_metadata() return True @staticmethod def delete_feedback(feedback): """Delete feedback and all associated files Args: feedback: Feedback instance to delete """ feedback_dir = Feedback._get_feedback_dir(feedback.product_id, feedback.feedback_id) if os.path.exists(feedback_dir): shutil.rmtree(feedback_dir) @staticmethod def load_feedback_list(product_ids=None, page=1, per_page=50, filters=None, search_query=None): """Load feedback list with filtering, searching, and pagination Args: product_ids: List of product IDs to load feedback for (None = all products) page: Page number (1-indexed) per_page: Items per page filters: Dict with filter criteria (category, status, language, date_range) search_query: Search query string Returns: dict: { 'items': List of feedback dicts, 'total': Total count, 'page': Current page, 'per_page': Items per page, 'pages': Total pages } """ data_dir = current_app.config['DATA_DIR'] products_dir = os.path.join(data_dir, 'products') all_feedback = [] # If no product_ids specified, load all products if product_ids is None: product_ids = [] if os.path.exists(products_dir): for item in os.listdir(products_dir): if os.path.isdir(os.path.join(products_dir, item)): product_ids.append(item) # Load feedback from each product for product_id in product_ids: feedback_dir = os.path.join(products_dir, product_id, 'feedback') if not os.path.exists(feedback_dir): continue for feedback_id in os.listdir(feedback_dir): feedback_path = os.path.join(feedback_dir, feedback_id) if not os.path.isdir(feedback_path): continue # Load metadata metadata_file = os.path.join(feedback_path, 'metadata.yaml') if not os.path.exists(metadata_file): continue with open(metadata_file, 'r') as f: metadata = yaml.safe_load(f) # Load content preview content_file = os.path.join(feedback_path, 'content.txt') content_preview = '' if os.path.exists(content_file): with open(content_file, 'r', encoding='utf-8') as f: content = f.read() content_preview = content[:200] # Add to list feedback_data = { 'feedback_id': feedback_id, 'product_id': product_id, 'status': metadata.get('status', 'new'), 'category': metadata.get('category', 'uncategorized'), 'original_language': metadata.get('original_language', 'unknown'), 'submitted_at': metadata.get('submitted_at'), 'has_attachments': metadata.get('has_attachments', False), 'attachment_count': metadata.get('attachment_count', 0), 'content_preview': content_preview } all_feedback.append(feedback_data) # Apply filters if filters: all_feedback = FeedbackStorageService._apply_filters(all_feedback, filters) # Apply search if search_query: all_feedback = FeedbackStorageService._apply_search(all_feedback, search_query) # Sort by timestamp (newest first) all_feedback.sort(key=lambda x: x.get('submitted_at', ''), reverse=True) # Calculate pagination total = len(all_feedback) total_pages = (total + per_page - 1) // per_page if total > 0 else 1 start_idx = (page - 1) * per_page end_idx = start_idx + per_page # Get page items items = all_feedback[start_idx:end_idx] return { 'items': items, 'total': total, 'page': page, 'per_page': per_page, 'pages': total_pages } @staticmethod def _apply_filters(feedback_list, filters): """Apply filters to feedback list Args: feedback_list: List of feedback dicts filters: Dict with filter criteria Returns: list: Filtered feedback list """ filtered = feedback_list # Filter by category if filters.get('category'): filtered = [f for f in filtered if f.get('category') == filters['category']] # Filter by status if filters.get('status'): filtered = [f for f in filtered if f.get('status') == filters['status']] # Filter by language if filters.get('language'): filtered = [f for f in filtered if f.get('original_language') == filters['language']] # Filter by date range if filters.get('date_from') or filters.get('date_to'): date_from = filters.get('date_from') date_to = filters.get('date_to') def in_date_range(feedback): submitted_at = feedback.get('submitted_at') if not submitted_at: return False if date_from and submitted_at < date_from: return False if date_to and submitted_at > date_to: return False return True filtered = [f for f in filtered if in_date_range(f)] return filtered @staticmethod def _apply_search(feedback_list, search_query): """Apply search query to feedback list Searches in content preview, category, and status Args: feedback_list: List of feedback dicts search_query: Search string Returns: list: Filtered feedback list """ if not search_query: return feedback_list query_lower = search_query.lower() def matches_search(feedback): # Search in content preview if query_lower in feedback.get('content_preview', '').lower(): return True # Search in category if query_lower in feedback.get('category', '').lower(): return True # Search in feedback ID if query_lower in feedback.get('feedback_id', '').lower(): return True return False return [f for f in feedback_list if matches_search(f)] @staticmethod def load_feedback_detail(product_id, feedback_id): """Load complete feedback details Args: product_id: Product ID feedback_id: Feedback ID Returns: dict: Complete feedback data or None if not found """ data_dir = current_app.config['DATA_DIR'] feedback_path = os.path.join(data_dir, 'products', product_id, 'feedback', feedback_id) if not os.path.exists(feedback_path): return None # Load metadata metadata_file = os.path.join(feedback_path, 'metadata.yaml') if not os.path.exists(metadata_file): return None with open(metadata_file, 'r') as f: metadata = yaml.safe_load(f) # Load content content_file = os.path.join(feedback_path, 'content.txt') content = '' if os.path.exists(content_file): with open(content_file, 'r', encoding='utf-8') as f: content = f.read() # Load analysis if exists analysis_file = os.path.join(feedback_path, 'analysis.md') analysis = '' if os.path.exists(analysis_file): with open(analysis_file, 'r', encoding='utf-8') as f: analysis = f.read() # List attachments attachments = [] attachments_dir = os.path.join(feedback_path, 'attachments') if os.path.exists(attachments_dir): attachments = os.listdir(attachments_dir) return { 'feedback_id': feedback_id, 'product_id': product_id, 'metadata': metadata, 'content': content, 'analysis': analysis, 'attachments': attachments } @staticmethod def update_feedback_status_by_id(product_id, feedback_id, new_status): """Update feedback status by IDs Args: product_id: Product ID feedback_id: Feedback ID new_status: New status value Returns: bool: True if updated successfully, False otherwise """ data_dir = current_app.config['DATA_DIR'] metadata_file = os.path.join( data_dir, 'products', product_id, 'feedback', feedback_id, 'metadata.yaml' ) if not os.path.exists(metadata_file): return False # Load metadata with open(metadata_file, 'r') as f: metadata = yaml.safe_load(f) # Update status if new_status not in Feedback.VALID_STATUSES: return False metadata['status'] = new_status # Save metadata with open(metadata_file, 'w') as f: yaml.dump(metadata, f) return True @staticmethod def get_attachment_path(product_id, feedback_id, filename): """Get path to attachment file Args: product_id: Product ID feedback_id: Feedback ID filename: Attachment filename Returns: str: Full path to attachment file or None if not found """ data_dir = current_app.config['DATA_DIR'] attachment_path = os.path.join( data_dir, 'products', product_id, 'feedback', feedback_id, 'attachments', filename ) if not os.path.exists(attachment_path): return None # Check for path traversal attachments_dir = os.path.join(data_dir, 'products', product_id, 'feedback', feedback_id, 'attachments') if not os.path.abspath(attachment_path).startswith(os.path.abspath(attachments_dir)): return None return attachment_path @staticmethod def save_analysis(product_id, feedback_id, analysis_result): """Save AI analysis results to filesystem (T082) Creates analysis.md file with formatted analysis results and updates metadata with category and language information. Args: product_id: Product ID feedback_id: Feedback ID analysis_result: AnalysisResult instance with analysis data Returns: bool: True if saved successfully, False otherwise """ data_dir = current_app.config['DATA_DIR'] feedback_dir = os.path.join(data_dir, 'products', product_id, 'feedback', feedback_id) if not os.path.exists(feedback_dir): return False # Create analysis.md file with formatted content (T083) analysis_file = os.path.join(feedback_dir, 'analysis.md') analysis_markdown = FeedbackStorageService._create_analysis_markdown(analysis_result) with open(analysis_file, 'w', encoding='utf-8') as f: f.write(analysis_markdown) # Update metadata with category and language (T089) metadata_file = os.path.join(feedback_dir, 'metadata.yaml') if os.path.exists(metadata_file): with open(metadata_file, 'r') as f: metadata = yaml.safe_load(f) # Store detected language metadata['original_language'] = analysis_result.original_language # Store category metadata['category'] = analysis_result.category with open(metadata_file, 'w') as f: yaml.dump(metadata, f) return True @staticmethod def _create_analysis_markdown(analysis_result): """Create formatted analysis markdown (T083) Args: analysis_result: AnalysisResult instance Returns: str: Formatted markdown content """ # Use the raw analysis from Claude, which is already formatted return analysis_result.raw_analysis @staticmethod def has_analysis(product_id, feedback_id): """Check if feedback has been analyzed Args: product_id: Product ID feedback_id: Feedback ID Returns: bool: True if analysis.md exists, False otherwise """ data_dir = current_app.config['DATA_DIR'] analysis_file = os.path.join( data_dir, 'products', product_id, 'feedback', feedback_id, 'analysis.md' ) return os.path.exists(analysis_file)