Oliver/ferrero-opentext
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ferrero-opentext/Python-Version/scripts/shared/box_client.py
DJP 80d5757bbb Add Box metadata extraction for CreativeX fields in A2→A3 workflow 
Major Feature: Box Metadata Integration

box_client.py:
✅ Added get_file_metadata() method
✅ Reads 'Ferrero-DAM-Metadata' template from Box files
✅ Extracts 'CreativeX Score' and 'CreativeX URL' fields
✅ Returns dict with score and url

a2_to_a3_upload_polling.py:
✅ Calls box.get_file_metadata() before download
✅ Logs Box metadata retrieved
✅ Passes box_metadata to build_mvp_asset_representation()

metadata_extractor_mvp.py:
✅ Added box_metadata parameter to build_mvp_asset_representation()
✅ Added _update_creativex_fields() method
✅ Updates FERRERO.FIELD.CREATIVEX LINK with URL from Box
✅ Logs CreativeX Score (tabular field - needs special handling)

Flow:
1. File uploaded to Box by agency
2. Agency adds metadata using Ferrero-DAM-Metadata template
3. Script reads CreativeX Score and URL from Box metadata
4. Updates MVP fields with Box metadata values
5. Uploads to DAM with CreativeX data

Field Mapping:
- Box: 'CreativeX URL' → DAM: FERRERO.FIELD.CREATIVEX LINK
- Box: 'CreativeX Score' → DAM: FERRERO.TAB.FIELD.CREATIVEX (logged, needs structure)

Next: Test with file that has Box metadata template applied

🤖 Generated with Claude Code

Co-Authored-By: Claude <noreply@anthropic.com>
2025-11-03 14:26:00 -05:00

247 lines
8.8 KiB
Python
Raw Blame History

"""
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='Ferrero-DAM-Metadata'):
"""
Get metadata from Box file using metadata template
Args:
file_id: Box file ID
template_name: Metadata template name (default: Ferrero-DAM-Metadata)
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
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
creativex_data = {}
if 'CreativeX Score' in metadata_dict:
creativex_data['score'] = metadata_dict['CreativeX Score']
logger.info("CreativeX Score: {}".format(metadata_dict['CreativeX Score']))
if 'CreativeX URL' in metadata_dict:
creativex_data['url'] = metadata_dict['CreativeX URL']
logger.info("CreativeX URL: {}".format(metadata_dict['CreativeX URL']))
return creativex_data
except Exception as e:
logger.warning("No metadata template found on file: {}".format(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
Reference in a new issue View git blame Copy permalink