<commit_before><?php

/*
	Accord-ion (c) CloudGarage Asia 2013

	Uni-directional synchronisation of MySQL to Salesforce and/or Salesforce to MySQL.

	Usage:
		php accordion.php epoch
                                    |_ Epoch (seconds) to retrieve rows based on their Last Modified timestamp

	Author(s):
		<NAME> <<EMAIL>>
	
*/

date_default_timezone_set('Hongkong');

function logger($indent, $msgs) {
	if($msgs != '') 
		foreach(explode(PHP_EOL, $msgs) as $msg)
			print date("d/m/Y H:i:s", time()).' - '.str_repeat("  ", $indent).$msg."\n";
}

/**
 * Retrieve the LastModified time from a given MySQL table
 *
 * @return string 
 */
function driverSalesforceGetLastModified($sf, $bac, $objName) {
	$soql = "SELECT LastModifiedDate FROM {$objName} ORDER BY LastModifiedDate DESC LIMIT 1";
	$response = $sf->query($soql);
	$result = $response->current()->LastModifiedDate;
	preg_match('/(\d+)-(\d+)-(\d+)T(\d+):(\d+):(\d+)\.\d+Z/', $result, $matches);
	$dt = new DateTime();
	$dt->setDate($matches[0], $matches[1], $matches[2]);
	$dt->setTime($matches[3], $matches[4], $matches[5]);
	return $dt;
}

/**
 * Utilises the Force.com Bulk API to perform SOQL queries, returning CSV output.
 *
 * @return string 
 */
function driverSalesforceGetData($sf, $bac, $objName, $fields, $lm) {
	$selectFields = join(",", $fields);
	$ts = $lm->format("Y-m-d\TH:i:s.000\Z");
	$soql = "SELECT {$selectFields} FROM {$objName} "; /*WHERE LastModifiedDate > '{$ts}' ORDER BY LastModifiedDate DESC"; */
	logger(3, 'SOQL:'.$soql);

	$job = new JobInfo();
	$job->setObject($objName);
	$job->setOperation("query");
	$job->setContentType("CSV");
	$job->setConcurrencyMode("Parallel");

	$job = $bac->createJob($job);
	$batch = $bac->createBatch($job, $soql);
	$bac->updateJobState($job->getId(), "Closed");

	logger(3, 'Waiting for query job completion...');
	while($batch->getState() == "Queued" || $batch->getState() == "InProgress") {
	    $batch = $bac->getBatchInfo($job->getId(), $batch->getId());
	    sleep(3);
	}

	if($batch->getState() == "Failed") {
		logger(4, "ERROR: ".$batch->getStateMessage());
	}

	logger(3, 'Job completed.');
	try {
		logger(3, 'Retrieving list of results...');
		$resultList = $bac->getBatchResultList($job->getId(), $batch->getId());
		$resultId = $resultList[0];
		logger(3, 'Retrieving result...');
		$results = $bac->getBatchResult($job->getId(), $batch->getId(), $resultId);
//		print "== SOQL QUERY STRING == \n" . htmlspecialchars($soql) . "\n\n";
//		print "== QUERY RESULTS == \n" . print_r($results, true) . "\n\n";
//		print "== CLIENT LOGS == \n" . $bac->getLogs() . "\n\n";
	} catch(Exception $e) {
		if(0 === strpos($e->getMessage(), 'No result-list'))
			return '';
		else
			throw $e;
	}
	return $results;
}

/**
 * Retrieve the LastModified time from a given MySQL table
 *
 * @return string 
 */
function driverMySQLGetLastModified($link, $objName, $lmField) {
	$sql = "SELECT {$lmField} FROM {$objName} ORDER BY {$lmField} DESC LIMIT 1";
	logger(2, 'LastModified SQL: '.$sql);
	$result = mysqli_query($link, $sql);
	if(!$result) { logger(2, 'ERROR: '.mysqli_error($link)); }
	$row = mysqli_fetch_array($result);
	$last_modified = $row[0];
	return $last_modified;
}

/**
 * Returns CSV output from a MySQL query
 *
 * @return string 
 */
