diff --git a/_php_code/Huojian.php b/_php_code/Huojian.php index e797ebb..fb23bba 100755 --- a/_php_code/Huojian.php +++ b/_php_code/Huojian.php @@ -172,8 +172,8 @@ class Huojian extends AbstractScrmSpider { $unifiedData = $this->unifiedData; - $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0); - + $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0) + (int) ($allListPagesData[0]['data']['counterWorker']['newRemovedTodayFriend'] ?? 0); + $count = 0; foreach ($allListPagesData as $pageRaw) { $records = $pageRaw['data']['counterCsAccountVo'] ?? []; diff --git a/_php_code/index.php b/_php_code/index.php index 99b9d18..61aad2c 100755 --- a/_php_code/index.php +++ b/_php_code/index.php @@ -85,7 +85,7 @@ try { // 长链:快速验证业务页拦截(推荐先测通) // $pageUrl = 'https://user.a2c.chat/visitors/counter/share?id=1b2cf9a91e2647c185b252e723c871de'; // 短链:与客户后台一致,上线前必须回归 - $pageUrl = 'https://yyk.ink/91SEWrL'; + $pageUrl = 'https://yyk.ink/3xj1B2'; $username = ""; $password = ""; $spider = (new A2c($pageUrl, $username, $password))->setVerbose(true); diff --git a/application/admin/command/Install/split_config_node_channel.sql b/application/admin/command/Install/split_config_node_channel.sql new file mode 100644 index 0000000..953393c --- /dev/null +++ b/application/admin/command/Install/split_config_node_channel.sql @@ -0,0 +1,19 @@ +-- 双通道定时同步:Node 通道上限与候选池配置(已有环境增量执行) +SET NAMES utf8mb4; + +INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`) +SELECT 'split_sync_max_per_cron', 'split', 'Node通道单次同步上限', '每分钟 cron 在 Node 通道最多处理几条到期工单(仅影响需调用 Node 的类型);纯 PHP 直连类型不受此限制', 'number', '', '5', '', '', '', NULL +FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_max_per_cron' LIMIT 1); + +INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`) +SELECT 'split_sync_node_candidate_multiplier', 'split', 'Node通道候选池倍数', 'Node 候选工单数 = 单次上限 × 本倍数,用于公平轮转', 'number', '', '10', '', '', '', NULL +FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_node_candidate_multiplier' LIMIT 1); + +UPDATE `fa_config` +SET `title` = 'Node通道单次同步上限', + `tip` = '每分钟 cron 在 Node 通道最多处理几条到期工单(仅影响需调用 Node 的类型);纯 PHP 直连类型不受此限制' +WHERE `name` = 'split_sync_max_per_cron'; + +UPDATE `fa_config` +SET `value` = '5' +WHERE `name` = 'split_sync_max_per_cron' AND (`value` = '' OR `value` = '2'); diff --git a/application/admin/command/Install/split_ticket_sync_log_enhance.sql b/application/admin/command/Install/split_ticket_sync_log_enhance.sql old mode 100644 new mode 100755 diff --git a/application/common/library/scrm/AntiBotConfigBuilder.php b/application/common/library/scrm/AntiBotConfigBuilder.php old mode 100644 new mode 100755 diff --git a/application/common/library/scrm/HuojianUrlHelper.php b/application/common/library/scrm/HuojianUrlHelper.php old mode 100644 new mode 100755 diff --git a/application/common/library/scrm/spider/HuojianSpider.php b/application/common/library/scrm/spider/HuojianSpider.php index 291b01e..786f5fb 100755 --- a/application/common/library/scrm/spider/HuojianSpider.php +++ b/application/common/library/scrm/spider/HuojianSpider.php @@ -80,7 +80,7 @@ class HuojianSpider extends AbstractScrmSpider protected function parseToUnifiedData($detailData, array $allListPagesData): UnifiedScrmData { $unifiedData = $this->unifiedData; - $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0); + $unifiedData->todayNewCount = (int) ($allListPagesData[0]['data']['counterWorker']['newTodayFriend'] ?? 0) + (int) ($allListPagesData[0]['data']['counterWorker']['newRemovedTodayFriend'] ?? 0); $count = 0; foreach ($allListPagesData as $pageRaw) { $records = $pageRaw['data']['counterCsAccountVo'] ?? []; diff --git a/application/common/service/SplitPageUrlResolver.php b/application/common/service/SplitPageUrlResolver.php old mode 100644 new mode 100755 diff --git a/application/common/service/SplitScrmSpiderFactory.php b/application/common/service/SplitScrmSpiderFactory.php index 479251e..e96981b 100755 --- a/application/common/service/SplitScrmSpiderFactory.php +++ b/application/common/service/SplitScrmSpiderFactory.php @@ -4,6 +4,7 @@ declare(strict_types=1); namespace app\common\service; +use app\common\library\scrm\AbstractScrmSpider; use app\common\library\scrm\ScrmSpiderInterface; use app\common\library\scrm\spider\A2cSpider; use app\common\library\scrm\spider\ChatknowSpider; @@ -59,6 +60,49 @@ class SplitScrmSpiderFactory return array_keys(self::MAP); } + /** + * 同步是否依赖 Node Headless(纯 PHP cURL 等为 false) + */ + public static function requiresNode(string $ticketType): bool + { + $class = self::resolveClass($ticketType); + if ($class === null) { + return true; + } + + return is_subclass_of($class, AbstractScrmSpider::class); + } + + /** + * @return list + */ + public static function listDirectSyncTypes(): array + { + $types = []; + foreach (self::listSupportedTypes() as $ticketType) { + if (!self::requiresNode($ticketType)) { + $types[] = $ticketType; + } + } + + return $types; + } + + /** + * @return list + */ + public static function listNodeSyncTypes(): array + { + $types = []; + foreach (self::listSupportedTypes() as $ticketType) { + if (self::requiresNode($ticketType)) { + $types[] = $ticketType; + } + } + + return $types; + } + /** * @return ScrmSpiderInterface|null */ diff --git a/application/common/service/SplitSyncConfigService.php b/application/common/service/SplitSyncConfigService.php index 3bf0d86..91b74c8 100755 --- a/application/common/service/SplitSyncConfigService.php +++ b/application/common/service/SplitSyncConfigService.php @@ -57,11 +57,23 @@ class SplitSyncConfigService { $value = self::getConfigValue('split_sync_max_per_cron'); if ($value === '') { - return 2; + return 5; } return max(1, min(20, (int) $value)); } + /** + * Node 通道候选池倍数(候选数 = maxPerRun × 本值) + */ + public static function getNodeCandidatePoolMultiplier(): int + { + $value = self::getConfigValue('split_sync_node_candidate_multiplier'); + if ($value === '') { + return 10; + } + return max(3, min(50, (int) $value)); + } + /** * 全局 cron 锁过期秒数,防止异常退出后永久占锁 */ @@ -86,6 +98,53 @@ class SplitSyncConfigService return max(0, (int) $value); } + /** + * 后台投递 / Cron 使用的 PHP CLI 可执行文件路径 + * + * Web(FPM) 环境下 PHP_BINARY 指向 php-fpm,不能用于 think 命令。 + */ + public static function getCliPhpBinary(): string + { + $configured = trim(self::getConfigValue('split_sync_cli_php')); + if ($configured !== '' && self::isUsableCliPhp($configured)) { + return $configured; + } + + if (defined('PHP_BINDIR') && PHP_BINDIR !== '') { + $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php'; + if (self::isUsableCliPhp($candidate)) { + return $candidate; + } + } + + if (PHP_SAPI === 'cli' && defined('PHP_BINARY') && PHP_BINARY !== '' && self::isUsableCliPhp(PHP_BINARY)) { + return PHP_BINARY; + } + + foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) { + if (self::isUsableCliPhp($candidate)) { + return $candidate; + } + } + + return 'php'; + } + + /** + * 判断路径是否为可用的 PHP CLI(排除 php-fpm) + */ + private static function isUsableCliPhp(string $path): bool + { + if (stripos($path, 'fpm') !== false) { + return false; + } + if ($path === 'php') { + return true; + } + + return is_file($path) && is_executable($path); + } + private static function getConfigValue(string $name): string { $site = Config::get('site.' . $name); diff --git a/application/common/service/SplitTicketSyncDispatchService.php b/application/common/service/SplitTicketSyncDispatchService.php old mode 100644 new mode 100755 index c62a386..cf3e131 --- a/application/common/service/SplitTicketSyncDispatchService.php +++ b/application/common/service/SplitTicketSyncDispatchService.php @@ -51,6 +51,7 @@ class SplitTicketSyncDispatchService 'queued' => $queued, 'skipped' => $skipped, 'failed' => $failed, + 'phpCli' => $php, ]); return [ @@ -130,13 +131,43 @@ class SplitTicketSyncDispatchService private function resolvePhpBinary(): string { - if (defined('PHP_BINARY') && PHP_BINARY !== '') { - return PHP_BINARY; + if (method_exists(SplitSyncConfigService::class, 'getCliPhpBinary')) { + return SplitSyncConfigService::getCliPhpBinary(); } + return $this->resolvePhpBinaryFallback(); + } + + /** + * 当 SplitSyncConfigService 未部署 getCliPhpBinary 时的兜底解析 + */ + private function resolvePhpBinaryFallback(): string + { + if (defined('PHP_BINDIR') && PHP_BINDIR !== '') { + $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php'; + if ($this->isUsableCliPhp($candidate)) { + return $candidate; + } + } + foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) { + if ($this->isUsableCliPhp($candidate)) { + return $candidate; + } + } return 'php'; } + private function isUsableCliPhp(string $path): bool + { + if (stripos($path, 'fpm') !== false) { + return false; + } + if ($path === 'php') { + return true; + } + return is_file($path) && is_executable($path); + } + private function resolveThinkScript(): string { $root = $this->resolveRootPath(); diff --git a/application/common/service/SplitTicketSyncService.php b/application/common/service/SplitTicketSyncService.php index d4c938a..03cca46 100755 --- a/application/common/service/SplitTicketSyncService.php +++ b/application/common/service/SplitTicketSyncService.php @@ -80,131 +80,49 @@ class SplitTicketSyncService } /** - * 扫描到期工单并同步(限流 + Node 队列感知) + * 扫描到期工单并同步(直连 PHP 与 Node 双通道) */ public function syncDueTickets(): int { - $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun(); $autoSyncTypes = $this->resolveAutoSyncTicketTypes(); - $failThreshold = SplitSyncConfigService::getFailPauseThreshold(); - if ($autoSyncTypes === []) { SplitTicketSyncLogger::logCron('scan skip', ['reason' => 'no auto-sync ticket types']); return 0; } - if (!SplitNodeHealthService::canAcceptWork()) { - SplitTicketSyncLogger::logCron('scan skip', [ - 'reason' => 'node queue busy', - 'queue' => SplitNodeHealthService::getQueueSnapshot(), - ]); - return 0; - } + $failThreshold = SplitSyncConfigService::getFailPauseThreshold(); + $directTypes = array_values(array_intersect( + $autoSyncTypes, + SplitScrmSpiderFactory::listDirectSyncTypes() + )); + $nodeTypes = array_values(array_intersect( + $autoSyncTypes, + SplitScrmSpiderFactory::listNodeSyncTypes() + )); - $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $autoSyncTypes); - if ($failThreshold > 0) { - $query->where('sync_fail_count', '<', $failThreshold); - } - $list = $query->order('sync_time', 'asc')->order('id', 'asc') - ->limit($maxPerRun * 5) - ->select(); - $list = $this->sortTicketsBySessionKey($list); - - $candidateCount = count($list); SplitTicketSyncLogger::logCron('scan start', [ - 'candidateCount' => $candidateCount, - 'maxPerRun' => $maxPerRun, - 'autoSyncTypes' => $autoSyncTypes, - 'failThreshold' => $failThreshold, - 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), - 'groupedBy' => 'sessionKey', - ]); - SplitTicketSyncLogger::log('cron', 'scan start', [ - 'candidateCount' => $candidateCount, - 'maxPerRun' => $maxPerRun, - 'autoSyncTypes' => $autoSyncTypes, - 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), + 'autoSyncTypes' => $autoSyncTypes, + 'directTypes' => $directTypes, + 'nodeTypes' => $nodeTypes, + 'failThreshold' => $failThreshold, + 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), ]); - /** @var list> $processed */ - $processed = []; - /** @var list> $skipped */ - $skipped = []; - $count = 0; - $stoppedByNodeBusy = false; + $directResult = $this->runDirectCronChannel($directTypes, $failThreshold); + $nodeResult = $this->runNodeCronChannel($nodeTypes, $failThreshold); - foreach ($list as $index => $ticket) { - $ticketId = (int) $ticket['id']; - $ticketUrl = (string) $ticket['ticket_url']; - $ticketType = (string) $ticket['ticket_type']; - $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl); - - if ($count >= $maxPerRun) { - $reason = '本轮处理上限已满'; - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason); - continue; - } - - $skip = $this->shouldSkip($ticket); - if ($skip !== null) { - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skip); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip); - continue; - } - - if (!SplitNodeHealthService::canAcceptWork()) { - $reason = 'Node 队列繁忙,停止本轮后续同步'; - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason); - $stoppedByNodeBusy = true; - for ($i = $index + 1; $i < $candidateCount; $i++) { - $remain = $list[$i]; - $remainReason = 'Node 队列繁忙,本轮未执行'; - $skipped[] = $this->buildCronTicketEntry( - (int) $remain['id'], - (string) $remain['ticket_type'], - (string) $remain['ticket_url'], - $remainReason - ); - } - SplitTicketSyncLogger::log('cron', 'node busy, stop batch', [ - 'processed' => $count, - 'queue' => SplitNodeHealthService::getQueueSnapshot(), - ]); - break; - } - - $result = $this->syncOne($ticketId, false, true, 'auto'); - if (!empty($result['skipped'])) { - $skipReason = (string) ($result['message'] ?? '已跳过'); - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason); - continue; - } - - $processed[] = [ - 'ticketId' => $ticketId, - 'ticketType' => $ticketType, - 'ticketUrl' => $ticketUrl, - 'sessionKey' => $sessionKey, - 'success' => !empty($result['success']), - 'message' => (string) ($result['message'] ?? ''), - ]; - $count++; - } - - $candidateIds = array_map(static function ($row) { - return (int) (is_array($row) ? ($row['id'] ?? 0) : ($row['id'] ?? 0)); - }, $list); + $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, - $maxPerRun, + SplitSyncConfigService::getMaxTicketsPerCronRun(), $candidateIds, $processed, $skipped, - $stoppedByNodeBusy + $nodeResult['stoppedByNodeBusy'] ); $successCount = count(array_filter($processed, static function (array $row): bool { @@ -213,21 +131,362 @@ class SplitTicketSyncService $failedCount = count($processed) - $successCount; SplitTicketSyncLogger::logCron('scan end', [ - 'processedCount' => $count, + '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' => $count, + 'processedCount' => $processedCount, 'successCount' => $successCount, 'failedCount' => $failedCount, ]); - return $count; + return $processedCount; + } + + /** + * 直连 PHP 通道:按周期同步,不占 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); + + SplitTicketSyncLogger::logCron('direct channel start', [ + 'types' => $directTypes, + 'candidateCount' => count($list), + ]); + + /** @var list> $processed */ + $processed = []; + /** @var list> $skipped */ + $skipped = []; + $count = 0; + + 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; + } + + $result = $this->syncOne($ticketId, false, true, 'auto'); + if (!empty($result['skipped'])) { + $skipReason = (string) ($result['message'] ?? '已跳过'); + $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason); + $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'direct'); + continue; + } + + $processed[] = [ + 'ticketId' => $ticketId, + 'ticketType' => $ticketType, + 'ticketUrl' => $ticketUrl, + 'channel' => 'direct', + 'success' => !empty($result['success']), + 'message' => (string) ($result['message'] ?? ''), + ]; + $count++; + } + + return [ + 'processedCount' => $count, + 'processed' => $processed, + 'skipped' => $skipped, + 'candidateIds' => $candidateIds, + 'summary' => [ + 'channel' => 'direct', + 'processedCount' => $count, + 'skippedCount' => count($skipped), + ], + 'stoppedByNodeBusy' => false, + ]; + } + + /** + * Node 通道:限流 + 类型公平轮转 + * + * @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(), + ]); + + if (!SplitNodeHealthService::canAcceptWork()) { + SplitTicketSyncLogger::logCron('node channel skip', [ + 'reason' => 'node queue busy', + 'queue' => SplitNodeHealthService::getQueueSnapshot(), + ]); + $skipped = []; + foreach ($picked as $ticket) { + $skipped[] = $this->buildCronTicketEntry( + (int) $ticket['id'], + (string) $ticket['ticket_type'], + (string) $ticket['ticket_url'], + 'Node 队列繁忙,本轮未执行' + ); + } + 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) + ); + } + + return [ + 'processedCount' => 0, + 'processed' => [], + 'skipped' => $skipped, + 'candidateIds' => $candidateIds, + 'summary' => [ + 'channel' => 'node', + 'processedCount' => 0, + 'skippedCount' => count($skipped), + 'reason' => 'node queue busy', + ], + 'stoppedByNodeBusy' => true, + ]; + } + + /** @var list> $processed */ + $processed = []; + /** @var list> $skipped */ + $skipped = []; + $count = 0; + $stoppedByNodeBusy = false; + + foreach ($pool as $ticket) { + $ticketId = (int) $ticket['id']; + $ticketType = (string) $ticket['ticket_type']; + $ticketUrl = (string) $ticket['ticket_url']; + + if (!isset($pickedIdMap[$ticketId])) { + $skip = $this->shouldSkip($ticket); + if ($skip !== null) { + continue; + } + $skipped[] = $this->buildCronTicketEntry( + $ticketId, + $ticketType, + $ticketUrl, + sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit) + ); + continue; + } + + if (!SplitNodeHealthService::canAcceptWork()) { + $stoppedByNodeBusy = true; + $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙,停止本轮后续同步'); + $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙,停止本轮后续同步', 'node'); + continue; + } + + $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl); + $result = $this->syncOne($ticketId, false, true, 'auto'); + if (!empty($result['skipped'])) { + $skipReason = (string) ($result['message'] ?? '已跳过'); + $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason); + $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'node'); + continue; + } + + $processed[] = [ + 'ticketId' => $ticketId, + 'ticketType' => $ticketType, + 'ticketUrl' => $ticketUrl, + 'sessionKey' => $sessionKey, + 'channel' => 'node', + 'success' => !empty($result['success']), + 'message' => (string) ($result['message'] ?? ''), + ]; + $count++; + } + + return [ + 'processedCount' => $count, + 'processed' => $processed, + 'skipped' => $skipped, + 'candidateIds' => $candidateIds, + 'summary' => [ + 'channel' => 'node', + 'processedCount' => $count, + 'skippedCount' => count($skipped), + 'maxPerRun' => $maxPerRun, + ], + 'stoppedByNodeBusy' => $stoppedByNodeBusy, + ]; + } + + /** + * @param list $types + * @return list + */ + private function loadOpenTicketsForTypes(array $types, int $failThreshold, int $limit): array + { + if ($types === []) { + return []; + } + + $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $types); + if ($failThreshold > 0) { + $query->where('sync_fail_count', '<', $failThreshold); + } + $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, + ]; } /** @@ -466,13 +725,14 @@ class SplitTicketSyncService ]; } - private function logCronCandidateSkipped(int $ticketId, string $ticketType, string $ticketUrl, string $reason): void + 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); @@ -515,14 +775,15 @@ class SplitTicketSyncService ->select(); $result = []; - foreach ($openList as $ticket) { - $ticketId = (int) $ticket['id']; - if (isset($handledIds[$ticketId])) { + 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']; + $ticketType = (string) ($ticket['ticket_type'] ?? ''); + $ticketUrl = (string) ($ticket['ticket_url'] ?? ''); $reason = $this->resolveOpenNotSyncedReason( $ticket, $autoSyncTypes, @@ -568,7 +829,13 @@ class SplitTicketSyncService return '该类型未配置自动同步周期'; } if (!isset($candidateIdMap[(int) ($ticket['id'] ?? 0)])) { - return sprintf('未进入本轮候选队列(仅扫描最久未同步的前%d条)', $maxPerRun * 5); + if (SplitScrmSpiderFactory::requiresNode($ticketType)) { + $poolSize = SplitSyncConfigService::getMaxTicketsPerCronRun() + * SplitSyncConfigService::getNodeCandidatePoolMultiplier(); + return sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $poolSize); + } + + return '直连通道:未到同步周期或本轮已处理'; } if ($stoppedByNodeBusy) { return 'Node 队列繁忙,本轮未执行'; diff --git a/application/extra/site.php b/application/extra/site.php index b5c6ea4..e1419ec 100755 --- a/application/extra/site.php +++ b/application/extra/site.php @@ -58,5 +58,6 @@ return array ( 'split_sync_interval_whatshub' => '5', 'split_sync_interval_sihai' => '0', 'split_sync_fail_pause_threshold' => '6', + 'split_sync_cli_php' => '', 'split_sync_interval_chatknow' => '5', ); diff --git a/patches/application/admin/command/Install/split_config_node_channel.sql b/patches/application/admin/command/Install/split_config_node_channel.sql new file mode 100644 index 0000000..953393c --- /dev/null +++ b/patches/application/admin/command/Install/split_config_node_channel.sql @@ -0,0 +1,19 @@ +-- 双通道定时同步:Node 通道上限与候选池配置(已有环境增量执行) +SET NAMES utf8mb4; + +INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`) +SELECT 'split_sync_max_per_cron', 'split', 'Node通道单次同步上限', '每分钟 cron 在 Node 通道最多处理几条到期工单(仅影响需调用 Node 的类型);纯 PHP 直连类型不受此限制', 'number', '', '5', '', '', '', NULL +FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_max_per_cron' LIMIT 1); + +INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`) +SELECT 'split_sync_node_candidate_multiplier', 'split', 'Node通道候选池倍数', 'Node 候选工单数 = 单次上限 × 本倍数,用于公平轮转', 'number', '', '10', '', '', '', NULL +FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_node_candidate_multiplier' LIMIT 1); + +UPDATE `fa_config` +SET `title` = 'Node通道单次同步上限', + `tip` = '每分钟 cron 在 Node 通道最多处理几条到期工单(仅影响需调用 Node 的类型);纯 PHP 直连类型不受此限制' +WHERE `name` = 'split_sync_max_per_cron'; + +UPDATE `fa_config` +SET `value` = '5' +WHERE `name` = 'split_sync_max_per_cron' AND (`value` = '' OR `value` = '2'); diff --git a/patches/application/admin/command/Install/split_config_sync.sql b/patches/application/admin/command/Install/split_config_sync.sql index 6fd68d1..55231e0 100755 --- a/patches/application/admin/command/Install/split_config_sync.sql +++ b/patches/application/admin/command/Install/split_config_sync.sql @@ -58,9 +58,13 @@ SELECT 'split_sync_interval_sihai', 'split', '四海云控同步周期(分钟)', FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_interval_sihai' LIMIT 1); INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`) -SELECT 'split_sync_max_per_cron', 'split', '单次定时同步上限', '每分钟 cron 最多处理几条到期工单,防止打满 Node Browser 槽位', 'number', '', '2', '', '', '', NULL +SELECT 'split_sync_max_per_cron', 'split', 'Node通道单次同步上限', '每分钟 cron 在 Node 通道最多处理几条到期工单(仅影响需调用 Node 的类型);纯 PHP 直连类型不受此限制', 'number', '', '5', '', '', '', NULL FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_max_per_cron' LIMIT 1); +INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`) +SELECT 'split_sync_node_candidate_multiplier', 'split', 'Node通道候选池倍数', 'Node 候选工单数 = 单次上限 × 本倍数,用于公平轮转', 'number', '', '10', '', '', '', NULL +FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_node_candidate_multiplier' LIMIT 1); + INSERT INTO `fa_config` (`name`, `group`, `title`, `tip`, `type`, `visible`, `value`, `content`, `rule`, `extend`, `setting`) SELECT 'split_sync_cron_lock_ttl', 'split', '定时同步全局锁TTL(秒)', '异常退出后自动释放全局锁,避免后续 cron 永久跳过', 'number', '', '1800', '', '', '', NULL FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM `fa_config` WHERE `name` = 'split_sync_cron_lock_ttl' LIMIT 1); diff --git a/patches/application/common/service/SplitScrmSpiderFactory.php b/patches/application/common/service/SplitScrmSpiderFactory.php index 479251e..e96981b 100755 --- a/patches/application/common/service/SplitScrmSpiderFactory.php +++ b/patches/application/common/service/SplitScrmSpiderFactory.php @@ -4,6 +4,7 @@ declare(strict_types=1); namespace app\common\service; +use app\common\library\scrm\AbstractScrmSpider; use app\common\library\scrm\ScrmSpiderInterface; use app\common\library\scrm\spider\A2cSpider; use app\common\library\scrm\spider\ChatknowSpider; @@ -59,6 +60,49 @@ class SplitScrmSpiderFactory return array_keys(self::MAP); } + /** + * 同步是否依赖 Node Headless(纯 PHP cURL 等为 false) + */ + public static function requiresNode(string $ticketType): bool + { + $class = self::resolveClass($ticketType); + if ($class === null) { + return true; + } + + return is_subclass_of($class, AbstractScrmSpider::class); + } + + /** + * @return list + */ + public static function listDirectSyncTypes(): array + { + $types = []; + foreach (self::listSupportedTypes() as $ticketType) { + if (!self::requiresNode($ticketType)) { + $types[] = $ticketType; + } + } + + return $types; + } + + /** + * @return list + */ + public static function listNodeSyncTypes(): array + { + $types = []; + foreach (self::listSupportedTypes() as $ticketType) { + if (self::requiresNode($ticketType)) { + $types[] = $ticketType; + } + } + + return $types; + } + /** * @return ScrmSpiderInterface|null */ diff --git a/patches/application/common/service/SplitSyncConfigService.php b/patches/application/common/service/SplitSyncConfigService.php index 3bf0d86..91b74c8 100755 --- a/patches/application/common/service/SplitSyncConfigService.php +++ b/patches/application/common/service/SplitSyncConfigService.php @@ -57,11 +57,23 @@ class SplitSyncConfigService { $value = self::getConfigValue('split_sync_max_per_cron'); if ($value === '') { - return 2; + return 5; } return max(1, min(20, (int) $value)); } + /** + * Node 通道候选池倍数(候选数 = maxPerRun × 本值) + */ + public static function getNodeCandidatePoolMultiplier(): int + { + $value = self::getConfigValue('split_sync_node_candidate_multiplier'); + if ($value === '') { + return 10; + } + return max(3, min(50, (int) $value)); + } + /** * 全局 cron 锁过期秒数,防止异常退出后永久占锁 */ @@ -86,6 +98,53 @@ class SplitSyncConfigService return max(0, (int) $value); } + /** + * 后台投递 / Cron 使用的 PHP CLI 可执行文件路径 + * + * Web(FPM) 环境下 PHP_BINARY 指向 php-fpm,不能用于 think 命令。 + */ + public static function getCliPhpBinary(): string + { + $configured = trim(self::getConfigValue('split_sync_cli_php')); + if ($configured !== '' && self::isUsableCliPhp($configured)) { + return $configured; + } + + if (defined('PHP_BINDIR') && PHP_BINDIR !== '') { + $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php'; + if (self::isUsableCliPhp($candidate)) { + return $candidate; + } + } + + if (PHP_SAPI === 'cli' && defined('PHP_BINARY') && PHP_BINARY !== '' && self::isUsableCliPhp(PHP_BINARY)) { + return PHP_BINARY; + } + + foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) { + if (self::isUsableCliPhp($candidate)) { + return $candidate; + } + } + + return 'php'; + } + + /** + * 判断路径是否为可用的 PHP CLI(排除 php-fpm) + */ + private static function isUsableCliPhp(string $path): bool + { + if (stripos($path, 'fpm') !== false) { + return false; + } + if ($path === 'php') { + return true; + } + + return is_file($path) && is_executable($path); + } + private static function getConfigValue(string $name): string { $site = Config::get('site.' . $name); diff --git a/patches/application/common/service/SplitTicketSyncDispatchService.php b/patches/application/common/service/SplitTicketSyncDispatchService.php index c62a386..cf3e131 100644 --- a/patches/application/common/service/SplitTicketSyncDispatchService.php +++ b/patches/application/common/service/SplitTicketSyncDispatchService.php @@ -51,6 +51,7 @@ class SplitTicketSyncDispatchService 'queued' => $queued, 'skipped' => $skipped, 'failed' => $failed, + 'phpCli' => $php, ]); return [ @@ -130,13 +131,43 @@ class SplitTicketSyncDispatchService private function resolvePhpBinary(): string { - if (defined('PHP_BINARY') && PHP_BINARY !== '') { - return PHP_BINARY; + if (method_exists(SplitSyncConfigService::class, 'getCliPhpBinary')) { + return SplitSyncConfigService::getCliPhpBinary(); } + return $this->resolvePhpBinaryFallback(); + } + + /** + * 当 SplitSyncConfigService 未部署 getCliPhpBinary 时的兜底解析 + */ + private function resolvePhpBinaryFallback(): string + { + if (defined('PHP_BINDIR') && PHP_BINDIR !== '') { + $candidate = rtrim(PHP_BINDIR, '/\\') . DIRECTORY_SEPARATOR . 'php'; + if ($this->isUsableCliPhp($candidate)) { + return $candidate; + } + } + foreach (['/usr/bin/php', '/usr/local/bin/php'] as $candidate) { + if ($this->isUsableCliPhp($candidate)) { + return $candidate; + } + } return 'php'; } + private function isUsableCliPhp(string $path): bool + { + if (stripos($path, 'fpm') !== false) { + return false; + } + if ($path === 'php') { + return true; + } + return is_file($path) && is_executable($path); + } + private function resolveThinkScript(): string { $root = $this->resolveRootPath(); diff --git a/patches/application/common/service/SplitTicketSyncService.php b/patches/application/common/service/SplitTicketSyncService.php index d4c938a..03cca46 100755 --- a/patches/application/common/service/SplitTicketSyncService.php +++ b/patches/application/common/service/SplitTicketSyncService.php @@ -80,131 +80,49 @@ class SplitTicketSyncService } /** - * 扫描到期工单并同步(限流 + Node 队列感知) + * 扫描到期工单并同步(直连 PHP 与 Node 双通道) */ public function syncDueTickets(): int { - $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun(); $autoSyncTypes = $this->resolveAutoSyncTicketTypes(); - $failThreshold = SplitSyncConfigService::getFailPauseThreshold(); - if ($autoSyncTypes === []) { SplitTicketSyncLogger::logCron('scan skip', ['reason' => 'no auto-sync ticket types']); return 0; } - if (!SplitNodeHealthService::canAcceptWork()) { - SplitTicketSyncLogger::logCron('scan skip', [ - 'reason' => 'node queue busy', - 'queue' => SplitNodeHealthService::getQueueSnapshot(), - ]); - return 0; - } + $failThreshold = SplitSyncConfigService::getFailPauseThreshold(); + $directTypes = array_values(array_intersect( + $autoSyncTypes, + SplitScrmSpiderFactory::listDirectSyncTypes() + )); + $nodeTypes = array_values(array_intersect( + $autoSyncTypes, + SplitScrmSpiderFactory::listNodeSyncTypes() + )); - $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $autoSyncTypes); - if ($failThreshold > 0) { - $query->where('sync_fail_count', '<', $failThreshold); - } - $list = $query->order('sync_time', 'asc')->order('id', 'asc') - ->limit($maxPerRun * 5) - ->select(); - $list = $this->sortTicketsBySessionKey($list); - - $candidateCount = count($list); SplitTicketSyncLogger::logCron('scan start', [ - 'candidateCount' => $candidateCount, - 'maxPerRun' => $maxPerRun, - 'autoSyncTypes' => $autoSyncTypes, - 'failThreshold' => $failThreshold, - 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), - 'groupedBy' => 'sessionKey', - ]); - SplitTicketSyncLogger::log('cron', 'scan start', [ - 'candidateCount' => $candidateCount, - 'maxPerRun' => $maxPerRun, - 'autoSyncTypes' => $autoSyncTypes, - 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), + 'autoSyncTypes' => $autoSyncTypes, + 'directTypes' => $directTypes, + 'nodeTypes' => $nodeTypes, + 'failThreshold' => $failThreshold, + 'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(), ]); - /** @var list> $processed */ - $processed = []; - /** @var list> $skipped */ - $skipped = []; - $count = 0; - $stoppedByNodeBusy = false; + $directResult = $this->runDirectCronChannel($directTypes, $failThreshold); + $nodeResult = $this->runNodeCronChannel($nodeTypes, $failThreshold); - foreach ($list as $index => $ticket) { - $ticketId = (int) $ticket['id']; - $ticketUrl = (string) $ticket['ticket_url']; - $ticketType = (string) $ticket['ticket_type']; - $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl); - - if ($count >= $maxPerRun) { - $reason = '本轮处理上限已满'; - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason); - continue; - } - - $skip = $this->shouldSkip($ticket); - if ($skip !== null) { - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skip); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip); - continue; - } - - if (!SplitNodeHealthService::canAcceptWork()) { - $reason = 'Node 队列繁忙,停止本轮后续同步'; - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $reason); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $reason); - $stoppedByNodeBusy = true; - for ($i = $index + 1; $i < $candidateCount; $i++) { - $remain = $list[$i]; - $remainReason = 'Node 队列繁忙,本轮未执行'; - $skipped[] = $this->buildCronTicketEntry( - (int) $remain['id'], - (string) $remain['ticket_type'], - (string) $remain['ticket_url'], - $remainReason - ); - } - SplitTicketSyncLogger::log('cron', 'node busy, stop batch', [ - 'processed' => $count, - 'queue' => SplitNodeHealthService::getQueueSnapshot(), - ]); - break; - } - - $result = $this->syncOne($ticketId, false, true, 'auto'); - if (!empty($result['skipped'])) { - $skipReason = (string) ($result['message'] ?? '已跳过'); - $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason); - $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason); - continue; - } - - $processed[] = [ - 'ticketId' => $ticketId, - 'ticketType' => $ticketType, - 'ticketUrl' => $ticketUrl, - 'sessionKey' => $sessionKey, - 'success' => !empty($result['success']), - 'message' => (string) ($result['message'] ?? ''), - ]; - $count++; - } - - $candidateIds = array_map(static function ($row) { - return (int) (is_array($row) ? ($row['id'] ?? 0) : ($row['id'] ?? 0)); - }, $list); + $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, - $maxPerRun, + SplitSyncConfigService::getMaxTicketsPerCronRun(), $candidateIds, $processed, $skipped, - $stoppedByNodeBusy + $nodeResult['stoppedByNodeBusy'] ); $successCount = count(array_filter($processed, static function (array $row): bool { @@ -213,21 +131,362 @@ class SplitTicketSyncService $failedCount = count($processed) - $successCount; SplitTicketSyncLogger::logCron('scan end', [ - 'processedCount' => $count, + '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' => $count, + 'processedCount' => $processedCount, 'successCount' => $successCount, 'failedCount' => $failedCount, ]); - return $count; + return $processedCount; + } + + /** + * 直连 PHP 通道:按周期同步,不占 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); + + SplitTicketSyncLogger::logCron('direct channel start', [ + 'types' => $directTypes, + 'candidateCount' => count($list), + ]); + + /** @var list> $processed */ + $processed = []; + /** @var list> $skipped */ + $skipped = []; + $count = 0; + + 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; + } + + $result = $this->syncOne($ticketId, false, true, 'auto'); + if (!empty($result['skipped'])) { + $skipReason = (string) ($result['message'] ?? '已跳过'); + $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason); + $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'direct'); + continue; + } + + $processed[] = [ + 'ticketId' => $ticketId, + 'ticketType' => $ticketType, + 'ticketUrl' => $ticketUrl, + 'channel' => 'direct', + 'success' => !empty($result['success']), + 'message' => (string) ($result['message'] ?? ''), + ]; + $count++; + } + + return [ + 'processedCount' => $count, + 'processed' => $processed, + 'skipped' => $skipped, + 'candidateIds' => $candidateIds, + 'summary' => [ + 'channel' => 'direct', + 'processedCount' => $count, + 'skippedCount' => count($skipped), + ], + 'stoppedByNodeBusy' => false, + ]; + } + + /** + * Node 通道:限流 + 类型公平轮转 + * + * @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(), + ]); + + if (!SplitNodeHealthService::canAcceptWork()) { + SplitTicketSyncLogger::logCron('node channel skip', [ + 'reason' => 'node queue busy', + 'queue' => SplitNodeHealthService::getQueueSnapshot(), + ]); + $skipped = []; + foreach ($picked as $ticket) { + $skipped[] = $this->buildCronTicketEntry( + (int) $ticket['id'], + (string) $ticket['ticket_type'], + (string) $ticket['ticket_url'], + 'Node 队列繁忙,本轮未执行' + ); + } + 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) + ); + } + + return [ + 'processedCount' => 0, + 'processed' => [], + 'skipped' => $skipped, + 'candidateIds' => $candidateIds, + 'summary' => [ + 'channel' => 'node', + 'processedCount' => 0, + 'skippedCount' => count($skipped), + 'reason' => 'node queue busy', + ], + 'stoppedByNodeBusy' => true, + ]; + } + + /** @var list> $processed */ + $processed = []; + /** @var list> $skipped */ + $skipped = []; + $count = 0; + $stoppedByNodeBusy = false; + + foreach ($pool as $ticket) { + $ticketId = (int) $ticket['id']; + $ticketType = (string) $ticket['ticket_type']; + $ticketUrl = (string) $ticket['ticket_url']; + + if (!isset($pickedIdMap[$ticketId])) { + $skip = $this->shouldSkip($ticket); + if ($skip !== null) { + continue; + } + $skipped[] = $this->buildCronTicketEntry( + $ticketId, + $ticketType, + $ticketUrl, + sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit) + ); + continue; + } + + if (!SplitNodeHealthService::canAcceptWork()) { + $stoppedByNodeBusy = true; + $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙,停止本轮后续同步'); + $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙,停止本轮后续同步', 'node'); + continue; + } + + $sessionKey = AntiBotConfigBuilder::resolveSessionKey($ticketType, $ticketUrl); + $result = $this->syncOne($ticketId, false, true, 'auto'); + if (!empty($result['skipped'])) { + $skipReason = (string) ($result['message'] ?? '已跳过'); + $skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, $skipReason); + $this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skipReason, 'node'); + continue; + } + + $processed[] = [ + 'ticketId' => $ticketId, + 'ticketType' => $ticketType, + 'ticketUrl' => $ticketUrl, + 'sessionKey' => $sessionKey, + 'channel' => 'node', + 'success' => !empty($result['success']), + 'message' => (string) ($result['message'] ?? ''), + ]; + $count++; + } + + return [ + 'processedCount' => $count, + 'processed' => $processed, + 'skipped' => $skipped, + 'candidateIds' => $candidateIds, + 'summary' => [ + 'channel' => 'node', + 'processedCount' => $count, + 'skippedCount' => count($skipped), + 'maxPerRun' => $maxPerRun, + ], + 'stoppedByNodeBusy' => $stoppedByNodeBusy, + ]; + } + + /** + * @param list $types + * @return list + */ + private function loadOpenTicketsForTypes(array $types, int $failThreshold, int $limit): array + { + if ($types === []) { + return []; + } + + $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $types); + if ($failThreshold > 0) { + $query->where('sync_fail_count', '<', $failThreshold); + } + $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, + ]; } /** @@ -466,13 +725,14 @@ class SplitTicketSyncService ]; } - private function logCronCandidateSkipped(int $ticketId, string $ticketType, string $ticketUrl, string $reason): void + 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); @@ -515,14 +775,15 @@ class SplitTicketSyncService ->select(); $result = []; - foreach ($openList as $ticket) { - $ticketId = (int) $ticket['id']; - if (isset($handledIds[$ticketId])) { + 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']; + $ticketType = (string) ($ticket['ticket_type'] ?? ''); + $ticketUrl = (string) ($ticket['ticket_url'] ?? ''); $reason = $this->resolveOpenNotSyncedReason( $ticket, $autoSyncTypes, @@ -568,7 +829,13 @@ class SplitTicketSyncService return '该类型未配置自动同步周期'; } if (!isset($candidateIdMap[(int) ($ticket['id'] ?? 0)])) { - return sprintf('未进入本轮候选队列(仅扫描最久未同步的前%d条)', $maxPerRun * 5); + if (SplitScrmSpiderFactory::requiresNode($ticketType)) { + $poolSize = SplitSyncConfigService::getMaxTicketsPerCronRun() + * SplitSyncConfigService::getNodeCandidatePoolMultiplier(); + return sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $poolSize); + } + + return '直连通道:未到同步周期或本轮已处理'; } if ($stoppedByNodeBusy) { return 'Node 队列繁忙,本轮未执行'; diff --git a/patches/puppeteer-api/server.js b/patches/puppeteer-api/server.js index 9a353ed..c0d45a6 100755 --- a/patches/puppeteer-api/server.js +++ b/patches/puppeteer-api/server.js @@ -60,6 +60,7 @@ const { isHuojianShortEntryUrl, applyHuojianBusinessHostHint, } = require('./lib/huojian-url'); +const { runUiPagination } = require('./lib/ui-pagination-runner'); const app = express(); app.use(express.json({ limit: '50mb' }));