baoshi-was/app/Services/StationTaskBatchService.php

<?php


namespace App\Services;


use App\Exceptions\ErrorException;
use App\StationTaskBatch;
use App\StationTaskCommodity;
use App\TaskTransaction;
use Carbon\Carbon;
use Exception;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use App\Traits\ServiceAppAop;
use Illuminate\Support\Facades\DB;


class StationTaskBatchService
{
    use ServiceAppAop;
    protected $modelClass=StationTaskBatch::class;
    /** @var StationService $stationService */
    private $stationService;
    /** @var StationTaskBatchTypeService $stationTaskBatchTypeService */
    private $stationTaskBatchTypeService;
    /** @var BatchService $batchService */
    private $batchService;
    /** @var StationTypeService $stationTypeService */
    private $stationTypeService;
    /** @var StationTaskService $stationTaskService */
    private $stationTaskService;
    /** @var ForeignHaiRoboticsService $foreignHaiRoboticsService */
    private $foreignHaiRoboticsService;

    public function __construct()
    {
        $this->stationService = null;
        $this->stationTypeService = null;
        $this->stationTaskBatchTypeService = null;
        $this->batchService = null;
        $this->stationTaskService = null;
        $this->foreignHaiRoboticsService = null;
    }

    /**
     * @param Collection $batches Batch[]
     * @param Collection $stationTasks_toAttach
     * @return Collection
     * @throws Exception
     */
    function createByBatches(Collection $batches, Collection $stationTasks_toAttach): Collection
    {
        $this->stationService = app('StationService');
        $this->stationTaskService = app('StationTaskService');
        $this->stationTypeService = app('StationTypeService');
        $this->stationTaskBatchTypeService = app('StationTaskBatchTypeService');
        $this->batchService = app('BatchService');

        $stationTaskBatches_toCreate = new Collection();
        $stationTaskBatchType = $this->stationTaskBatchTypeService->firstByWhere('name', 'U型线分捡');
        $id_stationTaskBatchType=$stationTaskBatchType['id']??'';

        $batches_handled = collect();
        foreach ($batches as $batch) {
            if ($batch['status'] != '已处理') {
                $stationType = $this->stationTypeService->getByBatch($batch);
                $station = $this->stationService->getStation_byType($stationType['name']);
                $stationTaskBatches_toCreate->push(
                    new StationTaskBatch([
                        'batch_id' => $batch['id'],
                        'station_id' => $station['id'],
                        'station_task_batch_type_id' => $id_stationTaskBatchType,
                        'status' => '待处理'
                    ])
                );
                $batches_handled->push($batch);
            }
        }
        $this->batchService->updateWhereIn('id', data_get($batches_handled, '*.id'), ['status' => '处理中']);
        $stationTaskBatches_toCreate=$stationTaskBatches_toCreate->reverse();//这里的波次顺序是反的，不知为什么，所以反向一次就好了。其他解耦的地方都是正序，必须保持一致
        $this->insert($stationTaskBatches_toCreate->toArray());

        $stationTaskBatches_toCreate=$this->getAndAttachIds($stationTaskBatches_toCreate);
        $this->stationTaskService->registerSubTasks(
            $stationTasks_toAttach,
            $stationTaskBatches_toCreate->map(function($taskBatch){
                return [$taskBatch];
            })
        );
        $this->stationTaskService->registerStations(
            $stationTasks_toAttach,
            data_get($stationTaskBatches_toCreate,'*.station_id')
        );
        return $stationTaskBatches_toCreate;
    }

    function getAndAttachIds($stationTaskBatches): Collection
    {
        $md5=md5(is_array($stationTaskBatches)
            ?$md5=json_encode($stationTaskBatches):$stationTaskBatches->toJson());

        return Cache::remember(
            'StationTaskBatch_'.$md5??md5(json_encode($stationTaskBatches->toArray()))
            ,config('cache.expirations.rarelyChange')
            ,function()use($stationTaskBatches){
            return StationTaskBatch::query()
                ->whereIn('status',data_get($stationTaskBatches,'*.status'))
                ->whereIn('batch_id',data_get($stationTaskBatches,'*.batch_id'))
                ->orderByDesc('id')
                ->limit(count($stationTaskBatches))
                ->get();
        });
    }


    function markManyExcepted(Collection $stationTaskBatches_failed)
    {
        foreach (
            $stationTaskBatches_failed
            as $stationTaskBatch) {
            if($stationTaskBatch['status']!='异常')
                $this->markExcepted($stationTaskBatch);
        }

        ($logAtFailings_andWait =
            function ($stationTaskBatches_failed) {
                if ($stationTaskBatches_failed->isEmpty()) return;
                throw new ErrorException('任务波次异常失败的');
            })($stationTaskBatches_failed);
    }

