SAAS-1071 - initial work on podcast celery backend; tweak SoundCloud service

airtime_mvc/application/configs/constants.php

@@ -122,11 +122,6 @@ define('CELERY_FAILED_STATUS', 'FAILED');
 // Celery Services
 define('SOUNDCLOUD_SERVICE_NAME', 'soundcloud');
 
-// Celery Task Types
-define('UPLOAD', 'upload');
-define('DOWNLOAD', 'download');
-define('DELETE', 'delete');
-
 // Podcast Types
 define('STATION_PODCAST', 0);
 define('IMPORTED_PODCAST', 1);

airtime_mvc/application/services/SoundcloudService.php

@@ -4,6 +4,10 @@ require_once "ThirdPartyCeleryService.php";
 
 class Application_Service_SoundcloudService extends Application_Service_ThirdPartyCeleryService implements OAuth2 {
 
+    const UPLOAD    = 'upload';
+    const DOWNLOAD  = 'download';
+    const DELETE    = 'delete';
+
     /**
      * @var string service access token for accessing remote API
      */
@@ -26,9 +30,9 @@ class Application_Service_SoundcloudService extends Application_Service_ThirdPar
     protected static $_CELERY_EXCHANGE_NAME = 'soundcloud';
 
     protected static $_CELERY_TASKS = [
-        UPLOAD      => 'soundcloud-upload',
+        self::UPLOAD      => 'soundcloud-upload',
-        DOWNLOAD    => 'soundcloud-download',
+        self::DOWNLOAD    => 'soundcloud-download',
-        DELETE      => 'soundcloud-delete'
+        self::DELETE      => 'soundcloud-delete'
     ];
 
     /**
@@ -117,7 +121,7 @@ class Application_Service_SoundcloudService extends Application_Service_ThirdPar
             'token' => $this->_accessToken,
             'file_path' => $file->getFilePaths()[0]
         );
-        $this->_executeTask(static::$_CELERY_TASKS[UPLOAD], $data, $fileId);
+        $this->_executeTask(static::$_CELERY_TASKS[self::UPLOAD], $data, $fileId);
     }
 
     /**
@@ -128,16 +132,16 @@ class Application_Service_SoundcloudService extends Application_Service_ThirdPar
      * @param int $trackId a track identifier
      */
     public function download($trackId = null) {
-        $namespace = new Zend_Session_Namespace('csrf_namespace');
+        $CC_CONFIG = Config::getConfig();
-        $csrfToken = $namespace->authtoken;
         $data = array(
-            'callback_url' => 'http' . (empty($_SERVER['HTTPS']) ? '' : 's') . '://' . $_SERVER['HTTP_HOST'] . '/media/post?csrf_token=' . $csrfToken,
+            'callback_url'  => Application_Common_HTTPHelper::getStationUrl() . '/rest/media',
-            'token' => $this->_accessToken,
+            'api_key'       => $apiKey = $CC_CONFIG["apiKey"][0],
-            'track_id' => $trackId
+            'token'         => $this->_accessToken,
+            'track_id'      => $trackId
         );
         // FIXME
         Logging::warn("FIXME: we can't create a task reference without a valid file ID");
-        $this->_executeTask(static::$_CELERY_TASKS[DOWNLOAD], $data, null);
+        $this->_executeTask(static::$_CELERY_TASKS[self::DOWNLOAD], $data, null);
     }
 
     /**
@@ -157,7 +161,7 @@ class Application_Service_SoundcloudService extends Application_Service_ThirdPar
             'token' => $this->_accessToken,
             'track_id' => $serviceId
         );
-        $this->_executeTask(static::$_CELERY_TASKS[DELETE], $data, $fileId);
+        $this->_executeTask(static::$_CELERY_TASKS[self::DELETE], $data, $fileId);
     }
 
     /**
@@ -243,7 +247,7 @@ class Application_Service_SoundcloudService extends Application_Service_ThirdPar
         // Pass the current URL in the state parameter in order to preserve it
         // in the redirect. This allows us to create a singular script to redirect
         // back to any station the request comes from.
-        $url = urlencode('http'.(empty($_SERVER['HTTPS'])?'':'s').'://'.$_SERVER['HTTP_HOST'].'/soundcloud/redirect');
+        $url = urlencode(Application_Common_HTTPHelper::getStationUrl() . '/soundcloud/redirect');
         return $this->_client->getAuthorizeUrl(array("state" => $url, "scope" => "non-expiring"));
     }
 

python_apps/airtime-celery/airtime-celery/celeryconfig.py

@@ -29,6 +29,7 @@ CELERY_TASK_RESULT_EXPIRES = 600          # Expire task results after 10 minutes
 CELERY_RESULT_EXCHANGE = 'celeryresults'  # Default exchange - needed due to php-celery
 CELERY_QUEUES = (
     Queue('soundcloud', exchange=Exchange('soundcloud'), routing_key='soundcloud'),
+    Queue('podcast', exchange=Exchange('podcast'), routing_key='podcast'),
     Queue(exchange=Exchange('celeryresults'), auto_delete=True),
 )
 CELERY_EVENT_QUEUE_EXPIRES = 600          # RabbitMQ x-expire after 10 minutes

python_apps/airtime-celery/airtime-celery/tasks.py

@@ -36,12 +36,14 @@ def soundcloud_upload(data, token, file_path):
 
 
 @celery.task(name='soundcloud-download', acks_late=True)
-def soundcloud_download(token, callback_url, track_id=None):
+def soundcloud_download(token, callback_url, api_key, track_id=None):
     """
     This is in stasis 
 
-    :param token:       OAuth2 client access token
+    :param token:        OAuth2 client access token
-    :param track_id:    SoundCloud track identifier
+    :param callback_url: callback URL to send the downloaded file to
+    :param api_key:      API key for callback authentication
+    :param track_id:     SoundCloud track identifier
     :rtype: None
     """
     client = soundcloud.Client(access_token=token)
@@ -51,7 +53,7 @@ def soundcloud_download(token, callback_url, track_id=None):
             if track.downloadable:
                 track_file = client.get(track.download_url)
                 with track_file as f:
-                    requests.post(callback_url, data=f)
+                    requests.post(callback_url, data=f, auth=requests.auth.HTTPBasicAuth(api_key, ''))
     except Exception as e:
         logger.info('Error during file download: {0}'.format(e.message))
         logger.info(str(e))
@@ -77,3 +79,25 @@ def soundcloud_delete(token, track_id):
         logger.info('Error deleting track!')
         raise e
     return json.dumps(track.fields())
+
+
+@celery.task(name='podcast-download', acks_late=True)
+def podcast_download(download_urls, callback_url, api_key):
+    """
+    Download a given podcast episode
+
+    :param download_urls:   array of download URLs for episodes to download
+    :param callback_url:    callback URL to send the downloaded file to
+    :param api_key:         API key for callback authentication
+    :rtype: None
+    """
+    try:
+        for url in download_urls:
+            r = requests.get(url, stream=True)
+            r.raise_for_status()
+            with r as f:
+                requests.post(callback_url, data=f, auth=requests.auth.HTTPBasicAuth(api_key, ''))
+    except Exception as e:
+        logger.info('Error during file download: {0}'.format(e.message))
+        logger.info(str(e))
+        raise e