<commit_before>﻿<?php
require_once("bootstrap.php");
ini_set('memory_limit', '128M');

/**
 * Email ingestion batch.
 * Scans an IMAP mail server and adds all found attachments as partner entries.
 *
 * @package Scheduler
 * @subpackage Email-Ingestion
 */
class KAsyncEmailIngestion extends KBatchBase {


	/***************/
	/* Definitions */
	/***************/

	// temp folder to place attachments during processing
	private $TEMP_FILE_DIR = ''; // will be overwritten in run()

	// mailbox folders
	const NO_ATTACHMENT      = 'invalid';
	const PARTNER_INVALID    = 'invalid';
	const ATTACHMENT_INVALID = 'invalid';
	const ADD_ENTRY_FAIL     = 'failed';
	const PROCESS_OK         = 'processed';
	const INCOMPLETE         = 'incomplete';
	const UNKNOWN            = 'unkown_error';


	private $temp_files = array();



	/***********************************************/
	/* Implementation of parent ABSTRACT functions */
	/***********************************************/

	/**
	 * @return <number> batch type
	 */
	public static function getType()
	{
		return KalturaBatchJobType::EMAIL_INGESTION;
	}


	/**
	 * Batch is done running
	 */
	public function done ()
	{
		KalturaLog::info("Email ingestion batch DONE");
	}

	public function __destruct()
	{
		// deleting all remaining temporary files
		if ($this->temp_files) {
			foreach ($this->temp_files as $filename) {
				if (file_exists($filename)) {
					unlink($filename);
				}
			}
		}
	}

	protected function init()
	{ /* non-relevant abstract function */ }

	protected function updateExclusiveJob($jobId, KalturaBatchJob $job, $entryStatus = null)
	{ /* non-relevant abstract function */ }

	protected function freeExclusiveJob(KalturaBatchJob $job)
	{ /* non-relevant abstract function */ }


	/********/
	/* Main */
	/********/

