ViolationStoreDataJobs.php 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389
  1. <?php
  2. namespace App\Jobs\Manager\Process;
  3. use Illuminate\Bus\Queueable;
  4. use Illuminate\Contracts\Queue\ShouldBeUnique;
  5. use Illuminate\Contracts\Queue\ShouldQueue;
  6. use Illuminate\Foundation\Bus\Dispatchable;
  7. use Illuminate\Queue\InteractsWithQueue;
  8. use Illuminate\Queue\SerializesModels;
  9. use App\Facades\Servers\Logs\Log;
  10. use App\Models\Manager\Process\ViolationStore as ViolationStoreModel;
  11. use App\Models\Manager\Personnel\Employee as EmployeeModel;
  12. use App\Models\Api\Process\ExecuteLog as ExecuteLogModel;
  13. use App\Models\Manager\Process\ScrapeData as ScrapeDataModel;
  14. use App\Models\Manager\WashConfig\ViolationCompanyMember as ViolationCompanyMemberModel;
  15. use App\Models\Manager\Citys as CitysModel;
  16. use App\Models\Manager\Personnel\EmployeePlatform as EmployeePlatformModel;
  17. use App\Models\Manager\Personnel\EmployeeArea as EmployeeAreaModel;
  18. use Illuminate\Support\Facades\DB;
  19. use Illuminate\Support\Carbon;
  20. /**
  21. * 数据清洗-违规店铺清洗数据队列
  22. * @author 唐远望
  23. * @version 1.0
  24. * @date 2025-12-11
  25. */
  26. class ViolationStoreDataJobs implements ShouldQueue
  27. {
  28. use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
  29. protected $message_data;
  30. /**
  31. * Create a new job instance.
  32. *
  33. * @return void
  34. */
  35. public function __construct(array $message_data)
  36. {
  37. $this->message_data = $message_data;
  38. }
  39. /**
  40. * Execute the job.
  41. *
  42. * @return void
  43. */
  44. public function handle()
  45. {
  46. try {
  47. $this->getViolationStoreData($this->message_data);
  48. } catch (\Exception $e) {
  49. Log::info('job_error', '数据清洗-违规店铺清洗数据队列失败', ['data' => $this->message_data, 'error' => $e->getMessage()]);
  50. }
  51. }
  52. /**
  53. * 采集商品数据清洗
  54. * @author 唐远望
  55. * @version 1.0
  56. * @date 2025-12-11
  57. */
  58. public function getViolationStoreData($message_data)
  59. {
  60. $CitysModel = new CitysModel();
  61. $EmployeeModel = new EmployeeModel();
  62. $ViolationStoreModel = new ViolationStoreModel();
  63. $ScrapeDataModel = new ScrapeDataModel();
  64. $ViolationCompanyMemberModel = new ViolationCompanyMemberModel();
  65. $EmployeePlatformModel = new EmployeePlatformModel();
  66. $EmployeeAreaModel = new EmployeeAreaModel();
  67. $platform = $message_data['platform']; //多个平台配置
  68. $executeLog_id = $message_data['executeLog_id'];
  69. $store_name = $message_data['store_name']; //店铺名称
  70. $specify_responsible_person = $message_data['specify_responsible_person']; //指派责任人 0=开启 1=关闭
  71. $limit = isset($message_data['limit']) ? $message_data['limit'] : 50;
  72. $page = isset($message_data['page']) ? $message_data['page'] : 1;
  73. $company_id = isset($message_data['company_id']) ? $message_data['company_id'] : 0; //品牌方公司ID
  74. $item_totle_page = $message_data['item_totle_page']; //清洗店铺总页数
  75. $item_now_page = $message_data['item_now_page']; //清洗店铺当前页
  76. $start_time = time() - 60 * 5; // 开始时间 5分钟之前
  77. $end_time = time();//结束时间
  78. $start_time_string = date('Y-m-d H:i:s', $start_time);
  79. $end_time_string = date('Y-m-d H:i:s', $end_time);
  80. $where_map[] = ['insert_time', '>=', $start_time_string];
  81. $where_map[] = ['insert_time', '<=', $end_time_string];
  82. if ($platform != '0' && !empty($platform)) {
  83. $platform = explode(',', $platform);
  84. $ScrapeDataModel = $ScrapeDataModel->where('store_name', $store_name)->where($where_map)->whereIn('platform_id', $platform);
  85. } else {
  86. $ScrapeDataModel = $ScrapeDataModel->where('store_name', $store_name)->where($where_map);
  87. }
  88. $product_data_info = $ScrapeDataModel->paginate($limit, ['*'], 'page', $page)->toarray();
  89. $product_datas = $product_data_info['data'];
  90. if (empty($product_datas)) {
  91. if ($executeLog_id && $item_now_page >= $item_totle_page) {
  92. (new ExecuteLogModel())->where('id', $executeLog_id)->update(['status' => 0, 'update_time' => time()]);
  93. }
  94. return true;
  95. }
  96. foreach ($product_datas as $product_data) {
  97. //-------------------------------------- 处理营业执照地区信息(开始) --------------------------------------
  98. $province_name = $product_data['province_name'];
  99. $city_name = $product_data['city_name'];
  100. //特殊地区1级移除市
  101. if ($province_name && in_array($province_name, ['北京市', '天津市', '上海市', '重庆市'])) {
  102. //移除市这个字符
  103. $province_name = trim(str_replace('市', '', $province_name));
  104. } else if ($province_name && in_array($province_name, ['北京', '天津', '上海', '重庆'])) {
  105. } else if ($province_name && in_array($province_name, ['内蒙古', '广西', '西藏', '新疆', '宁夏'])) {
  106. switch ($province_name) {
  107. case '内蒙古':
  108. $province_name = '内蒙古自治区';
  109. break;
  110. case '广西':
  111. $province_name = '广西壮族自治区';
  112. break;
  113. case '西藏':
  114. $province_name = '西藏自治区';
  115. break;
  116. case '新疆':
  117. $province_name = '新疆维吾尔自治区';
  118. break;
  119. case '宁夏':
  120. $province_name = '宁夏回族自治区';
  121. break;
  122. }
  123. } else if ($province_name && in_array($province_name, ['内蒙古自治区', '广西壮族自治区', '西藏自治区', '新疆维吾尔自治区', '宁夏回族自治区'])) {
  124. //完整匹配不做处理
  125. } else if (trim($province_name) != '' && strpos($province_name, '省') === false) {
  126. //是否存在市省,如果不存在则补全
  127. $province_name = $province_name . '省';
  128. }
  129. if (trim($province_name) != '') {
  130. //根据最新处理后的省份名称获取省份ID
  131. $db_province_id = $CitysModel->where([['name', '=', $province_name], ['level', '=', '1']])->value('id');
  132. if (!empty($db_province_id)) {
  133. $product_data['province_id'] = $db_province_id;
  134. }else{
  135. //逆推省份ID失败,则修改为未知名称
  136. $province_name ='未知';
  137. }
  138. } else {
  139. //尝试从地区详情里面匹配省份名称
  140. $db_province_data_list = $CitysModel->where([['level', '=', '1']])->select(['id','name'])->get()->toarray();
  141. foreach ($db_province_data_list as $db_province_data) {
  142. $db_province_name = $db_province_data['name'];
  143. if (strpos($product_data['area_info'], $db_province_name) !== false) {
  144. $province_name = $db_province_name;
  145. $product_data['province_id'] = $db_province_data['id'];
  146. break;
  147. }
  148. }
  149. }
  150. if (trim($city_name) != '') {
  151. //校验是否存在县这个字眼
  152. if (strpos($city_name, '县') !== false) {
  153. $db_city_id = $CitysModel->where([['name', '=', $city_name], ['level', '=', '2']])->value('id');
  154. if (!empty($db_city_id)) {
  155. $product_data['city_id'] = $db_city_id;
  156. }
  157. } elseif (trim($city_name) != '' && strpos($city_name, '市') === false) {
  158. //是否存在市
  159. $city_name = $city_name . '市';
  160. //根据最新处理后的市名称获取市ID
  161. $db_city_id = $CitysModel->where([['name', '=', $city_name], ['level', '=', '2']])->value('id');
  162. if (!empty($db_city_id)) {
  163. $product_data['city_id'] = $db_city_id;
  164. }else{
  165. $city_name ='未知';
  166. }
  167. } else {
  168. $db_city_id = $CitysModel->where([['name', '=', $city_name], ['level', '=', '2']])->value('id');
  169. if (!empty($db_city_id)) {
  170. $product_data['city_id'] = $db_city_id;
  171. }else{
  172. $city_name ='未知';
  173. }
  174. }
  175. } else {
  176. if (trim($product_data['area_info']) != '') {
  177. //尝试从地区详情里面匹配市名称
  178. $db_city_name_list = $CitysModel->where([['level', '=', '2']])->pluck('name')->toarray();
  179. foreach ($db_city_name_list as $db_city_name) {
  180. if (strpos($product_data['area_info'], $db_city_name) !== false) {
  181. $db_city_id = $CitysModel->where([['name', '=', $db_city_name], ['level', '=', '2']])->value('id');
  182. $product_data['city_id'] = $db_city_id;
  183. $product_data['city_name'] = $db_city_name;
  184. $city_name = $db_city_name;
  185. break;
  186. }
  187. }
  188. }
  189. }
  190. //如果存在市区city_id 则逆推省份ID以及名称
  191. if (!empty($product_data['city_id']) && empty($product_data['province_id'])) {
  192. $db_province_id = $CitysModel->where([['id', '=', $product_data['city_id']], ['level', '=', '2']])->value('pid');
  193. $db_province_name = $CitysModel->where([['id', '=', $db_province_id], ['level', '=', '1']])->value('name');
  194. $product_data['province_id'] = $db_province_id;
  195. $product_data['province_name'] = $db_province_name;
  196. $province_name = $db_province_name;
  197. }
  198. if (trim($product_data['area_info']) == '' && trim($city_name) != '' && trim($province_name) != '') {
  199. $product_data['area_info'] = $province_name . $city_name;
  200. }
  201. //-------------------------------------- 处理营业执照地区信息(结束) --------------------------------------
  202. //处理链接信息
  203. preg_match('/https?:\/\/[^\s\'"<>]+/i', $product_data['link_url'], $matches);
  204. if (!empty($matches)) {
  205. $product_data['link_url'] = $matches[0];
  206. }
  207. $store_name = $product_data['store_name'];
  208. $anonymous_store_name = $product_data['anonymous_store_name'];
  209. if ($anonymous_store_name != '') {
  210. $store_name = $anonymous_store_name;
  211. }
  212. $insert_product_data = [
  213. 'company_id' => $company_id,
  214. 'source_id' => $product_data['id'],
  215. 'platform' => $product_data['platform_id'],
  216. 'company_name' => $product_data['company_name'],
  217. 'link_url' => $product_data['link_url'],
  218. 'store_name' => $store_name,
  219. 'social_credit_code' => $product_data['qualification_number'],
  220. 'province_id' => $product_data['province_id'],
  221. 'province_name' => $province_name,
  222. 'city_id' => $product_data['city_id'],
  223. 'city_name' => $city_name,
  224. 'area_info' => $product_data['area_info'],
  225. 'company_category_name' => '',
  226. 'first_responsible_person' => '',
  227. 'responsible_person' => '',
  228. 'source_responsible_person' => '',
  229. 'scrape_date' => $product_data['scrape_date'],
  230. 'collection_time' => strtotime($product_data['insert_time']),
  231. ];
  232. //获取公司绑定责任人信息
  233. $company_data = $ViolationStoreModel->where('social_credit_code', $product_data['qualification_number'])
  234. ->where('process_violation_store.company_id', $company_id)->first();
  235. $employee_id_list = [];
  236. if ($company_data) {
  237. $employee_id_list = $ViolationCompanyMemberModel->where('company_logid', $company_data->id)->pluck('employee_id')->toarray();
  238. }
  239. //获取指定人员信息
  240. if ($specify_responsible_person == 0) {
  241. $where_query1 = [];
  242. $where_query2 = [];
  243. $where_query3 = [];
  244. //查询指定公司第一责任人
  245. if (!empty($employee_id_list)) {
  246. $where_query1[] = ['company_id', '=', $company_id];
  247. $where_query1[] = ['id', 'in', $employee_id_list];
  248. $where_query1[] = ['status', '=', 0];
  249. $where_query1[] = ['duty_type', '=', 1]; //责任类型1=第一责任人,2=责任人
  250. }
  251. //查询地区配置的第一责任人
  252. $employee_id_area = $EmployeeAreaModel->where('city_id', $product_data['city_id'])->pluck('employee_id')->toarray();
  253. if (!empty($employee_id_area)) {
  254. $where_query2[] = ['company_id', '=', $company_id];
  255. $where_query2[] = ['id', 'in', $employee_id_area];
  256. $where_query2[] = ['status', '=', 0];
  257. $where_query2[] = ['duty_type', '=', 1]; //责任类型1=第一责任人,2=责任人
  258. }
  259. //查询平台配置的第一责任人
  260. $employee_id_platform = $EmployeePlatformModel->where('platform_id', $product_data['platform_id'])->pluck('employee_id')->toarray();
  261. if (!empty($employee_id_platform)) {
  262. $where_query3[] = ['company_id', '=', $company_id];
  263. $where_query3[] = ['id', 'in', $employee_id_platform];
  264. $where_query3[] = ['status', '=', 0];
  265. $where_query3[] = ['duty_type', '=', 1]; //责任类型1=第一责任人,2=责任人
  266. }
  267. //并行查询第一责任人
  268. $EmployeeModel = new EmployeeModel();
  269. $EmployeeModel = $EmployeeModel
  270. ->orWhere(function ($q) use ($employee_id_list, $company_id) {
  271. if (!empty($employee_id_list)) {
  272. $q->where('company_id', $company_id)
  273. ->orWhereIn('id', $employee_id_list)
  274. ->where('status', 0)
  275. ->where('duty_type', 1);
  276. }
  277. })->orWhere(function ($q) use ($employee_id_area, $company_id) {
  278. if (!empty($employee_id_area)) {
  279. $q->where('company_id', $company_id)
  280. ->whereIn('id', $employee_id_area)
  281. ->where('status', 0)
  282. ->where('duty_type', 1);
  283. }
  284. })->orWhere(function ($q) use ($employee_id_platform, $company_id) {
  285. if (!empty($employee_id_platform)) {
  286. $q->where('company_id', $company_id)
  287. ->whereIn('id', $employee_id_platform)
  288. ->where('status', 0)
  289. ->where('duty_type', 1);
  290. }
  291. });
  292. $first_responsible_person = $EmployeeModel->pluck('id')->implode(',');
  293. //调试记录查询条件
  294. $insert_product_data['first_responsible_person'] = $first_responsible_person;
  295. //查询责任人
  296. $where_query1 = [];
  297. $where_query2 = [];
  298. $where_query3 = [];
  299. //查询指定公司责任人
  300. if (!empty($employee_id_list)) {
  301. $where_query1[] = ['company_id', '=', $company_id];
  302. $where_query1[] = ['id', 'in', $employee_id_list];
  303. $where_query1[] = ['status', '=', 0];
  304. $where_query1[] = ['duty_type', '=', 2]; //责任类型1=第一责任人,2=责任人
  305. }
  306. //查询地区配置的员工
  307. $employee_id_area = $EmployeeAreaModel->where('city_id', $product_data['city_id'])->pluck('employee_id')->toarray();
  308. if (!empty($employee_id_area)) {
  309. $where_query2[] = ['company_id', '=', $company_id];
  310. $where_query2[] = ['id', 'in', $employee_id_area];
  311. $where_query2[] = ['status', '=', 0];
  312. $where_query2[] = ['duty_type', '=', 2]; //责任类型1=第一责任人,2=责任人
  313. }
  314. //查询平台配置的员工
  315. $employee_id_platform = $EmployeePlatformModel->where('platform_id', $product_data['platform_id'])->pluck('employee_id')->toarray();
  316. if (!empty($employee_id_platform)) {
  317. $where_query3[] = ['company_id', '=', $company_id];
  318. $where_query3[] = ['id', 'in', $employee_id_platform];
  319. $where_query3[] = ['status', '=', 0];
  320. $where_query3[] = ['duty_type', '=', 2]; //责任类型1=第一责任人,2=责任人
  321. }
  322. //并行查询责任人
  323. $EmployeeModel = new EmployeeModel();
  324. $EmployeeModel = $EmployeeModel
  325. ->orWhere(function ($q) use ($employee_id_list, $company_id) {
  326. if (!empty($employee_id_list)) {
  327. $q->where('company_id', $company_id)
  328. ->orWhereIn('id', $employee_id_list)
  329. ->where('status', 0)
  330. ->where('duty_type', 2); //责任类型1=第一责任人,2=责任人
  331. }
  332. })->orWhere(function ($q) use ($employee_id_area, $company_id) {
  333. if (!empty($employee_id_area)) {
  334. $q->where('company_id', $company_id)
  335. ->whereIn('id', $employee_id_area)
  336. ->where('status', 0)
  337. ->where('duty_type', 2); //责任类型1=第一责任人,2=责任人
  338. }
  339. })->orWhere(function ($q) use ($employee_id_platform, $company_id) {
  340. if (!empty($employee_id_platform)) {
  341. $q->where('company_id', $company_id)
  342. ->whereIn('id', $employee_id_platform)
  343. ->where('status', 0)
  344. ->where('duty_type', 2); //责任类型1=第一责任人,2=责任人
  345. }
  346. });
  347. $responsible_person = $EmployeeModel->pluck('id')->implode(',');
  348. $insert_product_data['responsible_person'] = $responsible_person;
  349. //溯源责任人
  350. $source_responsible_person = '';
  351. if ($first_responsible_person && $responsible_person) {
  352. //转换成数组,合并后在去重
  353. $first_responsible_person = explode(',', $first_responsible_person);
  354. $responsible_person = explode(',', $responsible_person);
  355. $source_responsible_person = array_unique(array_merge($first_responsible_person, $responsible_person));
  356. $source_responsible_person = ',' . implode(',', $source_responsible_person) . ',';
  357. } else if ($first_responsible_person) {
  358. $source_responsible_person = $first_responsible_person;
  359. } else if ($responsible_person) {
  360. $source_responsible_person = $responsible_person;
  361. }
  362. $insert_product_data['source_responsible_person'] = $source_responsible_person;
  363. }
  364. //插入数据
  365. $ViolationStoreModel->addViolationStore($insert_product_data);
  366. }
  367. //继续执行下一页
  368. $message_data['page'] = $page + 1;
  369. $message_data['limit'] = $limit;
  370. ViolationStoreDataJobs::dispatch($message_data);
  371. }
  372. public function failed(\Throwable $exception)
  373. {
  374. Log::info('job_error', '数据清洗-违规店铺清洗数据队列完全失败', ['data' => $this->message_data, 'error' => $exception]);
  375. }
  376. }