b3fbb8b345
Box Metadata Extraction Fixed: Template Name: ❌ Was: 'Ferrero-DAM-Metadata' (with dashes and caps) ✅ Now: 'ferrerodammetadata' (lowercase, no dashes) Field Names: ❌ Was: 'CreativeX Score', 'CreativeX URL' ✅ Now: 'creativexScore', 'creativexUrl' (camelCase) Test Results with File 2035459900168: ✅ Template found successfully ✅ creativexScore: 90 ✅ creativexUrl: https://www.bbc.com ✅ Metadata extraction working! A2→A3 workflow will now: 1. Download file from Box 2. Read ferrerodammetadata template 3. Extract creativexScore and creativexUrl 4. Update CREATIVEX fields in asset representation 5. Upload to DAM with CreativeX data Test: Upload a file to Box with ferrerodammetadata template applied and the A2→A3 script will extract and use those values! 🤖 Generated with Claude Code
247 lines
8.8 KiB
Python
247 lines
8.8 KiB
Python
"""
|
|
Box Client - Box.com API Integration
|
|
Handles JWT authentication and Box operations
|
|
Compatible with Python 3.6+
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from boxsdk import Client, JWTAuth
|
|
|
|
logger = logging.getLogger('BoxClient')
|
|
|
|
class BoxClient:
|
|
def __init__(self, config, root_folder_id=None):
|
|
self.config = config
|
|
|
|
# Use provided folder ID or default to A1→A2 folder
|
|
if root_folder_id:
|
|
self.root_folder_id = root_folder_id
|
|
else:
|
|
self.root_folder_id = config['box'].get('root_folder_a1_a2',
|
|
config['box'].get('root_folder_id', '348304357505'))
|
|
|
|
# Load Box config for JWT
|
|
box_config_path = config['box']['rsa_private_key_path']
|
|
|
|
try:
|
|
with open(box_config_path, 'r') as f:
|
|
box_config = json.load(f)
|
|
|
|
# Initialize JWT authentication
|
|
auth = JWTAuth.from_settings_dictionary(box_config)
|
|
self.client = Client(auth)
|
|
|
|
logger.info("Box client initialized with JWT auth")
|
|
logger.info("Using Box root folder: {}".format(self.root_folder_id))
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to initialize Box client: {}".format(str(e)))
|
|
raise
|
|
|
|
def upload_with_tracking_id(self, file_path, campaign_id, campaign_name, tracking_id):
|
|
"""
|
|
Upload file to Box with tracking ID in filename
|
|
|
|
Args:
|
|
file_path: Path to local file
|
|
campaign_id: Campaign ID
|
|
campaign_name: Campaign name
|
|
tracking_id: 6-character tracking ID
|
|
|
|
Returns:
|
|
dict with file_id, url, folder_id
|
|
"""
|
|
try:
|
|
import os
|
|
|
|
# Create or find campaign folder
|
|
folder = self._get_or_create_campaign_folder(campaign_id, campaign_name)
|
|
|
|
# Get original filename
|
|
original_filename = os.path.basename(file_path)
|
|
name_without_ext, ext = os.path.splitext(original_filename)
|
|
|
|
# Add tracking ID to filename
|
|
box_filename = "{}_{}{}".format(name_without_ext, tracking_id, ext)
|
|
|
|
# Upload file
|
|
uploaded_file = folder.upload(file_path, box_filename)
|
|
|
|
# Try to set description (API may vary by SDK version)
|
|
try:
|
|
description = "Tracking ID: {}\nOriginal: {}".format(
|
|
tracking_id, original_filename
|
|
)
|
|
# boxsdk 3.x API
|
|
uploaded_file = uploaded_file.update_info(data={'description': description})
|
|
except Exception as e:
|
|
# Description update failed - not critical, file is uploaded
|
|
logger.warning("Could not set Box file description: {}".format(str(e)))
|
|
|
|
logger.info("Uploaded to Box: {} → File ID: {}".format(box_filename, uploaded_file.id))
|
|
|
|
# Get folder ID safely
|
|
folder_id = folder.object_id if hasattr(folder, 'object_id') else str(folder)
|
|
|
|
return {
|
|
'file_id': uploaded_file.id,
|
|
'url': 'https://app.box.com/file/{}'.format(uploaded_file.id),
|
|
'folder_id': folder_id,
|
|
'box_filename': box_filename
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Box upload failed: {}".format(str(e)))
|
|
raise
|
|
|
|
def _get_or_create_campaign_folder(self, campaign_id, campaign_name):
|
|
"""Get or create campaign folder in Box"""
|
|
try:
|
|
root_folder = self.client.folder(self.root_folder_id)
|
|
|
|
# Folder name format: C000000078-Campaign_Name
|
|
folder_name = "{}-{}".format(campaign_id, campaign_name.replace(' ', '_'))
|
|
|
|
# Check if folder exists
|
|
items = root_folder.get_items()
|
|
for item in items:
|
|
if item.type == 'folder' and item.name == folder_name:
|
|
logger.info("Using existing Box folder: {}".format(folder_name))
|
|
return self.client.folder(item.id)
|
|
|
|
# Create new folder
|
|
new_folder = root_folder.create_subfolder(folder_name)
|
|
logger.info("Created new Box folder: {}".format(folder_name))
|
|
return new_folder
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get/create Box folder: {}".format(str(e)))
|
|
raise
|
|
|
|
def get_file_metadata(self, file_id, template_name='ferrerodammetadata'):
|
|
"""
|
|
Get metadata from Box file using metadata template
|
|
|
|
Args:
|
|
file_id: Box file ID
|
|
template_name: Metadata template name (default: ferrerodammetadata)
|
|
|
|
Returns:
|
|
dict with metadata fields or empty dict if not found
|
|
"""
|
|
try:
|
|
file_obj = self.client.file(file_id)
|
|
|
|
# Try to get metadata from template (scope is enterprise_ENTERPRISE_ID)
|
|
try:
|
|
metadata_dict = file_obj.metadata(scope='enterprise', template=template_name).get()
|
|
logger.info("Retrieved Box metadata from template: {}".format(template_name))
|
|
|
|
# Extract CreativeX fields (camelCase field names)
|
|
creativex_data = {}
|
|
if 'creativexScore' in metadata_dict:
|
|
creativex_data['score'] = metadata_dict['creativexScore']
|
|
logger.info("CreativeX Score: {}".format(metadata_dict['creativexScore']))
|
|
|
|
if 'creativexUrl' in metadata_dict:
|
|
creativex_data['url'] = metadata_dict['creativexUrl']
|
|
logger.info("CreativeX URL: {}".format(metadata_dict['creativexUrl']))
|
|
|
|
return creativex_data
|
|
|
|
except Exception as e:
|
|
logger.warning("No metadata template found on file ({}): {}".format(template_name, str(e)))
|
|
return {}
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get file metadata: {}".format(str(e)))
|
|
return {}
|
|
|
|
def download_file(self, file_id, output_path):
|
|
"""
|
|
Download file from Box
|
|
|
|
Args:
|
|
file_id: Box file ID
|
|
output_path: Path to save file
|
|
|
|
Returns:
|
|
Path to downloaded file
|
|
"""
|
|
try:
|
|
import os
|
|
|
|
file_obj = self.client.file(file_id)
|
|
file_info = file_obj.get()
|
|
|
|
# Ensure output directory exists
|
|
os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else '.', exist_ok=True)
|
|
|
|
# Download file
|
|
with open(output_path, 'wb') as f:
|
|
file_obj.download_to(f)
|
|
|
|
file_size = os.path.getsize(output_path)
|
|
logger.info("Downloaded from Box: {} ({} bytes)".format(file_info.name, file_size))
|
|
|
|
return output_path
|
|
|
|
except Exception as e:
|
|
logger.error("Box download failed: {}".format(str(e)))
|
|
raise
|
|
|
|
def list_folder_files(self, folder_id):
|
|
"""
|
|
List all files in a Box folder
|
|
|
|
Args:
|
|
folder_id: Box folder ID
|
|
|
|
Returns:
|
|
List of file dictionaries
|
|
"""
|
|
try:
|
|
folder = self.client.folder(folder_id)
|
|
items = folder.get_items()
|
|
|
|
files = []
|
|
for item in items:
|
|
if item.type == 'file':
|
|
# Get full item details (boxsdk 3.x requires explicit .get())
|
|
try:
|
|
file_info = item.get()
|
|
files.append({
|
|
'id': file_info.id,
|
|
'name': file_info.name,
|
|
'size': getattr(file_info, 'size', 0),
|
|
'modified_at': getattr(file_info, 'modified_at', None),
|
|
'url': 'https://app.box.com/file/{}'.format(file_info.id)
|
|
})
|
|
except Exception as e:
|
|
# Fallback to basic info
|
|
logger.warning("Could not get full file info for {}: {}".format(item.name, str(e)))
|
|
files.append({
|
|
'id': item.id,
|
|
'name': item.name,
|
|
'size': 0,
|
|
'modified_at': None,
|
|
'url': 'https://app.box.com/file/{}'.format(item.id)
|
|
})
|
|
|
|
logger.info("Found {} files in Box folder {}".format(len(files), folder_id))
|
|
return files
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to list Box folder: {}".format(str(e)))
|
|
raise
|
|
|
|
def test_connection(self):
|
|
"""Test Box connection"""
|
|
try:
|
|
user = self.client.user().get()
|
|
logger.info("Box connection OK - User: {}".format(user.name))
|
|
return True
|
|
except Exception as e:
|
|
logger.error("Box connection failed: {}".format(str(e)))
|
|
return False
|