numberSync = new SplitTicketNumberSyncService(); $this->ruleService = new SplitTicketRuleService(); $this->lockService = new SplitTicketSyncLockService(); $this->dispatchService = new SplitTicketSyncDispatchService(); $this->failRetryService = new SplitTicketSyncFailRetryService(); } /** * 同步单条工单 * * @param bool $dueValidated 为 true 时跳过 shouldSkip(由 syncDueTickets 预筛后传入) * @param string $syncMode auto=定时自动 manual=手动触发 * @return array{success:bool,message:string,skipped?:bool} */ public function syncOne(int $ticketId, bool $force = false, bool $dueValidated = false, string $syncMode = 'manual'): array { $ticket = Ticket::get($ticketId); if (!$ticket) { SplitTicketSyncLogger::log('sync', 'ticket not found', ['ticketId' => $ticketId]); return ['success' => false, 'message' => '工单不存在']; } SplitTicketSyncLogger::setTicketContext($ticketId, (string) $ticket['ticket_type']); SplitTicketSyncLogger::log('sync', 'syncOne start', [ 'force' => $force, 'syncMode' => $syncMode, 'status' => (string) $ticket['status'], 'manualManage' => (int) ($ticket['manual_manage'] ?? 0), 'syncFailCount' => (int) ($ticket['sync_fail_count'] ?? 0), 'syncTime' => (int) ($ticket['sync_time'] ?? 0), 'pageUrl' => (string) $ticket['ticket_url'], 'nodeHost' => SplitSyncConfigService::getNodeHost(), ]); if ($this->ruleService->isManuallyClosed($ticket)) { SplitTicketSyncLogger::log('sync', 'skipped', ['reason' => '工单已手动关闭']); SplitTicketSyncLogger::clearTicketContext(); return ['success' => false, 'message' => '工单已手动关闭,暂停同步', 'skipped' => true]; } if (!$force && !$dueValidated) { $skip = $this->shouldSkip($ticket); if ($skip !== null) { SplitTicketSyncLogger::log('sync', 'skipped', ['reason' => $skip]); SplitTicketSyncLogger::clearTicketContext(); return ['success' => false, 'message' => $skip, 'skipped' => true]; } } if (!$this->lockService->acquire($ticketId)) { SplitTicketSyncLogger::log('sync', 'lock busy', ['ticketId' => $ticketId]); SplitTicketSyncLogger::clearTicketContext(); return ['success' => false, 'message' => '工单正在同步中', 'skipped' => true]; } try { $result = $this->doSync($ticket, $syncMode); SplitTicketSyncLogger::log('sync', 'syncOne end', $result); return $result; } finally { $this->lockService->release($ticketId); SplitTicketSyncLogger::clearTicketContext(); } } /** * 扫描到期工单并同步(直连 PHP 与 Node 双通道) */ public function syncDueTickets(): int { $autoSyncTypes = $this->resolveAutoSyncTicketTypes(); if ($autoSyncTypes === []) { SplitTicketSyncLogger::logCron('scan skip', ['reason' => 'no auto-sync ticket types']); return 0; } $failThreshold = SplitSyncConfigService::getFailPauseThreshold(); $directTypes = array_values(array_intersect( $autoSyncTypes, SplitScrmSpiderFactory::listDirectSyncTypes() )); $nodeTypes = array_values(array_intersect( $autoSyncTypes, SplitScrmSpiderFactory::listNodeSyncTypes() )); SplitTicketSyncLogger::logCron('scan start', [ 'autoSyncTypes' => $autoSyncTypes, 'directTypes' => $directTypes, 'nodeTypes' => $nodeTypes, 'failThreshold' => $failThreshold, 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), ]); $directResult = $this->runDirectCronChannel($directTypes, $failThreshold); $nodeResult = $this->runNodeCronChannel($nodeTypes, $failThreshold); $processedCount = $directResult['processedCount'] + $nodeResult['processedCount']; $processed = array_merge($directResult['processed'], $nodeResult['processed']); $skipped = array_merge($directResult['skipped'], $nodeResult['skipped']); $candidateIds = array_merge($directResult['candidateIds'], $nodeResult['candidateIds']); $openNotSynced = $this->buildOpenNotSyncedAudit( $autoSyncTypes, $failThreshold, SplitSyncConfigService::getMaxTicketsPerCronRun(), $candidateIds, $processed, $skipped, $nodeResult['stoppedByNodeBusy'] ); $successCount = count(array_filter($processed, static function (array $row): bool { return !empty($row['dispatched']); })); $failedCount = count($processed) - $successCount; SplitTicketSyncLogger::logCron('scan end', [ 'processedCount' => $processedCount, 'successCount' => $successCount, 'failedCount' => $failedCount, 'skippedCount' => count($skipped), 'directChannel' => $directResult['summary'], 'nodeChannel' => $nodeResult['summary'], 'processed' => $processed, 'skipped' => $skipped, 'openNotSynced' => $openNotSynced, ]); SplitTicketSyncLogger::log('cron', 'scan end', [ 'processedCount' => $processedCount, 'successCount' => $successCount, 'failedCount' => $failedCount, ]); return $processedCount; } /** * 直连 PHP 通道:挑选到期工单并投递后台 CLI,不占 Node 名额 * * @param list $directTypes * @return array{ * processedCount:int, * processed:list>, * skipped:list>, * candidateIds:list, * summary:array * } */ private function runDirectCronChannel(array $directTypes, int $failThreshold): array { if ($directTypes === []) { return $this->emptyCronChannelResult('direct'); } $list = $this->loadOpenTicketsForTypes($directTypes, $failThreshold, 0); $candidateIds = $this->extractTicketIds($list); /** @var list> $dueList */ $dueList = []; /** @var list> $skipped */ $skipped = []; foreach ($list as $ticket) { $ticketId = (int) $ticket['id']; $ticketType = (string) $ticket['ticket_type']; $ticketUrl = (string) $ticket['ticket_url']; $skip = $this->shouldSkip($ticket); if ($skip !== null) { $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skip); $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip, 'direct'); continue; } $dueList[] = $ticket; } SplitTicketSyncLogger::logCron('direct channel start', [ 'types' => $directTypes, 'candidateCount' => count($list), 'dueCount' => count($dueList), ]); $dispatchResult = $this->dispatchService->dispatchAuto($dueList); $skipped = array_merge($skipped, $this->mapDispatchSkippedToCron($dispatchResult['skipped'])); $failed = $this->mapDispatchFailedToCron($dispatchResult['failed']); foreach ($failed as $row) { $skipped[] = $row; $this->logCronCandidateSkipped( (int) $row['ticketId'], (string) $row['ticketType'], (string) $row['ticketUrl'], (string) $row['reason'], 'direct' ); } /** @var list> $processed */ $processed = []; foreach ($dispatchResult['queued'] as $row) { $processed[] = [ 'ticketId' => (int) ($row['ticketId'] ?? 0), 'ticketType' => (string) ($row['ticketType'] ?? ''), 'ticketUrl' => (string) ($row['ticketUrl'] ?? ''), 'channel' => 'direct', 'dispatched' => true, 'message' => '已投递后台同步', ]; } return [ 'processedCount' => count($processed), 'processed' => $processed, 'skipped' => $skipped, 'candidateIds' => $candidateIds, 'summary' => [ 'channel' => 'direct', 'processedCount' => count($processed), 'skippedCount' => count($skipped), 'dispatchMode' => 'async-cli', ], 'stoppedByNodeBusy' => false, ]; } /** * Node 通道:限流 + 类型公平轮转 + 异步 CLI 投递 * * @param list $nodeTypes * @return array{ * processedCount:int, * processed:list>, * skipped:list>, * candidateIds:list, * summary:array, * stoppedByNodeBusy:bool * } */ private function runNodeCronChannel(array $nodeTypes, int $failThreshold): array { if ($nodeTypes === []) { return $this->emptyCronChannelResult('node'); } $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun(); $poolMultiplier = SplitSyncConfigService::getNodeCandidatePoolMultiplier(); $candidateLimit = $maxPerRun * $poolMultiplier; $pool = $this->loadOpenTicketsForTypes($nodeTypes, $failThreshold, $candidateLimit); $candidateIds = $this->extractTicketIds($pool); $duePool = []; foreach ($pool as $ticket) { if ($this->shouldSkip($ticket) === null) { $duePool[] = $ticket; } } $picked = $this->pickFairNodeTickets($duePool, $maxPerRun); $picked = $this->sortTicketsBySessionKey($picked); $pickedIdMap = array_fill_keys($this->extractTicketIds($picked), true); SplitTicketSyncLogger::logCron('node channel start', [ 'types' => $nodeTypes, 'maxPerRun' => $maxPerRun, 'poolCount' => count($pool), 'dueCount' => count($duePool), 'pickedCount' => count($picked), 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), 'dispatchMode' => 'async-cli', ]); /** @var list> $skipped */ $skipped = []; foreach ($pool as $ticket) { $ticketId = (int) $ticket['id']; if (isset($pickedIdMap[$ticketId])) { continue; } $skip = $this->shouldSkip($ticket); if ($skip !== null) { continue; } $skipped[] = $this->buildCronTicketEntry( $ticketId, (string) $ticket['ticket_type'], (string) $ticket['ticket_url'], sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit) ); } $dispatchResult = $this->dispatchService->dispatchAuto($picked); $skipped = array_merge($skipped, $this->mapDispatchSkippedToCron($dispatchResult['skipped'])); $failed = $this->mapDispatchFailedToCron($dispatchResult['failed']); foreach ($failed as $row) { $skipped[] = $row; $this->logCronCandidateSkipped( (int) $row['ticketId'], (string) $row['ticketType'], (string) $row['ticketUrl'], (string) $row['reason'], 'node' ); } /** @var list> $processed */ $processed = []; foreach ($dispatchResult['queued'] as $row) { $processed[] = [ 'ticketId' => (int) ($row['ticketId'] ?? 0), 'ticketType' => (string) ($row['ticketType'] ?? ''), 'ticketUrl' => (string) ($row['ticketUrl'] ?? ''), 'sessionKey' => (string) ($row['sessionKey'] ?? ''), 'channel' => 'node', 'dispatched' => true, 'message' => '已投递后台同步', ]; } return [ 'processedCount' => count($processed), 'processed' => $processed, 'skipped' => $skipped, 'candidateIds' => $candidateIds, 'summary' => [ 'channel' => 'node', 'processedCount' => count($processed), 'skippedCount' => count($skipped), 'maxPerRun' => $maxPerRun, 'dispatchMode' => 'async-cli', ], 'stoppedByNodeBusy' => $dispatchResult['stoppedByNodeBusy'], ]; } /** * @param list> $rows * @return list */ private function mapDispatchSkippedToCron(array $rows): array { $result = []; foreach ($rows as $row) { $result[] = $this->buildCronTicketEntry( (int) ($row['ticketId'] ?? 0), (string) ($row['ticketType'] ?? ''), (string) ($row['ticketUrl'] ?? ''), (string) ($row['reason'] ?? '已跳过') ); } return $result; } /** * @param list> $rows * @return list */ private function mapDispatchFailedToCron(array $rows): array { return $this->mapDispatchSkippedToCron($rows); } /** * @param list $types * @return list */ private function loadOpenTicketsForTypes(array $types, int $failThreshold, int $limit): array { if ($types === []) { return []; } $query = Ticket::where('status', 'normal') ->where('manual_manage', 0) ->whereIn('ticket_type', $types); if ($failThreshold > 0) { $query->where(function ($query) use ($failThreshold): void { $query->where('sync_fail_count', '<', $failThreshold) ->whereOr(function ($query): void { $query->where('sync_fail_pause_at', '>', 0) ->where('sync_auto_sync_stopped', 0); }); }); } $query->order('sync_time', 'asc')->order('id', 'asc'); if ($limit > 0) { $query->limit($limit); } $list = $query->select(); return $list instanceof \think\Collection ? $list->all() : (array) $list; } /** * Node 通道:按工单类型公平轮转选取(每轮每类型优先取 1 条) * * @param list $dueTickets * @return list */ private function pickFairNodeTickets(array $dueTickets, int $maxPerRun): array { if ($dueTickets === [] || $maxPerRun <= 0) { return []; } /** @var array> $byType */ $byType = []; foreach ($dueTickets as $ticket) { $type = (string) $ticket['ticket_type']; $byType[$type][] = $ticket; } $typeKeys = array_keys($byType); sort($typeKeys, SORT_STRING); /** @var list $picked */ $picked = []; while (count($picked) < $maxPerRun) { $added = false; foreach ($typeKeys as $type) { if (count($picked) >= $maxPerRun) { break; } if ($byType[$type] === []) { continue; } $picked[] = array_shift($byType[$type]); $added = true; } if (!$added) { break; } } return $picked; } /** * @param list> $list * @return list */ private function extractTicketIds(array $list): array { $ids = []; foreach ($list as $row) { $ids[] = (int) (is_array($row) ? ($row['id'] ?? 0) : ($row['id'] ?? 0)); } return $ids; } /** * @return array{ * processedCount:int, * processed:list>, * skipped:list>, * candidateIds:list, * summary:array, * stoppedByNodeBusy:bool * } */ private function emptyCronChannelResult(string $channel): array { return [ 'processedCount' => 0, 'processed' => [], 'skipped' => [], 'candidateIds' => [], 'summary' => ['channel' => $channel, 'processedCount' => 0, 'skippedCount' => 0], 'stoppedByNodeBusy' => false, ]; } /** * 已配置自动同步周期且已实现蜘蛛的工单类型 * * @return list */ private function resolveAutoSyncTicketTypes(): array { $types = []; foreach (SplitScrmSpiderFactory::listSupportedTypes() as $ticketType) { if (SplitSyncConfigService::getIntervalMinutes($ticketType) > 0) { $types[] = $ticketType; } } return $types; } /** * @return array{success:bool,message:string} */ private function doSync(Ticket $ticket, string $syncMode): array { $freshTicket = Ticket::get((int) $ticket['id']); if (!$freshTicket) { return ['success' => false, 'message' => '工单不存在', 'skipped' => true]; } if ($this->ruleService->isManuallyClosed($freshTicket)) { return ['success' => false, 'message' => '工单已手动关闭,暂停同步', 'skipped' => true]; } $ticket = $freshTicket; $ticketType = (string) $ticket['ticket_type']; $pageUrl = trim((string) $ticket['ticket_url']); if ($pageUrl === '') { SplitTicketSyncLogger::log('sync', 'empty pageUrl'); $this->markFailure($ticket, '工单链接为空'); return ['success' => false, 'message' => '工单链接为空']; } if (!SplitScrmSpiderFactory::isSupported($ticketType)) { SplitTicketSyncLogger::log('sync', 'spider not supported', ['ticketType' => $ticketType]); $this->markFailure($ticket, '工单类型尚未实现蜘蛛'); return ['success' => false, 'message' => '工单类型尚未实现蜘蛛']; } SplitTicketSyncLogger::log('sync', 'create spider', [ 'ticketType' => $ticketType, 'hasAccount' => trim((string) ($ticket['account'] ?? '')) !== '', ]); $spider = SplitScrmSpiderFactory::create( $ticketType, $pageUrl, (string) ($ticket['account'] ?? ''), (string) ($ticket['password'] ?? '') ); if ($spider === null) { $this->markFailure($ticket, '无法创建蜘蛛实例'); return ['success' => false, 'message' => '无法创建蜘蛛实例']; } Db::startTrans(); try { SplitTicketSyncLogger::log('sync', 'spider run begin'); $finalData = $spider->run(); if (!$finalData instanceof UnifiedScrmData) { throw new Exception('蜘蛛返回数据无效'); } $this->numberSync->syncFromUnifiedData($ticket, $finalData); $completeCount = max(0, $finalData->todayNewCount); $this->ruleService->applyTicketStatusRules($ticket, $completeCount); $freshTicket = Ticket::get((int) $ticket['id']) ?: $ticket; if ((string) $freshTicket['status'] === 'hidden') { $this->ruleService->cascadeTicketClosedToNumbers($freshTicket); } else { $this->ruleService->applyNumberRules($freshTicket); } $ticket = $freshTicket; $inboundCount = $this->numberSync->sumInboundForTicket($ticket); $speed = $this->calcSpeedPerHour($ticket, $completeCount); $payload = [ 'complete_count' => $completeCount, 'inbound_count' => $inboundCount, 'speed_per_hour' => $speed['speed'], 'number_count' => max(0, $finalData->total), 'number_offline_count' => max(0, $finalData->totalOffline), 'number_banned_count' => 0, 'online_count' => max(0, $finalData->totalOnline), 'sync_fail_count' => 0, 'speed_snapshot_count' => $speed['snapshot_count'], 'speed_snapshot_time' => $speed['snapshot_time'], ]; $this->applySyncResult($ticket, $payload, true, '', $syncMode); Db::commit(); SplitTicketSyncLogger::log('sync', 'db commit ok', $payload); return ['success' => true, 'message' => '同步成功']; } catch (\Throwable $e) { Db::rollback(); $msg = $e->getMessage(); SplitTicketSyncLogger::log('sync', 'exception', [ 'type' => get_class($e), 'message' => $msg, 'file' => $e->getFile(), 'line' => $e->getLine(), ]); $this->markFailure($ticket, $msg); return ['success' => false, 'message' => $msg]; } } private function shouldSkip(Ticket $ticket): ?string { if ($this->ruleService->isManuallyClosed($ticket)) { return '工单已手动关闭,暂停同步'; } if ((string) $ticket['status'] === 'hidden') { return '工单已关闭'; } if ($this->failRetryService->isPermanentlyStopped($ticket)) { return '自动同步已终止(重试已用尽)'; } if ($this->failRetryService->isInRetryMode($ticket)) { $retrySkip = $this->failRetryService->shouldAllowAutoRetry($ticket); if ($retrySkip !== null) { return $retrySkip; } if (!SplitScrmSpiderFactory::isSupported((string) $ticket['ticket_type'])) { return '工单类型尚未实现'; } $interval = SplitSyncConfigService::getIntervalMinutes((string) $ticket['ticket_type']); if ($interval <= 0) { return '该类型未配置自动同步周期'; } return null; } $failThreshold = SplitSyncConfigService::getFailPauseThreshold(); if ($failThreshold > 0 && (int) ($ticket['sync_fail_count'] ?? 0) >= $failThreshold) { return sprintf('连续同步失败超过%d次已暂停', $failThreshold); } if (!SplitScrmSpiderFactory::isSupported((string) $ticket['ticket_type'])) { return '工单类型尚未实现'; } $interval = SplitSyncConfigService::getIntervalMinutes((string) $ticket['ticket_type']); if ($interval <= 0) { return '该类型未配置自动同步周期'; } $lastSync = (int) ($ticket['sync_time'] ?? 0); $elapsed = $lastSync > 0 ? (time() - $lastSync) : null; if ($lastSync > 0 && $elapsed !== null && $elapsed < ($interval * 60)) { SplitTicketSyncLogger::log('sync', 'interval not reached', [ 'intervalMinutes' => $interval, 'elapsedSeconds' => $elapsed, 'needSeconds' => $interval * 60, ]); return '未到同步周期'; } return null; } /** * @param array $payload */ public function applySyncResult(Ticket $ticket, array $payload, bool $success, string $message = '', string $syncMode = 'manual'): void { $now = time(); $data = [ 'complete_count' => max(0, (int) ($payload['complete_count'] ?? 0)), 'inbound_count' => max(0, (int) ($payload['inbound_count'] ?? 0)), 'speed_per_hour' => max(0, (float) ($payload['speed_per_hour'] ?? 0)), 'number_count' => max(0, (int) ($payload['number_count'] ?? 0)), 'number_offline_count' => max(0, (int) ($payload['number_offline_count'] ?? 0)), 'number_banned_count' => max(0, (int) ($payload['number_banned_count'] ?? 0)), 'online_count' => max(0, (int) ($payload['online_count'] ?? 0)), 'sync_status' => $success ? 'success' : 'error', 'sync_time' => $now, 'sync_message' => $success ? '' : $message, 'sync_fail_count' => $success ? 0 : ((int) ($ticket['sync_fail_count'] ?? 0) + 1), 'speed_snapshot_count' => (int) ($payload['speed_snapshot_count'] ?? $ticket['speed_snapshot_count'] ?? 0), 'speed_snapshot_time' => (int) ($payload['speed_snapshot_time'] ?? $ticket['speed_snapshot_time'] ?? 0), ]; if ($success) { $data['sync_success_time'] = $now; $data['sync_success_mode'] = in_array($syncMode, ['auto', 'manual'], true) ? $syncMode : 'manual'; $data = array_merge($data, $this->failRetryService->clearOnSuccessFields()); } if (!$ticket->allowField(array_keys($data))->save($data)) { throw new Exception('工单同步结果保存失败'); } } private function markFailure(Ticket $ticket, string $message): void { $failCount = (int) ($ticket['sync_fail_count'] ?? 0) + 1; $failThreshold = SplitSyncConfigService::getFailPauseThreshold(); $previousSyncStatus = (string) ($ticket['sync_status'] ?? 'pending'); $neverSyncedSuccessfully = $previousSyncStatus === 'pending' && (int) ($ticket['sync_time'] ?? 0) <= 0; $update = [ 'sync_status' => 'error', 'sync_time' => time(), 'sync_message' => $message, 'sync_fail_count' => $failCount, ]; if ($this->failRetryService->isInRetryMode($ticket)) { $update = array_merge($update, $this->failRetryService->onRetryFailure($ticket)); $ticket->save($update); return; } if ($failThreshold > 0 && $failCount >= $failThreshold && SplitSyncConfigService::hasFailRetrySchedule()) { $update = array_merge($update, $this->failRetryService->enterRetryIfNeeded($ticket, $failCount, $failThreshold)); $ticket->save($update); return; } if ($neverSyncedSuccessfully || ($failThreshold > 0 && $failCount >= $failThreshold)) { $update['status'] = 'hidden'; } $ticket->save($update); if (isset($update['status']) && $update['status'] === 'hidden') { $fresh = Ticket::get((int) $ticket['id']); if ($fresh) { $this->ruleService->cascadeTicketClosedToNumbers($fresh); } } } /** * @return array{speed:float,snapshot_count:int,snapshot_time:int} */ private function calcSpeedPerHour(Ticket $ticket, int $currentComplete): array { $now = time(); $snapshotTime = (int) ($ticket['speed_snapshot_time'] ?? 0); $snapshotCount = (int) ($ticket['speed_snapshot_count'] ?? 0); if ($snapshotTime <= 0) { return [ 'speed' => 0.0, 'snapshot_count' => $currentComplete, 'snapshot_time' => $now, ]; } $elapsed = $now - $snapshotTime; if ($elapsed >= 3600) { return [ 'speed' => 0.0, 'snapshot_count' => $currentComplete, 'snapshot_time' => $now, ]; } $hours = $elapsed > 0 ? ($elapsed / 3600) : 0; $delta = $currentComplete - $snapshotCount; $speed = ($delta < 0 || $hours <= 0) ? 0.0 : round($delta / $hours, 2); return [ 'speed' => $speed, 'snapshot_count' => $snapshotCount, 'snapshot_time' => $snapshotTime, ]; } /** * @return array{ticketId:int,ticketType:string,ticketUrl:string,reason:string} */ private function buildCronTicketEntry(int $ticketId, string $ticketType, string $ticketUrl, string $reason): array { return [ 'ticketId' => $ticketId, 'ticketType' => $ticketType, 'ticketUrl' => $ticketUrl, 'reason' => $reason, ]; } private function logCronCandidateSkipped(int $ticketId, string $ticketType, string $ticketUrl, string $reason, string $channel = 'legacy'): void { $ctx = [ 'ticketId' => $ticketId, 'ticketType' => $ticketType, 'ticketUrl' => $ticketUrl, 'reason' => $reason, 'channel' => $channel, ]; SplitTicketSyncLogger::logCron('candidate skipped', $ctx); SplitTicketSyncLogger::log('cron', 'candidate skipped', $ctx); } /** * 开启但未在本轮同步的工单及原因(供 cron 汇总) * * @param list $candidateIds * @param list> $processed * @param list> $skipped * @return list> */ private function buildOpenNotSyncedAudit( array $autoSyncTypes, int $failThreshold, int $maxPerRun, array $candidateIds, array $processed, array $skipped, bool $stoppedByNodeBusy ): array { $handledIds = []; foreach ($processed as $row) { $handledIds[(int) ($row['ticketId'] ?? 0)] = true; } foreach ($skipped as $row) { $handledIds[(int) ($row['ticketId'] ?? 0)] = true; } $candidateIdMap = array_fill_keys($candidateIds, true); $supportedTypes = SplitScrmSpiderFactory::listSupportedTypes(); if ($supportedTypes === []) { return []; } $openList = Ticket::where('status', 'normal') ->whereIn('ticket_type', $supportedTypes) ->field('id,ticket_type,ticket_url,sync_fail_count,sync_time,sync_status,sync_fail_pause_at,sync_fail_retry_index,sync_auto_sync_stopped,status,manual_manage') ->select(); $result = []; foreach ($openList as $ticketRow) { $ticket = $ticketRow instanceof Ticket ? $ticketRow->toArray() : (array) $ticketRow; $ticketId = (int) ($ticket['id'] ?? 0); if ($ticketId <= 0 || isset($handledIds[$ticketId])) { continue; } $ticketType = (string) ($ticket['ticket_type'] ?? ''); $ticketUrl = (string) ($ticket['ticket_url'] ?? ''); $reason = $this->resolveOpenNotSyncedReason( $ticket, $autoSyncTypes, $failThreshold, $maxPerRun, $candidateIdMap, $stoppedByNodeBusy ); if ($reason === null) { continue; } $result[] = [ 'ticketId' => $ticketId, 'ticketType' => $ticketType, 'ticketUrl' => $ticketUrl, 'reason' => $reason, ]; } return $result; } /** * @param array $ticket * @param array $candidateIdMap */ private function resolveOpenNotSyncedReason( array $ticket, array $autoSyncTypes, int $failThreshold, int $maxPerRun, array $candidateIdMap, bool $stoppedByNodeBusy ): ?string { $ticketType = (string) ($ticket['ticket_type'] ?? ''); $failCount = (int) ($ticket['sync_fail_count'] ?? 0); if ($this->failRetryService->isPermanentlyStopped($ticket)) { return '自动同步已终止(重试已用尽)'; } if ($this->failRetryService->isInRetryMode($ticket)) { $retrySkip = $this->failRetryService->shouldAllowAutoRetry($ticket); if ($retrySkip !== null) { return $retrySkip; } } if ($failThreshold > 0 && $failCount >= $failThreshold && !SplitSyncConfigService::hasFailRetrySchedule()) { return sprintf('连续同步失败超过%d次已暂停自动同步', $failThreshold); } if (!in_array($ticketType, $autoSyncTypes, true)) { return '该类型未配置自动同步周期'; } if (!isset($candidateIdMap[(int) ($ticket['id'] ?? 0)])) { if (SplitScrmSpiderFactory::requiresNode($ticketType)) { $poolSize = SplitSyncConfigService::getMaxTicketsPerCronRun() * SplitSyncConfigService::getNodeCandidatePoolMultiplier(); return sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $poolSize); } return '直连通道:未到同步周期或本轮已处理'; } if ($stoppedByNodeBusy) { return 'Node 队列繁忙,本轮未执行'; } $skip = $this->shouldSkip(Ticket::get((int) $ticket['id']) ?: new Ticket($ticket)); return $skip ?? '未知原因,未进入执行队列'; } /** * 按 sessionKey(ticketType:host)分组排序,使同域名工单连续执行以命中 Browser 温池 * * @param \think\Collection|array> $list * @return list> */ private function sortTicketsBySessionKey($list): array { $rows = $list instanceof \think\Collection ? $list->all() : (array) $list; usort($rows, static function ($a, $b): int { $typeA = (string) (is_array($a) ? ($a['ticket_type'] ?? '') : ($a['ticket_type'] ?? '')); $urlA = (string) (is_array($a) ? ($a['ticket_url'] ?? '') : ($a['ticket_url'] ?? '')); $typeB = (string) (is_array($b) ? ($b['ticket_type'] ?? '') : ($b['ticket_type'] ?? '')); $urlB = (string) (is_array($b) ? ($b['ticket_url'] ?? '') : ($b['ticket_url'] ?? '')); $keyA = AntiBotConfigBuilder::resolveSessionKey($typeA, $urlA); $keyB = AntiBotConfigBuilder::resolveSessionKey($typeB, $urlB); if ($keyA === $keyB) { $timeA = (int) (is_array($a) ? ($a['sync_time'] ?? 0) : ($a['sync_time'] ?? 0)); $timeB = (int) (is_array($b) ? ($b['sync_time'] ?? 0) : ($b['sync_time'] ?? 0)); if ($timeA === $timeB) { $idA = (int) (is_array($a) ? ($a['id'] ?? 0) : ($a['id'] ?? 0)); $idB = (int) (is_array($b) ? ($b['id'] ?? 0) : ($b['id'] ?? 0)); return $idA <=> $idB; } return $timeA <=> $timeB; } return strcmp($keyA, $keyB); }); return $rows; } }