Cavemanon/piratepoet
7
Fork 0
Code Pull Requests Actions Releases 16 Activity

Merge pull request #359 from mailpoet/import_batch_processing

Browse Source 
Import update
This commit is contained in:
Tautvidas Sipavičius
2016-02-23 13:04:48 +02:00
parent 3e7d1690bd 24b98a1154
commit a883e1176c
8 changed files with 305 additions and 5160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
assets/js/src/subscribers/importExport/import.js
View File

@ -6,7 +6,8 @@ define(
'mailpoet',
'handlebars',
'papaparse',
'select2'
'select2',
'asyncqueue'
],
function (
Backbone,
@ -14,12 +15,14 @@ define(
jQuery,
MailPoet,
Handlebars,
Papa
Papa,
AsyncQueue
) {
if (!jQuery('#mailpoet_subscribers_import').length) {
return;
}
jQuery(document).ready(function () {
var noticeTimeout = 3000;
jQuery('input[name="select_method"]').attr('checked', false);
// configure router
router = new (Backbone.Router.extend({
@ -124,7 +127,7 @@ define(
var pasteSize = encodeURI(pasteInputElement.val()).split(/%..|./).length - 1;
if (pasteSize > maxPostSizeBytes) {
MailPoet.Notice.error(MailPoetI18n.maxPostSizeNotice, {
timeout: 3000,
timeout: noticeTimeout,
});
return;
}
@ -144,7 +147,7 @@ define(
if (ext === null || ext[1].toLowerCase() !== 'csv') {
this.value = '';
MailPoet.Notice.error(MailPoetI18n.wrongFileFormat, {
timeout: 3000,
timeout: noticeTimeout,
});
}
@ -194,7 +197,7 @@ define(
if (response.result === false) {
MailPoet.Notice.hide();
MailPoet.Notice.error(response.errors, {
timeout: 3000,
timeout: noticeTimeout,
});
jQuery('.mailpoet_mailchimp-key-status')
.removeClass()
@ -219,7 +222,7 @@ define(
MailPoet.Modal.loading(false);
MailPoet.Notice.error(
MailPoetI18n.serverError + error.statusText.toLowerCase() + '.', {
timeout: 3000,
timeout: noticeTimeout,
}
);
});
@ -246,7 +249,7 @@ define(
else {
MailPoet.Notice.hide();
MailPoet.Notice.error(response.errors, {
timeout: 3000,
timeout: noticeTimeout,
});
}
MailPoet.Modal.loading(false);
@ -254,7 +257,7 @@ define(
MailPoet.Modal.loading(false);
MailPoet.Notice.error(
MailPoetI18n.serverError + result.statusText.toLowerCase() + '.', {
timeout: 3000,
timeout: noticeTimeout,
}
);
});
@ -346,7 +349,7 @@ define(
error: function () {
MailPoet.Notice.hide();
MailPoet.Notice.error(MailPoetI18n.dataProcessingError, {
timeout: 3000,
timeout: noticeTimeout,
});
},
complete: function (CSV) {
@ -430,7 +433,7 @@ define(
errorNotice = errorNotice.replace('[link]', MailPoetI18n.csvKBLink);
errorNotice = errorNotice.replace('[/link]', '</a>');
MailPoet.Notice.error(errorNotice, {
timeout: 3000,
timeout: noticeTimeout,
});
}
}
@ -561,7 +564,7 @@ define(
if (!segmentSelectionNotice.length) {
MailPoet.Notice.error(MailPoetI18n.segmentSelectionRequired, {
static: true,
timeout: 3000,
timeout: noticeTimeout,
scroll: true,
id: 'segmentSelection',
hideClose: true
@ -640,7 +643,7 @@ define(
MailPoet.Modal.close();
MailPoet.Notice.error(
MailPoetI18n.segmentCreateError + response.message + '.', {
timeout: 3000,
timeout: noticeTimeout,
}
);
}
@ -649,7 +652,7 @@ define(
MailPoet.Modal.close();
MailPoet.Notice.error(
MailPoetI18n.serverError + error.statusText.toLowerCase() + '.', {
timeout: 3000
timeout: noticeTimeout
}
);
});
@ -856,7 +859,7 @@ define(
}
else {
MailPoet.Notice.error(MailPoetI18n.customFieldCreateError, {
timeout: 3000,
timeout: noticeTimeout,
});
}
MailPoet.Modal.loading(false);
@ -865,7 +868,7 @@ define(
MailPoet.Modal.loading(false);
MailPoet.Notice.error(
MailPoetI18n.serverError + error.statusText.toLowerCase() + '.', {
timeout: 3000,
timeout: noticeTimeout,
}
);
});
@ -930,7 +933,7 @@ define(
if (!jQuery('[data-id="notice_invalidEmail"]').length) {
MailPoet.Notice.error(MailPoetI18n.columnContainsInvalidElement, {
static: true,
timeout: 3000,
timeout: noticeTimeout,
scroll: true,
hideClose: true,
id: 'invalidEmail'
@ -1010,7 +1013,7 @@ define(
if (preventNextStep && !jQuery('.mailpoet_invalidDate').length) {
MailPoet.Notice.error(MailPoetI18n.columnContainsInvalidDate, {
static: true,
timeout: 3000,
timeout: noticeTimeout,
scroll: true,
hideClose: true,
id: 'invalidDate'
@ -1050,64 +1053,98 @@ define(
}
MailPoet.Modal.loading(true);
var subscribers = {};
var columns = {},
queue = new jQuery.AsyncQueue(),
batchNumber = 0,
batchSize = 500,
timestamp = Date.now() / 1000,
subscribers = [],
importResults = {
'created': 0,
'updated': 0,
'errors': [],
'segments': []
},
splitSubscribers = function (subscribers, size) {
return subscribers.reduce(function (res, item, index) {
if (index % size === 0) {
res.push([]);
}
res[res.length - 1].push(item);
return res;
}, []);
},
subscribers = splitSubscribers(importData.step1.subscribers, batchSize);
_.each(jQuery('select.mailpoet_subscribers_column_data_match'),
function (column, index) {
var columnId = jQuery(column).data('column-id');
if (columnId === 'ignore') {
return;
}
subscribers[columnId] = [];
_.each(importData.step1.subscribers, function (subsciber) {
subscribers[columnId].push(
_.chain(subsciber)
.pick(index)
.toArray()
.flatten()
.value()
);
});
subscribers[columnId] = _.flatten(subscribers[columnId]);
});
function (column, columnIndex) {
var columnId = jQuery(column).data('column-id');
if (columnId === 'ignore') {
return;
}
columns[columnId] = columnIndex;
});
MailPoet.Ajax.post({
endpoint: 'ImportExport',
action: 'processImport',
data: JSON.stringify({
subscribers: subscribers,
segments: segmentSelectElement.val(),
updateSubscribers: (jQuery(':radio[name="subscriber_update_option"]:checked').val() === 'yes') ? true : false
_.each(subscribers, function () {
queue.add(function (queue) {
queue.pause();
MailPoet.Ajax
.post({
endpoint: 'ImportExport',
action: 'processImport',
data: JSON.stringify({
columns: columns,
subscribers: subscribers[batchNumber],
timestamp: timestamp,
segments: segmentSelectElement.val(),
updateSubscribers: (jQuery(':radio[name="subscriber_update_option"]:checked').val() === 'yes') ? true : false
})
})
.done(function (response) {
if (response.result === false) {
importResults.errors.push(response.errors);
} else {
importResults.created = response.data.created;
importResults.updated = response.data.updated;
importResults.segments = response.data.segments;
}
queue.run();
})
.error(function (error) {
importResults.errors.push(
MailPoetI18n.serverError + error.statusText.toLowerCase() + '.'
);
queue.run();
});
batchNumber++;
})
}).done(function (response) {
});
queue.run();
queue.onComplete(function () {
MailPoet.Modal.loading(false);
if (response.result === false) {
MailPoet.Notice.error(response.errors, {
timeout: 3000,
});
} else {
mailpoetSegments = response.data.segments;
response.data.segments = _.map(segmentSelectElement.select2('data'),
function (data) {
return data.name;
});
importData.step2 = response.data;
if (importResults.errors.length > 0 && !importResults.updated && !importResults.created) {
MailPoet.Notice.error(_.flatten(importResults.errors), {
timeout: noticeTimeout,
}
);
}
else {
mailpoetSegments = importResults.segments;
importResults.segments = _.map(segmentSelectElement.select2('data'),
function (data) {
return data.name;
});
importData.step2 = importResults;
enableSegmentSelection(mailpoetSegments);
router.navigate('step3', {trigger: true});
}
}).error(function (error) {
MailPoet.Modal.loading(false);
MailPoet.Notice.error(
MailPoetI18n.serverError + error.statusText.toLowerCase() + '.', {
timeout: 3000,
}
);
});
});
filterSubscribers();
enableSegmentSelection(mailpoetSegments);
});
router.on('route:step3', function () {
@ -1118,6 +1155,12 @@ define(
showCurrentStep();
if (importData.step2.errors.length > 0) {
MailPoet.Notice.error(_.flatten(importData.step2.errors), {
timeout: noticeTimeout,
});
}
// display statistics
var subscribersDataImportResultsTemplate =
Handlebars

79
assets/js/src/vendor/jquery.asyncqueue.js vendored Normal file
View File

@ -0,0 +1,79 @@
/*
* This file is part of the jquery plugin "asyncQueue".
*
* (c) Sebastien Roch <roch.sebastien@gmail.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
(function($){
$.AsyncQueue = function() {
var that = this,
queue = [],
failureFunc,
completeFunc,
paused = false,
lastCallbackData,
_run;
_run = function() {
var f = queue.shift();
if (f) {
f.apply(that, [that]);
if (paused === false) {
_run();
}
} else {
if(completeFunc){
completeFunc.apply(that);
}
}
}
this.onFailure = function(func) {
failureFunc = func;
}
this.onComplete = function(func) {
completeFunc = func;
}
this.add = function(func) {
queue.push(func);
return this;
}
this.storeData = function(dataObject) {
lastCallbackData = dataObject;
return this;
}
this.lastCallbackData = function () {
return lastCallbackData;
}
this.run = function() {
paused = false;
_run();
}
this.pause = function () {
paused = true;
return this;
}
this.failure = function() {
paused = true;
if (failureFunc) {
var args = [that];
for(i = 0; i < arguments.length; i++) {
args.push(arguments[i]);
}
failureFunc.apply(that, args);
}
}
return this;
}
})(jQuery);

4
lib/Models/Subscriber.php
View File

@ -509,7 +509,7 @@ class Subscriber extends Model {
);
}
static function updateMultiple($columns, $subscribers, $currentTime = false) {
static function updateMultiple($columns, $subscribers, $updated_at = false) {
$ignoreColumnsOnUpdate = array(
'email',
'created_at'
@ -549,7 +549,7 @@ class Subscriber extends Model {
return self::rawExecute(
'UPDATE `' . self::$_table . '` ' .
'SET ' . implode(', ', $sql('statement')) . ' '.
(($currentTime) ? ', updated_at = "' . $currentTime . '" ' : '') .
(($updated_at) ? ', updated_at = "' . $updated_at . '" ' : '') .
'WHERE email IN ' .
'(' . rtrim(str_repeat('?,', count($subscribers)), ',') . ')',
array_merge(

70
lib/Subscribers/ImportExport/Import/Import.php
View File

@ -14,21 +14,26 @@ class Import {
public $subscriber_fields;
public $subscriber_custom_fields;
public $subscribers_count;
public $import_time;
public $created_at;
public $updated_at;
public $profiler_start;
public function __construct($data) {
$this->subscribers_data = $data['subscribers'];
$this->subscribers_data = $this->transformSubscribersData(
$data['subscribers'],
$data['columns']
);
$this->segments = $data['segments'];
$this->update_subscribers = $data['updateSubscribers'];
$this->subscriber_fields = $this->getSubscriberFields(
array_keys($this->subscribers_data)
array_keys($data['columns'])
);
$this->subscriber_custom_fields = $this->getCustomSubscriberFields(
array_keys($this->subscribers_data)
array_keys($data['columns'])
);
$this->subscribers_count = count(reset($this->subscribers_data));
$this->import_time = date('Y-m-d H:i:s');
$this->created_at = date('Y-m-d H:i:s', (int) $data['timestamp']);
$this->updated_at = date('Y-m-d H:i:s', (int) $data['timestamp'] + 1);
$this->profiler_start = microtime(true);
}
@ -63,14 +68,6 @@ class Import {
$subscriber_fields,
$subscriber_custom_fields
);
if($created_subscribers) {
// subtract added from updated subscribers when DB operation takes <1s
$updated_subscribers = array_diff_key(
$updated_subscribers,
$created_subscribers,
$subscriber_custom_fields
);
}
}
} catch(\PDOException $e) {
return array(
@ -90,14 +87,22 @@ class Import {
);
}
function transformSubscribersData($subscribers, $columns) {
foreach($columns as $column => $index) {
$transformed_subscribers[$column] = Helpers::arrayColumn($subscribers, $index);
}
return $transformed_subscribers;
}
function filterExistingAndNewSubscribers($subscribers_data) {
$chunk_size = 200;
$existing_records = array_filter(
array_map(function($subscriber_emails) {
return Subscriber::selectMany(array('email'))
->whereIn('email', $subscriber_emails)
->whereNull('deleted_at')
->findArray();
}, array_chunk($subscribers_data['email'], 200))
}, array_chunk($subscribers_data['email'], $chunk_size))
);
if(!$existing_records) {
return array(
@ -144,17 +149,19 @@ class Import {
}
function deleteExistingTrashedSubscribers($subscribers_data) {
$chunk_size = 200;
$existing_trashed_records = array_filter(
array_map(function($subscriber_emails) {
return Subscriber::selectMany(array('id'))
->whereIn('email', $subscriber_emails)
->whereNotNull('deleted_at')
->findArray();
}, array_chunk($subscribers_data['email'], 200))
}, array_chunk($subscribers_data['email'], $chunk_size))
);
if(!$existing_trashed_records) return;
$existing_trashed_records = Helpers::flattenArray($existing_trashed_records);
foreach(array_chunk($existing_trashed_records, 200) as $subscriber_ids) {
foreach(array_chunk($existing_trashed_records, $chunk_size) as
$subscriber_ids) {
Subscriber::whereIn('id', $subscriber_ids)
->deleteMany();
SubscriberSegment::whereIn('subscriber_id', $subscriber_ids)
@ -163,7 +170,8 @@ class Import {
}
function extendSubscribersAndFields($subscribers_data, $subscriber_fields) {
$subscribers_data['created_at'] = $this->filterSubscriberCreatedAtDate();
$subscribers_data['created_at'] =
array_fill(0, $this->subscribers_count, $this->created_at);
$subscriber_fields[] = 'created_at';
return array(
$subscribers_data,
@ -191,10 +199,6 @@ class Import {
);
}
function filterSubscriberCreatedAtDate() {
return array_fill(0, $this->subscribers_count, $this->import_time);
}
function filterSubscriberStatus($subscribers_data, $subscriber_fields) {
if(!in_array('status', $subscriber_fields)) {
$subscribers_data['status'] =
@ -249,14 +253,14 @@ class Import {
$subscriber_fields,
$subscriber_custom_fields
) {
$chunk_size = 100;
$subscribers_count = count(reset($subscribers_data)) - 1;
$subscribers = array_map(function($index) use ($subscribers_data, $subscriber_fields) {
return array_map(function($field) use ($index, $subscribers_data) {
return $subscribers_data[$field][$index];
}, $subscriber_fields);
}, range(0, $subscribers_count));
$import_time = ($action === 'update') ? date('Y-m-d H:i:s') : $this->import_time;
foreach(array_chunk($subscribers, 100) as $data) {
foreach(array_chunk($subscribers, $chunk_size) as $data) {
if($action == 'create') {
Subscriber::createMultiple(
$subscriber_fields,
@ -267,18 +271,20 @@ class Import {
Subscriber::updateMultiple(
$subscriber_fields,
$data,
$import_time
$this->updated_at
);
}
}
$result = Helpers::arrayColumn( // return id=>email array of results
Subscriber::selectMany(
array(
'id',
'email'
))
->where(($action === 'create') ? 'created_at' : 'updated_at', $import_time)
->findArray(),
$query = Subscriber::selectMany(
array(
'id',
'email'
));
$query = ($action === 'update') ?
$query->where('updated_at', $this->updated_at) :
$query->where('created_at', $this->created_at);
$result = Helpers::arrayColumn(
$query->findArray(),
'email', 'id'
);
if($subscriber_custom_fields) {

135
tests/unit/Subscribers/ImportExport/Import/ImportCest.php
View File

@ -7,25 +7,33 @@ use MailPoet\Subscribers\ImportExport\Import\Import;
use MailPoet\Util\Helpers;
class ImportCest {
function __construct() {
$this->JSON_data = json_decode(file_get_contents(dirname(__FILE__) . '/ImportTestData.json'), true);
$this->subscribers_data = array(
'first_name' => array(
'Adam',
'Mary'
function _before() {
$this->data = array(
'subscribers' => array(
array(
'Adam',
'Smith',
'adam@smith.com',
'France'
),
array(
'Mary',
'Jane',
'mary@jane.com',
'Brazil'
)
),
'last_name' => array(
'Smith',
'Jane'
'columns' => array(
'first_name' => 0,
'last_name' => 1,
'email' => 2,
777 => 3
),
'email' => array(
'adam@smith.com',
'mary@jane.com'
'segments' => array(
195
),
777 => array(
'France',
'Brazil'
)
'timestamp' => time(),
'updateSubscribers' => true
);
$this->subscriber_fields = array(
'first_name',
@ -34,25 +42,36 @@ class ImportCest {
);
$this->segments = range(0, 1);
$this->subscriber_custom_fields = array(777);
$this->import = new Import($this->JSON_data);
$this->import = new Import($this->data);
$this->subscribers_data = $this->import->transformSubscribersData(
$this->data['subscribers'],
$this->data['columns']
);
}
function itCanConstruct() {
expect($this->import->subscribers_data)->equals($this->JSON_data['subscribers']);
expect($this->import->segments)->equals($this->JSON_data['segments']);
expect(is_array($this->import->subscribers_data))->true();
expect($this->import->segments)->equals($this->data['segments']);
expect(is_array($this->import->subscriber_fields))->true();
expect(is_array($this->import->subscriber_custom_fields))->true();
expect($this->import->subscribers_count)->equals(
count($this->JSON_data['subscribers']['email'])
);
expect(
preg_match(
'/\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}/',
$this->import->import_time)
)->equals(1);
expect($this->import->subscribers_count)->equals(2);
expect($this->import->created_at)->notEmpty();
expect($this->import->updated_at)->notEmpty();
}
function itCanTransformSubscribers() {
expect($this->import->subscribers_data['first_name'][0])
->equals($this->data['subscribers'][0][0]);
expect($this->import->subscribers_data['last_name'][0])
->equals($this->data['subscribers'][0][1]);
expect($this->import->subscribers_data['email'][0])
->equals($this->data['subscribers'][0][2]);
expect($this->import->subscribers_data['777'][0])
->equals($this->data['subscribers'][0][3]);
}
function itCanFilterExistingAndNewSubscribers() {
$subscribers_data = $this->subscribers_data;
$subscriber = Subscriber::create();
$subscriber->hydrate(
array(
@ -62,10 +81,10 @@ class ImportCest {
));
$subscriber->save();
list($existing, $new) = $this->import->filterExistingAndNewSubscribers(
$this->subscribers_data
$subscribers_data
);
expect($existing['email'][0])->equals($this->subscribers_data['email'][0]);
expect($new['email'][0])->equals($this->subscribers_data['email'][1]);
expect($existing['email'][0])->equals($subscribers_data['email'][0]);
expect($new['email'][0])->equals($subscribers_data['email'][1]);
}
function itCanExtendSubscribersAndFields() {
@ -106,17 +125,17 @@ class ImportCest {
}
function itCanFilterSubscriberStatus() {
$subscibers_data = $this->subscribers_data;
$subscribers_data = $this->subscribers_data;
$subscriber_fields = $this->subscriber_fields;
list($subscibers_data, $subsciber_fields) =
$this->import->filterSubscriberStatus($subscibers_data, $subscriber_fields);
list($subscribers_data, $subsciber_fields) =
$this->import->filterSubscriberStatus($subscribers_data, $subscriber_fields);
// subscribers' status was set to "subscribed" & status column was added
// to subscribers fields
expect(array_pop($subsciber_fields))->equals('status');
expect($subscibers_data['status'][0])->equals('subscribed');
expect(count($subscibers_data['status']))->equals(2);
expect($subscribers_data['status'][0])->equals('subscribed');
expect(count($subscribers_data['status']))->equals(2);
$subscriber_fields[] = 'status';
$subscibers_data = array(
$subscribers_data = array(
'status' => array(
#subscribed
'subscribed',
@ -135,9 +154,9 @@ class ImportCest {
'false'
),
);
list($subscibers_data, $subsciber_fields) =
$this->import->filterSubscriberStatus($subscibers_data, $subscriber_fields);
expect($subscibers_data)->equals(
list($subscribers_data, $subsciber_fields) =
$this->import->filterSubscriberStatus($subscribers_data, $subscriber_fields);
expect($subscribers_data)->equals(
array(
'status' => array(
'subscribed',
@ -185,8 +204,8 @@ class ImportCest {
$subscribers_data = $this->subscribers_data;
$subscriber_fields = $this->subscriber_fields;
$subscribers_data['deleted_at'] = array(
null,
date('Y-m-d H:i:s')
null,
date('Y-m-d H:i:s')
);
$subscriber_fields[] = 'deleted_at';
$this->import->createOrUpdateSubscribers(
@ -255,6 +274,7 @@ class ImportCest {
->equals($subscribers_data[777][1]);
}
function itCanaddSubscribersToSegments() {
$subscribers_data = $this->subscribers_data;
$this->import->createOrUpdateSubscribers(
@ -293,26 +313,27 @@ class ImportCest {
);
}
function itCanProcess() {
$import = clone($this->import);
$result = $import->process();
expect($result['data']['created'])->equals(997);
function itCanUpdateSubscribers() {
$result = $this->import->process();
expect($result['data']['updated'])->equals(0);
$result = $import->process();
expect($result['data']['created'])->equals(0);
expect($result['data']['updated'])->equals(997);
Subscriber::where('email', 'mbanks4@blinklist.com')
$result = $this->import->process();
expect($result['data']['updated'])->equals(2);
$this->import->update_subscribers = false;
$result = $this->import->process();
expect($result['data']['updated'])->equals(0);
}
function itCanProcess() {
$result = $this->import->process();
expect($result['data']['created'])->equals(2);
Subscriber::where('email', 'mary@jane.com')
->findOne()
->delete();
// TODO: find a more elegant way to test this
$import->import_time = date('Y-m-d 12:i:s');
$result = $import->process();
$timestamp = time() + 1;
$this->import->created_at = date('Y-m-d H:i:s', $timestamp);
$this->import->updated_at = date('Y-m-d H:i:s', $timestamp + 1);
$result = $this->import->process();
expect($result['data']['created'])->equals(1);
expect($result['data']['updated'])->equals(996);
$import->update_subscribers = false;
$result = $import->process();
expect($result['data']['created'])->equals(0);
expect($result['data']['updated'])->equals(0);
}
function _after() {

5003
tests/unit/Subscribers/ImportExport/Import/ImportTestData.json
View File

File diff suppressed because it is too large Load Diff

2
views/subscribers/importExport/import/step1.html
View File

@ -60,8 +60,6 @@
</th>
<td>
<input type="file" id="file_local" accept=".csv" />
&nbsp;
<%= __('total max upload file size : %s')|format(maxPostSize) %>
</td>
</tr>
</tbody>

5
webpack.config.js
View File

@ -17,7 +17,7 @@ baseConfig = {
resolve: {
modulesDirectories: [
'node_modules',
'assets/js/src'
'assets/js/src',
],
alias: {
'handlebars': 'handlebars/dist/handlebars.js',
@ -30,7 +30,8 @@ baseConfig = {
'filesaver$': 'filesaver/FileSaver.js',
'papaparse': 'papaparse/papaparse.min.js',
'helpscout': 'helpscout.js',
'html2canvas': 'html2canvas/dist/html2canvas.js'
'html2canvas': 'html2canvas/dist/html2canvas.js',
'asyncqueue': 'vendor/jquery.asyncqueue.js'
},
},
node: {