drb/app/Jobs/ProcessWMSArchiveDataJob.php

<?php

namespace App\Jobs;

use App\Model\SyncTempRecordProduct;
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\DB;
use Illuminate\Support\Facades\Log;
use Symfony\Component\Console\Output\ConsoleOutput;
use Symfony\Component\Console\Output\OutputInterface;

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

    protected $data;
    public $timeout = 30;

    public function __construct($data)
    {
        //record表
        $this->data = $data;
    }

    public function handle()
    {
        try {
            list($bool, $msg) = $this->settle();
            if(! $bool) $this->finalDo($msg);
        } catch (\Throwable $e) {
            $this->finalDo("异常：" . $e->getMessage());
            $this->delete();
        }
    }

    private function finalDo($msg){
        SyncTempRecordProduct::where('id', $this->data['id'])
            ->update(['error_msg' => $msg, 'status' => SyncTempRecordProduct::status_two]);
    }

    private function settle()
    {
        $id = $this->data['id'];
        $record = SyncTempRecordProduct::where('id', $id)->first();

        // 1. 安全检查：如果记录不存在，或者已经是成功状态(status_one)，则退出
        if (!$record || $record->status != SyncTempRecordProduct::status_zero) {
            return [true, '已处理或记录不存在'];
        }

        // 2. 解析 U8 原始数据
        $u8Data = is_array($record->payload) ? $record->payload : json_decode($record->payload, true);

        $jsonBody = [
            'operationType' => SyncTempRecordProduct::$opMapping[$record->op_type],
            'materialCode'  => $u8Data['product_code'],
            'materialName'  => $u8Data['product_name'],
            'materialSpec'  => $u8Data['product_size'] ?? '',
            'materialTypeCode' => $u8Data['product_category_code'],
            'materialTypeName' => $u8Data['product_category_name'],
            'unitCode'       => $u8Data['product_unit_title'],
            'netWeight'      => 0,
            'grossWeight'   => $u8Data['grossWeight'] ?? 0,
        ];

        // 4. 接口路由
        $path = '/erp/material';
        $apiUrl = config('wms.api_url') . $path;

        // 5. 调用
        list($status, $result) = $this->post_helper($apiUrl, $jsonBody, ['Content-Type: application/json']);

        if (! $status) {
            // 接口返回失败或网络失败
            $record->update(['status' => SyncTempRecordProduct::status_two, 'error_msg' => $result]);
            return [false, $result];
        }

        // 6. 接口返回成功 (200) 的后续操作
        DB::transaction(function () use ($record, $u8Data) {
            // 更新本地流水状态为成功
            $record->update([
                'status' => SyncTempRecordProduct::status_one,
            ]);

            // 同步维护快照表，保证下次 Command 比对正确
            if ($record->op_type == SyncTempRecordProduct::opt_two) {
                DB::table('sync_snapshot_product')
                    ->where('code', $u8Data['product_code'])
                    ->where('ufts', $u8Data['ufts'])
                    ->delete();
            } else {
                DB::table('sync_snapshot_product')->updateOrInsert(
                    ['code' => $u8Data['product_code']],
                    ['code' => $u8Data['product_code'], 'ufts' => $u8Data['ufts'], 'payload' => json_encode($u8Data)]
                );
            }
        });

        return [true, ''];
    }

    public function post_helper($url, $data, $header = [], $timeout = 30)
    {
        Log::channel('apiLog')->info('存货同步', ["api" => $url, "param" => $data]);

        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_POST, 1);
        curl_setopt($ch, CURLOPT_HTTPHEADER, $header);
        curl_setopt($ch, CURLOPT_SSL_VERIFYPEER, false);
        curl_setopt($ch, CURLOPT_TIMEOUT, $timeout);
        if(!is_null($data)) curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($data));

        $r = curl_exec($ch);

        if ($r === false) {
            $errorMessage = curl_error($ch);
            curl_close($ch);
            Log::channel('apiLog')->error('存货同步异常:WMS_CURL_ERROR', ["msg" => $errorMessage]);
            return [false, "网络错误: " . $errorMessage];
        }
        curl_close($ch);

        $return = json_decode($r, true);
        Log::channel('apiLog')->info('存货同步返回结果', ["res" => $return]);

        // 判断逻辑：code 存在且为 200 才算 true
        if (isset($return['code']) && $return['code'] == 200) {
            return [true, $return];
        }

        $msg = $return['message'] ?? '未知接口错误';
        return [false, $msg];
    }

    protected function echoMessage(OutputInterface $output)
    {
        //输出消息
        $output->writeln(json_encode($this->data));
    }
}