creative-x-ferrero/creativex-automation/utils/file_handler.py
DJP b20119b383 Add complete mapping system and automated Box.com monitoring service
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>
2026-01-29 09:51:16 -05:00

170 lines
4 KiB
Python

"""
File handling utilities for CreativeX API Integration
"""
import mimetypes
import os
from pathlib import Path
from typing import Optional
class FileHandler:
"""Utilities for file operations and validation"""
# Supported file formats
SUPPORTED_VIDEO_FORMATS = ['.mp4', '.mov']
SUPPORTED_IMAGE_FORMATS = ['.jpg', '.jpeg', '.png']
@staticmethod
def get_file_extension(file_path: str) -> str:
"""
Get file extension in lowercase including dot
Args:
file_path: Path to file
Returns:
str: Lowercase extension including dot (e.g., '.mp4')
"""
return Path(file_path).suffix.lower()
@staticmethod
def get_file_size_mb(file_path: str) -> float:
"""
Get file size in megabytes
Args:
file_path: Path to file
Returns:
float: File size in MB
"""
size_bytes = os.path.getsize(file_path)
return size_bytes / (1024 * 1024)
@staticmethod
def is_video_file(file_path: str) -> bool:
"""
Check if file is a supported video format
Args:
file_path: Path to file
Returns:
bool: True if video file
"""
ext = FileHandler.get_file_extension(file_path)
return ext in FileHandler.SUPPORTED_VIDEO_FORMATS
@staticmethod
def is_image_file(file_path: str) -> bool:
"""
Check if file is a supported image format
Args:
file_path: Path to file
Returns:
bool: True if image file
"""
ext = FileHandler.get_file_extension(file_path)
return ext in FileHandler.SUPPORTED_IMAGE_FORMATS
@staticmethod
def is_supported_format(file_path: str) -> bool:
"""
Check if file format is supported
Args:
file_path: Path to file
Returns:
bool: True if supported format
"""
return (FileHandler.is_video_file(file_path) or
FileHandler.is_image_file(file_path))
@staticmethod
def get_content_type(file_path: str) -> Optional[str]:
"""
Get MIME content type for file
Args:
file_path: Path to file
Returns:
str: MIME type (e.g., 'video/mp4', 'image/jpeg')
None: If type cannot be determined
"""
# Try to guess from extension
content_type, _ = mimetypes.guess_type(file_path)
if content_type:
return content_type
# Fallback based on extension
ext = FileHandler.get_file_extension(file_path)
content_type_map = {
'.mp4': 'video/mp4',
'.mov': 'video/quicktime',
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.png': 'image/png',
}
return content_type_map.get(ext, 'application/octet-stream')
@staticmethod
def file_exists(file_path: str) -> bool:
"""
Check if file exists
Args:
file_path: Path to file
Returns:
bool: True if file exists
"""
return Path(file_path).exists()
@staticmethod
def is_readable(file_path: str) -> bool:
"""
Check if file is readable
Args:
file_path: Path to file
Returns:
bool: True if file is readable
"""
try:
return os.access(file_path, os.R_OK)
except Exception:
return False
@staticmethod
def get_filename(file_path: str) -> str:
"""
Get filename from path (with extension)
Args:
file_path: Path to file
Returns:
str: Filename with extension
"""
return Path(file_path).name
@staticmethod
def get_absolute_path(file_path: str) -> str:
"""
Get absolute path of file
Args:
file_path: Path to file
Returns:
str: Absolute path
"""
return str(Path(file_path).absolute())