import argparse import fnmatch import hashlib from queue import Queue import os import requests import shutil import threading import time import logging from db.art_relations import conn as ar_conn q = Queue() lock = threading.Lock() logging.basicConfig(filename='errors.log',level=logging.DEBUG) def download_track(upc, unique_track_id, track_id, cd, stream_base_url, physical_location, target_dir, backup_stream_base_url, backup_stream_key_cookie, backup_stream_remember_me_cookie, vendor_id, isrc): """download a single track from streaming service and save to directory Args: upc (str): release uid unique_track_id: track.id primary key track_id: track.track_id cd: track.cd stream_base_url (str): streaming svc such as http://dc.streams.devorch.com/stream physical_location (str): location to pass to streaming service such as "2" target_dir (str): directory to download files to backup_stream_base_url (str): backup streaming service if other one fails backup_stream_key_cookie (str): key cookie for backup service backup_stream_remember_me_cookie (str): remember_me cookie for backup service vendor_id (int): id to use in user param for backup service isrc (str): isrc to use in backup service request """ upc_cd_track_id = '{}_{}_{}'.format(upc, cd, track_id) token = hashlib.md5(upc_cd_track_id.encode('utf-8')).hexdigest() url = '{}/{}/{}/{}/{}'.format(stream_base_url, upc, physical_location, unique_track_id, token) s = requests.Session() r = s.get(url, stream=True) if r.status_code != 200: # streaming service failed, try backup payload = {'user': vendor_id, 'upc': upc, 'isrc': isrc} cookies = {'remember_me': backup_stream_remember_me_cookie, 'key': backup_stream_key_cookie} try: r = s.get(backup_stream_base_url, cookies=cookies, params=payload, stream=True) # write the file path = '{}/{}.mp3'.format(target_dir, unique_track_id) with open(path, 'wb') as f: r.raw.decode_content = True shutil.copyfileobj(r.raw, f) with lock: # lock to avoid output interleaving print('unique_track_id {} downloaded'.format(unique_track_id)) except requests.exceptions.RequestException as e: print('could not download: ' + r.url) logging.info('ERROR: COULD NOT LOAD FILE: ' + r.url) def download_tracks(vendor_id, db_limit, stream_base_url, physical_location, target_dir, backup_stream_base_url, backup_stream_key_cookie, backup_stream_remember_me_cookie, prev_downloaded_track_uids): """get track info from db and push that info onto the queue to be processed by thread workers Args: vendor_id (int): artist_info.vendor_id to select db_limit (int): limit arg for query stream_base_url (str): streaming svc such as http://dc.streams.devorch.com/stream physical_location (str): location to pass to streaming service such as "2" target_dir (str): directory to download files to backup_stream_base_url (str): backup streaming service if other one fails backup_stream_key_cookie (str): key cookie for backup service backup_stream_remember_me_cookie (str): remember_me cookie for backup service prev_downloaded_track_uids (list of int): list of track_uids to skip """ query = """SELECT r.upc, t.track_id, t.id AS unique_track_id, t.cd, t.isrc FROM releases r JOIN track t ON r.release_id = t.release_id JOIN artist_info ai ON r.artist_id = ai.artist_id WHERE r.release_status = 'in_content' AND ai.vendor_id = {} """.format(vendor_id) if len(prev_downloaded_track_uids) > 0: query += " and t.id not in (" for track_uid in prev_downloaded_track_uids: query += "{},".format(track_uid) query = query[:-1] # remove trailing comma query += ") " if db_limit is not None: query += " limit {}".format(db_limit) cursor = ar_conn.cursor() cursor.execute(query) for (upc, track_id, unique_track_id, cd, isrc) in cursor: params = {'upc': upc, 'track_id': track_id, 'unique_track_id': unique_track_id, 'cd': cd, 'stream_base_url': stream_base_url, 'physical_location': physical_location, 'target_dir': target_dir, 'backup_stream_base_url': backup_stream_base_url, 'backup_stream_key_cookie': backup_stream_key_cookie, 'backup_stream_remember_me_cookie': backup_stream_remember_me_cookie, 'vendor_id': vendor_id, 'isrc': isrc} q.put(params) def worker(): """Pull dictionaries from queue and call download_track()""" while True: item = q.get() download_track(**item) q.task_done() def create_threads(thread_count): """ create and start a bunch of threads Args: thread_count (int): number of threads to create """ for _ in range(thread_count): t = threading.Thread(target=worker) t.daemon = True t.start() def get_cl_args(): """Process command-line args Args: Argparse Namespace object containing args and/or defaults """ help_str = """ Download files from streaming service and save in specified directory. Examples: python download_track.py -h python download_track.py --vendor_id=16915 --db_limit=5 --backup_stream_key_cookie=db30a84 --backup_stream_remember_me_cookie=myuser python download_track.py --vendor_id=16915 --stream_base_url=http://dc.streams.devorch.com/stream --backup_stream_key_cookie=db30a84 --backup_stream_remember_me_cookie=myuser python download_track.py --vendor_id=16915 --physical_location=2 --backup_stream_key_cookie=db30a84 --backup_stream_remember_me_cookie=myuser python download_track.py --vendor_id=16915 --thread_count=5 --backup_stream_key_cookie=db30a84 --backup_stream_remember_me_cookie=myuser python download_track.py --vendor_id=16915 --target_dir=/tmp --backup_stream_key_cookie=db30a84 --backup_stream_remember_me_cookie=myuser python download_track.py --vendor_id=16915 --db_limit=5 --physical_location=2\ --stream_base_url=http://dc.streams.devorch.com/stream --thread_count=5 --target_dir=/tmp --backup_stream_key_cookie=db30a84 --backup_stream_remember_me_cookie=myuser To get backup_stream_key_cookie and backup_stream_remember_me_cookie values, log in to oa.theorchard.com with "Remember my user name and password" selected. In Chrome Developer Tools, get the key and remember_me values in Resources -> Cookies """ parser = argparse.ArgumentParser(help_str) parser.add_argument('--vendor_id', type=int, required=True, help='artist_info.vendor_id such as 16915', dest='vendor_id') parser.add_argument('--stream_base_url', type=str, required=False, help='streaming svc such as http://dc.streams.devorch.com/stream', default='http://dc.streams.devorch.com/stream', dest='stream_base_url'), parser.add_argument('--backup_stream_base_url', type=str, required=False, help='backup streaming service to call if stream_base_url fails', default='https://oa.theorchard.com/cont_mgmt/download_track_audio.php', dest='backup_stream_base_url') parser.add_argument('--backup_stream_remember_me_cookie', type=str, required=True, help='remember_me cookie for backup stream', dest='backup_stream_remember_me_cookie') parser.add_argument('--backup_stream_key_cookie', type=str, required=True, help='key cookie for backup stream', dest='backup_stream_key_cookie') parser.add_argument('--physical_location', type=str, required=False, help='location to pass to streaming service such as "2"', default='2', dest='physical_location') parser.add_argument('--thread_count', type=int, required=False, help='number of threads to use for file download', default=10, dest='thread_count') parser.add_argument('--db_limit', type=int, required=False, help='limit arg to database query. no limit if not provided.', dest='db_limit') parser.add_argument('--target_dir', type=str, required=False, help='directory to download files to', default='../audio', dest='target_dir') return parser.parse_args() def get_prev_downloaded_track_uids(target_dir): """ get a list of previously downloaded track uids to allow script to be re-run and pick up where it left off. Assumption: file names in target_dir are named .mp3 Skip any files that are less than 100k, as these might be errors or partial downloads. Args: target_dir: directory to check for prev downloaded files """ track_uids = [] for file in os.listdir(target_dir): if fnmatch.fnmatch(file, "*.mp3"): file_size_bytes = os.path.getsize("{}/{}".format(target_dir, file)) a = file.split(".") # ignore files of <= 100k (will re-download) if file_size_bytes > 100000: track_uid = int(a[0]) track_uids.append(track_uid) return track_uids def main(): start = time.time() args = get_cl_args() prev_downloaded_track_uids=get_prev_downloaded_track_uids(args.target_dir) print("args: {}".format(args)) create_threads(args.thread_count) download_tracks(args.vendor_id, args.db_limit, args.stream_base_url, args.physical_location, args.target_dir, args.backup_stream_base_url, args.backup_stream_key_cookie, args.backup_stream_remember_me_cookie, prev_downloaded_track_uids) q.join() # wait for workers to complete elapsed_time = time.time() - start print('elapsed time: {} seconds'.format(round(elapsed_time, 3))) main()