| @@ -2,18 +2,24 @@ | ||
| 2 | 2 | |
| 3 | 3 | namespace ImportWP\Common\Importer; |
| 4 | 4 | |
| 5 | 5 | use ImportWP\Common\Importer\ConfigInterface; |
| 6 | +use ImportWP\Common\Importer\Exception\FileException; | |
| 7 | +use ImportWP\Common\Importer\Exception\MapperException; | |
| 8 | +use ImportWP\Common\Importer\Exception\ParserException; | |
| 9 | +use ImportWP\Common\Importer\Exception\RecordUpdatedSkippedException; | |
| 6 | 10 | use ImportWP\Common\Importer\File\CSVFile; |
| 11 | +use ImportWP\Common\Importer\File\JSONFile; | |
| 7 | 12 | use ImportWP\Common\Importer\File\XMLFile; |
| 8 | 13 | use ImportWP\Common\Importer\MapperInterface; |
| 9 | 14 | use ImportWP\Common\Importer\Parser\CSVParser; |
| 15 | +use ImportWP\Common\Importer\Parser\JSONParser; | |
| 10 | 16 | use ImportWP\Common\Importer\Parser\XMLParser; |
| 11 | 17 | use ImportWP\Common\Importer\ParserInterface; |
| 12 | 18 | use ImportWP\Common\Importer\State\ImporterState; |
| 13 | 19 | use ImportWP\Common\Properties\Properties; |
| 14 | -use ImportWP\Common\Runner\ImporterRunner; | |
| 15 | 20 | use ImportWP\Common\Runner\ImporterRunnerState; |
| 21 | +use ImportWP\Common\Util\Logger; | |
| 16 | 22 | use ImportWP\Common\Util\Util; |
| 17 | 23 | use ImportWP\Container; |
| 18 | 24 | |
| 19 | 25 | class Importer |
| @@ -18,8 +24,12 @@ | ||
| 18 | 24 | |
| 19 | 25 | class Importer |
| 20 | 26 | { |
| 21 | 27 | /** |
| 28 | + * @var int | |
| 29 | + */ | |
| 30 | + protected $memory_limit; | |
| 31 | + /** | |
| 22 | 32 | * @var ConfigInterface $config |
| 23 | 33 | */ |
| 24 | 34 | public $config; |
| 25 | 35 | |
| @@ -123,8 +133,23 @@ | ||
| 123 | 133 | return $this; |
| 124 | 134 | } |
| 125 | 135 | |
| 126 | 136 | /** |
| 137 | + * Load JSON File | |
| 138 | + * | |
| 139 | + * @param string $file_path | |
| 140 | + * | |
| 141 | + * @return $this | |
| 142 | + */ | |
| 143 | + public function jsonFile($file_path) | |
| 144 | + { | |
| 145 | + $file = new JSONFile($file_path, $this->config); | |
| 146 | + $this->parser = new JSONParser($file); | |
| 147 | + | |
| 148 | + return $this; | |
| 149 | + } | |
| 150 | + | |
| 151 | + /** | |
| 127 | 152 | * Set record to start importing from |
| 128 | 153 | * |
| 129 | 154 | * @param int $start |
| 130 | 155 | */ |
| @@ -214,13 +239,13 @@ | ||
| 214 | 239 | */ |
| 215 | 240 | public function import($id, $user, $importer_state) |
| 216 | 241 | { |
| 217 | 242 | if ($this->parser == null) { |
| 218 | - throw new \Exception("Parser Not Loaded."); | |
| 243 | + throw new \Exception(__("Parser Not Loaded.", 'jc-importer')); | |
| 219 | 244 | } |
| 220 | 245 | |
| 221 | 246 | if ($this->mapper == null) { |
| 222 | - throw new \Exception("Mapper Not Loaded."); | |
| 247 | + throw new \Exception(__("Mapper Not Loaded.", 'jc-importer')); | |
| 223 | 248 | } |
| 224 | 249 | |
| 225 | 250 | $this->mapper->setup(); |
| 226 | 251 | |
| @@ -225,8 +250,10 @@ | ||
| 225 | 250 | $this->mapper->setup(); |
| 226 | 251 | |
| 227 | 252 | $this->register_shutdown($importer_state); |
| 228 | 253 | |
| 254 | + $this->disable_caching(); | |
| 255 | + | |
| 229 | 256 | /** |
| 230 | 257 | * @var Util $util |
| 231 | 258 | */ |
| 232 | 259 | $util = Container::getInstance()->get('util'); |
| @@ -231,20 +258,360 @@ | ||
| 231 | 258 | */ |
| 232 | 259 | $util = Container::getInstance()->get('util'); |
| 233 | 260 | $util->set_time_limit(); |
| 234 | 261 | |
| 262 | + // TODO: | |
| 263 | + // $runner = new ImporterRunner($properties, $this); | |
| 264 | + // $runner->process($id, $user, $importer_state); | |
| 265 | + $this->process_chunk($id, $user, $importer_state); | |
| 266 | + | |
| 267 | + $this->mapper->teardown(); | |
| 268 | + $this->unregister_shutdown(); | |
| 269 | + } | |
| 270 | + | |
| 271 | + protected function disable_caching() | |
| 272 | + { | |
| 273 | + if (!defined('WP_IMPORTING')) { | |
| 274 | + define('WP_IMPORTING', true); | |
| 275 | + } | |
| 276 | + | |
| 277 | + // WP Rocket Integration | |
| 278 | + add_filter('rocket_is_importing', '__return_true'); | |
| 279 | + } | |
| 280 | + | |
| 281 | + protected function process_chunk($id, $user, $importer_state) | |
| 282 | + { | |
| 283 | + // Introduce new running state, to stop cron running duplicates | |
| 284 | + $importer_state->populate([ | |
| 285 | + 'status' => 'processing' | |
| 286 | + ]); | |
| 287 | + ImporterState::set_state($id, $importer_state->get_raw()); | |
| 288 | + | |
| 235 | 289 | /** |
| 236 | 290 | * @var Properties $properties |
| 237 | 291 | */ |
| 238 | 292 | $properties = Container::getInstance()->get('properties'); |
| 293 | + $time_limit = $properties->get_setting('timeout'); | |
| 294 | + Logger::info('time_limit ' . $time_limit . 's'); | |
| 239 | 295 | |
| 240 | - $runner = new ImporterRunner($properties, $this); | |
| 241 | - $runner->process($id, $user, $importer_state); | |
| 296 | + $start = microtime(true); | |
| 297 | + $max_record_time = 0; | |
| 298 | + $memory_max_usage = 0; | |
| 242 | 299 | |
| 300 | + $progress = $importer_state->get_progress(); | |
| 301 | + $session = $importer_state->get_session(); | |
| 302 | + $max_total = $progress['end'] - 1; | |
| 303 | + $i = $progress['start'] + $progress['current_row'] - 1; | |
| 304 | + | |
| 305 | + // limit to max 20 rows per chunk | |
| 306 | + $i_max = $i + apply_filters('iwp/chunk_max_records', 20); | |
| 307 | + | |
| 308 | + while ( | |
| 309 | + $i < $max_total | |
| 310 | + && (!defined('REST_REQUEST') || !REST_REQUEST || $i < $i_max) | |
| 311 | + && ( | |
| 312 | + $time_limit === 0 || $this->has_enough_time($start, $time_limit, $max_record_time) | |
| 313 | + ) | |
| 314 | + && $this->has_enough_memory($memory_max_usage) | |
| 315 | + ) { | |
| 316 | + $i++; | |
| 317 | + | |
| 318 | + $flag = ImporterState::get_flag($id); | |
| 319 | + | |
| 320 | + if (ImporterState::is_paused($flag)) { | |
| 321 | + | |
| 322 | + $importer_state->populate([ | |
| 323 | + 'status' => 'paused' | |
| 324 | + ]); | |
| 325 | + | |
| 326 | + ImporterState::set_state($id, $importer_state->get_raw()); | |
| 327 | + Util::write_status_session_to_file($id, $importer_state); | |
| 328 | + return; | |
| 329 | + } | |
| 330 | + | |
| 331 | + if (ImporterState::is_cancelled($flag)) { | |
| 332 | + $importer_state->populate([ | |
| 333 | + 'status' => 'cancelled' | |
| 334 | + ]); | |
| 335 | + | |
| 336 | + ImporterState::set_state($id, $importer_state->get_raw()); | |
| 337 | + Util::write_status_session_to_file($id, $importer_state); | |
| 338 | + return; | |
| 339 | + } | |
| 340 | + | |
| 341 | + $stats = [ | |
| 342 | + 'inserts' => 0, | |
| 343 | + 'updates' => 0, | |
| 344 | + 'deletes' => 0, | |
| 345 | + 'skips' => 0, | |
| 346 | + 'errors' => 0, | |
| 347 | + ]; | |
| 348 | + | |
| 349 | + $record_time = microtime(true); | |
| 350 | + | |
| 351 | + if ($importer_state->get_section() === 'import') { | |
| 352 | + | |
| 353 | + /** | |
| 354 | + * @var ParsedData $data | |
| 355 | + */ | |
| 356 | + $data = null; | |
| 357 | + | |
| 358 | + $data_parser = new DataParser($this->getParser(), $this->getMapper(), $this->config->getData()); | |
| 359 | + | |
| 360 | + try { | |
| 361 | + | |
| 362 | + $data = $data_parser->get($i); | |
| 363 | + do_action('iwp/importer/before_row', $data); | |
| 364 | + | |
| 365 | + $skip_record = $this->filterRecords(); | |
| 366 | + $skip_record = apply_filters('iwp/importer/skip_record', $skip_record, $data, $this); | |
| 367 | + | |
| 368 | + if ($skip_record) { | |
| 369 | + | |
| 370 | + Logger::write('import -skip-record=' . $i); | |
| 371 | + | |
| 372 | + $stats['skips']++; | |
| 373 | + | |
| 374 | + // set data to null, to flag chunk as skipped | |
| 375 | + $message = apply_filters('iwp/status/record_skipped', "Skipped Record"); | |
| 376 | + Util::write_status_log_file_message($id, $session, $message, 'S', $progress['current_row']); | |
| 377 | + | |
| 378 | + $data = null; | |
| 379 | + } else { | |
| 380 | + | |
| 381 | + // import | |
| 382 | + $data = apply_filters('iwp/importer/before_mapper', $data, $this); | |
| 383 | + $data->map(); | |
| 384 | + | |
| 385 | + $unique_identifier_str = $this->get_unique_identifier_log_text(); | |
| 386 | + | |
| 387 | + if ($data->isInsert()) { | |
| 388 | + | |
| 389 | + Logger::write('import:' . $i . ' -success -insert'); | |
| 390 | + | |
| 391 | + $stats['inserts']++; | |
| 392 | + | |
| 393 | + $message = apply_filters('iwp/status/record_inserted', 'Record Inserted: #' . $data->getId(), $data->getId(), $data); | |
| 394 | + Util::write_status_log_file_message($id, $session, $message . $unique_identifier_str, 'S', $progress['current_row']); | |
| 395 | + } | |
| 396 | + | |
| 397 | + if ($data->isUpdate()) { | |
| 398 | + | |
| 399 | + Logger::write('import:' . $i . ' -success -update'); | |
| 400 | + | |
| 401 | + $stats['updates']++; | |
| 402 | + | |
| 403 | + $message = apply_filters('iwp/status/record_updated', 'Record Updated: #' . $data->getId(), $data->getId(), $data); | |
| 404 | + Util::write_status_log_file_message($id, $session, $message . $unique_identifier_str, 'S', $progress['current_row']); | |
| 405 | + } | |
| 406 | + } | |
| 407 | + } catch (RecordUpdatedSkippedException $e) { | |
| 408 | + | |
| 409 | + Logger::write('import:' . $i . ' -success -update -skipped="hash"'); | |
| 410 | + $stats['updates']++; | |
| 411 | + $message = 'Record Update Skipped: #' . $data->getId() . ' ' . $e->getMessage(); | |
| 412 | + $unique_identifier_str = $this->get_unique_identifier_log_text(); | |
| 413 | + | |
| 414 | + Util::write_status_log_file_message($id, $session, $message . $unique_identifier_str, 'S', $progress['current_row']); | |
| 415 | + } catch (ParserException $e) { | |
| 416 | + | |
| 417 | + $stats['errors']++; | |
| 418 | + Logger::error('import:' . $i . ' -parser-error=' . $e->getMessage()); | |
| 419 | + Util::write_status_log_file_message($id, $session, $e->getMessage(), 'E', $progress['current_row']); | |
| 420 | + } catch (MapperException $e) { | |
| 421 | + | |
| 422 | + $stats['errors']++; | |
| 423 | + Logger::error('import:' . $i . ' -mapper-error=' . $e->getMessage()); | |
| 424 | + Util::write_status_log_file_message($id, $session, $e->getMessage(), 'E', $progress['current_row']); | |
| 425 | + } catch (FileException $e) { | |
| 426 | + | |
| 427 | + $stats['errors']++; | |
| 428 | + Logger::error('import:' . $i . ' -file-error=' . $e->getMessage()); | |
| 429 | + Util::write_status_log_file_message($id, $session, $e->getMessage(), 'E', $progress['current_row']); | |
| 430 | + } | |
| 431 | + | |
| 432 | + do_action('iwp/importer/after_row'); | |
| 433 | + } elseif ($importer_state->get_section() === 'delete') { | |
| 434 | + | |
| 435 | + if ($this->getMapper()->permission() && $this->getMapper()->permission()->allowed_method('remove')) { | |
| 436 | + | |
| 437 | + try { | |
| 438 | + $GLOBALS['wp_object_cache']->delete('iwp_importer_config_' . $id, 'options'); | |
| 439 | + $config = get_option('iwp_importer_config_' . $id); | |
| 440 | + | |
| 441 | + $object_ids = $config['delete_ids']; | |
| 442 | + if ($object_ids && count($object_ids) > $i) { | |
| 443 | + | |
| 444 | + $object_id = $object_ids[$i]; | |
| 445 | + | |
| 446 | + if (apply_filters('iwp/importer/enable_custom_delete_action', false, $id)) { | |
| 447 | + | |
| 448 | + Logger::write('custom_delete_action:' . $i . ' -object=' . $object_id); | |
| 449 | + do_action('iwp/importer/custom_delete_action', $id, $object_id); | |
| 450 | + } else { | |
| 451 | + | |
| 452 | + Logger::write('delete:' . $i . ' -object=' . $object_id); | |
| 453 | + $this->getMapper()->delete($object_id); | |
| 454 | + } | |
| 455 | + | |
| 456 | + $message = apply_filters('iwp/status/record_deleted', 'Record Deleted: #' . $object_id, $object_id); | |
| 457 | + $stats['deletes']++; | |
| 458 | + | |
| 459 | + Util::write_status_log_file_message($id, $session, $message, 'D', $progress['current_row']); | |
| 460 | + } | |
| 461 | + } catch (MapperException $e) { | |
| 462 | + | |
| 463 | + $stats['errors']++; | |
| 464 | + Logger::error('delete:' . $i . ' -mapper-error=' . $e->getMessage()); | |
| 465 | + Util::write_status_log_file_message($id, $session, $e->getMessage(), 'E', $progress['current_row']); | |
| 466 | + } | |
| 467 | + } | |
| 468 | + } | |
| 469 | + | |
| 470 | + $importer_state->update_importer_stats($stats); | |
| 471 | + Util::write_status_session_to_file($id, $importer_state); | |
| 472 | + | |
| 473 | + $importer_state->increment_current_row(); | |
| 474 | + $progress = $importer_state->get_progress(); | |
| 475 | + | |
| 476 | + ImporterState::set_state($id, $importer_state->get_raw()); | |
| 477 | + | |
| 478 | + $max_record_time = max($max_record_time, microtime(true) - $record_time); | |
| 479 | + } | |
| 480 | + | |
| 481 | + // TODO: need a new state that will stop the running from happening more than once. | |
| 482 | + // if returning timeout then the cron will stop on older versions | |
| 483 | + if (defined('IWP_PRO_VERSION') && version_compare(IWP_PRO_VERSION, '2.8.0', '>')) { | |
| 484 | + // default status to idle after run | |
| 485 | + $importer_state->populate([ | |
| 486 | + 'status' => 'timeout' | |
| 487 | + ]); | |
| 488 | + } else { | |
| 489 | + $importer_state->populate([ | |
| 490 | + 'status' => 'running' | |
| 491 | + ]); | |
| 492 | + } | |
| 493 | + | |
| 494 | + $state_data = $importer_state->get_raw(); | |
| 495 | + | |
| 496 | + $progress = $importer_state->get_progress(); | |
| 497 | + if ($progress['end'] - $progress['start'] <= $progress['current_row']) { | |
| 498 | + | |
| 499 | + switch ($importer_state->get_section()) { | |
| 500 | + case 'import': | |
| 501 | + | |
| 502 | + | |
| 503 | + if ($this->getMapper()->permission() && $this->getMapper()->permission()->allowed_method('remove')) { | |
| 504 | + | |
| 505 | + // importer delete | |
| 506 | + $state_data['section'] = 'delete'; | |
| 507 | + | |
| 508 | + // generate list of items to be deleted | |
| 509 | + $object_ids = $this->getMapper()->get_objects_for_removal(); | |
| 510 | + if (!empty($object_ids)) { | |
| 511 | + | |
| 512 | + $config = get_option('iwp_importer_config_' . $id); | |
| 513 | + $config['delete_ids'] = $object_ids; | |
| 514 | + update_option('iwp_importer_config_' . $id, $config); | |
| 515 | + | |
| 516 | + $state_data['progress']['delete']['start'] = 0; | |
| 517 | + $state_data['progress']['delete']['end'] = $object_ids ? count($object_ids) : 0; | |
| 518 | + } else { | |
| 519 | + $state_data['section'] = ''; | |
| 520 | + $state_data['status'] = 'complete'; | |
| 521 | + } | |
| 522 | + } else { | |
| 523 | + $state_data['section'] = ''; | |
| 524 | + $state_data['status'] = 'complete'; | |
| 525 | + } | |
| 526 | + | |
| 527 | + break; | |
| 528 | + case 'delete': | |
| 529 | + | |
| 530 | + // importer complete | |
| 531 | + $state_data['section'] = ''; | |
| 532 | + $state_data['status'] = 'complete'; | |
| 533 | + | |
| 534 | + break; | |
| 535 | + } | |
| 536 | + } | |
| 537 | + | |
| 538 | + ImporterState::set_state($id, $state_data); | |
| 539 | + $importer_state->populate($state_data); | |
| 540 | + | |
| 243 | 541 | Util::write_status_session_to_file($id, $importer_state); |
| 542 | + } | |
| 244 | 543 | |
| 245 | - $this->mapper->teardown(); | |
| 246 | - $this->unregister_shutdown(); | |
| 544 | + function get_unique_identifier_log_text() | |
| 545 | + { | |
| 546 | + $unique_identifier_str = ''; | |
| 547 | + | |
| 548 | + $unqiue_identifier_settings = $this->getMapper()->get_unqiue_identifier_settings(); | |
| 549 | + if (!empty($unqiue_identifier_settings) && isset($unqiue_identifier_settings['field'], $unqiue_identifier_settings['value'])) { | |
| 550 | + | |
| 551 | + $unique_identifier_str = ' using unique identifier '; | |
| 552 | + if ($unqiue_identifier_settings['field'] === '_iwp_ref_uid') { | |
| 553 | + $unique_identifier_str .= sprintf('("_iwp_ref_uid" = "%s")', $unqiue_identifier_settings['value']); | |
| 554 | + } else { | |
| 555 | + $unique_identifier_str .= sprintf('("%s" = "%s")', $unqiue_identifier_settings['field'], $unqiue_identifier_settings['value']); | |
| 556 | + } | |
| 557 | + } | |
| 558 | + | |
| 559 | + return $unique_identifier_str; | |
| 560 | + } | |
| 561 | + | |
| 562 | + function has_enough_time($start, $time_limit, $max_record_time) | |
| 563 | + { | |
| 564 | + return (microtime(true) - $start) < $time_limit - $max_record_time; | |
| 565 | + } | |
| 566 | + | |
| 567 | + function get_memory_usage() | |
| 568 | + { | |
| 569 | + return memory_get_usage(true); | |
| 570 | + } | |
| 571 | + | |
| 572 | + function has_enough_memory($memory_max_usage) | |
| 573 | + { | |
| 574 | + $limit = $this->get_memory_limit(); | |
| 575 | + | |
| 576 | + // Has unlimited memory | |
| 577 | + if ($limit == '-1') { | |
| 578 | + return true; | |
| 579 | + } | |
| 580 | + | |
| 581 | + $limit *= 0.9; | |
| 582 | + $current_usage = $this->get_memory_usage(); | |
| 583 | + | |
| 584 | + if ($current_usage + $memory_max_usage < $limit) { | |
| 585 | + return true; | |
| 586 | + } | |
| 587 | + | |
| 588 | + Logger::error(sprintf("Not Enough Memory left to use %s, %s/%s", Logger::formatBytes($memory_max_usage, 2), Logger::formatBytes($current_usage, 2), Logger::formatBytes($limit, 2))); | |
| 589 | + | |
| 590 | + return false; | |
| 591 | + } | |
| 592 | + | |
| 593 | + function get_memory_limit($force = false) | |
| 594 | + { | |
| 595 | + if ($force || is_null($this->memory_limit)) { | |
| 596 | + | |
| 597 | + $memory_limit = ini_get('memory_limit'); | |
| 598 | + if (preg_match('/^(\d+)(.)$/', $memory_limit, $matches)) { | |
| 599 | + if ($matches[2] == 'G') { | |
| 600 | + $memory_limit = $matches[1] * 1024 * 1024 * 1024; // nnnM -> nnn MB | |
| 601 | + } elseif ($matches[2] == 'M') { | |
| 602 | + $memory_limit = $matches[1] * 1024 * 1024; // nnnM -> nnn MB | |
| 603 | + } else if ($matches[2] == 'K') { | |
| 604 | + $memory_limit = $matches[1] * 1024; // nnnK -> nnn KB | |
| 605 | + } | |
| 606 | + } | |
| 607 | + | |
| 608 | + $this->memory_limit = $memory_limit; | |
| 609 | + | |
| 610 | + Logger::info('memory_limit ' . $this->memory_limit . ' bytes'); | |
| 611 | + } | |
| 612 | + | |
| 613 | + return $this->memory_limit; | |
| 247 | 614 | } |
| 248 | 615 | |
| 249 | 616 | /** |
| 250 | 617 | * Apply any importer filters to skip records |
