db->beginTransaction(); try { $basePath = \Yii::$app->basePath; if (empty($_FILES) || !isset($_FILES['file'])) throw new Exception('请上传csv文件'); $fileName = $_FILES['file']['name']; $tmpName = $_FILES['file']['tmp_name']; $basePath = $basePath . '/web/uploads/addons/ScoreExpansion/'; if (!is_dir($basePath)) mkdir($basePath, 0775, true); $ext = strtolower(pathinfo($fileName, PATHINFO_EXTENSION)); if ($ext != 'csv') throw new Exception('请上传csv文件'); $uploadFile = $basePath . 'ScoreExpansion-' . date('Y-m-d His') . '.' . $ext; //上传 move_uploaded_file($tmpName, $uploadFile); $success_counts = $fail_counts = 0; $counts = 0; $param['counts'] = $counts; $param['success_counts'] = $success_counts; $param['fail_counts'] = $fail_counts; $param['upload_file'] = $uploadFile; $param['mall_id'] = $params['mall_id']; $param['source'] = ScoreExpansionWallet::class; $import_log_model = ImportLog::setData($param); if (!$import_log_model) throw new Exception(ImportLog::getStaticError()); $t->commit(); $params = ['id' => $import_log_model['id']]; Yii::$app->services->rabbitMq->push(RabbitMqEnum::EXCHANGE_LONG_DURATION, RabbitMqEnum::LONG_DURATION_QUEUE, $params, self::class, 'syncImport'); return ['counts' => $counts, 'success_counts' => $success_counts, 'fail_counts' => $fail_counts]; } catch (Exception $e) { self::setError($e->getMessage()); $t->rollBack(); return false; } } public static function handleImportData($params, $data, $sync = 1) { $success_counts = $fail_counts = 0; $counts = count($data); $fail_arr = []; $userMap = User::find() ->where([ 'status' => StatusEnum::ENABLED, 'mobile' => array_column($data, 'mobile'), 'mall_id' => $params['mall_id'] ]) ->select('id, mobile') ->indexBy('mobile') ->asArray() ->all(); foreach ($data as $val) { array_walk($val, function (&$v) { $v = trim($v); }); $val['mall_id'] = $params['mall_id']; $val['source_table'] = ImportLog::tableName(); $val['source_table_id'] = $params['import_log_id']; if (empty($val['mobile'])) { $val['reason'] = '用户手机号为空'; [$fail_arr, $fail_counts] = ImportFailLog::packFailArr($val, $fail_arr, $fail_counts); continue; } if (!isset($userMap[$val['mobile']])) { $val['reason'] = '手机号错误,未查询到该用户'; [$fail_arr, $fail_counts] = ImportFailLog::packFailArr($val, $fail_arr, $fail_counts); continue; } $val['user_id'] = $userMap[$val['mobile']]['id']; if (!isset($val['integral']) || !is_numeric($val['integral'])) { $val['integral'] = 0; } if (!isset($val['frozen_integral']) || !is_numeric($val['frozen_integral'])) { $val['frozen_integral'] = 0; } if (self::handleScore($val)) { $success_counts += 1; } else { $val['reason'] = self::getStaticError(); [$fail_arr, $fail_counts] = ImportFailLog::packFailArr($val, $fail_arr, $fail_counts); } } $param['counts'] = $counts; $param['success_counts'] = $success_counts; $param['fail_counts'] = $fail_counts; $param['source'] = ScoreExpansionWallet::class; $param['mall_id'] = $params['mall_id']; if ($sync == 0) { $param['fail_arr'] = $fail_arr; // 异步返回数据 return $param; } $import_log_model = ImportLog::setData($param); if (!$import_log_model) throw new Exception(ImportLog::getStaticError()); if (!empty($fail_arr)) { $res = ImportFailLog::setData($fail_arr, $import_log_model->id); if (!$res) throw new Exception(ImportFailLog::getStaticError()); } } /** * @param $params * * @return void */ public static function syncImport($params) { try { if (empty($params['id'])) throw new Exception('缺少参数id'); $model = ImportLog::findOne(['id' => $params['id']]); if (empty($model['upload_file'])) throw new Exception('缺少参数upload_file'); $upload_file = $model['upload_file']; $mall_id = $model['mall_id']; $cvsFile = fopen($upload_file, 'r'); //开始读取csv文件数据 $i = 0; $rows = []; while ($fileData = fgetcsv($cvsFile)) { $i++; if ($i == 1) continue; //过滤表头 $temp_row = []; foreach ($fileData as $k => $val) { $val = mb_convert_encoding(trim($val), "UTF-8", "GBK"); if ($k == 0) { $temp_row['mobile'] = $val; } else if ($k == 1) { $temp_row['integral'] = $val; } else if ($k == 2) { $temp_row['frozen_integral'] = $val; } } $rows[] = $temp_row; } $result = self::handleImportData(['mall_id' => $mall_id, 'import_log_id' => $params['id']], $rows, 0); $model->counts = $result['counts']; $model->success_counts = $result['success_counts']; $model->fail_counts = $result['fail_counts']; if (!$model->save()) throw new RuntimeException($model->getErrorMessage()); $fail_arr = $result['fail_arr'] ?? []; if (!empty($fail_arr)) { $res = ImportFailLog::setData($fail_arr, $model->id); if (!$res) throw new RuntimeException(ImportFailLog::getStaticError()); } } catch (Exception $e) { Yii::$app->custom->logs(FormatHelper::exception($e, '【积分拓客】充值积分失败')); } } /** * @param array $params * * @return bool */ private static function handleScore(array $params): bool { $integralFrozenText = ScoreExpansionEnum::getRedpackText(ScoreExpansionEnum::WALLET_INTEGRAL_FROZEN, $params['mall_id']); $integralText = ScoreExpansionEnum::getRedpackText(ScoreExpansionEnum::WALLET_INTEGRAL, $params['mall_id']); $t = Yii::$app->db->beginTransaction(); try { // 加激活积分 if ($params['integral']) { $res = ScoreExpansionWallet::setData([ 'mall_id' => $params['mall_id'], 'user_id' => $params['user_id'], 'is_frozen' => false, 'is_deduct' => true, 'integral' => $params['integral'], 'from_type' => ScoreExpansionEnum::FROM_IMPORT, 'desc' => '手动导入' . $integralText, ]); if ($res === false) throw new RuntimeException(ScoreExpansionWallet::getStaticError()); } // 加冻结积分 if ($params['frozen_integral']) { $res = ScoreExpansionWallet::setData([ 'mall_id' => $params['mall_id'], 'user_id' => $params['user_id'], 'is_frozen' => true, 'is_deduct' => true, 'integral' => $params['frozen_integral'], 'from_type' => ScoreExpansionEnum::FROM_IMPORT, 'desc' => '手动导入' . $integralFrozenText, ]); if ($res === false) throw new RuntimeException(ScoreExpansionWallet::getStaticError()); } $t->commit(); } catch (Exception $e) { $t->rollBack(); self::setStaticError(FormatHelper::exception($e, '【积分拓客】充值积分失败')); return false; } return true; } }