ferrero-opentext/Python-Version/scripts/b1_to_b2_download.py
nickviljoen 6d6213024a Fix: Merge A+B live campaigns into single CSV for OMG
OMG's Box automation treats each new live_campaigns_*.csv as a full-list
replacement, so the per-series global CSV introduced 2026-04-30 stomped
the local list whenever a B1→B2 ran. Collapse to one combined CSV
(A-series + B-series) emitted by every handler.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-04 17:36:43 +02:00

615 lines
25 KiB
Python
Executable file

#!/usr/bin/env python3
"""
B1→B2 Master Asset Downloader
Polls DAM for campaigns with status B1, downloads master assets, uploads to Box
Updates status to B2 only when ALL assets successfully processed
Supports OAuth2 (default) and mTLS (--auth-pfx) authentication
Compatible with Python 3.6+
"""
import sys
import os
import time
import csv
import logging
import argparse
from datetime import datetime, timezone
# Add shared library to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from shared.config_loader import load_config, load_field_mappings
from shared.dam_client import DAMClient
from shared.box_client import BoxClient
from shared.database import Database
from shared.notifier import Notifier
from shared.common import sanitize_box_item_name
# Setup logging with rotation
from logging.handlers import RotatingFileHandler
# Create logs directory if it doesn't exist
os.makedirs('logs', exist_ok=True)
os.makedirs('logs/backup', exist_ok=True)
# Configure logging with rotation
# Keep 1 week of active logs (7 days * 10MB = 70MB)
# Backup rotates keep 4 weeks (28 backups * 10MB = 280MB total)
log_handler = RotatingFileHandler(
'logs/b1_to_b2.log',
maxBytes=10*1024*1024, # 10MB per file
backupCount=28 # Keep 28 rotated files (approximately 1 month)
)
log_handler.setLevel(logging.INFO)
log_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logging.basicConfig(
level=logging.INFO,
handlers=[log_handler, console_handler]
)
logger = logging.getLogger('B1toB2')
def _walk_metadata_elements(elements):
"""Recursively yield every element in nested metadata_element_list arrays.
Categories and tables both nest fields underneath them, so a flat walk
misses anything below the top level."""
for e in elements or []:
if not isinstance(e, dict):
continue
yield e
nested = e.get('metadata_element_list')
if isinstance(nested, list):
for sub in _walk_metadata_elements(nested):
yield sub
def extract_creativex_from_dam_metadata(asset_metadata):
"""
Extract CreativeX score and URL from DAM asset metadata if present.
Walks the metadata_element_list recursively because the score field
(FERRERO.TAB.FIELD.CREATIVEX) is nested at depth 2 under its parent
table FERRERO.TABULAR.FIELD.CREATIVEX, not at the top level.
"""
try:
top = (asset_metadata or {}).get('metadata', {}).get('metadata_element_list', [])
cx = {'score': None, 'url': None}
for element in _walk_metadata_elements(top):
element_id = element.get('id')
if element_id == 'FERRERO.TAB.FIELD.CREATIVEX':
values = element.get('values', [])
if values:
value_obj = values[0].get('value', {})
if isinstance(value_obj, dict):
field_value = value_obj.get('field_value', {})
if isinstance(field_value, dict):
score = field_value.get('value')
if score:
cx['score'] = str(score)
elif element_id == 'FERRERO.FIELD.CREATIVEX LINK':
value_obj = element.get('value', {})
if isinstance(value_obj, dict):
nested_value = value_obj.get('value', {})
if isinstance(nested_value, dict):
url = nested_value.get('value')
if url:
cx['url'] = url
return cx
except Exception as e:
logger.warning("Failed to extract CreativeX from metadata: {}".format(str(e)))
return {'score': None, 'url': None}
def generate_and_upload_csv(db, box, config):
"""
Generate the combined live-campaigns CSV (A-series + B-series) and upload
to Box. OMG's automation treats each new file as a full replacement of
its live list, so we always emit the complete list under one filename.
"""
try:
logger.info("Generating live campaigns CSV...")
campaigns = db.get_all_live_campaigns()
if not campaigns:
logger.warning("No live campaigns found to report")
logger.info("Found {} live campaigns".format(len(campaigns)))
timestamp = datetime.now(timezone.utc).strftime('%Y-%m-%d_%H%M%S_UTC')
csv_filename = 'live_campaigns_{}.csv'.format(timestamp)
csv_path = os.path.join('temp', csv_filename)
os.makedirs('temp', exist_ok=True)
with open(csv_path, 'w', newline='') as csvfile:
fieldnames = ['code', 'description']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for camp in campaigns:
writer.writerow({
'code': "{}-{}".format(camp['campaign_number'], camp['campaign_name']),
'description': camp['campaign_name']
})
logger.info("Generated CSV: {}".format(csv_path))
folder_id = config['box'].get('live_campaigns_folder_id')
if not folder_id:
logger.error("Box live_campaigns_folder_id not configured")
return False
upload_result = box.upload_file(
file_path=csv_path,
folder_id=folder_id,
target_filename=csv_filename
)
logger.info("Uploaded CSV to Box: {} (File ID: {})".format(
csv_filename, upload_result['file_id']
))
os.remove(csv_path)
return True
except Exception as e:
logger.error("Failed to generate/upload CSV: {}".format(str(e)))
return False
def format_cx_score_for_display(raw_score):
"""DAM stores the CreativeX score as a tabular cell that concatenates
platform and score with a caret, e.g. 'DV360^100'. Convert to
'100 (DV360)' for human-readable email output. Returns the raw value
unchanged if it doesn't match the expected pattern."""
if not raw_score:
return raw_score
if '^' in raw_score:
platform, _, score = raw_score.partition('^')
platform = platform.strip()
score = score.strip()
if platform and score:
return "{} ({})".format(score, platform)
return raw_score
def process_campaign(campaign, dam, box, db, notifier, config):
"""
Process single campaign - download all master assets
Returns:
dict with success, processed_count, failed_count
"""
campaign_id = campaign['asset_id']
campaign_name = campaign['campaign_name']
campaign_number = campaign.get('campaign_id', 'N/A')
logger.info("=" * 60)
logger.info("Processing campaign: {} ({})".format(campaign_name, campaign_number))
logger.info("=" * 60)
try:
# Get master assets (B1 campaigns use Final Assets folder, is_global=True)
master_assets = dam.get_master_assets(campaign_id, is_global=True)
total_assets = len(master_assets)
logger.info("Found {} master assets".format(total_assets))
if total_assets == 0:
logger.warning("No master assets found in Final Assets folder")
# Send email notification about empty campaign
notifier.send_email(
template_name='b1_to_b2_no_assets',
recipients=config['notifications']['recipients']['errors'],
data={
'campaign_name': campaign_name,
'campaign_id': campaign_id,
'campaign_number': campaign_number
}
)
logger.info("✓ Email sent: No assets found notification")
return {'success': False, 'processed': 0, 'failed': 0}
# Track results
processed_assets = []
failed_assets = []
# Get Final Assets folder for upload directory
final_folder_id = dam.find_final_assets_folder(campaign_id)
if not final_folder_id:
logger.error("Final Assets folder not found")
return {'success': False, 'processed': 0, 'failed': total_assets}
# Process each asset
skipped_count = 0
for asset in master_assets:
asset_id = asset['asset_id']
asset_name = asset.get('name', 'unknown')
folder_path = asset.get('folder_path', '') # Get subfolder path from recursive search
try:
if folder_path:
logger.info("Processing: {} (from subfolder: {})".format(asset_name, folder_path))
else:
logger.info("Processing: {}".format(asset_name))
# SAFEGUARD: Check if it's a folder (should be handled by dam_client, but double check)
asset_type = asset.get('asset_type', {})
type_name = asset_type.get('name', '') if isinstance(asset_type, dict) else str(asset_type)
if 'folder' in type_name.lower():
logger.warning("Skipping item identified as folder: {} (Type: {})".format(asset_name, type_name))
continue
# SAFEGUARD 2: Check for missing extension (likely a container/folder)
_, ext = os.path.splitext(asset_name)
if not ext or len(ext) < 2: # No extension or just a dot
logger.warning("Skipping item with no extension (likely folder/container): {}".format(asset_name))
continue
# SKIP CHECK: If this asset was already processed (exists in DB), skip re-downloading
existing_tracking_id = db.find_global_master_by_opentext_id(asset_id)
if existing_tracking_id:
existing_asset = db.get_master_asset(existing_tracking_id)
if existing_asset and existing_asset.get('box_url'):
skipped_count += 1
logger.info("⏭ Already processed: {}{} (skipping)".format(asset_name, existing_tracking_id))
cx = extract_creativex_from_dam_metadata(existing_asset.get('full_metadata') or {})
if cx['score'] or cx['url']:
db.store_creativex_score(
filename=asset_name,
creativex_id='',
creativex_url=cx['url'] or '',
quality_score=cx['score'] or '',
box_file_id=existing_asset.get('box_file_id', ''),
full_extraction_data={'master_metadata': True, 'source': 'b1_to_b2', 'data': cx},
tracking_id=existing_tracking_id,
status='b1-master-cx-score'
)
processed_assets.append({
'asset_id': asset_id,
'asset_name': asset_name,
'tracking_id': existing_tracking_id,
'box_file_id': existing_asset.get('box_file_id', ''),
'box_url': existing_asset.get('box_url', ''),
'creativex_score': format_cx_score_for_display(cx['score']),
'creativex_url': cx['url'],
'is_existing': True
})
continue
# 1. Download from DAM
file_path = dam.download_asset(
asset_id,
output_dir='temp/downloads/{}'.format(campaign_id)
)
# 2. Generate tracking ID (master files always start with 'M')
tracking_id = db.generate_unique_tracking_id(is_master=True)
# 3. Upload to Box (B1 uses MASTERS_CampaignNumber format, preserve folder structure)
# If campaign_number exists, use it; otherwise use MASTERS prefix only
box_campaign_id = "MASTERS_{}".format(campaign_number) if campaign_number and campaign_number != 'N/A' else "MASTERS"
box_result = box.upload_with_tracking_id(
file_path=file_path,
campaign_id=box_campaign_id, # MASTERS_C000000068
campaign_name=sanitize_box_item_name(campaign_name).replace(' ', '_'), # NUTELLA_PLANT_BASED_LAUNCH
tracking_id=tracking_id,
subfolder_path=folder_path # Preserve DAM folder structure
)
# Result: MASTERS_C000000068-NUTELLA_PLANT_BASED_LAUNCH/[subfolder_path]
# 4. Store in database with FULL metadata
db_result = db.store_master_asset(
tracking_id=tracking_id,
opentext_id=asset_id,
asset_data=asset,
box_file_id=box_result['file_id'],
box_url=box_result['url'],
upload_folder_id=final_folder_id
)
if db_result['success']:
cx = extract_creativex_from_dam_metadata(asset)
if cx['score']:
logger.info("CreativeX score on master {}: {}".format(asset_name, cx['score']))
if cx['score'] or cx['url']:
db.store_creativex_score(
filename=asset_name,
creativex_id='',
creativex_url=cx['url'] or '',
quality_score=cx['score'] or '',
box_file_id=box_result['file_id'],
full_extraction_data={'master_metadata': True, 'source': 'b1_to_b2', 'data': cx},
tracking_id=tracking_id,
status='b1-master-cx-score'
)
processed_assets.append({
'asset_id': asset_id,
'asset_name': asset_name,
'tracking_id': tracking_id,
'box_file_id': box_result['file_id'],
'box_url': box_result['url'],
'creativex_score': format_cx_score_for_display(cx['score']),
'creativex_url': cx['url'],
'is_existing': False
})
logger.info("✓ Success: {}{}".format(asset_name, tracking_id))
else:
raise Exception("Database storage failed")
# Clean up temp file
os.remove(file_path)
except Exception as e:
logger.error("✗ Failed: {} - {}".format(asset_name, str(e)))
failed_assets.append({
'asset_id': asset_id,
'asset_name': asset_name,
'error': str(e)
})
# CHECK: All assets processed successfully?
all_done = len(processed_assets) == total_assets
# Split new vs existing for reporting
new_assets = [a for a in processed_assets if not a.get('is_existing')]
existing_assets = [a for a in processed_assets if a.get('is_existing')]
logger.info("")
logger.info("Campaign {} Results:".format(campaign_id))
logger.info(" Total: {}".format(total_assets))
logger.info(" Successful: {}".format(len(processed_assets)))
logger.info(" Skipped (already done): {}".format(skipped_count))
logger.info(" New this run: {}".format(len(new_assets)))
logger.info(" Failed: {}".format(len(failed_assets)))
logger.info(" All Done: {}".format("YES" if all_done else "NO"))
logger.info("")
if all_done:
# ALL assets processed - update status
logger.info("All assets processed - Updating status B1 → B2")
status_result = dam.update_campaign_status(campaign_id, "B2")
if status_result['success']:
logger.info("✓ Status updated successfully")
# Record campaign status in database — marks it as LIVE so the
# global CSV picks it up. B4 closure (or A4 with prior B-status)
# later flips this to NO.
logger.info("Recording campaign status in database (Live: YES, status B2)...")
db.record_campaign_status(
campaign_id=campaign_id,
campaign_number=campaign_number,
campaign_name=campaign_name,
live_campaign='YES',
status='B2',
webhook_sent=False # B-series workflow doesn't send a webhook
)
# Regenerate and upload the combined live campaigns CSV to Box.
# Box automation forwards it to OMG as a full-list replacement.
logger.info("Generating and uploading live campaigns CSV...")
csv_success = generate_and_upload_csv(db, box, config)
if csv_success:
logger.info("✓ CSV report uploaded successfully")
else:
logger.error("✗ CSV report generation/upload failed")
# NOTE: B1→B2 workflow does NOT send webhook (only email notification)
# Webhook is only used for A1→A2 workflow
# Generate CSV Report
import csv
try:
csv_filename = "B1_Campaign_{}_Assets.csv".format(campaign_number)
csv_path = os.path.join("temp", csv_filename)
if not os.path.exists("temp"):
os.makedirs("temp")
with open(csv_path, 'w', newline='') as csvfile:
fieldnames = ['Filename', 'Tracking ID', 'Campaign Number', 'Status']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for asset in processed_assets:
writer.writerow({
'Filename': asset['asset_name'],
'Tracking ID': asset['tracking_id'],
'Campaign Number': campaign_number,
'Status': 'Existing' if asset.get('is_existing') else 'New'
})
logger.info("Generated CSV report: {}".format(csv_path))
attachments = [csv_path]
except Exception as csv_error:
logger.error("Failed to generate CSV report: {}".format(str(csv_error)))
attachments = None
# Send success email with asset details
notifier.send_email(
template_name='b1_to_b2_complete',
recipients=config['notifications']['recipients']['success'],
data={
'campaign_name': campaign_name,
'campaign_id': campaign_id,
'campaign_number': campaign_number,
'asset_count': len(processed_assets),
'new_asset_count': len(new_assets),
'existing_asset_count': len(existing_assets),
'processed_assets': processed_assets,
'new_assets': new_assets,
'existing_assets': existing_assets
},
attachments=attachments
)
# Clean up CSV
if attachments and os.path.exists(csv_path):
try:
os.remove(csv_path)
logger.info("Cleaned up CSV report")
except Exception as e:
logger.warning("Failed to remove temp CSV: {}".format(str(e)))
return {'success': True, 'processed': len(processed_assets), 'failed': 0}
else:
logger.error("✗ Status update failed: {}".format(status_result.get('error')))
# Don't send success notification if status update failed
return {'success': False, 'processed': len(processed_assets), 'failed': 0}
else:
# NOT all done - some failed
logger.warning("Campaign incomplete - NOT updating status (remains A1)")
# Send partial completion email with asset details
notifier.send_email(
template_name='b1_to_b2_partial',
recipients=config['notifications']['recipients']['errors'],
data={
'campaign_name': campaign_name,
'campaign_id': campaign_id,
'campaign_number': campaign_number,
'total_assets': total_assets,
'successful': len(processed_assets),
'failed': len(failed_assets),
'processed_assets': processed_assets,
'failed_assets': failed_assets
}
)
return {'success': False, 'processed': len(processed_assets), 'failed': len(failed_assets)}
except Exception as e:
logger.error("Campaign processing failed: {}".format(str(e)))
# Return 0 for failed count since we don't know how many assets there were
return {'success': False, 'processed': 0, 'failed': 0}
def main():
"""Main polling loop"""
# Parse command-line arguments
parser = argparse.ArgumentParser(description='Ferrero B1→B2 Global Master Asset Downloader')
parser.add_argument('--auth-pfx', action='store_true',
help='Use mTLS certificate authentication (Legacy APIM)')
parser.add_argument('--auth-pfx-v2', action='store_true',
help='Use mTLS V2 (Hybrid) authentication')
args = parser.parse_args()
logger.info("=" * 60)
logger.info("Ferrero B1→B2 Master Asset Downloader Starting")
# Determine auth mode
auth_mode = 'oauth'
if args.auth_pfx_v2:
auth_mode = 'mtls_v2'
logger.info("Authentication: mTLS V2 (Hybrid)")
elif args.auth_pfx:
auth_mode = 'mtls'
logger.info("Authentication: mTLS Certificate (Legacy)")
else:
logger.info("Authentication: OAuth2 (default)")
logger.info("=" * 60)
# Load configuration
config = load_config('config/config.yaml')
# Initialize clients (pass mTLS flag to DAM)
dam = DAMClient(config, auth_mode=auth_mode)
# Use B1→B2 specific Box folder (349261192115)
box = BoxClient(config, root_folder_id=config['box'].get('root_folder_b1_b2'))
db = Database(config)
notifier = Notifier(config)
# Test connections
logger.info("Testing connections...")
if not dam.test_connection():
logger.error("DAM connection failed - exiting")
sys.exit(1)
if not box.test_connection():
logger.error("Box connection failed - exiting")
sys.exit(1)
if not db.test_connection():
logger.error("Database connection failed - exiting")
sys.exit(1)
logger.info("All connections OK")
logger.info("")
# SINGLE RUN MODE - Process ONE campaign and exit
# Cron will run this script every 5 minutes, processing one campaign at a time
try:
logger.info("Searching for B1 Global campaigns...")
# Search for campaigns with status B1 (Global comm campaigns)
campaigns = dam.search_campaigns(status="B1", campaign_type="Global comm")
if not campaigns:
logger.info("No B1 campaigns found - exiting")
db.close()
sys.exit(0)
# Process ONLY THE FIRST campaign
campaign = campaigns[0]
logger.info("Found {} A1 campaigns - processing first one only".format(len(campaigns)))
logger.info("")
result = process_campaign(campaign, dam, box, db, notifier, config)
if result['success']:
logger.info("")
logger.info("=" * 60)
logger.info("✓ Campaign completed successfully")
logger.info(" Processed: {} assets".format(result['processed']))
logger.info(" Status updated: B1 → B2")
logger.info("=" * 60)
db.close()
sys.exit(0)
else:
logger.warning("")
logger.warning("=" * 60)
logger.warning("✗ Campaign incomplete or failed")
logger.warning(" Processed: {} assets".format(result['processed']))
logger.warning(" Failed: {} assets".format(result['failed']))
logger.warning(" Status NOT updated (remains A1)")
logger.warning("=" * 60)
db.close()
sys.exit(1)
except Exception as e:
logger.critical("Script error: {}".format(str(e)))
# Send critical error notification
notifier.send_email(
template_name='upload_failed',
recipients=config['notifications']['recipients']['critical'],
data={
'filename': 'B1→B2 Script',
'tracking_id': 'N/A',
'error': str(e)
}
)
db.close()
sys.exit(1)
if __name__ == '__main__':
main()