function driverMySQLGetData($link, $objName, $fields, $lmField, $lm) {
	$selectFields = join(",", $fields);
	$mylm = $lm->format('Y-m-d H:i:s');
	$sql = "SELECT {$selectFields} FROM {$objName} WHERE {$lmField} > '{$mylm}' ORDER BY {$lmField} DESC";
	print $sql."\n";

	$results = array();

	$stmt = mysqli_query($link, $sql);
	if(!$stmt) { logger(2, 'ERROR: '.mysqli_error($link)); }
	$records = 0;
	while($result = mysqli_fetch_array($stmt, MYSQLI_ASSOC)) {
		// The dict contains two arrays - src (the original data), and dst (which will
		// be populated later, and become the transformed data, ready for insertion)
		array_push($results, array('src' => $result, 'dst' => NULL));
		$records++;
	}
	logger(2, "Retrieved {$records} records");
	return $results;
}

require_once('soapclient/SforcePartnerClient.php');
require_once('bulkapiclient/BulkApiClient.php');

logger(0, 'Accordion starting...');

/* Configuration Loader */
logger(0, 'Accordion loading configuration...');
$json_raw = file_get_contents('accordion.cfg');
$cfg = json_decode($json_raw);
$objs = $cfg->{'objects'};

define("USERNAME", $cfg->{'sf'}->{'username'});
define("PASSWORD", $cfg->{'sf'}->{'password'});
define("SECURITY_TOKEN", $cfg->{'sf'}->{'security_token'});

define("MYSQL_HOST", $cfg->{'db'}->{'host'});
define("MYSQL_USERNAME", $cfg->{'db'}->{'username'});
define("MYSQL_PASSWORD", $cfg->{'db'}->{'password'});
define("MYSQL_DATABASE", $cfg->{'db'}->{'database'});

/* MySQL Database Connection */
logger(0, 'Connecting to MySQL...');
$link = mysqli_connect(MYSQL_HOST, MYSQL_USERNAME, MYSQL_PASSWORD, MYSQL_DATABASE);
if(mysqli_connect_errno()) { die("Failed to connect to MySQL: ".mysqli_connect_error()."\n"); }

/* Force.com API Connection */
logger(0, 'Connecting to Salesforce...');
$sf = new SforcePartnerClient();
$sf->createConnection("partner.wsdl.xml");
$sf->login(USERNAME, PASSWORD.SECURITY_TOKEN);
<commit_msg>Throw error on config JSON parse error<commit_after><?php

/*
	Accord-ion (c) CloudGarage Asia 2013

	Uni-directional synchronisation of MySQL to Salesforce and/or Salesforce to MySQL.

	Usage:
		php accordion.php epoch
                                    |_ Epoch (seconds) to retrieve rows based on their Last Modified timestamp

	Author(s):
		<NAME> <<EMAIL>>
	
*/

date_default_timezone_set('Hongkong');

function logger($indent, $msgs) {
	if($msgs != '') 
		foreach(explode(PHP_EOL, $msgs) as $msg)
			print date("d/m/Y H:i:s", time()).' - '.str_repeat("  ", $indent).$msg."\n";
}

/**
 * Retrieve the LastModified time from a given MySQL table
 *
 * @return string 
 */
function driverSalesforceGetLastModified($sf, $bac, $objName) {
	$soql = "SELECT LastModifiedDate FROM {$objName} ORDER BY LastModifiedDate DESC LIMIT 1";
	$response = $sf->query($soql);
	$result = $response->current()->LastModifiedDate;
	preg_match('/(\d+)-(\d+)-(\d+)T(\d+):(\d+):(\d+)\.\d+Z/', $result, $matches);
	$dt = new DateTime();
	$dt->setDate($matches[0], $matches[1], $matches[2]);
	$dt->setTime($matches[3], $matches[4], $matches[5]);
	return $dt;
}

/**
 * Utilises the Force.com Bulk API to perform SOQL queries, returning CSV output.
 *
 * @return string 
 */
