""" 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