zhijiayun/app/Jobs/Manager/Process/ViolationProductDataJobs.php

<?php

namespace App\Jobs\Manager\Process;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use App\Facades\Servers\Logs\Log;
use App\Models\Manager\Process\ViolationProduct as ViolationProductModel;
use App\Models\Manager\Personnel\Employee as EmployeeModel;
use App\Models\Api\Process\ExecuteLog as ExecuteLogModel;
use App\Models\Manager\WashConfig\ViolationCompany as ViolationCompanyModel;
use App\Models\Manager\WashConfig\ViolationCompanyMember as ViolationCompanyMemberModel;
use App\Models\Manager\Citys as CitysModel;
use App\Models\Manager\Personnel\EmployeePlatform as EmployeePlatformModel;
use App\Models\Manager\Personnel\EmployeeArea as EmployeeAreaModel;
use App\Models\Manager\WashConfig\ViolationStore as ViolationStoreModel;

/**
 * 数据清洗-违规挂网商品数据队列
 * @author  唐远望
 * @version 1.0
 * @date  2025-12-10
 */
class ViolationProductDataJobs implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public $tries = 3; // 限制重试次数
    public $timeout = 600; // 5分钟超时

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

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        try {
            $this->getViolationProductData($this->message_data);
        } catch (\Exception $e) {
            Log::info('job_error', '数据清洗-违规挂网商品数据队列失败', ['data' => $this->message_data, 'error' => $e->getMessage()]);
        }
    }


    /**
     * 采集商品数据清洗
     * @author  唐远望
     * @version 1.0
     * @date  2025-12-10
     */
    public function getViolationProductData($message_data)
    {
        $CitysModel = new CitysModel();
        $EmployeeModel = new EmployeeModel();
        $ViolationProductModel = new ViolationProductModel();
        $ViolationCompanyModel = new ViolationCompanyModel();
        $ViolationCompanyMemberModel = new ViolationCompanyMemberModel();
        $EmployeePlatformModel = new EmployeePlatformModel();
        $EmployeeAreaModel = new EmployeeAreaModel();
        $platform = $message_data['platform']; //多个平台配置
        $product_name = $message_data['product_name']; //商品名称
        $enable_full_quantity = $message_data['enable_full_quantity']; //是否启用全量采集0=是 1=否
        $product_specs = $message_data['product_specs']; //商品规格
        $store_scope = $message_data['store_scope']; //店铺范围1=全部店铺2=指定店铺
        $company_scope = $message_data['company_scope']; //公司范围1=全部公司2=指定公司
        $social_credit_code = $message_data['social_credit_code']; //社会信用代码
        $executeLog_id = $message_data['executeLog_id'];
        $category_name = $message_data['category_name']; //产品分类名称
        $specify_responsible_person = $message_data['specify_responsible_person']; //指派责任人 0=开启 1=关闭
        $company_id = isset($message_data['company_id']) ? $message_data['company_id'] : 0; //品牌方公司ID
        $product_brand = $message_data['product_brand']; //商品品牌名称
        $product_keywords = $message_data['product_keyword']; //关键字
        $product_data = isset($this->message_data['product_data']) ? $this->message_data['product_data'] : [];
        $index_number = isset($this->message_data['index_number']) ? $this->message_data['index_number'] : 0;
        $data_totle = isset($this->message_data['data_totle']) ? $this->message_data['data_totle'] : 0;

        //-------------------------------------- 清洗规则(开始) --------------------------------------
        //过滤没有链接数据
        if (trim($product_data['link_url']) == '')  return true;
        //数据是否命中商品名称
        $hit_product_name = false;
        if (strpos($product_data['product_name'], $product_name) !== false) {
            $hit_product_name = true;
        }
        if ($hit_product_name != true) return true;
        //数据是否命中规格
        $hit_product_specs = false;
        if ($enable_full_quantity == 1 && $product_specs) {
            //product_name 包含规格
            if (strpos($product_data['product_name'], $product_specs) !== false) {
                $hit_product_specs = true;
            } else if (strpos($product_data['product_specs'], $product_specs) !== false) {
                $hit_product_specs = true;
            }
        }
        if ($hit_product_specs != true) return true;
        //数据是否命中关键字
        $hit_product_keywords = false;
        if (!empty($product_keywords)) {
            $product_keywords = explode(',', $product_keywords);
            foreach ($product_keywords as $product_keyword) {
                if (empty($product_keyword)) continue;
                if (strpos($product_data['product_name'], $product_keyword) !== false) {
                    $hit_product_keywords = true;
                    break;
                }
            }
        }
        if ($hit_product_keywords != true) return true;
        //数据是否命中指定公司
        $hit_company_scope = false;
        if ($company_scope == 2 && $social_credit_code != '') {
            if ($product_data['qualification_number'] == $social_credit_code) {
                $hit_company_scope = true;
            }
        }
        if ($hit_company_scope != true) return true;
        //数据是否命中平台
        $hit_platform = false;
        if ($platform != '0' && !empty($platform)) {
            $platform = explode(',', $platform);
            if (in_array($product_data['platform_id'], $platform)) {
                $hit_platform = true;
            }
        }
        if ($hit_platform != true) return true;
        //-------------------------------------- 清洗规则(结束) --------------------------------------



        //-------------------------------------- 处理营业执照地区信息(开始) --------------------------------------
        $province_name = $product_data['province_name'];
        $city_name = $product_data['city_name'];
        //特殊地区1级移除市
        if ($province_name && in_array($province_name, ['北京市', '天津市', '上海市', '重庆市'])) {
            //移除市这个字符
            $province_name = trim(str_replace('市', '', $province_name));
        } else if ($province_name && in_array($province_name, ['北京', '天津', '上海', '重庆'])) {
        } else if ($province_name && in_array($province_name, ['内蒙古', '广西', '西藏', '新疆', '宁夏'])) {
            switch ($province_name) {
                case '内蒙古':
                    $province_name = '内蒙古自治区';
                    break;
                case '广西':
                    $province_name = '广西壮族自治区';
                    break;
                case '西藏':
                    $province_name = '西藏自治区';
                    break;
                case '新疆':
                    $province_name = '新疆维吾尔自治区';
                    break;
                case '宁夏':
                    $province_name = '宁夏回族自治区';
                    break;
            }
        } else if ($province_name && in_array($province_name, ['内蒙古自治区', '广西壮族自治区', '西藏自治区', '新疆维吾尔自治区', '宁夏回族自治区'])) {
            //完整匹配不做处理
        } else if (trim($province_name) != '' && strpos($province_name, '省') === false) {
            //是否存在市省，如果不存在则补全
            $province_name = $province_name . '省';
        }
        if (trim($province_name) != '') {
            //根据最新处理后的省份名称获取省份ID
            $db_province_id = $CitysModel->where([['name', '=', $province_name], ['level', '=', '1']])->value('id');
            if (!empty($db_province_id)) {
                $product_data['province_id'] = $db_province_id;
            } else {
                //逆推省份ID失败，则修改为未知名称
                $province_name = '未知';
            }
        } else {
            //尝试从地区详情里面匹配省份名称
            $db_province_data_list = $CitysModel->where([['level', '=', '1']])->select(['id', 'name'])->get()->toarray();
            foreach ($db_province_data_list as $db_province_data) {
                $db_province_name = $db_province_data['name'];
                if (strpos($product_data['area_info'], $db_province_name) !== false) {
                    $province_name = $db_province_name;
                    $product_data['province_id'] = $db_province_data['id'];
                    break;
                }
            }
        }
        if (trim($city_name) != '') {
            //校验是否存在县这个字眼
            if (strpos($city_name, '县') !== false) {
                $db_city_id = $CitysModel->where([['name', '=', $city_name], ['level', '=', '2']])->value('id');
                if (!empty($db_city_id)) {
                    $product_data['city_id'] = $db_city_id;
                }
            } elseif (trim($city_name) != '' && strpos($city_name, '市') === false) {
                //是否存在市
                $city_name = $city_name . '市';
                //根据最新处理后的市名称获取市ID
                $db_city_id = $CitysModel->where([['name', '=', $city_name], ['level', '=', '2']])->value('id');
                if (!empty($db_city_id)) {
                    $product_data['city_id'] = $db_city_id;
                } else {
                    $city_name = '未知';
                }
            } else {
                $db_city_id = $CitysModel->where([['name', '=', $city_name], ['level', '=', '2']])->value('id');
                if (!empty($db_city_id)) {
                    $product_data['city_id'] = $db_city_id;
                } else {
                    $city_name = '未知';
                }
            }
        } else {
            if (trim($product_data['area_info']) != '') {
                //尝试从地区详情里面匹配市名称
                $db_city_name_list = $CitysModel->where([['level', '=', '2']])->pluck('name')->toarray();
                foreach ($db_city_name_list as $db_city_name) {
                    if (strpos($product_data['area_info'], $db_city_name) !== false) {
                        $db_city_id = $CitysModel->where([['name', '=', $db_city_name], ['level', '=', '2']])->value('id');
                        $product_data['city_id'] = $db_city_id;
                        $product_data['city_name'] = $db_city_name;
                        $city_name = $db_city_name;
                        break;
                    }
                }
            }
        }
        //如果存在市区city_id 则逆推省份ID以及名称
        if (!empty($product_data['city_id']) && empty($product_data['province_id'])) {
            $db_province_id = $CitysModel->where([['id', '=', $product_data['city_id']], ['level', '=', '2']])->value('pid');
            $db_province_name = $CitysModel->where([['id', '=', $db_province_id], ['level', '=', '1']])->value('name');
            $product_data['province_id'] = $db_province_id;
            $product_data['province_name'] = $db_province_name;
            $province_name = $db_province_name;
        }
        if (trim($product_data['area_info']) == '' && trim($city_name) != '' && trim($province_name) != '') {
            $product_data['area_info'] = $province_name . $city_name;
        }
        //-------------------------------------- 处理营业执照地区信息(结束) --------------------------------------

        //-------------------------------------- 处理发货省份地区信息(开始) --------------------------------------
        $shipment_province_name = $product_data['shipment_province_name'];
        $shipment_city_name = $product_data['shipment_city_name'];
        //特殊地区1级移除市
        if ($shipment_province_name && in_array($shipment_province_name, ['北京市', '天津市', '上海市', '重庆市'])) {
            //移除市这个字符
            $shipment_province_name = trim(str_replace('市', '', $shipment_province_name));
        } else if ($shipment_province_name && in_array($shipment_province_name, ['北京', '天津', '上海', '重庆'])) {
        } else if ($shipment_province_name && in_array($shipment_province_name, ['内蒙古', '广西', '西藏', '新疆', '宁夏'])) {
            switch ($shipment_province_name) {
                case '内蒙古':
                    $shipment_province_name = '内蒙古自治区';
                    break;
                case '广西':
                    $shipment_province_name = '广西壮族自治区';
                    break;
                case '西藏':
                    $shipment_province_name = '西藏自治区';
                    break;
                case '新疆':
                    $shipment_province_name = '新疆维吾尔自治区';
                    break;
                case '宁夏':
                    $shipment_province_name = '宁夏回族自治区';
                    break;
            }
        } else if ($shipment_province_name && in_array($shipment_province_name, ['内蒙古自治区', '广西壮族自治区', '西藏自治区', '新疆维吾尔自治区', '宁夏回族自治区'])) {
            //完整匹配不做处理
        } else if (trim($shipment_province_name) != '' && strpos($shipment_province_name, '省') === false) {
            //是否存在市省，如果不存在则补全
            $shipment_province_name = $shipment_province_name . '省';
        }
        if (trim($shipment_province_name) != '') {
            //根据最新处理后的省份名称获取省份ID
            $db_shipment_province_id = $CitysModel->where([['name', '=', $shipment_province_name], ['level', '=', '1']])->value('id');
            if (!empty($db_shipment_province_id)) {
                $product_data['shipment_province_id'] = $db_shipment_province_id;
            } else {
                //逆推省份ID失败，则修改为未知名称
                $shipment_province_name = '未知';
            }
        }
        if (trim($shipment_city_name) != '') {
            //校验是否存在县这个字眼
            if (strpos($shipment_city_name, '县') !== false) {
                $db_shipment_city_id = $CitysModel->where([['name', '=', $shipment_city_name], ['level', '=', '2']])->value('id');
                if (!empty($db_shipment_city_id)) {
                    $product_data['shipment_city_id'] = $db_shipment_city_id;
                }
            } elseif (trim($shipment_city_name) != '' && strpos($shipment_city_name, '市') === false) {
                //是否存在市
                $shipment_city_name = $shipment_city_name . '市';
                //根据最新处理后的市名称获取市ID
                $db_shipment_city_id = $CitysModel->where([['name', '=', $shipment_city_name], ['level', '=', '2']])->value('id');
                if (!empty($db_shipment_city_id)) {
                    $product_data['shipment_city_id'] = $db_shipment_city_id;
                } else {
                    //逆推市ID失败，则修改为未知名称
                    $shipment_city_name = '未知';
                }
            } else {
                $db_shipment_city_id = $CitysModel->where([['name', '=', $shipment_city_name], ['level', '=', '2']])->value('id');
                if (!empty($db_shipment_city_id)) {
                    $product_data['shipment_city_id'] = $db_shipment_city_id;
                } else {
                    //逆推市ID失败，则修改为未知名称
                    $shipment_city_name = '未知';
                }
            }
        }
        //如果存在市区shipment_city_id 则逆推省份ID以及名称
        if (!empty($product_data['shipment_city_id']) && empty($product_data['shipment_province_id'])) {
            $db_shipment_province_id = $CitysModel->where([['id', '=', $product_data['shipment_city_id']], ['level', '=', '2']])->value('pid');
            $db_shipment_province_name = $CitysModel->where([['id', '=', $db_shipment_province_id], ['level', '=', '1']])->value('name');
            $product_data['shipment_province_id'] = $db_shipment_province_id;
            $product_data['shipment_province_name'] = $db_shipment_province_name;
        }
        //-------------------------------------- 处理发货省份地区信息(结束) --------------------------------------
        //处理链接信息
        preg_match('/https?:\/\/[^\s\'"<>]+/i', $product_data['link_url'], $matches);
        if (!empty($matches)) {
            $product_data['link_url'] = $matches[0];
        }
        //合并省份市区ID
        $merge_city_id = 0;
        $merge_province_id = 0;
        $merge_city_name = '';
        $merge_province_name = '';
        if (!empty($product_data['city_id']) && !empty($product_data['province_id'])) {
            $merge_city_id = $product_data['city_id'];
            $merge_province_id = $product_data['province_id'];
            $merge_city_name = $city_name;
            $merge_province_name = $province_name;
        } else if (!empty($product_data['shipment_city_id']) && !empty($product_data['shipment_province_id'])) {
            $merge_city_id = $product_data['shipment_city_id'];
            $merge_province_id = $product_data['shipment_province_id'];
            $merge_city_name = $shipment_city_name;
            $merge_province_name = $shipment_province_name;
        }
        $insert_product_data = [
            'company_id' => $company_id,
            'source_id' => $product_data['id'],
            'platform'    => $product_data['platform_id'],
            'company_name'    => $product_data['company_name'],
            'product_brand'    => $product_brand != '' ? $product_brand : $product_data['product_brand'],
            'product_name'    => $product_name,
            'product_specs'    => $product_specs,
            'inventory'    => $product_data['inventory'] ? $product_data['inventory'] : '',
            'sales'    => $product_data['sales'] ? $product_data['sales'] : '',
            'snapshot_url' => $product_data['snapshot_url'] ? $product_data['snapshot_url'] : '',
            'online_posting_count'    => $product_data['online_posting_count'],
            'continuous_listing_count'    => $product_data['continuous_listing_count'],
            'link_url'    => $product_data['link_url'],
            'store_name'    => $product_data['store_name'],
            'anonymous_store_name'    => $product_data['anonymous_store_name'],
            'social_credit_code'    => $product_data['qualification_number'],
            'province_id'    => $product_data['province_id'],
            'province_name' => $province_name,
            'city_id' => $product_data['city_id'],
            'city_name' => $city_name,
            'area_info' => $product_data['area_info'],
            'category_name' => $category_name,
            'company_category_name' => '',
            'first_responsible_person' => '',
            'responsible_person' => '',
            'source_responsible_person' => '',
            'scrape_date' => $product_data['scrape_date'],
            'collection_time' => strtotime($product_data['insert_time']),
            'shipment_province_id' => $product_data['shipment_province_id'],
            'shipment_province_name' => $product_data['shipment_province_name'],
            'shipment_city_id' => $product_data['shipment_city_id'],
            'shipment_city_name' => $product_data['shipment_city_name'],
            'collect_config_info' => $product_data['collect_config_info'],
            'merge_city_id' => $merge_city_id,
            'merge_province_id' => $merge_province_id,
            'merge_city_name' => $merge_city_name,
            'merge_province_name' => $merge_province_name,
        ];
        //获取指定人员信息
        if ($specify_responsible_person == 0) {
            //获取店铺责任人信息
            $ViolationStoreModel = new ViolationStoreModel();
            $store_data_info = $ViolationStoreModel->where('store_name', $product_data['store_name'])->first();
            if (!empty($store_data_info) && trim($store_data_info->employee_ids) != '') {
                //查询店铺第一责任人信息
                $EmployeeModel = new EmployeeModel();
                $store_where_query = [];
                $employee_id_store = explode(',', $store_data_info->employee_ids);
                $store_where_query[] = ['company_id', '=', $company_id];
                $store_where_query[] = ['id', 'in', $employee_id_store];
                $store_where_query[] = ['status', '=', 0];
                $store_where_query[] = ['duty_type', '=', 1]; //责任类型1=第一责任人，2=责任人
                $first_responsible_person = $EmployeeModel->pluck('id')->implode(',');
                $insert_product_data['first_responsible_person'] = $first_responsible_person;
                //查询店铺责任人信息
                $EmployeeModel = new EmployeeModel();
                $store_where_query = [];
                $employee_id_store = explode(',', $store_data_info->employee_ids);
                $store_where_query[] = ['company_id', '=', $company_id];
                $store_where_query[] = ['id', 'in', $employee_id_store];
                $store_where_query[] = ['status', '=', 0];
                $store_where_query[] = ['duty_type', '=', 2]; //责任类型1=第一责任人，2=责任人
                $responsible_person = $EmployeeModel->pluck('id')->implode(',');
                $insert_product_data['responsible_person'] = $responsible_person;
                //溯源责任人
                $source_responsible_person = '';
                if ($first_responsible_person && $responsible_person) {
                    //转换成数组，合并后在去重
                    $first_responsible_person = explode(',', $first_responsible_person);
                    $responsible_person = explode(',', $responsible_person);
                    $source_responsible_person = array_unique(array_merge($first_responsible_person, $responsible_person));
                    $source_responsible_person = ',' . implode(',', $source_responsible_person) . ',';
                } else if ($first_responsible_person) {
                    $source_responsible_person = $first_responsible_person;
                } else if ($responsible_person) {
                    $source_responsible_person = $responsible_person;
                }
                $insert_product_data['source_responsible_person'] = $source_responsible_person;
            } else {
                //获取公司绑定责任人信息
                $company_data = $ViolationCompanyModel->leftjoin('washconfig_company_category', 'washconfig_company_category.id', '=', 'washconfig_violation_company.category_id')
                    ->where('washconfig_violation_company.social_credit_code', $product_data['qualification_number'])
                    ->where('washconfig_violation_company.company_id', $company_id)
                    ->select(['washconfig_violation_company.id', 'washconfig_company_category.name as category_name'])->first();
                $employee_id_list = [];
                if ($company_data) {
                    $employee_id_list = $ViolationCompanyMemberModel->where('company_logid', $company_data->id)->pluck('employee_id')->toarray();
                    $insert_product_data['company_category_name'] = $company_data->category_name ? $company_data->category_name : '';
                }
                $where_query1 = [];
                $where_query2 = [];
                $where_query3 = [];
                //查询指定公司第一责任人
                if (!empty($employee_id_list)) {
                    $where_query1[] = ['company_id', '=', $company_id];
                    $where_query1[] = ['id', 'in', $employee_id_list];
                    $where_query1[] = ['status', '=', 0];
                    $where_query1[] = ['duty_type', '=', 1]; //责任类型1=第一责任人，2=责任人
                }
                //查询地区配置的第一责任人
                $employee_id_area = $EmployeeAreaModel->where('city_id', $product_data['city_id'])->pluck('employee_id')->toarray();
                if (!empty($employee_id_area)) {
                    $where_query2[] = ['company_id', '=', $company_id];
                    $where_query2[] = ['id', 'in', $employee_id_area];
                    $where_query2[] = ['status', '=', 0];
                    $where_query2[] = ['duty_type', '=', 1]; //责任类型1=第一责任人，2=责任人
                }
                //查询平台配置的第一责任人
                $employee_id_platform = $EmployeePlatformModel->where('platform_id', $product_data['platform_id'])->pluck('employee_id')->toarray();
                if (!empty($employee_id_platform)) {
                    $where_query3[] = ['company_id', '=', $company_id];
                    $where_query3[] = ['id', 'in', $employee_id_platform];
                    $where_query3[] = ['status', '=', 0];
                    $where_query3[] = ['duty_type', '=', 1]; //责任类型1=第一责任人，2=责任人
                }
                //并行查询第一责任人
                $EmployeeModel = new EmployeeModel();
                $EmployeeModel = $EmployeeModel
                    ->orWhere(function ($q) use ($employee_id_list, $company_id) {
                        if (!empty($employee_id_list)) {
                            $q->where('company_id', $company_id)
                                ->orWhereIn('id', $employee_id_list)
                                ->where('duty_type', 1)
                                ->where('status', 0);
                        }
                    })->orWhere(function ($q) use ($employee_id_area, $company_id) {
                        if (!empty($employee_id_area)) {
                            $q->where('company_id', $company_id)
                                ->whereIn('id', $employee_id_area)
                                ->where('duty_type', 1)
                                ->where('status', 0);
                        }
                    })->orWhere(function ($q) use ($employee_id_platform, $company_id) {
                        if (!empty($employee_id_platform)) {
                            $q->where('company_id', $company_id)
                                ->whereIn('id', $employee_id_platform)
                                ->where('duty_type', 1)
                                ->where('status', 0);
                        }
                    });
                $first_responsible_person = $EmployeeModel->pluck('id')->implode(',');

                //当以上规则匹配不到责任人时，则去查询发货地区信息关联责任人
                if (trim($first_responsible_person) == '' && trim($product_data['shipment_city_id']) != '') {
                    $employee_id_area = $EmployeeAreaModel->where('city_id', $product_data['city_id'])->pluck('employee_id')->toarray();
                    if (!empty($employee_id_area)) {
                        $where_city[] = ['company_id', '=', $company_id];
                        $where_city[] = ['id', 'in', $employee_id_area];
                        $where_city[] = ['status', '=', 0];
                        $where_city[] = ['duty_type', '=', 1]; //责任类型1=第一责任人，2=责任人
                        $EmployeeModel = new EmployeeModel();
                        $first_responsible_person = $EmployeeModel->where($where_city)->pluck('id')->implode(',');
                    }
                }

                //调试记录查询条件
                $insert_product_data['first_responsible_person'] = $first_responsible_person;
                //查询责任人
                $where_query1 = [];
                $where_query2 = [];
                $where_query3 = [];
                //查询指定公司责任人
                if (!empty($employee_id_list)) {
                    $where_query1[] = ['company_id', '=', $company_id];
                    $where_query1[] = ['id', 'in', $employee_id_list];
                    $where_query1[] = ['status', '=', 0];
                    $where_query1[] = ['duty_type', '=', 2]; //责任类型1=第一责任人，2=责任人
                }
                //查询地区配置的员工
                $employee_id_area = $EmployeeAreaModel->where('city_id', $product_data['city_id'])->pluck('employee_id')->toarray();
                if (!empty($employee_id_area)) {
                    $where_query2[] = ['company_id', '=', $company_id];
                    $where_query2[] = ['id', 'in', $employee_id_area];
                    $where_query2[] = ['status', '=', 0];
                    $where_query2[] = ['duty_type', '=', 2]; //责任类型1=第一责任人，2=责任人
                }
                //查询平台配置的员工
                $employee_id_platform = $EmployeePlatformModel->where('platform_id', $product_data['platform_id'])->pluck('employee_id')->toarray();
                if (!empty($employee_id_platform)) {
                    $where_query3[] = ['company_id', '=', $company_id];
                    $where_query3[] = ['id', 'in', $employee_id_platform];
                    $where_query3[] = ['status', '=', 0];
                    $where_query3[] = ['duty_type', '=', 2]; //责任类型1=第一责任人，2=责任人
                }
                //并行查询责任人
                $EmployeeModel = new EmployeeModel();
                $EmployeeModel = $EmployeeModel
                    ->orWhere(function ($q) use ($employee_id_list, $company_id) {
                        if (!empty($employee_id_list)) {
                            $q->where('company_id', $company_id)
                                ->orWhereIn('id', $employee_id_list)
                                ->where('duty_type', 2)
                                ->where('status', 0);
                        }
                    })->orWhere(function ($q) use ($employee_id_area, $company_id) {
                        if (!empty($employee_id_area)) {
                            $q->where('company_id', $company_id)
                                ->whereIn('id', $employee_id_area)
                                ->where('duty_type', 2)
                                ->where('status', 0);
                        }
                    })->orWhere(function ($q) use ($employee_id_platform, $company_id) {
                        if (!empty($employee_id_platform)) {
                            $q->where('company_id', $company_id)
                                ->whereIn('id', $employee_id_platform)
                                ->where('duty_type', 2)
                                ->where('status', 0);
                        }
                    });
                $responsible_person = $EmployeeModel->pluck('id')->implode(',');

                //当以上规则匹配不到责任人时，则去查询发货地区信息关联责任人
                if (trim($responsible_person) == '' && trim($product_data['shipment_city_id']) != '') {
                    $employee_id_area = $EmployeeAreaModel->where('city_id', $product_data['city_id'])->pluck('employee_id')->toarray();
                    if (!empty($employee_id_area)) {
                        $where_city[] = ['company_id', '=', $company_id];
                        $where_city[] = ['id', 'in', $employee_id_area];
                        $where_city[] = ['status', '=', 0];
                        $where_city[] = ['duty_type', '=', 2]; //责任类型1=第一责任人，2=责任人
                        $EmployeeModel = new EmployeeModel();
                        $responsible_person = $EmployeeModel->where($where_city)->pluck('id')->implode(',');
                    }
                }
                $insert_product_data['responsible_person'] = $responsible_person;

                //溯源责任人
                $source_responsible_person = '';
                if ($first_responsible_person && $responsible_person) {
                    //转换成数组，合并后在去重
                    $first_responsible_person = explode(',', $first_responsible_person);
                    $responsible_person = explode(',', $responsible_person);
                    $source_responsible_person = array_unique(array_merge($first_responsible_person, $responsible_person));
                    $source_responsible_person = ',' . implode(',', $source_responsible_person) . ',';
                } else if ($first_responsible_person) {
                    $source_responsible_person = $first_responsible_person;
                } else if ($responsible_person) {
                    $source_responsible_person = $responsible_person;
                }
                $insert_product_data['source_responsible_person'] = $source_responsible_person;
            }
        }

        //插入数据
        $ViolationProductModel->addViolationProduct($insert_product_data);
        if ($index_number == $data_totle) {
            (new ExecuteLogModel())->where('id', $executeLog_id)->update(['status' => 0, 'update_time' => time()]);
        }
    }


    public function failed(\Throwable $exception)
    {
        Log::info('job_error', '数据清洗-违规挂网商品数据队列完全失败', ['data' => $this->message_data, 'error' => $exception->getMessage()]);
    }
}