import os from multiprocessing.pool import ThreadPool import random import requests import time from dotenv import load_dotenv import boto3 from owslogger import logger load_dotenv('.env') VENDOR_ID = int(os.environ.get('VENDOR_ID')) USER_ID = os.environ.get('ORCHARD_USER_ID') LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO') projects_to_create = int(os.environ.get('PROJECTS_NUM', 5)) products_per_project = int(os.environ.get('PRODUCTS_PER_NUM', 10)) script_logger = logger.setup( 'dev', 'content-review-test-gen-products', LOG_LEVEL, 'content-review-test', '1.0') request_headers = { 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': str(VENDOR_ID), 'Correlation-Id': 'content-review-gen', 'Orchard-User-Id': USER_ID, 'Content-Type': 'application/json' } OWS_PROJECT_URL = 'https://qa-ows-project-manager.theorchard.io' OWS_PRODUCT_DIG_URL = 'https://qa-ows-product-digital.theorchard.io' OWS_PRODUCT_URL = 'https://qa-ows-product.theorchard.io' OWS_ASSETS_URL = 'https://qa-ows-assets.theorchard.io' OWS_TRACK_URL = 'https://qa-ows-track.theorchard.io' def create_project(): code = str(int(random.random() * 1000000)) data = { 'project_name': 'content-review test project {}'.format(code), 'project_code': code, 'artist_id': 842651 } post_url = '{}/project'.format(OWS_PROJECT_URL) response = requests.post( post_url, headers=request_headers, json=data ) project_id = response.json().get('project_id') script_logger.debug('project {} created'.format(project_id)) return project_id def get_upload_token(asset_type): post_url = '{}/upload-token'.format(OWS_ASSETS_URL) response = requests.post( post_url, headers=request_headers, json={ 'asset_type': asset_type } ) return response.json() def upload_art(product_id, upc): token = get_upload_token('image') upload_creds = token['credentials'] s3_client = boto3.client( 's3', aws_access_key_id=upload_creds.get('aws_access_key_id'), aws_secret_access_key=upload_creds.get('aws_secret_access_key'), aws_session_token=upload_creds.get('token') ) file = s3_client.upload_file( 'original.jpg', token.get('bucket'), '{}.jpg'.format(token.get('filename')), ExtraArgs={ 'ContentType': 'image/jpeg', 'Metadata': { 'asset_type': 'jpg', 'product_id': str(product_id), 'upc': str(upc), 'track_unique_id': '0', 'original_filename': 'temp_asset.jpg' } } ) return file def put_subgenre_and_artist(product_id): put_url = '{}/product/audio/{}'.format(OWS_PRODUCT_DIG_URL, product_id) data = { "subgenre_id": 513, "product_artists": [{"role": "primary_artist", "name": "Immense", "artist_info_id": "842651"}] } response = requests.put( put_url, headers=request_headers, json=data ) return response.json() def create_product(project_id): post_url = '{}/product/audio'.format(OWS_PRODUCT_DIG_URL) data = { "artist_id": 842651, "c_line": "2020 test imprint", "distribution_format_id": 1, "format": "Full Length", "genre_id": 1, "imprint": "Cool for Cats", "manufacturer_upc": None, "meta_language": "ENG", "not_for_distribution": "N", "p_line": None, "preorder_date": None, "previewable": None, "product_code": None, "product_name": "CR Test Product {}".format( int(random.random() * 1000)), "product_type_id": 1, "release_status": "label_processing", "special_instructions": None, "subaccount_id": None, "version": "Standard", "release_date": "2023-02-01", "sale_start_date": "2023-02-01", "account_type": "vendor", "account_id": VENDOR_ID, "project_id": project_id } response = requests.post( post_url, headers=request_headers, json=data ) formatted_response = response.json() product_id = formatted_response['product_id'] upc = formatted_response['upc'] upload_art(product_id, upc) updated_product = put_subgenre_and_artist(product_id) return updated_product def generate_isrc(track_id, track): create_isrc_url = '{}/track/{}'.format( OWS_TRACK_URL, track_id) updated_track = requests.put( create_isrc_url, headers=request_headers, json=track, params={ 'generate_isrc': 1 } ) return updated_track.json() def add_performers(track_id): performer_url = '{}/track/{}/performers'.format( OWS_TRACK_URL, track_id) performer_data = { 'performers': [ { 'birth_name': 'test performer 21', 'performer_role_id': 11, 'type': 'primary' } ] } response = requests.put( performer_url, headers=request_headers, json=performer_data ) return response.json() def add_publishing_obligation(product_id, track_id): pub_obl_url = '{}/product/{}/tracks/publishing-obligation'.format( OWS_TRACK_URL, product_id) pub_obl_data = [ { 'us_publishing_obligation': 'ControlledByYourLabel', 'third_party_publisher': 'N', 'tuid': track_id } ] response = requests.put( pub_obl_url, headers=request_headers, json=pub_obl_data ) return response.json() def upload_track(product_id, upc, track_id): token = get_upload_token('audio') upload_creds = token['credentials'] s3_client = boto3.client( 's3', aws_access_key_id=upload_creds.get('aws_access_key_id'), aws_secret_access_key=upload_creds.get('aws_secret_access_key'), aws_session_token=upload_creds.get('token') ) file = s3_client.upload_file( 'a.wav', token.get('bucket'), '{}.wav'.format(token.get('filename')), ExtraArgs={ 'ContentType': 'audio/wav', 'Metadata': { 'asset_type': 'wav', 'product_id': str(product_id), 'track_unique_id': str(track_id), 'upc': str(upc), 'original_filename': 'temp_asset.wav' } } ) return file def create_track(product_id, upc): create_track_url = '{}/product/{}/tracks'.format( OWS_TRACK_URL, product_id) track_body = { 'tracks': [ { 'track_name': 'test_track', 'p_info': '2020 test label', 'version': 'Standard', 'original_rights_holder_country_id': 3, 'meta_language_code': 'ENG', 'explicit': 'N', 'ownership_rights': 'original_owner', 'recording_country_id': 3, 'writers': [ { 'type': 'writer', 'name': 'test_songwriter' } ], 'artists': [ { 'name': 'test_primary_artist', 'type': 'performer' }, { 'name': 'Immense', 'type': 'performer' } ] } ] } track_response = requests.put( create_track_url, headers=request_headers, json=track_body ) track_id = track_response.json()['items'][0]['tuid'] generate_isrc(track_id, track_body['tracks'][0]) add_performers(track_id) add_publishing_obligation(product_id, track_id) upload_track(product_id, upc, track_id) return track_id def validate_product(product_id, count): get_url = '{}/product/audio/{}/validate'.format( OWS_PRODUCT_DIG_URL, product_id) response = requests.get( get_url, headers=request_headers) validation = response.json()['validations'] artwork_valid = validation['artwork']['valid'] track_valid = validation['tracks']['valid'] if (artwork_valid and track_valid): return True if count > 25: return False time.sleep(5) return validate_product(product_id, count + 1) def validate_submit_product(product_id): is_valid = validate_product(product_id, 0) if is_valid: put_url = '{}/product/audio/{}/status'.format( OWS_PRODUCT_DIG_URL, product_id) response = requests.put( put_url,headers=request_headers, json={}) submitted = response.json() return submitted['release_status'] script_logger.error( 'product {} is invalid'.format(product_id)) return None def create_product_for_review(project_id): product = create_product(project_id) product_id = product.get('product_id') upc = product.get('upc') track_id = create_track(product_id, upc) status = validate_submit_product(product_id) script_logger.debug( 'product {} with track {} created with status: {}'.format( product_id, track_id, status)) if status == 'transfer_to_content': return product return None def copy_product(product_id): post_url = '{}/product/{}/copy/digital'.format( OWS_PRODUCT_URL, product_id) data = { "format": "Full Length", "product_name": "CR Test Product {}".format( int(random.random() * 1000)), } response = requests.post( post_url, headers=request_headers, json=data) return response.json() def create_products_for_project(project_id): product_ids = [] invalid_product_ids = [] for _ in range(products_per_project): try: if len(product_ids) == 0: product = create_product(project_id) product_id = product.get('product_id') upc = product.get('upc') track_id = create_track(product_id, upc) script_logger.debug( 'product {} with track {} created'.format( product_id, track_id)) else: copy_product_id = product_ids[0] product = copy_product(copy_product_id) product_id = product.get('product_id') script_logger.debug( 'product {} copied from {}'.format( product_id, copy_product_id)) status = validate_submit_product(product_id) if status == 'transfer_to_content': product_ids.append(product_id) else: invalid_product_ids.append(product_id) except Exception as e: script_logger.error('error creating product {}'.format(e)) return product_ids, invalid_product_ids def create_project_products(i): script_logger.debug('started generating project {}'.format(i)) try: project_id = create_project() product_ids, invalid_product_ids = create_products_for_project(project_id) script_logger.debug('finished generating project {}'.format(i)) message = "PROJECT {}:\n valid products: {}\n invalid products: {}".format( project_id, product_ids, invalid_product_ids ) script_logger.info(message) except Exception as e: script_logger.error('failed generating project {}: {}'.format(i, e)) def main(): with ThreadPool(projects_to_create) as pool: pool.map(create_project_products, range(projects_to_create)) if __name__ == '__main__': main()