daengdeni/webstatement
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(webstatement): optimize ProcessAccountDataJob logic with batch processing

Browse Source 
- Menambahkan properti baru `CHUNK_SIZE` untuk mengelola proses data dalam batch sehingga mengurangi penggunaan memori.
- Memperkenalkan batching untuk penyimpanan akun (`accountBatch`) dan balance (`balanceBatch`) untuk mengurangi beban query database.
- Mengganti penyimpanan langsung di database dengan metode bulk insert menggunakan `upsert` pada tabel Account dan AccountBalance.
- Menambahkan logging untuk mencatat setiap pemrosesan batch dan memberikan deskripsi jumlah chunk yang telah diproses.
- Memindahkan proses normalization dan balance data ke sistem batching alih-alih penyimpanan langsung.
- Menambahkan validasi tambahan untuk pengolahan data row dan mencegah error data yang memiliki kolom tidak sesuai.
- Memperbarui log pesan untuk menampilkan jumlah record yang diproses dan error secara lebih rinci.
- Memodularisasi fungsi agar lebih readable dan maintainable dengan beberapa fungsi baru seperti `addToBatch` dan `saveBatch`.
This commit is contained in:
daengdeni
2025-05-28 08:21:25 +07:00
parent 60e60b4fef
commit a8dafb23c5
1 changed files with 257 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files

468
app/Jobs/ProcessAccountDataJob.php
View File