	public function run ()
	{
		KalturaLog::info("Email ingestion batch is running");

		if($this->taskConfig->isInitOnly()) {
			return $this->init();
		}

		// get parameters from ini file
		try {
			$this->TEMP_FILE_DIR = $this->taskConfig->params->localTempPath;
		}
		catch (Exception $e) {
			KalturaLog::crit("Cannot find all required parameters from config file");
		}

		// create a temp file path
		if ( !self::createDir( $this->TEMP_FILE_DIR ) )
		{
			KalturaLog::crit( "Cannot continue email ingestion without a temp directory");
			return false; // quit run()
		}


		// ----------------------------------
		// loop through all defined mailboxes
		// ----------------------------------

		$mailboxNumber = 0;	

		while ( isset($this->taskConfig->params->{'mailbox'.($mailboxNumber+1)}) ) {

			$mailboxNumber++;
			$mailesProcessed = 0;
			$keepCurMailbox = true;

			$params = $this->taskConfig->params->{'mailbox'.$mailboxNumber};

			// get parameters
			try {
				$host      = $params->hostname;
				$port      = $params->port;
				$user      = $params->user;
				$pass      = $<PASSWORD>;
				$options   = $params->options;
				$maxMails  = $params->maxMailsPerRun;
			}
			catch (Exception $e) {
				KalturaLog::crit("Cannot find all required parameters from config file for mailbox number [$mailboxNumber]");
				continue; // skip current mailbox
			}


			// connect to current mailbox
			$mailChecker = new KMailChecker($host, $port, $user, $pass, $options);
			if (!$mailChecker->connect()) {
				KalturaLog::crit("Error connecting to [$host:$port] as [$user] - ".imap_last_error());
				continue; // skip current mailbox
			}
			KalturaLog::info("Sucessfuly connected to [$host:$port] as [$user]");
				
			// check for unread mails
			$newMails = $mailChecker->getUnreadIds();
			if (!$newMails || count($newMails) <= 0) {
				// no new mail availble in current mailbox
				KalturaLog::info("No new mails found on [$user@$host]");
				continue; // skip current mailbox
			}
			KalturaLog::info('['.count($newMails)."] unread mails found on [$user@$host]");

			// -----------------------------------------
			// loop through all mails in current mailbox
			// -----------------------------------------
			while ($keepCurMailbox && (list(,$curId) = each($newMails))) {

				if ($mailesProcessed >= $maxMails) {
					KalturaLog::info("Reached the max mails per job for current mailbox [$mailboxNumber] - skipping to next mailbox");
					$keepCurMailbox = false; // skip current mailbox
					continue; // skip current mail --> skip current mailbox
				}

				$mailesProcessed++;

				// fetch current message
				$curMail = $mailChecker->fetchMsg($curId);
				if (!$curMail) {
					KalturaLog::err("Error fetching message with folder ID [$curId] from [$user@$host] - ".imap_last_error());
					continue; // skip current mail - error fetching
				}

				// check if mail contains attachments
				if (!$curMail->attachments || count($curMail->attachments) == 0) {
					// no attachments found
					KalturaLog::info('No attachments found for mail ['.$curMail->header->msgid."] on [$user@$host] from [".$curMail->header->fromadd.'] with subject ['.$curMail->header->subject.']');
					if (!$mailChecker->moveMsg($curId, self::NO_ATTACHMENT)) {
						KalturaLog::err('Failed moving msg ['.$curMail->header->msgid.'] to the ['.self::NO_ATTACHMENT.'] folder - '.imap_last_error());
					}
					continue; // skip current mail - no attachments
				}
				
				
				// validate partner and get email profile
				$email_profiles = $this->validePartnerAndGetProfile($curMail->header->toadd, $user.'@'.$host);
				if (!$email_profiles) {
					// error validating partner
					KalturaLog::err('Partner validation failed for ['.$curMail->header->msgid."] on [$user@$host] from [".$curMail->header->fromadd.'] with subject ['.$curMail->header->subject.']');
					if (!$mailChecker->moveMsg($curId, self::_INVALID)) {
						KalturaLog::err('Failed moving msg ['.$curMail->header->msgid.'] to the ['.self::PARTNER_INVALID.'] folder - '.imap_last_error());
					}
					continue; // skip current mail - partner invalid
				}

				// create a new media entry from data in mail body text
				$mediaEntry = $this->createMediaEntry($curMail->header, $curMail->body);<commit_msg>bug fix<commit_after>﻿<?php
require_once("bootstrap.php");
ini_set('memory_limit', '128M');

/**
 * Email ingestion batch.
 * Scans an IMAP mail server and adds all found attachments as partner entries.
 *
 * @package Scheduler
 * @subpackage Email-Ingestion
 */
class KAsyncEmailIngestion extends KBatchBase {


	/***************/
	/* Definitions */
	/***************/

	// temp folder to place attachments during processing
	private $TEMP_FILE_DIR = ''; // will be overwritten in run()

	// mailbox folders
	const NO_ATTACHMENT      = 'invalid';
	const PARTNER_INVALID    = 'invalid';
	const ATTACHMENT_INVALID = 'invalid';
	const ADD_ENTRY_FAIL     = 'failed';
	const PROCESS_OK         = 'processed';
	const INCOMPLETE         = 'incomplete';
	const UNKNOWN            = 'unkown_error';


	private $temp_files = array();



	/***********************************************/
	/* Implementation of parent ABSTRACT functions */
	/***********************************************/

	/**
	 * @return <number> batch type
	 */
	public static function getType()
	{
		return KalturaBatchJobType::EMAIL_INGESTION;
	}


	/**
	 * Batch is done running
	 */
	public function done ()
	{
		KalturaLog::info("Email ingestion batch DONE");
	}

	public function __destruct()
	{
		// deleting all remaining temporary files
		if ($this->temp_files) {
			foreach ($this->temp_files as $filename) {
				if (file_exists($filename)) {
					unlink($filename);
				}
			}
		}
	}

	protected function init()
	{ /* non-relevant abstract function */ }

	protected function updateExclusiveJob($jobId, KalturaBatchJob $job, $entryStatus = null)
	{ /* non-relevant abstract function */ }

	protected function freeExclusiveJob(KalturaBatchJob $job)
	{ /* non-relevant abstract function */ }


	/********/
	/* Main */
	/********/

	public function run ()
	{
		KalturaLog::info("Email ingestion batch is running");

		if($this->taskConfig->isInitOnly()) {
			return $this->init();
		}

		// get parameters from ini file
		try {
			$this->TEMP_FILE_DIR = $this->taskConfig->params->localTempPath;
		}
		catch (Exception $e) {
			KalturaLog::crit("Cannot find all required parameters from config file");
		}

		// create a temp file path
		if ( !self::createDir( $this->TEMP_FILE_DIR ) )
		{
			KalturaLog::crit( "Cannot continue email ingestion without a temp directory");
			return false; // quit run()
		}


		// ----------------------------------
		// loop through all defined mailboxes
		// ----------------------------------

		$mailboxNumber = 0;	

		while ( isset($this->taskConfig->params->{'mailbox'.($mailboxNumber+1)}) ) {

			$mailboxNumber++;
			$mailesProcessed = 0;
			$keepCurMailbox = true;

			$params = $this->taskConfig->params->{'mailbox'.$mailboxNumber};

			// get parameters
			try {
				$host      = $params->hostname;
				$port      = $params->port;
				$user      = $params->user;
				$pass      = $params->pass;
				$options   = $params->options;
				$maxMails  = $params->maxMailsPerRun;
			}
			catch (Exception $e) {
				KalturaLog::crit("Cannot find all required parameters from config file for mailbox number [$mailboxNumber]");
				continue; // skip current mailbox
			}


			// connect to current mailbox
			$mailChecker = new KMailChecker($host, $port, $user, $pass, $options);
			if (!$mailChecker->connect()) {
				KalturaLog::crit("Error connecting to [$host:$port] as [$user] - ".imap_last_error());
				continue; // skip current mailbox
			}
			KalturaLog::info("Sucessfuly connected to [$host:$port] as [$user]");
				
			// check for unread mails
			$newMails = $mailChecker->getUnreadIds();
			if (!$newMails || count($newMails) <= 0) {
				// no new mail availble in current mailbox
				KalturaLog::info("No new mails found on [$user@$host]");
				continue; // skip current mailbox
			}
			KalturaLog::info('['.count($newMails)."] unread mails found on [$user@$host]");

			// -----------------------------------------
			// loop through all mails in current mailbox
			// -----------------------------------------
			while ($keepCurMailbox && (list(,$curId) = each($newMails))) {

				if ($mailesProcessed >= $maxMails) {
					KalturaLog::info("Reached the max mails per job for current mailbox [$mailboxNumber] - skipping to next mailbox");
					$keepCurMailbox = false; // skip current mailbox
					continue; // skip current mail --> skip current mailbox
				}

				$mailesProcessed++;

				// fetch current message
				$curMail = $mailChecker->fetchMsg($curId);
				if (!$curMail) {
					KalturaLog::err("Error fetching message with folder ID [$curId] from [$user@$host] - ".imap_last_error());
					continue; // skip current mail - error fetching
				}

				// check if mail contains attachments
				if (!$curMail->attachments || count($curMail->attachments) == 0) {
					// no attachments found
					KalturaLog::info('No attachments found for mail ['.$curMail->header->msgid."] on [$user@$host] from [".$curMail->header->fromadd.'] with subject ['.$curMail->header->subject.']');
					if (!$mailChecker->moveMsg($curId, self::NO_ATTACHMENT)) {
						KalturaLog::err('Failed moving msg ['.$curMail->header->msgid.'] to the ['.self::NO_ATTACHMENT.'] folder - '.imap_last_error());
					}
					continue; // skip current mail - no attachments
				}
				
				
				// validate partner and get email profile
				$email_profiles = $this->validePartnerAndGetProfile($curMail->header->toadd, $user.'@'.$host);
				if (!$email_profiles) {
					// error validating partner
					KalturaLog::err('Partner validation failed for ['.$curMail->header->msgid."] on [$user@$host] from [".$curMail->header->fromadd.'] with subject ['.$curMail->header->subject.']');
					if (!$mailChecker->moveMsg($curId, self::PARTNER_INVALID_INVALID)) {
						KalturaLog::err('Failed moving msg ['.$curMail->header->msgid.'] to the ['.self::PARTNER_INVALID.'] folder - '.imap_last_error());
					}
					continue; // skip current mail - partner invalid
				}

				// create a new media entry from data in mail body text
				$mediaEntry = $this->createMediaEntry($curMail->header, $curMail->body);
{'fraction_non_alphanumeric': 0.1918169095249856, 'fraction_numerical': 0.00131719766197415, 'mean_word_length': 3.075142569607514, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 111, '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': '4041708', 'n_tokens_mistral': 4287, 'n_tokens_neox': 3646, 'n_words': 932}