Initial
Takes in a file, process it and output a new file with additional columns.
This commit is contained in:
Stefan Lazic
@@ -0,0 +1,457 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Import;
|
||||
|
||||
use RuntimeException;
|
||||
use Throwable;
|
||||
|
||||
final class ImportService
|
||||
{
|
||||
private const REQUIRED_COLUMNS = [
|
||||
'Accrual Category',
|
||||
'Local Selling Company (S/T)',
|
||||
'Contracting Party',
|
||||
'Rights Holder Liable Flag',
|
||||
'Global Transfer Pricing Flag',
|
||||
'Domestic/ Foreign Flag',
|
||||
'Ownership Type',
|
||||
];
|
||||
|
||||
public function __construct(
|
||||
private readonly SpreadsheetReader $reader,
|
||||
private readonly SpreadsheetRowProcessor $processor,
|
||||
private readonly ImportRepository $repository,
|
||||
private readonly int $chunkSize = 5,
|
||||
) {
|
||||
}
|
||||
|
||||
public function requiredColumns(): array
|
||||
{
|
||||
return self::REQUIRED_COLUMNS;
|
||||
}
|
||||
|
||||
public function import(string $filePath, string $sourceFilename): ImportResult
|
||||
{
|
||||
$state = [
|
||||
'filePath' => $filePath,
|
||||
'sourceFilename' => $sourceFilename,
|
||||
'sourceExtension' => strtolower(pathinfo($filePath, PATHINFO_EXTENSION)),
|
||||
'status' => 'pending',
|
||||
'importedRows' => 0,
|
||||
'warningsCount' => 0,
|
||||
'warnings' => [],
|
||||
];
|
||||
|
||||
while (!in_array(($state['status'] ?? 'pending'), ['completed', 'error'], true)) {
|
||||
$state = $this->processJobState($state, $this->chunkSize);
|
||||
}
|
||||
|
||||
if (($state['status'] ?? '') === 'error') {
|
||||
throw new RuntimeException('Import failed: ' . ($state['error'] ?? 'Unknown error'));
|
||||
}
|
||||
|
||||
return new ImportResult(
|
||||
isset($state['batchId']) ? (int) $state['batchId'] : null,
|
||||
(int) ($state['importedRows'] ?? 0),
|
||||
(int) ($state['warningsCount'] ?? 0),
|
||||
is_array($state['warnings'] ?? null) ? array_values($state['warnings']) : [],
|
||||
is_array($state['headers'] ?? null) ? array_values($state['headers']) : [],
|
||||
$this->processor->calculatedHeaders()
|
||||
);
|
||||
}
|
||||
|
||||
public function processJobState(array $state, ?int $chunkSize = null): array
|
||||
{
|
||||
$chunkSize = max(1, $chunkSize ?? $this->chunkSize);
|
||||
$extension = strtolower((string) ($state['sourceExtension'] ?? pathinfo((string) ($state['filePath'] ?? ''), PATHINFO_EXTENSION)));
|
||||
if ($extension === 'xls') {
|
||||
return $this->processLegacyJobState($state, $chunkSize);
|
||||
}
|
||||
|
||||
return $this->processStreamingJobState($state, $chunkSize);
|
||||
}
|
||||
|
||||
private function processStreamingJobState(array $state, int $chunkSize): array
|
||||
{
|
||||
if (($state['status'] ?? 'pending') === 'completed' || ($state['status'] ?? 'pending') === 'error') {
|
||||
return $state;
|
||||
}
|
||||
|
||||
try {
|
||||
$state['importedRows'] = (int) ($state['importedRows'] ?? 0);
|
||||
$state['warningsCount'] = (int) ($state['warningsCount'] ?? 0);
|
||||
$state['warnings'] = is_array($state['warnings'] ?? null) ? array_values($state['warnings']) : [];
|
||||
|
||||
if (!isset($state['cacheBasePath'])) {
|
||||
$state['cacheBasePath'] = $this->cacheBasePath($state);
|
||||
$cache = $this->reader->buildChunkCache(
|
||||
(string) $state['filePath'],
|
||||
(string) $state['cacheBasePath'],
|
||||
self::REQUIRED_COLUMNS,
|
||||
$chunkSize
|
||||
);
|
||||
$this->reader->validateRequiredColumns($cache['headers'], self::REQUIRED_COLUMNS);
|
||||
|
||||
$state['headers'] = $cache['headers'];
|
||||
$state['headerLookup'] = $cache['header_lookup'];
|
||||
$state['headerRowIndex'] = (int) $cache['header_row_index'];
|
||||
$state['totalRows'] = (int) $cache['total_rows'];
|
||||
$state['chunkCount'] = (int) $cache['chunk_count'];
|
||||
$state['chunkSize'] = (int) $cache['chunk_size'];
|
||||
$state['nextChunkIndex'] = (int) ($state['nextChunkIndex'] ?? 0);
|
||||
$state['status'] = 'cached';
|
||||
$state['done'] = false;
|
||||
$state['message'] = 'Workbook cached and ready for chunked processing.';
|
||||
|
||||
return $state;
|
||||
}
|
||||
|
||||
if (!isset($state['batchId'])) {
|
||||
$state['batchId'] = $this->repository->createBatch(
|
||||
(string) $state['sourceFilename'],
|
||||
is_array($state['headers'] ?? null) ? $state['headers'] : [],
|
||||
$this->processor->calculatedHeaders(),
|
||||
'processing'
|
||||
);
|
||||
}
|
||||
|
||||
$chunkCount = (int) ($state['chunkCount'] ?? 0);
|
||||
$nextChunkIndex = (int) ($state['nextChunkIndex'] ?? 0);
|
||||
|
||||
if ($chunkCount <= 0 || $nextChunkIndex >= $chunkCount) {
|
||||
$this->repository->updateBatchSummary(
|
||||
(int) $state['batchId'],
|
||||
(int) $state['importedRows'],
|
||||
(int) $state['warningsCount'],
|
||||
'complete'
|
||||
);
|
||||
$this->repository->pruneExceptBatch((int) $state['batchId']);
|
||||
$state['status'] = 'completed';
|
||||
$state['done'] = true;
|
||||
$state['message'] = 'Import complete.';
|
||||
|
||||
return $state;
|
||||
}
|
||||
|
||||
$chunk = $this->reader->readChunkFromCache((string) $state['cacheBasePath'], $nextChunkIndex);
|
||||
$rows = is_array($chunk['rows'] ?? null) ? $chunk['rows'] : [];
|
||||
|
||||
$processedInChunk = 0;
|
||||
$this->repository->beginTransaction();
|
||||
try {
|
||||
foreach ($rows as $row) {
|
||||
if (!is_array($row) || !isset($row['row_number'], $row['data']) || !is_array($row['data'])) {
|
||||
continue;
|
||||
}
|
||||
|
||||
$processedInChunk++;
|
||||
$rowNumber = (int) $row['row_number'];
|
||||
$sourceRow = $row['data'];
|
||||
|
||||
try {
|
||||
$processed = $this->processor->process($sourceRow);
|
||||
$this->repository->insertRow(
|
||||
(int) $state['batchId'],
|
||||
$rowNumber,
|
||||
$sourceRow,
|
||||
$processed['normalized'],
|
||||
$processed['calculated'],
|
||||
$processed['merged']
|
||||
);
|
||||
$state['importedRows']++;
|
||||
} catch (Throwable $throwable) {
|
||||
$state['warningsCount']++;
|
||||
if (count($state['warnings']) < 100) {
|
||||
$state['warnings'][] = sprintf('Row %d skipped: %s', $rowNumber, $throwable->getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
$state['processedRows'] = (int) ($state['processedRows'] ?? 0) + $processedInChunk;
|
||||
$state['nextChunkIndex'] = $nextChunkIndex + 1;
|
||||
|
||||
$isComplete = ((int) $state['nextChunkIndex']) >= $chunkCount;
|
||||
$this->repository->updateBatchSummary(
|
||||
(int) $state['batchId'],
|
||||
(int) $state['importedRows'],
|
||||
(int) $state['warningsCount'],
|
||||
$isComplete ? 'complete' : 'processing'
|
||||
);
|
||||
$this->repository->commit();
|
||||
|
||||
if ($isComplete) {
|
||||
$this->repository->pruneExceptBatch((int) $state['batchId']);
|
||||
$state['status'] = 'completed';
|
||||
$state['done'] = true;
|
||||
$state['message'] = 'Import complete.';
|
||||
} else {
|
||||
$state['status'] = 'running';
|
||||
$state['done'] = false;
|
||||
$state['message'] = sprintf('Processed %d row(s) in the latest chunk.', $processedInChunk);
|
||||
}
|
||||
} catch (Throwable $throwable) {
|
||||
$this->repository->rollBack();
|
||||
throw $throwable;
|
||||
}
|
||||
} catch (Throwable $throwable) {
|
||||
$state['status'] = 'error';
|
||||
$state['done'] = true;
|
||||
$state['message'] = 'Import failed: ' . $throwable->getMessage();
|
||||
$state['error'] = $throwable->getMessage();
|
||||
}
|
||||
|
||||
return $state;
|
||||
}
|
||||
|
||||
private function processLegacyJobState(array $state, int $chunkSize): array
|
||||
{
|
||||
if (($state['status'] ?? 'pending') === 'completed' || ($state['status'] ?? 'pending') === 'error') {
|
||||
return $state;
|
||||
}
|
||||
|
||||
try {
|
||||
if (!isset($state['batchId'])) {
|
||||
$headerInfo = $this->reader->detectHeader((string) $state['filePath'], self::REQUIRED_COLUMNS);
|
||||
$this->reader->validateRequiredColumns($headerInfo['headers'], self::REQUIRED_COLUMNS);
|
||||
|
||||
$state['batchId'] = $this->repository->createBatch(
|
||||
(string) $state['sourceFilename'],
|
||||
$headerInfo['headers'],
|
||||
$this->processor->calculatedHeaders(),
|
||||
'processing'
|
||||
);
|
||||
$state['headers'] = $headerInfo['headers'];
|
||||
$state['headerLookup'] = $headerInfo['header_lookup'];
|
||||
$state['headerRowIndex'] = (int) $headerInfo['header_row_index'];
|
||||
$state['highestDataRow'] = (int) $headerInfo['highest_data_row'];
|
||||
$state['nextRow'] = ((int) $headerInfo['header_row_index']) + 1;
|
||||
$state['totalRows'] = max(0, (int) $headerInfo['highest_data_row'] - (int) $headerInfo['header_row_index']);
|
||||
$state['importedRows'] = (int) ($state['importedRows'] ?? 0);
|
||||
$state['warningsCount'] = (int) ($state['warningsCount'] ?? 0);
|
||||
$state['warnings'] = is_array($state['warnings'] ?? null) ? array_values($state['warnings']) : [];
|
||||
}
|
||||
|
||||
$nextRow = (int) ($state['nextRow'] ?? 0);
|
||||
$highestDataRow = (int) ($state['highestDataRow'] ?? 0);
|
||||
|
||||
if ($nextRow > $highestDataRow) {
|
||||
$this->repository->updateBatchSummary(
|
||||
(int) $state['batchId'],
|
||||
(int) ($state['importedRows'] ?? 0),
|
||||
(int) ($state['warningsCount'] ?? 0),
|
||||
'complete'
|
||||
);
|
||||
$this->repository->pruneExceptBatch((int) $state['batchId']);
|
||||
$state['status'] = 'completed';
|
||||
$state['done'] = true;
|
||||
$state['message'] = 'Import complete.';
|
||||
|
||||
return $state;
|
||||
}
|
||||
|
||||
$rowsInWindow = max(0, min($chunkSize, $highestDataRow - $nextRow + 1));
|
||||
$chunk = $this->reader->readChunk(
|
||||
(string) $state['filePath'],
|
||||
$nextRow,
|
||||
$chunkSize,
|
||||
$state['headerLookup']
|
||||
);
|
||||
|
||||
$processedInChunk = 0;
|
||||
$this->repository->beginTransaction();
|
||||
try {
|
||||
foreach ($chunk['rows'] as $row) {
|
||||
$processedInChunk++;
|
||||
$rowNumber = (int) $row['row_number'];
|
||||
$sourceRow = $row['data'];
|
||||
|
||||
try {
|
||||
$processed = $this->processor->process($sourceRow);
|
||||
$this->repository->insertRow(
|
||||
(int) $state['batchId'],
|
||||
$rowNumber,
|
||||
$sourceRow,
|
||||
$processed['normalized'],
|
||||
$processed['calculated'],
|
||||
$processed['merged']
|
||||
);
|
||||
$state['importedRows'] = (int) ($state['importedRows'] ?? 0) + 1;
|
||||
} catch (Throwable $throwable) {
|
||||
$state['warningsCount'] = (int) ($state['warningsCount'] ?? 0) + 1;
|
||||
if (!isset($state['warnings']) || !is_array($state['warnings'])) {
|
||||
$state['warnings'] = [];
|
||||
}
|
||||
if (count($state['warnings']) < 100) {
|
||||
$state['warnings'][] = sprintf('Row %d skipped: %s', $rowNumber, $throwable->getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
$state['nextRow'] = $nextRow + $rowsInWindow;
|
||||
$state['processedRows'] = (int) ($state['processedRows'] ?? 0) + $rowsInWindow;
|
||||
|
||||
$isComplete = (int) $state['nextRow'] > $highestDataRow;
|
||||
$this->repository->updateBatchSummary(
|
||||
(int) $state['batchId'],
|
||||
(int) $state['importedRows'],
|
||||
(int) $state['warningsCount'],
|
||||
$isComplete ? 'complete' : 'processing'
|
||||
);
|
||||
$this->repository->commit();
|
||||
|
||||
if ($isComplete) {
|
||||
$this->repository->pruneExceptBatch((int) $state['batchId']);
|
||||
$state['status'] = 'completed';
|
||||
$state['done'] = true;
|
||||
$state['message'] = 'Import complete.';
|
||||
} else {
|
||||
$state['status'] = 'running';
|
||||
$state['done'] = false;
|
||||
$state['message'] = sprintf('Processed %d row(s) in the latest chunk.', $processedInChunk);
|
||||
}
|
||||
} catch (Throwable $throwable) {
|
||||
$this->repository->rollBack();
|
||||
throw $throwable;
|
||||
}
|
||||
} catch (Throwable $throwable) {
|
||||
$state['status'] = 'error';
|
||||
$state['done'] = true;
|
||||
$state['message'] = 'Import failed: ' . $throwable->getMessage();
|
||||
$state['error'] = $throwable->getMessage();
|
||||
}
|
||||
|
||||
return $state;
|
||||
}
|
||||
|
||||
public function latestBatch(): ?array
|
||||
{
|
||||
return $this->repository->getLatestBatch();
|
||||
}
|
||||
|
||||
public function clear(): void
|
||||
{
|
||||
$this->repository->beginTransaction();
|
||||
try {
|
||||
$this->repository->clearAll();
|
||||
$this->repository->commit();
|
||||
} catch (Throwable $throwable) {
|
||||
$this->repository->rollBack();
|
||||
throw new RuntimeException('Clear failed: ' . $throwable->getMessage(), 0, $throwable);
|
||||
}
|
||||
}
|
||||
|
||||
public function getPreview(int $batchId, int $limit, int $offset, string $sortDirection = 'asc'): array
|
||||
{
|
||||
return $this->repository->paginateRows($batchId, $limit, $offset, $sortDirection);
|
||||
}
|
||||
|
||||
public function getFilteredPreview(
|
||||
int $batchId,
|
||||
int $limit,
|
||||
int $offset,
|
||||
string $query,
|
||||
array $tableHeaders = [],
|
||||
array $columnSearches = [],
|
||||
?int $orderColumn = null,
|
||||
string $sortDirection = 'asc'
|
||||
): array
|
||||
{
|
||||
return $this->repository->paginateRowsFiltered(
|
||||
$batchId,
|
||||
$limit,
|
||||
$offset,
|
||||
$query,
|
||||
$tableHeaders,
|
||||
$columnSearches,
|
||||
$orderColumn,
|
||||
$sortDirection
|
||||
);
|
||||
}
|
||||
|
||||
public function getGridRows(
|
||||
int $batchId,
|
||||
int $limit,
|
||||
int $offset,
|
||||
string $query = '',
|
||||
array $tableHeaders = [],
|
||||
array $filterModel = [],
|
||||
array $sortModel = []
|
||||
): array
|
||||
{
|
||||
return $this->repository->paginateRowsGrid(
|
||||
$batchId,
|
||||
$limit,
|
||||
$offset,
|
||||
$query,
|
||||
$tableHeaders,
|
||||
$filterModel,
|
||||
$sortModel
|
||||
);
|
||||
}
|
||||
|
||||
public function countRowsForBatch(int $batchId): int
|
||||
{
|
||||
return $this->repository->countRowsForBatch($batchId);
|
||||
}
|
||||
|
||||
/**
|
||||
* @return array<int, array<string, mixed>>
|
||||
*/
|
||||
public function getExportRows(int $batchId, int $limit, int $offset): array
|
||||
{
|
||||
return $this->repository->exportRowsForBatch($batchId, $limit, $offset);
|
||||
}
|
||||
|
||||
public function getDistinctValuesForBatch(int $batchId, string $header, int $limit = 200): array
|
||||
{
|
||||
return $this->repository->getDistinctValuesForBatch($batchId, $header, $limit);
|
||||
}
|
||||
|
||||
public function getDistinctValuesForBatchGrid(
|
||||
int $batchId,
|
||||
array $tableHeaders,
|
||||
string $columnKey,
|
||||
string $distinctSearch = '',
|
||||
int $limit = 500
|
||||
): array
|
||||
{
|
||||
return $this->repository->getDistinctValuesForBatchGrid($batchId, $tableHeaders, $columnKey, $distinctSearch, $limit);
|
||||
}
|
||||
|
||||
public function countFilteredRowsForBatch(int $batchId, string $query, array $tableHeaders = [], array $columnSearches = []): int
|
||||
{
|
||||
return $this->repository->countRowsForBatchFiltered($batchId, $query, $tableHeaders, $columnSearches);
|
||||
}
|
||||
|
||||
public function countGridRows(
|
||||
int $batchId,
|
||||
string $query = '',
|
||||
array $tableHeaders = [],
|
||||
array $filterModel = []
|
||||
): int
|
||||
{
|
||||
return $this->repository->countRowsForBatchGrid($batchId, $query, $tableHeaders, $filterModel);
|
||||
}
|
||||
|
||||
public function chunkSize(): int
|
||||
{
|
||||
return $this->chunkSize;
|
||||
}
|
||||
|
||||
public function deleteCache(string $cacheBasePath): void
|
||||
{
|
||||
$this->reader->deleteCache($cacheBasePath);
|
||||
}
|
||||
|
||||
private function cacheBasePath(array $state): string
|
||||
{
|
||||
$token = preg_replace('/[^a-zA-Z0-9_-]/', '', (string) ($state['token'] ?? ''));
|
||||
if ($token === '') {
|
||||
throw new RuntimeException('Missing import token.');
|
||||
}
|
||||
|
||||
return sys_get_temp_dir() . '/warner-import-cache/' . $token;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user