gurixandClaude
adbfd23c26
Add complete dashboard functionality for product owners and administrators to view, filter, search, and manage feedback submissions following Test-First Discipline. Tests (T093-T105): - Add 12 contract tests for dashboard routes (authentication, listing, filtering, search, detail view, status updates, attachment downloads, access control) - Add 2 integration tests for complete dashboard workflow and access control enforcement - All tests written first and verified to fail before implementation Services (T111-T118): - Enhance FeedbackStorageService with load_feedback_list() for pagination, filtering, searching, and sorting - Add load_feedback_detail() to load complete feedback with attachments and analysis - Add update_feedback_status_by_id() for status management - Add get_attachment_path() with path traversal prevention Routes (T119-T134): - Implement GET /dashboard with filters, search, and pagination (50 items/page) - Implement GET /feedback/<id> detail view with role-based access control - Implement POST /feedback/<id>/status for status updates - Implement GET /feedback/<id>/attachment/<filename> for secure file downloads - Add access control helpers (administrators see all products, owners see only assigned) Templates (T135-T136): - Create dashboard/list.html with filter form, search, and pagination - Create dashboard/detail.html with status update form and attachment links - Create error_403.html for access denied - Create error_404.html for not found Integration & Bug Fixes: - Update auth routes to remove /auth prefix and redirect to dashboard after login - Update Feedback.VALID_STATUSES to include dashboard statuses (in_progress, resolved, closed) - Register error handlers for 403 and 404 in app factory - Fix test fixtures to use correct users.yaml format and User.hash_password() Test Results: 39 passed, 1 skipped (all Phase 5 tests passing) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
474 lines
14 KiB
Python
474 lines
14 KiB
Python
"""Feedback storage service"""
|
|
import os
|
|
import shutil
|
|
import yaml
|
|
from datetime import datetime
|
|
from flask import current_app
|
|
from app.models.feedback import Feedback
|
|
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
|