@@ -1,223 +1,269 @@
<?php
namespace Modules\Webstatement\Jobs;
namespace Modules\Webstatement\Jobs;
use Exception;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Storage;
use Modules\Webstatement\Models\Account;
use Modules\Webstatement\Models\AccountBalance;
use Exception;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Storage;
use Modules\Webstatement\Models\Account;
use Modules\Webstatement\Models\AccountBalance;
class ProcessAccountDataJob implements ShouldQueue
class ProcessAccountDataJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
private const CSV_DELIMITER = '~';
private const MAX_EXECUTION_TIME = 86400; // 24 hours in seconds
private const FILENAME = 'ST.ACCOUNT.csv';
private const DISK_NAME = 'sftpStatement';
private const CHUNK_SIZE = 1000; // Process data in chunks to reduce memory usage
private string $period = '';
private int $processedCount = 0;
private int $errorCount = 0;
private $balanceData = [];
private $accountBatch = [];
private $balanceBatch = [];
/**
* Create a new job instance.
*/
public function __construct(string $period = '')
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
$this->period = $period;
}
private const CSV_DELIMITER = '~';
private const MAX_EXECUTION_TIME = 86400; // 24 hours in seconds
private const FILENAME = 'ST.ACCOUNT.csv';
private const DISK_NAME = 'sftpStatement';
private string $period = '';
private int $processedCount = 0;
private int $errorCount = 0;
/**
* Execute the job.
*/
public function handle()
: void
{
try {
$this->initializeJob();
private $balanceData = [];
/**
* Create a new job instance.
*/
public function __construct(string $period = '')
{
$this->period = $period;
}
/**
* Execute the job.
*/
public function handle()
: void
{
try {
$this->initializeJob();
if ($this->period === '') {
Log::warning('No period provided for account data processing');
return;
}
$this->processPeriod();
$this->logJobCompletion();
} catch (Exception $e) {
Log::error('Error in ProcessAccountDataJob: ' . $e->getMessage());
throw $e;
}
}
private function initializeJob()
: void
{
set_time_limit(self::MAX_EXECUTION_TIME);
$this->processedCount = 0;
$this->errorCount = 0;
}
private function processPeriod()
: void
{
$disk = Storage::disk(self::DISK_NAME);
$filename = "{$this->period}." . self::FILENAME;
$filePath = "{$this->period}/$filename";
if (!$this->validateFile($disk, $filePath)) {
if ($this->period === '') {
Log::warning('No period provided for account data processing');
return;
}
$tempFilePath = $this->createTemporaryFile($disk, $filePath, $filename);
$this->processFile($tempFilePath, $filePath);
$this->cleanup($tempFilePath);
}
private function validateFile($disk, string $filePath)
: bool
{
Log::info("Processing account file: $filePath");
if (!$disk->exists($filePath)) {
Log::warning("File not found: $filePath");
return false;
}
return true;
}
private function createTemporaryFile($disk, string $filePath, string $filename)
: string
{
$tempFilePath = storage_path("app/temp_$filename");
file_put_contents($tempFilePath, $disk->get($filePath));
return $tempFilePath;
}
private function processFile(string $tempFilePath, string $filePath)
: void
{
$handle = fopen($tempFilePath, "r");
if ($handle === false) {
Log::error("Unable to open file: $filePath");
return;
}
$headers = (new Account())->getFillable();
Log::info('Headers: ' . implode(", ", $headers));
$rowCount = 0;
while (($row = fgetcsv($handle, 0, self::CSV_DELIMITER)) !== false) {
$rowCount++;
$this->processRow($row, $headers, $rowCount, $filePath);
}
fclose($handle);
Log::info("Completed processing $filePath. Processed {$this->processedCount} records with {$this->errorCount} errors.");
}
private function processRow(array $row, array $headers, int $rowCount, string $filePath)
: void
{
if (count($headers) !== count($row)) {
Log::warning("Row $rowCount in $filePath has incorrect column count. Expected: " .
count($headers) . ", Got: " . count($row));
return;
}
$data = array_combine($headers, $row);
$this->normalizeData($data);
$this->saveRecord($data, $rowCount, $filePath);
}
private function normalizeData(array &$data)
: void
{
// Check if start_year_bal is empty and set it to 0 if so
if (empty($data['start_year_bal']) || $data['start_year_bal'] == "" || $data['start_year_bal'] == null) {
$data['start_year_bal'] = 0;
}
if (empty($data['closure_date']) || $data['closure_date'] == "" || $data['closure_date'] == null) {
$data['closure_date'] = null;
}
// Store balance data separately before removing from Account data
$this->balanceData = [
'open_actual_bal' => empty($data['open_actual_bal']) ? 0 : $data['open_actual_bal'],
'open_cleared_bal' => empty($data['open_cleared_bal']) ? 0 : $data['open_cleared_bal'],
];
// Remove balance fields from Account data
unset($data['open_actual_bal']);
unset($data['open_cleared_bal']);
}
private function saveRecord(array $data, int $rowCount, string $filePath)
: void
{
try {
if ($data['account_number'] !== 'account_number') {
// Use firstOrNew instead of updateOrCreate
$account = Account::firstOrNew(['account_number' => $data['account_number']]);
$account->fill($data);
$account->save();
$this->saveAccountBalance($data['account_number']);
$this->processedCount++;
}
} catch (Exception $e) {
$this->errorCount++;
Log::error("Error processing Account at row $rowCount in $filePath: " . $e->getMessage());
}
}
private function saveAccountBalance(string $accountNumber)
: void
{
// Store the opening balances in the AccountBalance model for this period
if (isset($this->balanceData['open_actual_bal']) || isset($this->balanceData['open_cleared_bal'])) {
// Prepare balance data for bulk insert/update
$balanceData = [
'account_number' => $accountNumber,
'period' => $this->period,
'actual_balance' => $this->balanceData['open_actual_bal'],
'cleared_balance' => $this->balanceData['open_cleared_bal'],
'created_at' => now(),
'updated_at' => now()
];
// Use updateOrInsert to reduce queries
AccountBalance::updateOrInsert(
[
'account_number' => $accountNumber,
'period' => $this->period
],
$balanceData
);
}
}
private function cleanup(string $tempFilePath)
: void
{
if (file_exists($tempFilePath)) {
unlink($tempFilePath);
}
}
private function logJobCompletion()
: void
{
Log::info("Account data processing completed. " .
"Total processed: {$this->processedCount}, Total errors: {$this->errorCount}");
$this->processPeriod();
$this->logJobCompletion();
} catch (Exception $e) {
Log::error('Error in ProcessAccountDataJob: ' . $e->getMessage());
throw $e;
}
}
private function initializeJob()
: void
{
set_time_limit(self::MAX_EXECUTION_TIME);
$this->processedCount = 0;
$this->errorCount = 0;
$this->accountBatch = [];
$this->balanceBatch = [];
}
private function processPeriod()
: void
{
$disk = Storage::disk(self::DISK_NAME);
$filename = "{$this->period}." . self::FILENAME;
$filePath = "{$this->period}/$filename";
if (!$this->validateFile($disk, $filePath)) {
return;
}
$tempFilePath = $this->createTemporaryFile($disk, $filePath, $filename);
$this->processFile($tempFilePath, $filePath);
$this->cleanup($tempFilePath);
}
private function validateFile($disk, string $filePath)
: bool
{
Log::info("Processing account file: $filePath");
if (!$disk->exists($filePath)) {
Log::warning("File not found: $filePath");
return false;
}
return true;
}
private function createTemporaryFile($disk, string $filePath, string $filename)
: string
{
$tempFilePath = storage_path("app/temp_$filename");
file_put_contents($tempFilePath, $disk->get($filePath));
return $tempFilePath;
}
private function processFile(string $tempFilePath, string $filePath)
: void
{
$handle = fopen($tempFilePath, "r");
if ($handle === false) {
Log::error("Unable to open file: $filePath");
return;
}
$headers = (new Account())->getFillable();
Log::info('Headers: ' . implode(", ", $headers));
$rowCount = 0;
$chunkCount = 0;
while (($row = fgetcsv($handle, 0, self::CSV_DELIMITER)) !== false) {
$rowCount++;
$this->processRow($row, $headers, $rowCount, $filePath);
// Process in chunks to avoid memory issues
if (count($this->accountBatch) >= self::CHUNK_SIZE) {
$this->saveBatch();
$chunkCount++;
Log::info("Processed chunk $chunkCount ({$this->processedCount} records so far)");
}
}
// Process any remaining records
if (!empty($this->accountBatch) || !empty($this->balanceBatch)) {
$this->saveBatch();
}
fclose($handle);
Log::info("Completed processing $filePath. Processed {$this->processedCount} records with {$this->errorCount} errors.");
}
private function processRow(array $row, array $headers, int $rowCount, string $filePath)
: void
{
if (count($headers) !== count($row)) {
Log::warning("Row $rowCount in $filePath has incorrect column count. Expected: " .
count($headers) . ", Got: " . count($row));
return;
}
$data = array_combine($headers, $row);
$this->normalizeData($data);
$this->addToBatch($data, $rowCount, $filePath);
}
private function normalizeData(array &$data)
: void
{
// Check if start_year_bal is empty and set it to 0 if so
if (empty($data['start_year_bal']) || $data['start_year_bal'] == "" || $data['start_year_bal'] == null) {
$data['start_year_bal'] = 0;
}
if (empty($data['closure_date']) || $data['closure_date'] == "" || $data['closure_date'] == null) {
$data['closure_date'] = null;
}
// Store balance data separately before removing from Account data
$this->balanceData = [
'open_actual_bal' => empty($data['open_actual_bal']) ? 0 : $data['open_actual_bal'],
'open_cleared_bal' => empty($data['open_cleared_bal']) ? 0 : $data['open_cleared_bal'],
];
// Remove balance fields from Account data
unset($data['open_actual_bal']);
unset($data['open_cleared_bal']);
}
/**
* Add record to batch instead of saving immediately
*/
private function addToBatch(array $data, int $rowCount, string $filePath)
: void
{
try {
if ($data['account_number'] !== 'account_number') {
// Add timestamp fields
$now = now();
$data['created_at'] = $now;
$data['updated_at'] = $now;
// Add to account batch
$this->accountBatch[] = $data;
// Add to balance batch
if (isset($this->balanceData['open_actual_bal']) || isset($this->balanceData['open_cleared_bal'])) {
$this->balanceBatch[] = [
'account_number' => $data['account_number'],
'period' => $this->period,
'actual_balance' => $this->balanceData['open_actual_bal'],
'cleared_balance' => $this->balanceData['open_cleared_bal'],
'created_at' => $now,
'updated_at' => $now
];
}
$this->processedCount++;
}
} catch (Exception $e) {
$this->errorCount++;
Log::error("Error processing Account at row $rowCount in $filePath: " . $e->getMessage());
}
}
/**
* Save batched records to the database
*/
private function saveBatch()
: void
{
try {
if (!empty($this->accountBatch)) {
// Bulk insert/update accounts
Account::upsert(
$this->accountBatch,
['account_number'], // Unique key
array_diff((new Account())->getFillable(), ['account_number']) // Update columns
);
// Reset account batch after processing
$this->accountBatch = [];
}
if (!empty($this->balanceBatch)) {
// Bulk insert/update account balances
AccountBalance::upsert(
$this->balanceBatch,
['account_number', 'period'], // Composite unique key
['actual_balance', 'cleared_balance', 'updated_at'] // Update columns
);
// Reset balance batch after processing
$this->balanceBatch = [];
}
} catch (Exception $e) {
Log::error("Error in saveBatch: " . $e->getMessage());
$this->errorCount += count($this->accountBatch);
}
}
private function cleanup(string $tempFilePath)
: void
{
if (file_exists($tempFilePath)) {
unlink($tempFilePath);
}
}
private function logJobCompletion()
: void
{
Log::info("Account data processing completed. " .
"Total processed: {$this->processedCount}, Total errors: {$this->errorCount}");
}
}