function driverSalesforceGetData($sf, $bac, $objName, $fields, $lm) {
	$selectFields = join(",", $fields);
	$ts = $lm->format("Y-m-d\TH:i:s.000\Z");
	$soql = "SELECT {$selectFields} FROM {$objName} "; /*WHERE LastModifiedDate > '{$ts}' ORDER BY LastModifiedDate DESC"; */
	logger(3, 'SOQL:'.$soql);

	$job = new JobInfo();
	$job->setObject($objName);
	$job->setOperation("query");
	$job->setContentType("CSV");
	$job->setConcurrencyMode("Parallel");

	$job = $bac->createJob($job);
	$batch = $bac->createBatch($job, $soql);
	$bac->updateJobState($job->getId(), "Closed");

	logger(3, 'Waiting for query job completion...');
	while($batch->getState() == "Queued" || $batch->getState() == "InProgress") {
	    $batch = $bac->getBatchInfo($job->getId(), $batch->getId());
	    sleep(3);
	}

	if($batch->getState() == "Failed") {
		logger(4, "ERROR: ".$batch->getStateMessage());
	}

	logger(3, 'Job completed.');
	try {
		logger(3, 'Retrieving list of results...');
		$resultList = $bac->getBatchResultList($job->getId(), $batch->getId());
		$resultId = $resultList[0];
		logger(3, 'Retrieving result...');
		$results = $bac->getBatchResult($job->getId(), $batch->getId(), $resultId);
//		print "== SOQL QUERY STRING == \n" . htmlspecialchars($soql) . "\n\n";
//		print "== QUERY RESULTS == \n" . print_r($results, true) . "\n\n";
//		print "== CLIENT LOGS == \n" . $bac->getLogs() . "\n\n";
	} catch(Exception $e) {
		if(0 === strpos($e->getMessage(), 'No result-list'))
			return '';
		else
			throw $e;
	}
	return $results;
}

/**
 * Retrieve the LastModified time from a given MySQL table
 *
 * @return string 
 */
function driverMySQLGetLastModified($link, $objName, $lmField) {
	$sql = "SELECT {$lmField} FROM {$objName} ORDER BY {$lmField} DESC LIMIT 1";
	logger(2, 'LastModified SQL: '.$sql);
	$result = mysqli_query($link, $sql);
	if(!$result) { logger(2, 'ERROR: '.mysqli_error($link)); }
	$row = mysqli_fetch_array($result);
	$last_modified = $row[0];
	return $last_modified;
}

/**
 * Returns CSV output from a MySQL query
 *
 * @return string 
 */
function driverMySQLGetData($link, $objName, $fields, $lmField, $lm) {
	$selectFields = join(",", $fields);
	$mylm = $lm->format('Y-m-d H:i:s');
	$sql = "SELECT {$selectFields} FROM {$objName} WHERE {$lmField} > '{$mylm}' ORDER BY {$lmField} DESC";
	print $sql."\n";

	$results = array();

	$stmt = mysqli_query($link, $sql);
	if(!$stmt) { logger(2, 'ERROR: '.mysqli_error($link)); }
	$records = 0;
	while($result = mysqli_fetch_array($stmt, MYSQLI_ASSOC)) {
		// The dict contains two arrays - src (the original data), and dst (which will
		// be populated later, and become the transformed data, ready for insertion)
		array_push($results, array('src' => $result, 'dst' => NULL));
		$records++;
	}
	logger(2, "Retrieved {$records} records");
	return $results;
}

require_once('soapclient/SforcePartnerClient.php');
require_once('bulkapiclient/BulkApiClient.php');

logger(0, 'Accordion starting...');

/* Configuration Loader */
logger(0, 'Accordion loading configuration...');
$json_raw = file_get_contents('accordion.cfg');
$cfg = json_decode($json_raw);
if(!$cfg) { die("Error parsing accordion.cfg!\n"); }
$objs = $cfg->{'objects'};

define("USERNAME", $cfg->{'sf'}->{'username'});
define("PASSWORD", $cfg->{'sf'}->{'password'});
define("SECURITY_TOKEN", $cfg->{'sf'}->{'security_token'});

define("MYSQL_HOST", $cfg->{'db'}->{'host'});
define("MYSQL_USERNAME", $cfg->{'db'}->{'username'});
define("MYSQL_PASSWORD", $cfg->{'db'}->{'password'});
define("MYSQL_DATABASE", $cfg->{'db'}->{'database'});

/* MySQL Database Connection */
logger(0, 'Connecting to MySQL...');
$link = mysqli_connect(MYSQL_HOST, MYSQL_USERNAME, MYSQL_PASSWORD, MYSQL_DATABASE);
if(mysqli_connect_errno()) { die("Failed to connect to MySQL: ".mysqli_connect_error()."\n"); }

/* Force.com API Connection */
logger(0, 'Connecting to Salesforce...');
$sf = new SforcePartnerClient();
$sf->createConnection("partner.wsdl.xml");
$sf->login(USERNAME, PASSWORD.SECURITY_TOKEN);

{'fraction_non_alphanumeric': 0.21596783457782884, 'fraction_numerical': 0.006509668772735976, 'mean_word_length': 4.846110800223839, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 111, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '598315', 'n_tokens_mistral': 3899, 'n_tokens_neox': 3783, 'n_words': 828}