<commit_before><?php
/**
 * batch service lets you handle different batch process from remote machines.
 * As oppesed to other ojects in the system, locking mechanism is critical in this case.
 * For this reason the GetExclusiveXX, UpdateExclusiveXX and FreeExclusiveXX actions are important for the system's intergity.
 * In general - updating batch object should be done only using the UpdateExclusiveXX which in turn can be called only after 
 * acuiring a batch objet properly (using  GetExclusiveXX).
 * If an object was aquired and should be returned to the pool in it's initial state - use the FreeExclusiveXX action 
 *
 *	Terminology:
 *		LocationId
 *		ServerID
 *		ParternGroups 
 * 
 * @service batchcontrol
 * @package api
 * @subpackage services
 */
class BatchControlService extends KalturaBaseService 
{
	// use initService to add a peer to the partner filter
	/**
	 * @ignore
	 */
	public function initService($serviceId, $serviceName, $actionName)
	{
		parent::initService($serviceId, $serviceName, $actionName);
//		parent::applyPartnerFilterForClass ( new BatchJobPeer() ); 	
//		parent::applyPartnerFilterForClass ( new notificationPeer() );
	}
	

		
	
// --------------------------------- scheduler support functions 	--------------------------------- //

		
	
	/**
	 * batch reportStatus action saves the a status attribute from a remote scheduler and returns pending commands for the scheduler
	 * 
	 * @action reportStatus
	 * @param KalturaScheduler $scheduler The scheduler
	 * @param KalturaSchedulerStatusArray $schedulerStatuses A scheduler status array
	 * @param KalturaWorkerQueueFilterArray $workerQueueFilters Filters list to get queues
	 * @return KalturaSchedulerStatusResponse
	 */
	function reportStatusAction(KalturaScheduler $scheduler, KalturaSchedulerStatusArray $schedulerStatuses, KalturaWorkerQueueFilterArray $workerQueueFilters)
	{
		$schedulerDb = $this->getOrCreateScheduler($scheduler);
		$schedulerChanged = false;
		
		// saves the statuses to the DB
		foreach($schedulerStatuses as $schedulerStatus)
		{
			$schedulerStatus->schedulerId = $schedulerDb->getId();
			$schedulerStatus->schedulerConfiguredId = $scheduler->configuredId;
			
			if($schedulerStatus->workerConfiguredId)
			{
				$worker = $this->getOrCreateWorker($schedulerDb, $schedulerStatus->workerConfiguredId, $schedulerStatus->workerType);
				$worker->setStatus($schedulerStatus->type, $schedulerStatus->value);
				$worker->save();
				$schedulerStatus->workerId = $worker->getId();
			}
			else
			{
				$schedulerChanged = true;
				$schedulerDb->setStatus($schedulerStatus->type, $schedulerStatus->value);
			}
			
			$schedulerStatusDb = new SchedulerStatus();
			$schedulerStatus->toObject($schedulerStatusDb);
			$schedulerStatusDb->save();
		}
		if($schedulerChanged)
			$schedulerDb->save();
		
		
		// creates a response
		$schedulerStatusResponse = new KalturaSchedulerStatusResponse();
		
		// gets the control pannel commands
		$c = new Criteria();
		$c->add(ControlPanelCommandPeer::SCHEDULER_ID, $schedulerDb->getId());
		$c->add(ControlPanelCommandPeer::TYPE, KalturaControlPanelCommandType::CONFIG, Criteria::NOT_EQUAL);
		$c->add(ControlPanelCommandPeer::STATUS, KalturaControlPanelCommandStatus::PENDING);
		$commandsList = ControlPanelCommandPeer::doSelect($c);
		foreach($commandsList as $command)
		{
			$command->setStatus(KalturaControlPanelCommandStatus::HANDLED);
			$command->save();
		}
		$schedulerStatusResponse->controlPanelCommands = KalturaControlPanelCommandArray::fromControlPanelCommandArray($commandsList);
		
		// gets new configs
		$c = new Criteria();
		$c->add(SchedulerConfigPeer::SCHEDULER_ID, $schedulerDb->getId());
		$c->add(SchedulerConfigPeer::COMMAND_STATUS, KalturaControlPanelCommandStatus::PENDING);
		$configList = SchedulerConfigPeer::doSelect($c);
		foreach($configList as $config)
		{
			$config->setCommandStatus(KalturaControlPanelCommandStatus::HANDLED);
			$config->save();
		}
		$schedulerStatusResponse->schedulerConfigs = KalturaSchedulerConfigArray::fromSchedulerConfigArray($configList);
		
		// gets queues length
		$schedulerStatusResponse->queuesStatus = new KalturaBatchQueuesStatusArray();
		foreach($workerQueueFilters as $workerQueueFilter)
		{
			$dbJobType = kPluginableEnumsManager::apiToCore('BatchJobType', $workerQueueFilter->jobType);
			$filter = $workerQueueFilter->filter->toFilter($dbJobType);
			
			$batchQueuesStatus = new KalturaBatchQueuesStatus();
			$batchQueuesStatus->jobType = $workerQueueFilter->jobType;
			$batchQueuesStatus->workerId = $workerQueueFilter->workerId;
			$batchQueuesStatus->size = kBatchManager::getQueueSize($workerQueueFilter->schedulerId, $workerQueueFilter->workerId, $dbJobType, $filter);
			
			$schedulerStatusResponse->queuesStatus[] = $batchQueuesStatus;
		}
		
		return $schedulerStatusResponse;
	}
	
	
	/**
	 * batch getOrCreateScheduler returns a scheduler by name, create it if doesnt exist
	 * 
	 * @param KalturaScheduler $scheduler
	 * @return Scheduler
	 */
	private function getOrCreateScheduler(KalturaScheduler $scheduler)
	{
		$c = new Criteria();
		$c->add ( SchedulerPeer::CONFIGURED_ID, $scheduler->configuredId);
		$schedulerDb = SchedulerPeer::doSelectOne($c, myDbHelper::getConnection(myDbHelper::DB_HELPER_CONN_PROPEL2));
		
		if($schedulerDb)
		{
			if(strlen($schedulerDb->getHost()) && $schedulerDb->getHost() != $scheduler->host)
				throw new KalturaAPIException(KalturaErrors::SCHEDULER_HOST_CONFLICT, $scheduler->configuredId, $scheduler->host, $schedulerDb->getHost());
			
			if($schedulerDb->getName() != $scheduler->name || $schedulerDb->getHost() != $scheduler->host)
			{
				$schedulerDb->setName($scheduler->name);
				$schedulerDb->setHost($scheduler->host);
				$schedulerDb->save();
			}
			
			return $schedulerDb;
		}
			
		$schedulerDb = new Scheduler();
		$schedulerDb->setLastStatus(time());
		$schedulerDb->setName($scheduler->name);
		$schedulerDb->setHost($scheduler->host);
		$schedulerDb->setConfiguredId($scheduler->configuredId);
		$schedulerDb->setDescription('');
		
		$schedulerDb->save();
		
		return $schedulerDb;
	}
	
	
	/**
	 * batch getOrCreateWorker returns a worker by name, create it if doesnt exist
	 * 
	 * @param Scheduler $scheduler The scheduler object
	 * @param int $workerConfigId The worker configured id
	 * @param KalturaBatchJobType $workerType The type of the remote worker
	 * @param string $workerName The name of the remote worker
	 * @return Worker
	 */
	private function getOrCreateWorker(Scheduler $scheduler, $workerConfigId, $workerType = null, $workerName = null)
	{
		$c = new Criteria();
		$c->add ( SchedulerWorkerPeer::SCHEDULER_CONFIGURED_ID, $scheduler->getConfiguredId());
		$c->add ( SchedulerWorkerPeer::CONFIGURED_ID, $workerConfigId);
		$workerDb = SchedulerWorkerPeer::doSelectOne($c, myDbHelper::getConnection(myDbHelper::DB_HELPER_CONN_PROPEL2));
		
		if($workerDb)
		{
			$shouldSave = false;
			
			if(!is_null($workerName) && $workerDb->getName() != $workerName)
			{
				$workerDb->setName($workerName);
				$shouldSave = true;
			}
			
			if(!is_null($workerType) && $workerDb->getType() != $workerType)
			{
				$workerDb->setType($workerType);
				$shouldSave = true;
			}
			
			if($shouldSave)
				$workerDb->save();
			
			return $workerDb;
		}
			
		$workerDb = new SchedulerWorker();
		$workerDb->setLastStatus(time());<commit_msg>Translate job type to core value<commit_after><?php
/**
 * batch service lets you handle different batch process from remote machines.
 * As oppesed to other ojects in the system, locking mechanism is critical in this case.
 * For this reason the GetExclusiveXX, UpdateExclusiveXX and FreeExclusiveXX actions are important for the system's intergity.
 * In general - updating batch object should be done only using the UpdateExclusiveXX which in turn can be called only after 
 * acuiring a batch objet properly (using  GetExclusiveXX).
 * If an object was aquired and should be returned to the pool in it's initial state - use the FreeExclusiveXX action 
 *
 *	Terminology:
 *		LocationId
 *		ServerID
 *		ParternGroups 
 * 
 * @service batchcontrol
 * @package api
 * @subpackage services
 */
class BatchControlService extends KalturaBaseService 
{
	// use initService to add a peer to the partner filter
	/**
	 * @ignore
	 */
	public function initService($serviceId, $serviceName, $actionName)
	{
		parent::initService($serviceId, $serviceName, $actionName);
//		parent::applyPartnerFilterForClass ( new BatchJobPeer() ); 	
//		parent::applyPartnerFilterForClass ( new notificationPeer() );
	}
	

		
	
// --------------------------------- scheduler support functions 	--------------------------------- //

		
	
	/**
	 * batch reportStatus action saves the a status attribute from a remote scheduler and returns pending commands for the scheduler
	 * 
	 * @action reportStatus
	 * @param KalturaScheduler $scheduler The scheduler
	 * @param KalturaSchedulerStatusArray $schedulerStatuses A scheduler status array
	 * @param KalturaWorkerQueueFilterArray $workerQueueFilters Filters list to get queues
	 * @return KalturaSchedulerStatusResponse
	 */
	function reportStatusAction(KalturaScheduler $scheduler, KalturaSchedulerStatusArray $schedulerStatuses, KalturaWorkerQueueFilterArray $workerQueueFilters)
	{
		$schedulerDb = $this->getOrCreateScheduler($scheduler);
		$schedulerChanged = false;
		
		// saves the statuses to the DB
		foreach($schedulerStatuses as $schedulerStatus)
		{
			$schedulerStatus->schedulerId = $schedulerDb->getId();
			$schedulerStatus->schedulerConfiguredId = $scheduler->configuredId;
			
			if($schedulerStatus->workerConfiguredId)
			{
				$worker = $this->getOrCreateWorker($schedulerDb, $schedulerStatus->workerConfiguredId, $schedulerStatus->workerType);
				$worker->setStatus($schedulerStatus->type, $schedulerStatus->value);
				$worker->save();
				$schedulerStatus->workerId = $worker->getId();
			}
			else
			{
				$schedulerChanged = true;
				$schedulerDb->setStatus($schedulerStatus->type, $schedulerStatus->value);
			}
			
			$schedulerStatusDb = new SchedulerStatus();
			$schedulerStatus->toObject($schedulerStatusDb);
			$schedulerStatusDb->save();
		}
		if($schedulerChanged)
			$schedulerDb->save();
		
		
		// creates a response
		$schedulerStatusResponse = new KalturaSchedulerStatusResponse();
		
		// gets the control pannel commands
		$c = new Criteria();
		$c->add(ControlPanelCommandPeer::SCHEDULER_ID, $schedulerDb->getId());
		$c->add(ControlPanelCommandPeer::TYPE, KalturaControlPanelCommandType::CONFIG, Criteria::NOT_EQUAL);
		$c->add(ControlPanelCommandPeer::STATUS, KalturaControlPanelCommandStatus::PENDING);
		$commandsList = ControlPanelCommandPeer::doSelect($c);
		foreach($commandsList as $command)
		{
			$command->setStatus(KalturaControlPanelCommandStatus::HANDLED);
			$command->save();
		}
		$schedulerStatusResponse->controlPanelCommands = KalturaControlPanelCommandArray::fromControlPanelCommandArray($commandsList);
		
		// gets new configs
		$c = new Criteria();
		$c->add(SchedulerConfigPeer::SCHEDULER_ID, $schedulerDb->getId());
		$c->add(SchedulerConfigPeer::COMMAND_STATUS, KalturaControlPanelCommandStatus::PENDING);
		$configList = SchedulerConfigPeer::doSelect($c);
		foreach($configList as $config)
		{
			$config->setCommandStatus(KalturaControlPanelCommandStatus::HANDLED);
			$config->save();
		}
		$schedulerStatusResponse->schedulerConfigs = KalturaSchedulerConfigArray::fromSchedulerConfigArray($configList);
		
		// gets queues length
		$schedulerStatusResponse->queuesStatus = new KalturaBatchQueuesStatusArray();
		foreach($workerQueueFilters as $workerQueueFilter)
		{
			$dbJobType = kPluginableEnumsManager::apiToCore('BatchJobType', $workerQueueFilter->jobType);
			$filter = $workerQueueFilter->filter->toFilter($dbJobType);
			
			$batchQueuesStatus = new KalturaBatchQueuesStatus();
			$batchQueuesStatus->jobType = $workerQueueFilter->jobType;
			$batchQueuesStatus->workerId = $workerQueueFilter->workerId;
			$batchQueuesStatus->size = kBatchManager::getQueueSize($workerQueueFilter->schedulerId, $workerQueueFilter->workerId, $dbJobType, $filter);
			
			$schedulerStatusResponse->queuesStatus[] = $batchQueuesStatus;
		}
		
		return $schedulerStatusResponse;
	}
	
	
	/**
	 * batch getOrCreateScheduler returns a scheduler by name, create it if doesnt exist
	 * 
	 * @param KalturaScheduler $scheduler
	 * @return Scheduler
	 */
	private function getOrCreateScheduler(KalturaScheduler $scheduler)
	{
		$c = new Criteria();
		$c->add ( SchedulerPeer::CONFIGURED_ID, $scheduler->configuredId);
		$schedulerDb = SchedulerPeer::doSelectOne($c, myDbHelper::getConnection(myDbHelper::DB_HELPER_CONN_PROPEL2));
		
		if($schedulerDb)
		{
			if(strlen($schedulerDb->getHost()) && $schedulerDb->getHost() != $scheduler->host)
				throw new KalturaAPIException(KalturaErrors::SCHEDULER_HOST_CONFLICT, $scheduler->configuredId, $scheduler->host, $schedulerDb->getHost());
			
			if($schedulerDb->getName() != $scheduler->name || $schedulerDb->getHost() != $scheduler->host)
			{
				$schedulerDb->setName($scheduler->name);
				$schedulerDb->setHost($scheduler->host);
				$schedulerDb->save();
			}
			
			return $schedulerDb;
		}
			
		$schedulerDb = new Scheduler();
		$schedulerDb->setLastStatus(time());
		$schedulerDb->setName($scheduler->name);
		$schedulerDb->setHost($scheduler->host);
		$schedulerDb->setConfiguredId($scheduler->configuredId);
		$schedulerDb->setDescription('');
		
		$schedulerDb->save();
		
		return $schedulerDb;
	}
	
	
	/**
	 * batch getOrCreateWorker returns a worker by name, create it if doesnt exist
	 * 
	 * @param Scheduler $scheduler The scheduler object
	 * @param int $workerConfigId The worker configured id
	 * @param KalturaBatchJobType $workerType The type of the remote worker
	 * @param string $workerName The name of the remote worker
	 * @return Worker
	 */
	private function getOrCreateWorker(Scheduler $scheduler, $workerConfigId, $workerType = null, $workerName = null)
	{
		$workerType = kPluginableEnumsManager::apiToCore('BatchJobType', $workerType);
		
		$c = new Criteria();
		$c->add ( SchedulerWorkerPeer::SCHEDULER_CONFIGURED_ID, $scheduler->getConfiguredId());
		$c->add ( SchedulerWorkerPeer::CONFIGURED_ID, $workerConfigId);
		$workerDb = SchedulerWorkerPeer::doSelectOne($c, myDbHelper::getConnection(myDbHelper::DB_HELPER_CONN_PROPEL2));
		
		if($workerDb)
		{
			$shouldSave = false;
			
			if(!is_null($workerName) && $workerDb->getName() != $workerName)
			{
				$workerDb->setName($workerName);
				$shouldSave = true;
			}
			
			if(!is_null($workerType) && $workerDb->getType() != $workerType)
			{
				$workerDb->setType($workerType);
				$shouldSave = true;
			}
			
			if($shouldSave)
				$workerDb->save();
			
			return $workerDb;
		}
			
		$workerDb = new SchedulerWorker();
		$workerDb->setLastStatus(time());
{'fraction_non_alphanumeric': 0.1275808729571592, 'fraction_numerical': 0.0002690160737104042, 'mean_word_length': 4.898452994843316, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 165, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1353302', 'n_tokens_mistral': 5018, 'n_tokens_neox': 4439, 'n_words': 903}