b4e004c822
MAJOR MILESTONE: Complete Python automation system created! Components Implemented: ✅ Box Client (box_client.py) - JWT authentication via boxsdk - Upload with tracking ID suffix - Download files - Campaign folder creation - Connection testing ✅ Database Client (database.py) - PostgreSQL connection pooling - generate_unique_tracking_id() - store_master_asset() with full_metadata JSONB - get_master_asset(tracking_id) - check_campaign_upload_complete() - ALL-DONE CHECK! - store_derivative_asset() - Connection testing ✅ Filename Parser (filename_parser.py) - V2 naming convention parser (ported from PHP) - parse_filename() - 10 components - strip_upload_components() - Remove Job# and Tracking ID - Strict validation with detailed errors ✅ Metadata Extractor MVP (metadata_extractor_mvp.py) - Extract 28 MVP fields from master - Update fields from V2 filename (Description, Language, State) - Add missing fields with defaults - Build asset representation for upload ✅ Notifier (notifier.py) - Mailgun email integration - Outgoing webhook sender - Email templates (success, error, partial, critical) - Configurable recipients Main Scripts: ✅ A1→A2 Download (a1_to_a2_download.py) - Poll DAM every 5 minutes for A1 campaigns - Download all master assets - Upload to Box with tracking IDs - Store in DB with full metadata - ALL-DONE CHECK before status update - Update A1→A2 only if all assets successful - Send webhook with campaign ID/number - Email notifications ✅ A2→A3 Upload (a2_to_a3_upload.py) - Flask webhook receiver for Box uploads - Signature validation - Async task queue processing - Parse V2 filenames - Load master metadata - Extract MVP fields - Upload to DAM - ALL-DONE CHECK for campaign - Update A2→A3 when all assets uploaded - Send webhook notifications ✅ Test Connection Script (test_connection.py) - Verify DAM, Box, Database connectivity - Quick health check ✅ README.md - Complete setup guide - Usage instructions - Configuration examples - Troubleshooting Key Features: - Python 3.6+ compatible (server requirement) - Virtual environment isolated - Configuration-driven (YAML files) - Easy field updates (no code changes) - Environment switching (staging/production) - Comprehensive error handling - Email + webhook notifications - Retry logic - All-done checks before status updates - Campaign webhook notifications Ready for testing locally with Python 3.10! 🤖 Generated with Claude Code Co-Authored-By: Claude <noreply@anthropic.com>
76 lines
2 KiB
Python
Executable file
76 lines
2 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
Test Connections - Verify DAM, Box, and Database connectivity
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
|
|
|
from shared.config_loader import load_config
|
|
from shared.dam_client import DAMClient
|
|
from shared.box_client import BoxClient
|
|
from shared.database import Database
|
|
|
|
def main():
|
|
print("=" * 60)
|
|
print("Testing Ferrero Automation Connections")
|
|
print("=" * 60)
|
|
print("")
|
|
|
|
# Load config
|
|
try:
|
|
config = load_config('config/config.yaml')
|
|
print("✓ Configuration loaded")
|
|
except Exception as e:
|
|
print("✗ Configuration failed: {}".format(e))
|
|
sys.exit(1)
|
|
|
|
# Test DAM
|
|
print("")
|
|
print("Testing DAM connection...")
|
|
try:
|
|
dam = DAMClient(config)
|
|
if dam.test_connection():
|
|
print("✓ DAM connection OK")
|
|
print(" URL: {}".format(config['dam']['base_url']))
|
|
else:
|
|
print("✗ DAM connection failed")
|
|
except Exception as e:
|
|
print("✗ DAM error: {}".format(e))
|
|
|
|
# Test Box
|
|
print("")
|
|
print("Testing Box connection...")
|
|
try:
|
|
box = BoxClient(config)
|
|
if box.test_connection():
|
|
print("✓ Box connection OK")
|
|
print(" Enterprise ID: {}".format(config['box']['enterprise_id']))
|
|
else:
|
|
print("✗ Box connection failed")
|
|
except Exception as e:
|
|
print("✗ Box error: {}".format(e))
|
|
|
|
# Test Database
|
|
print("")
|
|
print("Testing Database connection...")
|
|
try:
|
|
db = Database(config)
|
|
if db.test_connection():
|
|
print("✓ Database connection OK")
|
|
print(" Host: {}:{}".format(config['database']['host'], config['database']['port']))
|
|
else:
|
|
print("✗ Database connection failed")
|
|
db.close()
|
|
except Exception as e:
|
|
print("✗ Database error: {}".format(e))
|
|
|
|
print("")
|
|
print("=" * 60)
|
|
print("Testing complete!")
|
|
print("=" * 60)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|