drb/app/Jobs/ProcessWMSDataJob.php

<?php

namespace App\Jobs;

use App\Model\SyncTempRecord;
use App\Service\U8ThirdPartyService;
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 ProcessWMSDataJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $data;
    public $timeout = 60;

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

    public function handle()
    {
        try {
            list($bool, $msg) = $this->settle();
            if(! $bool) $this->finalDo($msg);
            Log::channel('queue_daily')->info('队列执行任务');
        } catch (\Throwable $e) {
            $this->finalDo("异常：" . $e->getMessage());
            Log::channel('queue_daily')->info('队列异常', ['msg' => $e->getMessage() . '|' . $e->getLine()]);
            $this->delete();
        }
    }

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

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

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

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

        $type = $record->type;
        if($type == SyncTempRecord::type_eight){
            //检验单
            list($status, $msg) = $this->bjOrder($record, $u8Data);
        }else{
            //同步到音飞
            list($status, $msg) = $this->orderInsert($record, $u8Data);
        }

        return [$status, $msg];
    }

    public function orderInsert($record, $u8Data){
        $orderType = SyncTempRecord::$map[$record->type];
        if(is_array($orderType)) $orderType = $orderType[$record->type_2];

        // 3. 组装报文
        $jsonBody = [
            'orderType'     => $orderType,
            'orderNo'       => $record->u8_no,
            'orderId'       => $record->u8_id,
            'operationType' => SyncTempRecord::$opMapping[$record->op_type],
            'lottar1'       => '',
            'lottar2'       => '',
            'lottar3'       => '',
            'lottar4'       => '',
            'lottar5'       => '',
            'lottar6'       => '',
            'lottar7'       => '',
            'lottar8'       => '',
            'lottar9'       => '',
            'details'       => []
        ];

        // 4. 接口路由
        $inboundTypes = [SyncTempRecord::type_one, SyncTempRecord::type_three, SyncTempRecord::type_six, SyncTempRecord::type_five];
        $path = in_array($record->type, $inboundTypes) ? '/erp/inbound' : '/erp/outbound';
        if($record->type_2 == 1 && $record->type == SyncTempRecord::type_one) $path = '/erp/outbound';
        $apiUrl = config('wms.api_url') . $path;

        if (isset($u8Data['details']) && is_array($u8Data['details'])) {
            if($path == '/erp/inbound'){ //入
                foreach ($u8Data['details'] as $item) {
                    $productDate = "";
                    if(! empty($item['dMDate'])) $productDate = date("Y-m-d", strtotime($item['dMDate']));
                    $validTime = 0;
                    if(! empty($item['iMassDate'])) $validTime = $item['iMassDate'];
                    $lot = "";
                    if(! empty($item['cBatch'])) $lot = $item['cBatch'];
                    $jsonBody['details'][] = [
                        'lineNum'       => $item['lineNum'] ?? 0,
                        'materialCode'  => $item['materialCode'] ?? '',
                        'planQty'       => (float)abs($item['planQty']),
                        'productDate'       => $productDate,
                        'validTime'       => $validTime,
                        'lot'       => $lot,
                        'lottar1'       => $item['cDefine23'] ?? "",
                        'lottar2'       => "",
                        'lottar3'       => '',
                        'lottar4'       => '',
                        'lottar5'       => '',
                        'lottar6'       => '',
                        'lottar7'       => '',
                        'lottar8'       => '',
                        'lottar9'       => '',
                    ];
                }
            }else{//出
                foreach ($u8Data['details'] as $item) {
                    $jsonBody['details'][] = [
                        'lineNum'       => $item['lineNum'] ?? 0,
                        'materialCode'  => $item['materialCode'] ?? '',
                        'planQty'       => (float)abs($item['planQty']),
                        'lottar1'       => "",
                        'lottar2'       => $item['cBatch'] ?? "",
                        'lottar3'       => '',
                        'lottar4'       => '',
                        'lottar5'       => '',
                        'lottar6'       => '',
                        'lottar7'       => '',
                        'lottar8'       => '',
                        'lottar9'       => '',
                    ];
                }
            }
        }

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

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

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

            // 同步维护快照表，保证下次 Command 比对正确
            if ($record->op_type == SyncTempRecord::opt_two) {
                DB::table('sync_snapshot')
                    ->where('type', $record->type)
                    ->where('u8_no', $record->u8_no)
                    ->where('u8_id', $record->u8_id)
                    ->delete();
            } else {
                DB::table('sync_snapshot')->updateOrInsert(
                    ['type' => $record->type, 'u8_no' => $record->u8_no],
                    ['u8_id' => $record->u8_id,'last_upd_time' => $record->u8_upd, 'order_date' => $u8Data['order_date'], 'payload' => json_encode($u8Data)]
                );
            }
        });

        return [true, ''];
    }

    //代码是有bug的 因为业务
    public function bjOrder($record, $u8Data){
        $hg_quantity = $u8Data['hg_quantity'] ?? 0;
        $rb_quantity = $u8Data['rb_quantity'] ?? 0;
        $hg_not_quantity = $u8Data['hg_not_quantity'] ?? 0;

        if($hg_quantity > 0){
            list($status, $msg) = $this->zj($record, $u8Data, 1);
        }elseif ($rb_quantity > 0) {
            list($status, $msg) = $this->zj($record, $u8Data,3);
        }elseif($hg_not_quantity > 0){
            list($status, $msg) = $this->zj($record, $u8Data,2);
        }

        return [$status ?? true, $msg ?? ""];
    }

    public function zj($record, $u8Data, $type){
        if($record['type_2'] == 1){
            $orderNo = $u8Data['from_order'];
        }else{
            $orderNo = $u8Data['bj_order'];
        }
        // 3. 组装报文
        $jsonBody = [
            'orderNo'       => $orderNo,
            'materialCode' => $u8Data['materialCode'],
            'lot' => $u8Data['lot'] ?? "",
            'inspectionStatus' => $type,
        ];

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

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

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

        //生成用友单据
        $service = new U8ThirdPartyService();
        //报检单类型
        $bg_type = $record['type_2'];
        if($bg_type == 1){
            //采购
            if($type == 1 || $type == 3){
                //合格 让步 =》 采购入库
                list($status, $msg) = $service->purchaseInByZj(['record' => $record, 'payload' => $u8Data,'cvouchtype' => 'QM03','type' => $type]);
            }else{
                //不合格 =》其他入库单
                list($status, $msg) = $service->otherInByZj(['record' => $record, 'payload' => $u8Data, 'cvouchtype' => 'QM03']);
            }
        }elseif ($bg_type == 2){
            //产品
            if($type == 1 || $type == 3){
                //合格 让步 =》 产成品入库
                list($status, $msg) = $service->productInByZj(['record' => $record, 'payload' => $u8Data, 'cvouchtype' => 'QM04','type' => $type]);
            }else{
                //不合格 =》其他入库单
                list($status, $msg) = $service->otherInByZj(['record' => $record, 'payload' => $u8Data, 'cvouchtype' => 'QM04']);
            }
        }else{
            //退货 不需要了
            list($status, $msg) = [true, ''];
        }

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

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

            // 同步维护快照表，保证下次 Command 比对正确
            if ($record->op_type == SyncTempRecord::opt_two) {
                DB::table('sync_snapshot')
                    ->where('type', $record->type)
                    ->where('u8_no', $record->u8_no)
                    ->where('u8_id', $record->u8_id)
                    ->delete();
            } else {
                DB::table('sync_snapshot')->updateOrInsert(
                    ['type' => $record->type, 'u8_no' => $record->u8_no],
                    ['u8_id' => $record->u8_id,'last_upd_time' => $record->u8_upd, 'order_date' => $u8Data['order_date'], '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('CURL错误', ["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));
    }
}