Major Features: - Complete Ferrero ↔ CreativeX mapping system with 93 brands - Automated Box.com folder monitoring service - Email notifications with score breakdowns - Database integration for result storage Mapping System (v2.0.0): - mappings.json: 93 brand mappings, 44+ channel mappings - core/mapping_resolver.py: Translates Ferrero codes to CreativeX format - scripts/validate_mappings.py: Validation tool for brand/channel support - scripts/generate_brand_mappings.py: Auto-mapping tool - scripts/download_reports.py: Scorecard PDF download tool - Updated scripts/upload.py: Integrated mapping validation - Updated scripts/check_status.py: Added detailed score display with guidelines Documentation: - Updated README.md: Complete user guide with mapping system - Updated STATUS.md: Production-ready status with test results - MAPPINGS_GUIDE.md: Complete mapping documentation - MAPPING_IMPLEMENTATION.md: Implementation summary - BRAND_MAPPINGS_REVIEW.md: Brand mapping validation guide - PRODUCTION_BRANDS_SUMMARY.md: Production brand catalog - PRODUCTION_MAPPING_COMPLETE.md: Mapping completion summary Automation Service (New): - creativex-automation/: Complete automated Box monitoring service - Monitors Box Ferrero-In folder (363284027140) for new files - Automatically uploads to CreativeX - Polls for completion (30 min intervals) - Extracts scores and stores in PostgreSQL creativex_scores table - Sends formatted emails to file uploader + daveporter@oliver.agency - Moves processed files to Processed subfolder Service Components: - automation/box_monitor.py: Box folder monitoring with uploader detection - automation/upload_processor.py: CreativeX upload integration - automation/status_poller.py: CreativeX status polling - automation/result_handler.py: Score extraction and email sending - automation/orchestrator.py: Service coordination - automation/processing_queue.py: JSON-based processing queue - service.py: Main service entry point - config.py: Service configuration loader - requirements.txt: All dependencies - deployment/systemd/: Systemd service unit file - Updated shared/notifier.py: Added creativex_upload_complete and creativex_upload_failed templates Testing: - Supports --dry-run mode for configuration testing - Supports --scan-once mode for Box folder testing - Manual run mode for development/testing - Comprehensive logging with rotation (10MB, 28 backups) Database Integration: - Uses existing creativex_scores table (no migrations needed) - Compatible with existing Ferrero-Opentext workflows - Stores full CreativeX API responses in JSONB Email Templates: - Matches Ferrero-Opentext styling (#9c27b0 purple for CreativeX) - Includes score, tier, guidelines breakdown, scorecard URL - Recipients: Box uploader + CC to daveporter@oliver.agency Deployment: - Runs locally for dev/testing - Systemd service for production - Auto-restart on failure - Complete documentation in creativex-automation/README.md Co-Authored-By: Claude Sonnet 4.5 (1M context) <noreply@anthropic.com>
129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""
|
|
Pre-upload validation for files and metadata
|
|
"""
|
|
|
|
from typing import Tuple, List
|
|
from core.data_loader import DataLoader
|
|
from utils.file_handler import FileHandler
|
|
|
|
|
|
class UploadValidator:
|
|
"""Pre-upload validation to ensure data quality"""
|
|
|
|
def __init__(self, data_loader: DataLoader, max_file_size_mb: int = 500):
|
|
"""
|
|
Initialize validator
|
|
|
|
Args:
|
|
data_loader: DataLoader instance for validation
|
|
max_file_size_mb: Maximum allowed file size in MB
|
|
"""
|
|
self.data_loader = data_loader
|
|
self.max_file_size_mb = max_file_size_mb
|
|
|
|
def validate_file(self, file_path: str) -> Tuple[bool, List[str]]:
|
|
"""
|
|
Validate file exists, is readable, and has supported format
|
|
|
|
Args:
|
|
file_path: Path to file
|
|
|
|
Returns:
|
|
tuple: (is_valid: bool, errors: List[str])
|
|
"""
|
|
errors = []
|
|
|
|
# Check file exists
|
|
if not FileHandler.file_exists(file_path):
|
|
errors.append(f"File not found: {file_path}")
|
|
return False, errors
|
|
|
|
# Check readable
|
|
if not FileHandler.is_readable(file_path):
|
|
errors.append(f"File not readable (permission denied): {file_path}")
|
|
return False, errors
|
|
|
|
# Check format
|
|
if not FileHandler.is_supported_format(file_path):
|
|
ext = FileHandler.get_file_extension(file_path)
|
|
supported = (FileHandler.SUPPORTED_VIDEO_FORMATS +
|
|
FileHandler.SUPPORTED_IMAGE_FORMATS)
|
|
errors.append(
|
|
f"Unsupported file format: {ext}. "
|
|
f"Supported formats: {', '.join(supported)}"
|
|
)
|
|
return False, errors
|
|
|
|
# Check file size
|
|
size_mb = FileHandler.get_file_size_mb(file_path)
|
|
if size_mb > self.max_file_size_mb:
|
|
errors.append(
|
|
f"File too large: {size_mb:.2f}MB "
|
|
f"(max: {self.max_file_size_mb}MB)"
|
|
)
|
|
return False, errors
|
|
|
|
return True, []
|
|
|
|
def validate_parsed_metadata(self, metadata: dict) -> Tuple[bool, List[str]]:
|
|
"""
|
|
Validate all required fields are present and valid
|
|
|
|
Args:
|
|
metadata: Parsed metadata dict
|
|
|
|
Returns:
|
|
tuple: (is_valid: bool, errors: List[str])
|
|
"""
|
|
errors = []
|
|
|
|
# Check required fields
|
|
required_fields = [
|
|
'brand_code', 'brand_name',
|
|
'subject',
|
|
'asset_type',
|
|
'aspect_ratio',
|
|
'country_code', 'country_name',
|
|
'language_code',
|
|
'channel'
|
|
]
|
|
|
|
for field in required_fields:
|
|
if not metadata.get(field):
|
|
errors.append(f"Missing required field: {field}")
|
|
|
|
# Validate brand
|
|
if metadata.get('brand_code') and not self.data_loader.validate_brand_code(metadata['brand_code']):
|
|
errors.append(f"Invalid brand code: {metadata['brand_code']}")
|
|
|
|
# Validate country
|
|
if metadata.get('country_code') and not self.data_loader.validate_country_code(metadata['country_code']):
|
|
errors.append(f"Invalid country code: {metadata['country_code']}")
|
|
|
|
# Validate language
|
|
if metadata.get('language_code') and not self.data_loader.validate_language_code(metadata['language_code']):
|
|
errors.append(f"Invalid language code: {metadata['language_code']}")
|
|
|
|
# Validate asset type
|
|
if metadata.get('asset_type') and not self.data_loader.validate_asset_type(metadata['asset_type']):
|
|
errors.append(f"Invalid asset type: {metadata['asset_type']}")
|
|
|
|
# Validate social media (if present)
|
|
if metadata.get('social_media'):
|
|
if not self.data_loader.validate_social_code(metadata['social_media']):
|
|
errors.append(f"Invalid social media code: {metadata['social_media']}")
|
|
|
|
is_valid = len(errors) == 0
|
|
return is_valid, errors
|
|
|
|
def get_content_type(self, file_path: str) -> str:
|
|
"""
|
|
Get MIME content type for file
|
|
|
|
Args:
|
|
file_path: Path to file
|
|
|
|
Returns:
|
|
str: MIME type
|
|
"""
|
|
return FileHandler.get_content_type(file_path)
|