diff --git a/app/command/InquiryImportFbCommand.php b/app/command/InquiryImportFbCommand.php index 3c9a3c0..e90884c 100644 --- a/app/command/InquiryImportFbCommand.php +++ b/app/command/InquiryImportFbCommand.php @@ -11,6 +11,9 @@ use PhpOffice\PhpSpreadsheet\IOFactory; use app\inquiry\utils\FbLeadHeaderMapper; use support\exception\BusinessException; +/** + * 从Excel文件导入fb_leads + */ class InquiryImportFbCommand extends Command { protected static $defaultName = 'iq-fb'; diff --git a/app/inquiry/service/FbFileImportService.php b/app/inquiry/service/FbFileImportService.php new file mode 100644 index 0000000..becbcc4 --- /dev/null +++ b/app/inquiry/service/FbFileImportService.php @@ -0,0 +1,153 @@ +importOneFile($filePath, $channelId); + $count++; + + // 导入成功后，把文件重命名为 .xlsx.done，避免下次重复导入 + $newPath = $filePath . '.done'; + if (!@rename($filePath, $newPath)) { + Log::warning("FbFileImportService: 文件导入成功，但重命名失败：{$filePath} -> {$newPath}"); + } else { + Log::info("FbFileImportService: 文件导入并标记完成：{$newPath}（行数：{$rows}）"); + } + } + + return $count; + } + + /** + * 导入单个 Excel 到 fb_leads + * + * @param string $filePath + * @param int $channelId 渠道ID + * @return int 导入的记录条数 + * @throws BusinessException + */ + public function importOneFile(string $filePath, int $channelId): int + { + if (!is_file($filePath)) { + throw new BusinessException("Excel 文件不存在：{$filePath}"); + } + + $spreadsheet = IOFactory::load($filePath); + $sheet = $spreadsheet->getActiveSheet(); + $rows = $sheet->toArray(null, true, true, true); + + if (empty($rows)) { + throw new BusinessException('Excel 内容为空：' . $filePath); + } + + // 表头 + $headerRow = array_shift($rows); + + $mapper = new FbLeadHeaderMapper(); + // 如有未知字段，会在这里抛异常 + $colToField = $mapper->buildColumnFieldMap($headerRow); + + $insertData = []; + $now = time(); + $sourceFile = basename($filePath); + + foreach ($rows as $row) { + $item = []; + + foreach ($colToField as $colIndex => $field) { + $raw = $row[$colIndex] ?? null; + + switch ($field) { + case 'is_organic': + $item['is_organic'] = $this->normalizeIsOrganic($raw); + break; + case 'fb_created_time': + $item['fb_created_time'] = $this->normalizeCreatedTime($raw); + break; + default: + $item[$field] = $raw; + break; + } + } + + $item['source_file'] = $sourceFile; + $item['channel'] = $channelId; // 👈 渠道ID记录到 fb_leads + + $item['created_at'] = $now; + $item['updated_at'] = $now; + $item['sync_status'] = 0; // 未同步到 inquiry_log + + $insertData[] = $item; + } + + if ($insertData) { + Db::table('fb_leads')->insert($insertData); + } + + return count($insertData); + } + + /** + * 规范化 is_organic 字段为 0/1 + */ + protected function normalizeIsOrganic($raw): int + { + if (is_bool($raw)) { + return $raw ? 1 : 0; + } + $str = strtolower((string)$raw); + if ($str === '') { + return 0; + } + return in_array($str, ['1', 'true', 'yes', 'organic'], true) ? 1 : 0; + } + + /** + * FB 创建时间目前按原始字符串存储 + */ + protected function normalizeCreatedTime($raw) + { + return $raw; + } +} diff --git a/app/inquiry/service/FbLeadSyncService.php b/app/inquiry/service/FbLeadSyncService.php new file mode 100644 index 0000000..9bd33b0 --- /dev/null +++ b/app/inquiry/service/FbLeadSyncService.php @@ -0,0 +1,148 @@ +where('sync_status', 0) + ->where('channel', $channelId) + ->get(); + + if ($leads->isEmpty()) { + return 0; + } + + $now = time(); + $count = 0; + + Db::transaction(function () use ($leads, $enterUserId, $channelId, $now, &$count) { + $insert = []; + $ids = []; + + foreach ($leads as $lead) { + $leadArr = (array)$lead; + $insert[] = $this->mapLeadToInquiry($leadArr, $now, $enterUserId, $channelId); + $ids[] = $leadArr['id']; + } + + if ($insert) { + Db::table('inquiry_log')->insert($insert); + + Db::table('fb_leads') + ->whereIn('id', $ids) + ->update([ + 'sync_status' => 1, + 'updated_at' => $now, + ]); + + $count = count($insert); + } + }); + + return $count; + } + + /** + * 映射一条 fb_leads -> inquiry_log + */ + protected function mapLeadToInquiry(array $lead, int $now, int $enterUserId, int $channelId): array + { + $email = (string)($lead['email'] ?? ''); + $tel = (string)($lead['phone'] ?? ''); + $wa = $tel; + + $countryId = PhoneCountryHelper::detectCountryId($tel ?: $wa); + + return [ + 'order_no' => $this->orderNum(), + 'enter_user_id' => $enterUserId, + 'recipient' => 0, + 'quality' => 2, + 'channel' => $channelId, // 👈 inquiry_log 里同样用这个渠道ID + + 'email' => $email, + 'tel' => $tel, + 'whatsapp' => $wa, + 'country' => $countryId ?: 0, + 'contacts' => (string)($lead['full_name'] ?? ''), + + 'content' => $this->buildContentHtml($lead), + + 'status' => 0, + 'r_time' => 0, + 't_time' => 0, + 'sync_crm' => 0, + + 'created_at' => $now, + 'updated_at' => $now, + ]; + } + + protected function orderNum(): string + { + return 'SN' . date('Ymd') . str_pad(mt_rand(1, 99999), 5, '0', STR_PAD_LEFT); + } + + /** + * 富文本内容拼装 —— 基于你之前的版本 + */ + protected function buildContentHtml(array $lead): string + { + $e = function ($v) { + return htmlspecialchars((string)$v, ENT_QUOTES | ENT_SUBSTITUTE, 'UTF-8'); + }; + + $lines = []; + + if (!empty($lead['car_model'])) { + $lines[] = '

意向车型：' . $e($lead['car_model']) . '

'; + } + if (!empty($lead['car_quantity']) || !empty($lead['car_quantity_text'])) { + $qty = $lead['car_quantity'] ?? $lead['car_quantity_text'] ?? ''; + $lines[] = '

意向数量：' . $e($qty) . '

'; + } + if (!empty($lead['favorite_color'])) { + $lines[] = '

意向颜色：' . $e($lead['favorite_color']) . '

'; + } + if (!empty($lead['usage_type'])) { + $lines[] = '

用途：' . $e($lead['usage_type']) . '

'; + } + if (!empty($lead['seat_preference'])) { + $lines[] = '

座位数偏好：' . $e($lead['seat_preference']) . '

'; + } + + $lines[] = '
'; + $lines[] = '

广告与表单信息：

'; + $lines[] = '

平台：' . $e($lead['platform'] ?? '') . '

'; + $lines[] = '

Campaign：' . $e($lead['campaign_name'] ?? '') . ' (#' . $e($lead['campaign_id'] ?? '') . ')

'; + $lines[] = '

Adset：' . $e($lead['adset_name'] ?? '') . ' (#' . $e($lead['adset_id'] ?? '') . ')

'; + $lines[] = '

Ad：' . $e($lead['ad_name'] ?? '') . ' (#' . $e($lead['ad_id'] ?? '') . ')

'; + $lines[] = '

Form：' . $e($lead['form_name'] ?? '') . ' (#' . $e($lead['form_id'] ?? '') . ')

'; + + $lines[] = '
'; + $lines[] = '

Meta 信息：

'; + $lines[] = '

Lead ID：' . $e($lead['fb_lead_id'] ?? '') . '

'; + $lines[] = '

FB 创建时间原始值：' . $e($lead['fb_created_time'] ?? '') . '

'; + + if (!empty($lead['inbox_url'])) { + $url = $e($lead['inbox_url']); + $lines[] = '

Inbox 链接：' . $url . '

'; + } + + return implode("\n", $lines); + } +} diff --git a/app/inquiry/service/InquiryAssignService.php b/app/inquiry/service/InquiryAssignService.php new file mode 100644 index 0000000..272f009 --- /dev/null +++ b/app/inquiry/service/InquiryAssignService.php @@ -0,0 +1,99 @@ + $v > 0)); + if (empty($userIds)) { + throw new BusinessException('分配失败：业务员列表为空'); + } + + $inquiries = Db::table('inquiry_log') + ->where('status', 0) + ->where('recipient', 0) + ->where('channel', $channelId) + ->orderBy('id') + ->limit($limit) + ->get(); + + if ($inquiries->isEmpty()) { + return 0; + } + + $now = time(); + $count = 0; + + foreach ($inquiries as $row) { + $inquiryId = (int)$row->id; + $recipientUserId = $this->nextRecipientId($userIds, $roundRobinKey); + + Db::transaction(function () use ($inquiryId, $recipientUserId, $now, &$count) { + $affected = Db::table('inquiry_log') + ->where('id', $inquiryId) + ->update([ + 'recipient' => $recipientUserId, + 'status' => 2, + 'r_time' => $now, + 't_time' => $now, + 'updated_at' => $now, + ]); + + if ($affected === 0) { + // 可能被其他进程抢先处理了，直接跳过 + return; + } + + $res = [ + 'id' => $inquiryId, + 'recipient' => $recipientUserId, + 'dispatched_at' => $now, + 'source' => 'fb_leads_inquiry', + ]; + + // 队列推送 —— 用你原来的写法 + RedisQueueRedis::send('inquiry-to-crm-new', $res); + + $count++; + }); + } + + return $count; + } + + /** + * 轮询获取下一个业务员ID + */ + protected function nextRecipientId(array $userIds, string $groupKey): int + { + $userIds = array_values(array_filter(array_map('intval', $userIds), fn($v) => $v > 0)); + if (empty($userIds)) { + throw new BusinessException('轮询分配失败：业务员列表为空'); + } + + $count = count($userIds); + $key = 'inquiry:assign:round_robin:' . $groupKey; + + $val = Redis::incr($key); // 原子自增 + $index = ($val - 1) % $count; + + return $userIds[$index]; + } +} diff --git a/app/process/InquiryAutoProcess.php b/app/process/InquiryAutoProcess.php new file mode 100644 index 0000000..502e977 --- /dev/null +++ b/app/process/InquiryAutoProcess.php @@ -0,0 +1,90 @@ +interval = $interval; + } + + public function onWorkerStart() + { + // 每 interval 秒执行一次 + Timer::add($this->interval, function () { + $this->runOnceForAllChannels(); + }); + } + + /** + * 一轮执行：扫描 config/inquiry_channel.php 中所有渠道，逐个跑一遍流水线 + */ + protected function runOnceForAllChannels() + { + $configAll = config('inquiry_channel'); + if (empty($configAll) || !is_array($configAll)) { + Log::error('InquiryAutoProcess: config("inquiry_channel") 为空或格式不正确'); + return; + } + + foreach ($configAll as $channelKey => $cfg) { + if (!is_array($cfg)) { + Log::error("InquiryAutoProcess: 渠道 {$channelKey} 配置不是数组，已跳过"); + continue; + } + + $this->runForChannel($channelKey, $cfg); + } + } + + /** + * 对单个渠道执行：导入 Excel -> fb_leads -> inquiry_log -> 分配 + 推队列 + */ + protected function runForChannel(string $channelKey, array $cfg) + { + $watchDir = public_path($cfg['watch_dir'] ?? 'excel'); + $enterUserId = (int)($cfg['enter_user_id'] ?? 0); + $channelId = (int)($cfg['channel_id'] ?? 0); + $salesUserIds = $cfg['sales_user_ids'] ?? []; + $roundRobinKey = $cfg['round_robin_key'] ?? $channelKey; + + if ($enterUserId <= 0 || $channelId <= 0) { + Log::error("InquiryAutoProcess: 渠道 {$channelKey} 配置缺少正确的 enter_user_id 或 channel_id，已跳过"); + return; + } + + $importService = new FbFileImportService(); + $syncService = new FbLeadSyncService(); + $assignService = new InquiryAssignService(); + + try { + // 1) 导入该渠道目录下的 Excel 到 fb_leads（fb_leads.channel = 渠道ID） + $filesImported = $importService->importDir($watchDir, $channelId); + + // 2) 只同步该渠道的 fb_leads -> inquiry_log（channel 过滤） + $synced = $syncService->syncToInquiryLog($enterUserId, $channelId); + + // 3) 只分配该渠道的 inquiry_log（channel 过滤） + $assigned = $assignService->assignBatch($salesUserIds, $roundRobinKey, $channelId, 100); + + Log::info("InquiryAutoProcess[{$channelKey}] files={$filesImported}, synced={$synced}, assigned={$assigned}"); + } catch (\Throwable $e) { + Log::error( + 'InquiryAutoProcess[' . $channelKey . '] error: ' . + $e->getMessage() . "\n" . $e->getTraceAsString() + ); + } + } +} diff --git a/config/inquiry_channel.php b/config/inquiry_channel.php new file mode 100644 index 0000000..fe89dc4 --- /dev/null +++ b/config/inquiry_channel.php @@ -0,0 +1,58 @@ + [ + // 监听目录（相对 public_path） + 'watch_dir' => 'excel/fb_paika_shemei', + + // 清洗到 inquiry_log 时使用的录入人 & 渠道 + 'enter_user_id' => 186, + 'channel_id' => 10, // 👈 这里就是你系统里的渠道ID + + // 参与轮询分配的业务员 ID 列表 + 'sales_user_ids' => [ + // 63, + // 64, + // 67, + // 93, + // 108, + // 132, + // 143, + // 145, + // 157, + // 160, + // 168, + // 172, + // 173, + // 191, + // 193, + // 194, + // 195, + // 196, + // 197, + 1, 74, 78 + ], + + // 轮询用的分组 key（Redis 记指针用） + 'round_robin_key'=> 'fb_paika_shemei_inquiry', + ], + + // 将来可以继续加其他渠道，如： + // 'tiktok' => [ + // // 监听的目录（相对 public_path） + // 'watch_dir' => 'excel/paika_shemei/11', + + // // 清洗到 inquiry_log 时使用 + // 'enter_user_id' => 186, + // 'channel' => 10, + + // // 参与轮询分配的业务员 ID 列表 + // 'sales_user_ids' => [ + // 1, 74, 78 + // ], + + // // 轮询用的分组 key（Redis 记指针用） + // 'round_robin_key'=> 'fb_main_inquiry', + // ], +]; diff --git a/config/process.php b/config/process.php index 8022ef3..e04f409 100644 --- a/config/process.php +++ b/config/process.php @@ -69,4 +69,12 @@ return [ 'checkweb' => [ 'handler' => app\process\CheckWeb::class ], + 'inquiry_auto_all' => [ + 'handler' => app\process\InquiryAutoProcess::class, + 'count' => 1, + 'reloadable' => true, + 'constructor' => [ + 'interval' => 30, // 每 30 秒跑一轮所有渠道 + ], + ], ];