Process BlueImp Uploaded Files

ProcessUploads.php

<?php

namespace App\Jobs\Gis;

use Log;

use Illuminate\Bus\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;

use Illuminate\Support\Facades\Storage;
use League\Csv\Reader;
use League\Csv\Statement;

use App\Gis\AddressBase;

use Carbon\Carbon;
use Illuminate\Support\Facades\Cache;

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

    protected $file;

    public $tries = 5;

    /**
     * Create a new job instance.
     *
     * @return void
     */
    public function __construct($file)
    {
        $this->file = $file;
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
      $fullpath = Storage::disk('local')->getDriver()
        ->getAdapter()->getPathPrefix();

      if (Storage::exists($this->file)) {
        try {
          $reader = Reader::createFromPath($fullpath . $this->file, 'r');

          $statement = (new Statement())
            ->offset(0)
            // ->limit(2000)
            ;

          $rows = $statement->process($reader);

          // Use the headers from AddressBase
          $header = explode(',', strtolower(
            'UPRN,OS_ADDRESS_TOID,UDPRN,ORGANISATION_NAME,DEPARTMENT_NAME,PO_BOX_NUMBER,SUB_BUILDING_NAME,BUILDING_NAME,BUILDING_NUMBER,DEPENDENT_THOROUGHFARE,THOROUGHFARE,POST_TOWN,DOUBLE_DEPENDENT_LOCALITY,DEPENDENT_LOCALITY,POSTCODE,POSTCODE_TYPE,X_COORDINATE,Y_COORDINATE,LATITUDE,LONGITUDE,RPC,COUNTRY,CHANGE_TYPE,LA_START_DATE,RM_START_DATE,LAST_UPDATE_DATE,CLASS'
          ));

          if (count(array_keys($rows->fetchOne())) == count($header)) {

              $cnt = 0;

              foreach($rows as $row) {
                // Create keyed array to make it easier to write to the model
                $data = array_combine($header, $row);

                // Fix the numeric requirement for building_number to handle null strings "" as null
                $data['building_number'] = ($data['building_number'] === "") ? null : $data['building_number'];

                switch ($data['change_type']) {
                  case 'I':
                    // Insert
                    $this->insertRow($data);
                    break;
                  case 'D':
                    // Delete
                    $this->deleteRow($data);
                    break;
                  case 'U':
                    $this->updateRow($data);
                    // Update
                    break;
                }
                $cnt++;
                if ($cnt % 1000 == 0)
                  Log::info($cnt++." Rows processed.");
              }

          }
          Storage::delete($this->file);
          Log::info('All rows processed');
          Cache::remember('gis.processuploads', 60 * 12, function() {
            return Carbon::Now();
          });
        } catch (\Exception $e) {
          Log::error($e->getMessage());
          Cache::forget('gis.processuploads');
        }
      }

  }

  public function buildAddress($row) {
    $address = '';
    if (trim($row['organisation_name']) !== '')
      $address .= $row['organisation_name'].', ';
    if (trim($row['department_name']) !== '')
      $address .= $row['department_name'].', ';
    if (trim($row['po_box_number']) !== '')
      $address .= $row['po_box_number'].', ';
    if (trim($row['sub_building_name']) !== '')
      $address .= $row['sub_building_name'].', ';
    if (trim($row['building_name']) !== '')
      $address .= $row['building_name'].', ';
    if(trim($row['building_number']) !== '')
      $address .= $row['building_number'].' ';
    if(trim($row['dependent_thoroughfare']) !== '')
      $address .= $row['dependent_thoroughfare'].', ';
    if (trim($row['thoroughfare']) !== '')
      $address .= $row['thoroughfare'].', ';
    if (trim($row['dependent_locality']) !== '')
      $address .= $row['dependent_locality'].', ';
    if (trim($row['double_dependent_locality']) !== '')
      $address .= $row['double_dependent_locality'].', ';
    $address .= $row['post_town'].', ';
    $address .= $row['postcode'];

    return $address;

  }

  public function insertRow($data) {
    // Try to insert row. If it already exists then do an update
    try {
      $addressbase = new AddressBase();
      foreach(array_keys($data) as $key) {
        $addressbase[$key] = $data[$key];
      }
      // $addressbase['full_address'] = $this->buildAddress($data);
      // $addressbase['geom'] = \DB::raw('public.ST_SetSRID(public.ST_MakePoint('.$data['latitude'].', '.$data['longitude'].'), 4326)');

      $addressbase->save();
    } catch (\PDOException $e) {
      if ($e->getCode() == 23505) { // Duplicate key
        $this->updateRow($data);
      } else {
        Log::error($e->getMessage());
        //$this->error($e->getMessage());
      }
    }

  }

  public function updateRow($data) {
    // Only update if the new row has a newer date
    // $addressbase['full_address'] = $this->buildAddress($data);
    // $addressbase['geom'] = \DB::raw('public.ST_SetSRID(public.ST_MakePoint('.$data['latitude'].', '.$data['longitude'].'), 4326)');

    $addressbase = AddressBase::where('uprn', '=', $data['uprn'])
      ->where('last_update_date', '<', $data['last_update_date'])
      ->update($data);
  }

  public function deleteRow($data) {
    // Only delete older records
    AddressBase::where('uprn', '=', $data['uprn'])
      ->where('last_update_date', '<', $data['last_update_date'])
      ->delete();
  }

  public function failed(FatalThrowableError $exception) {
    Log::error('failed triggered');
  }

}