/**
  * @param string $entryId
  * @param FileSync $object
  * @param int $fileSyncId
  * @param string $sourceFileUrl
  * @return BatchJob
  */
 public static function addFileSyncImportJob($entryId, FileSync $fileSync, $sourceFileUrl, BatchJob $parentJob = null, $fileSize = null)
 {
     $partnerId = $fileSync->getPartnerId();
     $fileSyncId = $fileSync->getId();
     $dc = $fileSync->getDc();
     KalturaLog::log(__METHOD__ . " entryId[{$entryId}], partnerId[{$partnerId}], fileSyncId[{$fileSyncId}], sourceFileUrl[{$sourceFileUrl}]");
     $fileSyncImportData = new kFileSyncImportJobData();
     $fileSyncImportData->setSourceUrl($sourceFileUrl);
     $fileSyncImportData->setFilesyncId($fileSyncId);
     $fileSyncImportData->setFileSize($fileSize);
     // tmpFilePath and destFilePath will be set later during get exlusive call on the target data center
     $batchJob = null;
     if ($parentJob) {
         $batchJob = $parentJob->createChild(BatchJobType::FILESYNC_IMPORT, null, true, $dc);
     } else {
         $batchJob = new BatchJob();
         $batchJob->setDc($dc);
         $batchJob->setEntryId($entryId);
         $batchJob->setPartnerId($partnerId);
     }
     $batchJob->setObjectId($fileSyncId);
     $batchJob->setObjectType(BatchJobObjectType::FILE_SYNC);
     //In case file sync is of type data and holds flavor asset than we need to check if its the source asset that is being synced and raise it sync priority
     if ($fileSync->getObjectType() == FileSyncObjectType::FLAVOR_ASSET && $fileSync->getObjectSubType() == entry::FILE_SYNC_ENTRY_SUB_TYPE_DATA && $fileSync->getFileType() == FileSync::FILE_SYNC_FILE_TYPE_FILE) {
         $assetdb = assetPeer::retrieveById($fileSync->getObjectId());
         if ($assetdb) {
             $isSourceAsset = $assetdb->getIsOriginal();
             if ($isSourceAsset) {
                 $fileSyncImportData->setIsSourceAsset(true);
             }
         }
     }
     KalturaLog::log("Creating Filesync Import job, with file sync id: {$fileSyncId} size: {$fileSize}");
     return kJobsManager::addJob($batchJob, $fileSyncImportData, BatchJobType::FILESYNC_IMPORT);
 }

 /**
  * @param BaseObject $object
  */
 public function attachCreatedObject(BaseObject $object)
 {
     $dropFolderFile = DropFolderFilePeer::retrieveByPK($this->getDropFolderFileId());
     $dropFolder = DropFolderPeer::retrieveByPK($dropFolderFile->getDropFolderId());
     $entryId = $asset = null;
     // create import job for remote drop folder files
     if ($dropFolder instanceof RemoteDropFolder) {
         // get params
         if ($object instanceof asset) {
             $entryId = $object->getEntryId();
             $asset = $object;
         } else {
             if ($object instanceof entry) {
                 $entryId = $object->getId();
                 $asset = null;
             } else {
                 return;
             }
         }
         $importUrl = $dropFolder->getFolderUrl();
         $importUrl .= '/' . $dropFolderFile->getFileName();
         $jobData = $dropFolder->getImportJobData();
         $jobData->setDropFolderFileId($this->getDropFolderFileId());
         // add job
         kJobsManager::addImportJob(null, $entryId, $dropFolderFile->getPartnerId(), $importUrl, $asset, $dropFolder->getFileTransferMgrType(), $jobData);
         // set file status to DOWNLOADING
         $dropFolderFile->setStatus(DropFolderFileStatus::DOWNLOADING);
         $dropFolderFile->save();
     }
 }

 /**
  * Add bulk upload job
  * @param DropFolder $folder
  * @param DropFolderFile $leadDropFolderFile
  * @throws Exception
  */
 private function addXMLBulkUploadJob(DropFolder $folder, DropFolderFile $leadDropFolderFile)
 {
     /* @var $leadDropFolderFile FeedDropFolderFile */
     KalturaLog::info('Adding BulkUpload job');
     try {
         $coreBulkUploadType = BulkUploadXmlPlugin::getBulkUploadTypeCoreValue(BulkUploadXmlType::XML);
         $objectId = $leadDropFolderFile->getId();
         $objectType = DropFolderXmlBulkUploadPlugin::getBatchJobObjectTypeCoreValue(DropFolderBatchJobObjectType::DROP_FOLDER_FILE);
         $partner = PartnerPeer::retrieveByPK($folder->getPartnerId());
         $data = KalturaPluginManager::loadObject('kBulkUploadJobData', $coreBulkUploadType);
         /* @var $data kBulkUploadJobData */
         $data->setUploadedBy(kDropFolderXmlEventsConsumer::UPLOADED_BY);
         KalturaLog::info("Feed XML path: " . $leadDropFolderFile->getFeedXmlPath());
         $data->setFilePath($leadDropFolderFile->getFeedXmlPath());
         $data->setFileName(basename($data->getFilePath()) . '.xml');
         $objectData = new kBulkUploadEntryData();
         KalturaLog::info('Conversion profile id: ' . $folder->getConversionProfileId());
         $objectData->setConversionProfileId($folder->getConversionProfileId());
         $data->setObjectData($objectData);
         $job = kJobsManager::addBulkUploadJob($partner, $data, $coreBulkUploadType, $objectId, $objectType);
         $this->setFileToProcessing($leadDropFolderFile);
         return $job;
     } catch (Exception $e) {
         KalturaLog::err("Error adding BulkUpload job -" . $e->getMessage());
         throw new Exception(DropFolderXmlBulkUploadPlugin::ERROR_ADDING_BULK_UPLOAD_MESSAGE, DropFolderXmlBulkUploadPlugin::getErrorCodeCoreValue(DropFolderXmlBulkUploadErrorCode::ERROR_ADDING_BULK_UPLOAD));
     }
 }

 /**
  * @param CaptionAsset $captionAsset
  * @param BatchJob $parentJob
  * @throws kCoreException FILE_NOT_FOUND
  * @return BatchJob
  */
 public function addParseCaptionAssetJob(CaptionAsset $captionAsset, BatchJob $parentJob = null)
 {
     $syncKey = $captionAsset->getSyncKey(asset::FILE_SYNC_ASSET_SUB_TYPE_ASSET);
     $fileSync = kFileSyncUtils::getReadyInternalFileSyncForKey($syncKey);
     if (!$fileSync) {
         if (!PermissionPeer::isValidForPartner(CaptionPermissionName::IMPORT_REMOTE_CAPTION_FOR_INDEXING, $captionAsset->getPartnerId())) {
             throw new kCoreException("File sync not found: {$syncKey}", kCoreException::FILE_NOT_FOUND);
         }
         $fileSync = kFileSyncUtils::getReadyExternalFileSyncForKey($syncKey);
         if (!$fileSync) {
             throw new kCoreException("File sync not found: {$syncKey}", kCoreException::FILE_NOT_FOUND);
         }
         $fullPath = myContentStorage::getFSUploadsPath() . '/' . $captionAsset->getId() . '.tmp';
         if (!kFile::downloadUrlToFile($fileSync->getExternalUrl($captionAsset->getEntryId()), $fullPath)) {
             throw new kCoreException("File sync not found: {$syncKey}", kCoreException::FILE_NOT_FOUND);
         }
         kFileSyncUtils::moveFromFile($fullPath, $syncKey, true, false, true);
     }
     $jobData = new kParseCaptionAssetJobData();
     $jobData->setCaptionAssetId($captionAsset->getId());
     $batchJob = null;
     if ($parentJob) {
         $batchJob = $parentJob->createChild();
     } else {
         $batchJob = new BatchJob();
         $batchJob->setEntryId($captionAsset->getEntryId());
         $batchJob->setPartnerId($captionAsset->getPartnerId());
     }
     return kJobsManager::addJob($batchJob, $jobData, CaptionSearchPlugin::getBatchJobTypeCoreValue(CaptionSearchBatchJobType::PARSE_CAPTION_ASSET));
 }

 public function sendEmailAlert()
 {
     $this->kuser = kuserPeer::retrieveByPK($this->getKuserId());
     if ($this->kuser) {
         kJobsManager::addMailJob(null, 0, $this->kuser->getPartnerId(), $this->getAlertType(), kMailJobData::MAIL_PRIORITY_NORMAL, kconf::get("batch_notification_sender_email"), kconf::get("batch_notification_sender_name"), $this->kuser->getEmail(), $this->getBodyParamsArray(), $this->getSubjectParamsArray());
     }
 }

 /**
  * @action notify
  * @disableTags TAG_WIDGET_SESSION,TAG_ENTITLEMENT_ENTRY,TAG_ENTITLEMENT_CATEGORY
  * @param int $id integration job id
  */
 public function notifyAction($id)
 {
     $coreType = IntegrationPlugin::getBatchJobTypeCoreValue(IntegrationBatchJobType::INTEGRATION);
     $batchJob = BatchJobPeer::retrieveByPK($id);
     $invalidJobId = false;
     $invalidKs = false;
     if (!self::validateKs($batchJob)) {
         $invalidKs = true;
         KalturaLog::err("ks not valid for notifying job [{$id}]");
     } elseif (!$batchJob) {
         $invalidJobId = true;
         KalturaLog::err("Job [{$id}] not found");
     } elseif ($batchJob->getJobType() != $coreType) {
         $invalidJobId = true;
         KalturaLog::err("Job [{$id}] wrong type [" . $batchJob->getJobType() . "] expected [" . $coreType . "]");
     } elseif ($batchJob->getStatus() != KalturaBatchJobStatus::ALMOST_DONE) {
         $invalidJobId = true;
         KalturaLog::err("Job [{$id}] wrong status [" . $batchJob->getStatus() . "] expected [" . KalturaBatchJobStatus::ALMOST_DONE . "]");
     } elseif ($batchJob->getPartnerId() != kCurrentContext::getCurrentPartnerId()) {
         $invalidKs = true;
         KalturaLog::err("Job [{$id}] of wrong partner [" . $batchJob->getPartnerId() . "] expected [" . kCurrentContext::getCurrentPartnerId() . "]");
     }
     if ($invalidJobId) {
         throw new KalturaAPIException(KalturaErrors::INVALID_BATCHJOB_ID, $id);
     }
     if ($invalidKs) {
         throw new KalturaAPIException(KalturaIntegrationErrors::INTEGRATION_NOTIFY_FAILED);
     }
     kJobsManager::updateBatchJob($batchJob, KalturaBatchJobStatus::FINISHED);
 }

 /**
  * Creates new download job for multiple entry ids (comma separated), an email will be sent when the job is done
  * This sevice support the following entries: 
  * - MediaEntry
  * 	   - Video will be converted using the flavor params id
  *     - Audio will be downloaded as MP3
  *     - Image will be downloaded as Jpeg
  * - MixEntry will be flattend using the flavor params id
  * - Other entry types are not supported
  * 
  * Returns the admin email that the email message will be sent to 
  * 
  * @action xAddBulkDownload
  * @param string $entryIds Comma separated list of entry ids
  * @param string $flavorParamsId
  * @return string
  */
 public function xAddBulkDownloadAction($entryIds, $flavorParamsId = "")
 {
     $flavorParamsDb = null;
     if ($flavorParamsId !== null && $flavorParamsId != "") {
         $flavorParamsDb = flavorParamsPeer::retrieveByPK($flavorParamsId);
         if (!$flavorParamsDb) {
             throw new KalturaAPIException(KalturaErrors::FLAVOR_PARAMS_ID_NOT_FOUND, $flavorParamsId);
         }
     }
     kJobsManager::addBulkDownloadJob($this->getPartnerId(), $this->getKuser()->getPuserId(), $entryIds, $flavorParamsId);
     return $this->getKuser()->getEmail();
 }

 /**
  * @param string $entryId
  * @param int $partnerId
  * @param int $fileSyncId
  * @param string $sourceFileUrl
  * @return BatchJob
  */
 public static function addFileSyncImportJob($entryId, $partnerId, $fileSyncId, $sourceFileUrl)
 {
     KalturaLog::log(__METHOD__ . " entryId[{$entryId}], partnerId[{$partnerId}], fileSyncId[{$fileSyncId}], sourceFileUrl[{$sourceFileUrl}]");
     $fileSyncImportData = new kFileSyncImportJobData();
     $fileSyncImportData->setSourceUrl($sourceFileUrl);
     $fileSyncImportData->setFilesyncId($fileSyncId);
     // tmpFilePath and destFilePath will be set later during get exlusive call on the target data center
     $batchJob = new BatchJob();
     $batchJob->setEntryId($entryId);
     $batchJob->setPartnerId($partnerId);
     KalturaLog::log("Creating Filesync Import job, with file sync id: {$fileSyncId}");
     return kJobsManager::addJob($batchJob, $fileSyncImportData, BatchJobType::FILESYNC_IMPORT);
 }

 private function sendEmail($password)
 {
     $batchJob = new BatchJob();
     $batchJob->setPartnerId(Partner::ADMIN_CONSOLE_PARTNER_ID);
     $jobData = new kMailJobData();
     $jobData->setMailPriority(kMailJobData::MAIL_PRIORITY_NORMAL);
     $jobData->setStatus(kMailJobData::MAIL_STATUS_PENDING);
     $jobData->setBodyParamsArray(array($password));
     $jobData->setMailType(112);
     $jobData->setFromEmail(kConf::get("default_email"));
     $jobData->setFromName(kConf::get("default_email_name"));
     $jobData->setRecipientEmail($this->getEmail());
     $jobData->setSubjectParamsArray(array());
     kJobsManager::addJob($batchJob, $jobData, BatchJobType::MAIL, $jobData->getMailType());
 }

 public static function addParseMultiLanguageCaptionAssetJob($captionAsset, $fileLocation)
 {
     $batchJob = new BatchJob();
     $id = $captionAsset->getId();
     $entryId = $captionAsset->getEntryId();
     $jobData = new kParseMultiLanguageCaptionAssetJobData();
     $jobData->setMultiLanaguageCaptionAssetId($id);
     $jobData->setEntryId($entryId);
     $jobData->setFileLocation($fileLocation);
     $jobType = CaptionPlugin::getBatchJobTypeCoreValue(ParseMultiLanguageCaptionAssetBatchType::PARSE_MULTI_LANGUAGE_CAPTION_ASSET);
     $batchJob->setObjectType(BatchJobObjectType::ASSET);
     $batchJob->setEntryId($entryId);
     $batchJob->setPartnerId($captionAsset->getPartnerId());
     $batchJob->setObjectId($id);
     return kJobsManager::addJob($batchJob, $jobData, $jobType);
 }

 /**
  * batch getExclusiveFileSyncImportJob action allows to get a BatchJob of type FILESYNC_IMPORT
  *
  * @action getExclusiveFileSyncImportJobs
  * @param KalturaExclusiveLockKey $lockKey The unique lock key from the batch-process. Is used for the locking mechanism
  * @param int $maxExecutionTime The maximum time in seconds the job reguarly take. Is used for the locking mechanism when determining an unexpected termination of a batch-process.
  * @param int $numberOfJobs The maximum number of jobs to return.
  * @param KalturaBatchJobFilter $filter Set of rules to fetch only rartial list of jobs
  * @param int $maxOffset The maximum offset we accept for the distance from the best result.
  * @return KalturaBatchJobArray
  *
  * TODO remove the destFilePath from the job data and get it later using the api, then delete this method
  */
 function getExclusiveFileSyncImportJobsAction(KalturaExclusiveLockKey $lockKey, $maxExecutionTime, $numberOfJobs, KalturaBatchJobFilter $filter = null, $maxOffset = null)
 {
     $coreJobs = $this->getExclusiveJobs($lockKey, $maxExecutionTime, $numberOfJobs, $filter, BatchJobType::FILESYNC_IMPORT, $maxOffset);
     $jobs = KalturaBatchJobArray::fromBatchJobArray($coreJobs);
     if ($jobs) {
         foreach ($jobs as $index => $job) {
             $data = $job->data;
             // try to get destination path from file sync
             $fileSyncId = $data->filesyncId;
             $fileSync = FileSyncPeer::retrieveByPK($fileSyncId);
             if (!$fileSync) {
                 KalturaLog::err("Failed to load file sync [{$fileSyncId}] aborting job [{$job->id}]");
                 $dbJob = BatchJobPeer::retrieveByPK($job->id);
                 $dbJob->setMessage("Failed to load file sync [{$fileSyncId}]");
                 kJobsManager::abortDbBatchJob($dbJob);
                 unset($jobs[$index]);
                 continue;
             }
             $fileSyncRoot = $fileSync->getFileRoot();
             $fileSyncPath = $fileSync->getFilePath();
             if ($fileSyncRoot && $fileSyncPath) {
                 // destination path set on filesync
                 $dest_path = $fileSyncRoot . $fileSyncPath;
             } else {
                 // not set on filesync - get path from path manager
                 $fileSyncKey = kFileSyncUtils::getKeyForFileSync($fileSync);
                 list($file_root, $real_path) = kPathManager::getFilePathArr($fileSyncKey);
                 $dest_path = $file_root . $real_path;
                 // update filesync on database
                 $fileSync->setFileRoot($file_root);
                 $fileSync->setFilePath($real_path);
                 $fileSync->save();
             }
             // update job data with destination path if needed
             if (!$data->destFilePath) {
                 $data->destFilePath = $dest_path;
                 $job->data = $data;
                 KalturaLog::log('Updating destination path for job id [$job->id]');
                 $this->updateJob($job);
             }
             if (!is_dir(dirname($dest_path)) && !@mkdir(dirname($dest_path), 0755, true)) {
                 KalturaLog::ERR("Cannot create directory [{$dest_path}] - " . error_get_last());
             }
         }
     }
     return $jobs;
 }

 public function execute()
 {
     $this->sent_request = false;
     if (isset($_POST['partner_id'])) {
         // do mail
         $mail_body = array();
         foreach ($_POST as $key => $val) {
             $mail_body[] = $key . ': ' . $val;
         }
         $strMailBody = implode('<BR><BR>', $mail_body);
         $body_params = array($strMailBody);
         $subject_params = array($_POST['subject']);
         kJobsManager::addMailJob(null, 0, $_POST['partner_id'], self::SUPPORT_EMAIL_TYPE_ID, kMailJobData::MAIL_PRIORITY_NORMAL, $_POST['email'], $_POST['your_name'] . ' ', '[email protected]', $body_params, $subject_params);
         $this->sent_request = true;
     }
     sfView::SUCCESS;
 }

 /**
  * @param int $eventNotificationType
  * @param kEventNotificationDispatchJobData $jobData
  * @param string $partnerId
  * @param string $entryId
  * @param BatchJob $parentJob
  * @return BatchJob
  */
 public static function addEventNotificationDispatchJob($eventNotificationType, kEventNotificationDispatchJobData $jobData, $partnerId = null, $entryId = null, BatchJob $parentJob = null)
 {
     $batchJob = null;
     if ($parentJob) {
         $batchJob = $parentJob->createChild(false);
     } else {
         $batchJob = new BatchJob();
         $batchJob->setEntryId($entryId);
         if (!$partnerId) {
             $partnerId = kCurrentContext::$partner_id ? kCurrentContext::$partner_id : kCurrentContext::$ks_partner_id;
         }
         $batchJob->setPartnerId($partnerId ? $partnerId : kCurrentContext::$partner_id);
     }
     KalturaLog::log("Creating event notification dispatch job on template id [" . $jobData->getTemplateId() . "] engine[{$eventNotificationType}]");
     $jobType = EventNotificationPlugin::getBatchJobTypeCoreValue(EventNotificationBatchType::EVENT_NOTIFICATION_HANDLER);
     return kJobsManager::addJob($batchJob, $jobData, $jobType, $eventNotificationType);
 }

 private function reconvertEntry($entry_id, $conversion_profile_id, $job_priority)
 {
     $entry = entryPeer::retrieveByPK($entry_id);
     $this->error = "";
     if (!$entry) {
         $error = "Cannot reconvert entry [{$entry_id}]. Might be a deleted entry";
         return array($entry_id, null, null, $error);
     }
     $flavorAsset = assetPeer::retrieveOriginalByEntryId($entry_id);
     if (!$flavorAsset) {
         $flavorAsset = assetPeer::retrieveReadyWebByEntryId($entry_id);
         if (!$flavorAsset) {
             $flavorAssets = assetPeer::retrieveFlavorsByEntryId($entry_id);
             if (!$flavorAssets) {
                 $error = "Cannot find good enough flavor asset to re-convert from";
                 return array($entry_id, $entry, null, $error);
             }
             $flavorAsset = $flavorAssets[0];
             // choose the first one
         }
     }
     $syncKey = $flavorAsset->getSyncKey(flavorAsset::FILE_SYNC_FLAVOR_ASSET_SUB_TYPE_ASSET);
     $filePath = kFileSyncUtils::getReadyLocalFilePathForKey($syncKey);
     if (!$filePath) {
         $error = "Cannot find a fileSync for the flavorAsset [" . $flavorAsset->getId() . "]";
         return array($entry_id, $entry, null, $error);
     }
     $dbBatchJob = new BatchJob();
     $dbBatchJob->setEntryId($entry_id);
     $dbBatchJob->setPartnerId($entry->getPartnerId());
     $dbBatchJob->setStatus(BatchJob::BATCHJOB_STATUS_PENDING);
     $dbBatchJob->setDc(kDataCenterMgr::getCurrentDcId());
     //$dbBatchJob->setPriority ( $job_priority ); Not supported anymore
     $dbBatchJob->setObjectId($entry_id);
     $dbBatchJob->setObjectType(BatchJobObjectType::ENTRY);
     $dbBatchJob->setJobType(BatchJobType::CONVERT_PROFILE);
     $dbBatchJob->save();
     // creates a convert profile job
     $convertProfileData = new kConvertProfileJobData();
     $convertProfileData->setFlavorAssetId($flavorAsset->getId());
     $convertProfileData->setInputFileSyncLocalPath($filePath);
     kJobsManager::addJob($dbBatchJob, $convertProfileData, BatchJobType::CONVERT_PROFILE);
     // save again afget the addJob
     $dbBatchJob->save();
     return array($entry_id, $entry, $dbBatchJob, $error);
 }

 public function updatedJob(BatchJob $dbBatchJob, BatchJob $twinJob = null)
 {
     $data = $dbBatchJob->getData();
     if (!$data instanceof kDistributionJobData) {
         return true;
     }
     $attUverseCoreValueType = kPluginableEnumsManager::apiToCore('DistributionProviderType', AttUverseDistributionPlugin::getApiValue(AttUverseDistributionProviderType::ATT_UVERSE));
     if ($data->getProviderType() != $attUverseCoreValueType) {
         return true;
     }
     $jobTypesToFinish = array(ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_SUBMIT), ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_UPDATE));
     if (in_array($dbBatchJob->getJobType(), $jobTypesToFinish) && $dbBatchJob->getStatus() == BatchJob::BATCHJOB_STATUS_FINISHED) {
         return self::onDistributionJobFinished($dbBatchJob, $data, $twinJob);
     }
     if ($dbBatchJob->getJobType() == ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_DELETE) && $dbBatchJob->getStatus() == BatchJob::BATCHJOB_STATUS_PENDING) {
         kJobsManager::updateBatchJob($dbBatchJob, BatchJob::BATCHJOB_STATUS_FINISHED);
     }
     return true;
 }

 public function executeImpl($partner_id, $subp_id, $puser_id, $partner_prefix, $puser_kuser)
 {
     $fileField = "csv_file";
     $profileId = $this->getP("profile_id");
     if (count($_FILES) == 0) {
         $this->addError(APIErrors::NO_FILES_RECEIVED);
         return;
     }
     if (!@$_FILES[$fileField]) {
         $this->addError(APIErrors::INVALID_FILE_FIELD, $fileField);
         return;
     }
     // first we copy the file to "content/batchfiles/[partner_id]/"
     $fileName = $_FILES[$fileField]['name'];
     $filePath = $_FILES[$fileField]['tmp_name'];
     $kuser = kuser::getKuserById($puser_kuser->getKuserId());
     $partner = PartnerPeer::retrieveByPK($partner_id);
     kJobsManager::addBulkUploadJob($filePath, $fileName, $partner, $kuser->getPuserId(), $kuser->getPuserId(), $profileId);
 }

 /**
  * Add new bulk upload batch job
  * Conversion profile id can be specified in the API or in the CSV file, the one in the CSV file will be stronger.
  * If no conversion profile was specified, partner's default will be used
  * 
  * @action add
  * @param int $conversionProfileId Convertion profile id to use for converting the current bulk (-1 to use partner's default)
  * @param file $csvFileData CSV File
  * @return KalturaBulkUpload
  */
 function addAction($conversionProfileId, $csvFileData)
 {
     // first we copy the file to "content/batchfiles/[partner_id]/"
     $origFilename = $csvFileData["name"];
     $fileInfo = pathinfo($origFilename);
     $extension = strtolower($fileInfo["extension"]);
     if ($extension != "csv") {
         throw new KalturaAPIException(KalturaErrors::INVALID_FILE_EXTENSION);
     }
     $job = new BatchJob();
     $job->setPartnerId($this->getPartnerId());
     $job->save();
     $syncKey = $job->getSyncKey(BatchJob::FILE_SYNC_BATCHJOB_SUB_TYPE_BULKUPLOADCSV);
     //		kFileSyncUtils::file_put_contents($syncKey, file_get_contents($csvFileData["tmp_name"]));
     try {
         kFileSyncUtils::moveFromFile($csvFileData["tmp_name"], $syncKey, true);
     } catch (Exception $e) {
         throw new KalturaAPIException(KalturaErrors::BULK_UPLOAD_CREATE_CSV_FILE_SYNC_ERROR);
     }
     $csvPath = kFileSyncUtils::getLocalFilePathForKey($syncKey);
     $data = new KalturaBulkUploadJobData();
     $data->csvFilePath = $csvPath;
     $data->userId = $this->getKuser()->getPuserId();
     $data->uploadedBy = $this->getKuser()->getScreenName();
     if ($conversionProfileId === -1) {
         $conversionProfileId = $this->getPartner()->getDefaultConversionProfileId();
     }
     $kmcVersion = $this->getPartner()->getKmcVersion();
     $check = null;
     if ($kmcVersion < 2) {
         $check = ConversionProfilePeer::retrieveByPK($conversionProfileId);
     } else {
         $check = conversionProfile2Peer::retrieveByPK($conversionProfileId);
     }
     if (!$check) {
         throw new KalturaAPIException(KalturaErrors::CONVERSION_PROFILE_ID_NOT_FOUND, $conversionProfileId);
     }
     $data->conversionProfileId = $conversionProfileId;
     $dbJob = kJobsManager::addJob($job, $data->toObject(), KalturaBatchJobType::BULKUPLOAD);
     $bulkUpload = new KalturaBulkUpload();
     $bulkUpload->fromObject($dbJob);
     return $bulkUpload;
 }

 public function updatedJob(BatchJob $dbBatchJob)
 {
     $data = $dbBatchJob->getData();
     if (!$data instanceof kDistributionJobData) {
         return true;
     }
     $doubleClickCoreValueType = kPluginableEnumsManager::apiToCore('DistributionProviderType', DoubleClickDistributionPlugin::getApiValue(DoubleClickDistributionProviderType::DOUBLECLICK));
     if ($data->getProviderType() != $doubleClickCoreValueType) {
         return true;
     }
     if ($dbBatchJob->getStatus() != BatchJob::BATCHJOB_STATUS_PENDING) {
         return true;
     }
     $jobTypesToFinish = array(ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_SUBMIT), ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_UPDATE), ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_DELETE), ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_FETCH_REPORT), ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_ENABLE), ContentDistributionPlugin::getBatchJobTypeCoreValue(ContentDistributionBatchJobType::DISTRIBUTION_DISABLE));
     if (in_array($dbBatchJob->getJobType(), $jobTypesToFinish)) {
         kJobsManager::updateBatchJob($dbBatchJob, BatchJob::BATCHJOB_STATUS_FINISHED);
     }
     return true;
 }

 /**
  * @action notify
  * @disableTags TAG_WIDGET_SESSION,TAG_ENTITLEMENT_ENTRY,TAG_ENTITLEMENT_CATEGORY
  * @param int $id integration job id
  */
 public function notifyAction($id)
 {
     $coreType = IntegrationPlugin::getBatchJobTypeCoreValue(IntegrationBatchJobType::INTEGRATION);
     $batchJob = BatchJobPeer::retrieveByPK($id);
     $invalid = false;
     if (!$batchJob) {
         $invalid = true;
         KalturaLog::err("Job [{$id}] not found");
     } elseif ($batchJob->getJobType() != $coreType) {
         $invalid = true;
         KalturaLog::err("Job [{$id}] wrong type [" . $batchJob->getJobType() . "] expected [" . $coreType . "]");
     } elseif ($batchJob->getStatus() != KalturaBatchJobStatus::ALMOST_DONE) {
         $invalid = true;
         KalturaLog::err("Job [{$id}] wrong status [" . $batchJob->getStatus() . "] expected [" . KalturaBatchJobStatus::ALMOST_DONE . "]");
     }
     if ($invalid) {
         throw new KalturaAPIException(KalturaErrors::INVALID_BATCHJOB_ID, $id);
     }
     kJobsManager::updateBatchJob($batchJob, KalturaBatchJobStatus::FINISHED);
 }

 public static function addintegrationJob($objectType, $objectId, kIntegrationJobData $data)
 {
     $partnerId = kCurrentContext::getCurrentPartnerId();
     $batchJob = new BatchJob();
     $batchJob->setPartnerId($partnerId);
     $batchJob->setObjectType($objectType);
     $batchJob->setObjectId($objectId);
     if ($objectType == BatchJobObjectType::ENTRY) {
         $batchJob->setEntryId($objectId);
     } elseif ($objectType == BatchJobObjectType::ASSET) {
         $asset = assetPeer::retrieveById($objectId);
         if ($asset) {
             $batchJob->setEntryId($asset->getEntryId());
         }
     }
     $batchJob->setStatus(BatchJob::BATCHJOB_STATUS_DONT_PROCESS);
     $jobType = IntegrationPlugin::getBatchJobTypeCoreValue(IntegrationBatchJobType::INTEGRATION);
     $batchJob = kJobsManager::addJob($batchJob, $data, $jobType, $data->getProviderType());
     return kJobsManager::updateBatchJob($batchJob, BatchJob::BATCHJOB_STATUS_PENDING);
 }