    /**
     * @param Collection|null $stationTaskBatches
     * @param string $locationType
     * @return Collection|\Tightenco\Collect\Support\Collection|null 返回执行失败的记录
     * @throws ErrorException
     */
    function runMany(?Collection $stationTaskBatches,string $locationType = 'OUTBIN-U-SHAPE-LINE'):?Collection
    {
        LogService::log(__METHOD__,'runMany','波次任务分配6.1：'.json_encode($stationTaskBatches));
        $stationTaskBatches_failed = null;
        ($execute = function(Collection $stationTaskBatches, &$stationTaskBatches_failed)use($locationType){
                if ($stationTaskBatches->isEmpty()) return; //波次任务不存在 跳出
                $stationTaskBatches_failed = collect();
                foreach ($stationTaskBatches as $stationTaskBatch){
                    $failed = !$this->run($stationTaskBatch, $locationType); //运行波次 获取执行结果
                    if ($failed) $stationTaskBatches_failed->push($stationTaskBatch);//执行失败  记录失败波次
                }
        })($stationTaskBatches, $stationTaskBatches_failed);

        (function ($stationTaskBatches_failed) {
                if ($stationTaskBatches_failed->isEmpty()) return;
                $retry_after_sec = config('task.batchTask.retry_after_sec');
                LogService::log(__METHOD__, __FUNCTION__,
                    '任务波次没有执行完的，' . $retry_after_sec . '秒后准备重试：' . $stationTaskBatches_failed->toJson()
                    . '调用堆栈r：' . json_encode(array_slice(debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS), 0, 3))
                );
                sleep($retry_after_sec);
            })($stationTaskBatches_failed);
        $execute($stationTaskBatches_failed, $stationTaskBatches_failed); //再次尝试
        $this->markManyExcepted($stationTaskBatches_failed);
        return  $stationTaskBatches_failed;
    }

    /**
     * 解析波次任务 任务下发海柔
     *
     * @param StationTaskBatch $stationTaskBatch
     * @param string $locationType
     * @return bool
     * @throws ErrorException
     */
    function run(StationTaskBatch $stationTaskBatch,string $locationType): bool
    {
        $this->instant($this->foreignHaiRoboticsService,'ForeignHaiRoboticsService');
        $this->instant($this->stationService,'StationService');
        $stationTaskBatch->loadMissing(['station','stationTask.stationTaskMaterialBoxes']);
        LogService::log(__METHOD__,'runMany','波次任务分配6.r2：'.json_encode($stationTaskBatch));

        $groupPrefix        = $stationTaskBatch['id'];//将波次任务ID当作组前缀
        $taskMaterialBoxes  = $stationTaskBatch['stationTask']['stationTaskMaterialBoxes'] ??
            (function () use ($stationTaskBatch) {
                LogService::log(__METHOD__,'runMany','波次任务分配6.r4：'.json_encode($stationTaskBatch));
                throw new Exception('找不到料箱:' . json_encode($stationTaskBatch));
            })();//存在任务返回任务 否则抛出无料箱异常
        DB::beginTransaction();
        try{
            //获取放线入口
            switch ($locationType){
                case 'OUTBIN-U-SHAPE-LINE'://U型线
                    $toLocation         = $this->stationService->getULineEntrance($stationTaskBatch['station'])['code'];
                    $isFetchedFromRobotics = $this->foreignHaiRoboticsService->
                    fetchGroup($toLocation, $taskMaterialBoxes, $groupPrefix);//执行料箱任务
                    break;
                case 'OUTBIN-CACHE-SHELF'://缓存架
                    list($toLocation, $taskMaterialBoxes, $map) = $this->apportionLocation($taskMaterialBoxes);
                    if ($toLocation->count()>0){
                        $isFetchedFromRobotics = $this->foreignHaiRoboticsService->
                        fetchGroup_multiLocation($toLocation, $taskMaterialBoxes, $groupPrefix, '立架出至缓存架',20,false);
                        foreach ($toLocation as $value){
                            app("CacheShelfService")->lightUp($value,'3','0',["title"=>"机器人取箱中，禁止操作"]);
                            Cache::forever("CACHE_SHELF_OCCUPANCY_{$map[$value]}",true);
                        }
                        app("StationService")->locationOccupyMulti($toLocation->toArray());
                    }else $isFetchedFromRobotics = true;
                    break;
                default:
                    $isFetchedFromRobotics = false;
            }
            DB::commit();
            LogService::log(__METHOD__,'runMany','波次任务分配6.r6：'.json_encode($stationTaskBatch));
        }catch(Exception $e){
            DB::rollBack();
            throw new ErrorException('$stationTaskBatch运行波次机器人任务失败，获取组失败: '.$stationTaskBatch->toJson() . $e->getMessage());
        }
        (function()use($isFetchedFromRobotics,$stationTaskBatch){
            $isFetchedFromRobotics?
                $this->markProcessing($stationTaskBatch)://执行成功标记已处理
                $this->markExcepted($stationTaskBatch);//执行失败标记失败
        })();
        return $isFetchedFromRobotics;//返回执行结果
    }

    function markProcessing($stationTaskBatch_orCollection)
    {
        if (get_class($stationTaskBatch_orCollection)==StationTaskBatch::class){
            $stationTaskBatch_orCollection = collect([$stationTaskBatch_orCollection]);
        }
        $taskIds = data_get($stationTaskBatch_orCollection, '*.id');
        $this->markProcessing_byIds($taskIds);
        //app("StorageService")->handleStorage($stationTaskBatch_orCollection);
    }
    function markProcessing_byIds($ids)
    {
        if(!$ids)$ids=[];
        if(!is_array($ids))$ids=[$ids];


        $hasProcessing=StationTaskBatch::query()
            ->whereNotIn('id', $ids)
            ->where(['status'=>'处理中'])
            ->where('created_at','>',Carbon::now()->subDay())
            ->get('id')->isNotEmpty();

        $status = '处理中';
        if($hasProcessing)
            $status = '处理队列';

        StationTaskBatch::query()
            ->whereIn('id', $ids)
            ->update(['status'=>$status]);
    }

    function markProcessed($stationTaskBatch_orCollection)
    {
        if (get_class($stationTaskBatch_orCollection)==StationTaskBatch::class){
            $stationTaskBatch_orCollection = collect([$stationTaskBatch_orCollection]);
        }
        $this->markProcessed_byIds(data_get($stationTaskBatch_orCollection, '*.id'));
    }
    function markProcessed_byIds($ids)
    {
        if(!$ids)$ids=[];
        if(!is_array($ids))$ids=[$ids];
        StationTaskBatch::query()
            ->whereIn('id', $ids)
            ->update(['status'=>'完成']);
    }

    function markExcepted(StationTaskBatch $stationTaskBatch)
    {
        $stationTaskBatch['status'] = '异常';
        $stationTaskBatch           ->update();
    }

    /**
     * 为任务分配缓存架库位
     *
     * @param \Illuminate\Database\Eloquent\Collection $taskMaterialBoxes
     *
     * @return array
     */
    public function apportionLocation(Collection $taskMaterialBoxes):array
    {
        $taskMaterialBoxes->loadMissing(["stationTaskCommodities.order","stationTaskCommodities.commodity.barcodes"]);
        //获取可用的库位 加行锁
        $stations = app("StationService")->getCacheShelf(true);
        $location = [];
        $map = [];
        foreach ($stations as $station){
            $location[] = $station->code;
            $map[$station->code] = $station->id;
        }
        /** @var Collection $handleTask */
        $handleTask = $taskMaterialBoxes->splice(0,count($location));
        $toLocation = collect();
        $updateTask = [["id","station_id"]];
        $insertTransaction = [];
        $dateTime = date("Y-m-d H:i:s");
        $exeInsert = function ($task,$taskCommodity,$status)use(&$insertTransaction,$dateTime){
            $insertTransaction[] = [
                "doc_code" => $taskCommodity->order->code ?? "",
                "bar_code" => $taskCommodity->commodity->barcodes[0]->code ?? "",
                "to_station_id" => $task->station_id,
                "material_box_id" => $task->material_box_id,
                "task_id" => $task->id,
                "commodity_id" => $taskCommodity->commodity_id,
                "amount" => $taskCommodity->amount,
                "type" => "出库",
                "status" => $status,
                "mark" => 2,
                "bin_number"=>$taskCommodity->bin_number,
                "created_at"=>$dateTime,
                "updated_at"=>$dateTime,
            ];
        };
        foreach ($handleTask as $index=>$task){
            $task->station_id = $map[$location[$index]];
            $toLocation->push($location[$index]);
            $handleTask->offsetSet($index,$task);
            $updateTask[] = ["id"=>$task->id,"station_id"=>$task->station_id];
            foreach ($task->stationTaskCommodities as $taskCommodity)$exeInsert($task,$taskCommodity,0);
        }
        if ($handleTask->count()>0)app("BatchUpdateService")->batchUpdate("station_task_material_boxes",$updateTask);
        foreach ($taskMaterialBoxes as $obj)foreach ($obj->stationTaskCommodities as $taskCommodity)$exeInsert($obj,$taskCommodity,3);
        TaskTransaction::query()->insert($insertTransaction);
        return array($toLocation, $handleTask, $map);
    }
}