"""Config for SoundCloud Statistics Workflow.""" import os import subprocess from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization from fangare import catchers FLOW_NAME = 'soundcloud_statistics' SOUNDCLOUD_LINK = 'https://soundcloud.com/' TRACKS_LINK = '/tracks' ENV = os.environ.get('Environment') if os.environ.get('Environment') in ['prod']: try: GIT_HASH_OF_LAST_COMMIT = ( subprocess.check_output( 'git rev-parse HEAD', shell=True).decode().strip()) except subprocess.CalledProcessError: GIT_HASH_OF_LAST_COMMIT = 'can_not_get_commit_hash' else: GIT_HASH_OF_LAST_COMMIT = 'soundcloud_statistics_dev_commit' # get Snowflake private_key # SNOWFLAKE_PRIVATE_KEY_PATH = os.getenv( # 'SNOWFLAKE_PRIVATE_KEY_PATH', # '{}/.ssh/snowflake/rsa_key.p8'.format(os.getenv('HOME'))) # SNOWFLAKE_KEY_PASSPHRASE = os.getenv('SNOWFLAKE_KEY_PASSPHRASE') # if SNOWFLAKE_KEY_PASSPHRASE: # # Snowflake key # with open(SNOWFLAKE_PRIVATE_KEY_PATH, 'rb') as key: # p_key = serialization.load_pem_private_key( # key.read(), # password=SNOWFLAKE_KEY_PASSPHRASE.encode(), # backend=default_backend() # ) # # PRIVATE_KEY = p_key.private_bytes( # encoding=serialization.Encoding.DER, # format=serialization.PrivateFormat.PKCS8, # encryption_algorithm=serialization.NoEncryption()) # else: PRIVATE_KEY = None SF_CONFIG = { 'role': os.environ.get('SNOWFLAKE_ROLE'), 'warehouse': os.environ.get('SNOWFLAKE_WAREHOUSE'), 'db': os.environ.get('SNOWFLAKE_DATABASE'), 'schema': os.environ.get('SNOWFLAKE_SCHEMA'), 'user': os.environ.get('SNOWFLAKE_USER'), 'password': catchers.get_secret('SNOWFLAKE_PASSWORD', 'youtube-statistics', ENV), 'account': os.environ.get('SNOWFLAKE_ACCOUNT'), 'private_key': PRIVATE_KEY } aws = { 'aws_key_id': os.environ.get('AWS_ACCESS_KEY_ID'), 'aws_secret_key': os.environ.get('AWS_SECRET_ACCESS_KEY') } # a DB table postfix for specifying country, team etc. (e.g. '_UK') postfix = os.environ.get('SNOWFLAKE_DATABASE_POSTFIX', '') snowflake_table_names = { 'soundcloud_artist_to_track': 'artist_to_track' + postfix, 'soundcloud_statistics': 'soundcloud_statistics' + postfix, 'staging_raw': 'staging_raw_soundcloud_statistics' + postfix } soundcloud = { 'username': os.environ.get('SOUNDCLOUD_USERNAME'), 'password': os.environ.get('SOUNDCLOUD_PASSWORD'), } xpaths = { 'followers': "//a[contains(@href, 'followers') " "and contains(@title, 'follow')]", 'followings': "//a[contains(@href, 'following')]/div", 'tracks_count': "//a[contains(@href, 'tracks')]/div", 'tracks': "//a[contains(@class, '" "soundTitle__title')]/span[@class='']", 'plays': "//li[contains(@title, ' plays')]/span[contains(" "@class, 'sc-ministats')]/span[@class='sc-visuallyhidden']", 'title': "//div[contains(@class, 'fullListenHero')]" "//div[contains(@class, 'soundTitle__usernameTitleContainer')]" "/span/span", 'likes': "//button[@aria-label='Like']", 'reposts': "//button[@aria-label='Repost']", } css_paths = { 'release': ".relativeTime.sc-text-secondary.sc-text-captions", 'followers': ".infoStats__value.sc-font-light", }