zhijiayun/app/Jobs/Manager/Process/ViolationStoreDataJobs.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\ViolationStore as ViolationStoreModel;
use App\Models\Manager\Personnel\Employee as EmployeeModel;
use App\Models\Api\Process\ExecuteLog as ExecuteLogModel;
use App\Models\Manager\Process\ScrapeData as ScrapeDataModel;
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 Illuminate\Support\Facades\DB;
use Illuminate\Support\Carbon;

/**
 * 数据清洗-违规店铺清洗数据队列
 * @author  唐远望
 * @version 1.0
 * @date  2025-12-11
 */
class ViolationStoreDataJobs 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->getViolationStoreData($this->message_data);
        } catch (\Exception $e) {
            Log::info('job_error', '数据清洗-违规店铺清洗数据队列失败', ['data' => $this->message_data, 'error' => $e->getMessage()]);
        }
    }


    /**
     * 采集商品数据清洗
     * @author  唐远望
     * @version 1.0
     * @date  2025-12-11
     */
    public function getViolationStoreData($message_data)
    {
        $CitysModel = new CitysModel();
        $EmployeeModel = new EmployeeModel();
        $ViolationStoreModel = new ViolationStoreModel();
        $ScrapeDataModel = new ScrapeDataModel();
        $ViolationCompanyMemberModel = new ViolationCompanyMemberModel();
        $EmployeePlatformModel = new EmployeePlatformModel();
        $EmployeeAreaModel = new EmployeeAreaModel();
        $platform = $message_data['platform']; //多个平台配置
        $executeLog_id = $message_data['executeLog_id'];
        $store_name = $message_data['store_name']; //店铺名称
        $specify_responsible_person = $message_data['specify_responsible_person']; //指派责任人 0=开启 1=关闭
        $limit = isset($message_data['limit']) ? $message_data['limit'] : 50;
        $page = isset($message_data['page']) ? $message_data['page'] : 1;
        $company_id = isset($message_data['company_id']) ? $message_data['company_id'] : 0; //品牌方公司ID
        $item_totle_page = $message_data['item_totle_page']; //清洗店铺总页数
        $item_now_page = $message_data['item_now_page']; //清洗店铺当前页
        $employee_ids = $message_data['employee_ids']; //指派责任人ID列表

        $start_time = time() - 60 * 5; // 开始时间 5分钟之前
        $end_time = time();//结束时间
        $start_time_string = date('Y-m-d H:i:s', $start_time);
        $end_time_string = date('Y-m-d H:i:s', $end_time);
        $where_map[] = ['insert_time', '>=', $start_time_string];
        $where_map[] = ['insert_time', '<=', $end_time_string];

        if ($platform != '0' && !empty($platform)) {
            $platform = explode(',', $platform);
            $ScrapeDataModel = $ScrapeDataModel->where('store_name', $store_name)->where($where_map)->whereIn('platform_id', $platform);
        } else {
            $ScrapeDataModel = $ScrapeDataModel->where('store_name', $store_name)->where($where_map);
        }
        $product_data_info = $ScrapeDataModel->paginate($limit, ['*'], 'page', $page)->toarray();
        $product_datas = $product_data_info['data'];
        if (empty($product_datas)) {
            if ($executeLog_id && $item_now_page >= $item_totle_page) {
                (new ExecuteLogModel())->where('id', $executeLog_id)->update(['status' => 0, 'update_time' => time()]);
            }
            return true;
        }
        foreach ($product_datas as $product_data) {
            //-------------------------------------- 处理营业执照地区信息(开始) --------------------------------------
            $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;
            }
            //-------------------------------------- 处理营业执照地区信息(结束) --------------------------------------

            //处理链接信息
            preg_match('/https?:\/\/[^\s\'"<>]+/i', $product_data['link_url'], $matches);
            if (!empty($matches)) {
                $product_data['link_url'] = $matches[0];
            }
            $store_name = $product_data['store_name'];
            $anonymous_store_name = $product_data['anonymous_store_name'];
            if ($anonymous_store_name != '') {
                $store_name = $anonymous_store_name;
            }
            $insert_product_data = [
                'company_id' => $company_id,
                'source_id' => $product_data['id'],
                'platform'    => $product_data['platform_id'],
                'company_name'    => $product_data['company_name'],
                'link_url'    => $product_data['link_url'],
                'store_name'    => $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'],
                'company_category_name' => '',
                'first_responsible_person' => '',
                'responsible_person' => '',
                'source_responsible_person' => '',
                'scrape_date' => $product_data['scrape_date'],
                'collection_time' => strtotime($product_data['insert_time']),
            ];
            $employee_id_list = explode(',', $employee_ids);
            //获取指定人员信息
            if ($specify_responsible_person == 0) {
                $first_responsible_person ='';
                //查询第一责任人
                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=责任人
                    $first_responsible_person = $EmployeeModel->where($where_query1)->pluck('id')->implode(',');
                }
                $insert_product_data['first_responsible_person'] = $first_responsible_person;
                //查询责任人
                $responsible_person ='';
                if (!empty($employee_id_list)) {
                    $where_query2[] = ['company_id', '=', $company_id];
                    $where_query2[] = ['id', 'in', $employee_id_list];
                    $where_query2[] = ['status', '=', 0];
                    $where_query2[] = ['duty_type', '=', 2]; //责任类型1=第一责任人，2=责任人
                    $responsible_person = $EmployeeModel->where($where_query2)->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;
            }

            //插入数据
            $ViolationStoreModel->addViolationStore($insert_product_data);
        }
        //继续执行下一页
        $message_data['page'] = $page + 1;
        $message_data['limit'] = $limit;
        ViolationStoreDataJobs::dispatch($message_data)->delay(now()->addSeconds(1));
    }


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