webstatement/app/Jobs/ProcessTransactionDataJob.php

<?php

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\TempTransaction;

class ProcessTransactionDataJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $periods;

    /**
     * Create a new job instance.
     */
    public function __construct(array $periods = [])
    {
        $this->periods = $periods;
    }

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        try {
            set_time_limit(24 * 60 * 60);
            $disk = Storage::disk('sftpStatement');
            $processedCount = 0;
            $errorCount = 0;

            if (empty($this->periods)) {
                Log::warning('No periods provided for transaction data processing');
                return;
            }

            foreach ($this->periods as $period) {
                // Skip the _parameter folder
                if ($period === '_parameter') {
                    Log::info("Skipping _parameter folder");
                    continue;
                }

                // Construct the filename based on the period folder name
                $filename = "$period.ST.TRANSACTION.csv";
                $filePath = "$period/$filename";

                Log::info("Processing transaction file: $filePath");

                if (!$disk->exists($filePath)) {
                    Log::warning("File not found: $filePath");
                    continue;
                }

                // Create a temporary local copy of the file
                $tempFilePath = storage_path("app/temp_$filename");
                file_put_contents($tempFilePath, $disk->get($filePath));

                $handle = fopen($tempFilePath, "r");

                if ($handle !== false) {
                    $headers = (new TempTransaction())->getFillable();
                    $rowCount = 0;

                    while (($row = fgetcsv($handle, 0, "~")) !== false) {
                        $rowCount++;

                        if (count($headers) === count($row)) {
                            $data = array_combine($headers, $row);
                            try {
                                if (isset($data['_id']) && $data['_id'] !== 'id' && $data['_id'] !== '_id') {
                                    TempTransaction::updateOrCreate(
                                        ['_id' => $data['_id']],
                                        $data
                                    );
                                    $processedCount++;
                                }
                            } catch (Exception $e) {
                                $errorCount++;
                                Log::error("Error processing Transaction at row $rowCount in $filePath: " . $e->getMessage());
                            }
                        } else {
                            Log::warning("Row $rowCount in $filePath has incorrect column count. Expected: " . count($headers) . ", Got: " . count($row));
                        }
                    }

                    fclose($handle);
                    Log::info("Completed processing $filePath. Processed $processedCount records with $errorCount errors.");

                    // Clean up the temporary file
                    unlink($tempFilePath);
                } else {
                    Log::error("Unable to open file: $filePath");
                }
            }

            Log::info("Transaction data processing completed. Total processed: $processedCount, Total errors: $errorCount");

        } catch (Exception $e) {
            Log::error('Error in ProcessTransactionDataJob: ' . $e->getMessage());
            throw $e;
        }
    }
}