root
568 lines
18 KiB
PHP
Executable File
568 lines
18 KiB
PHP
Executable File
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace app\common\service;
|
||
|
||
use app\common\library\scrm\AntiBotConfigBuilder;
|
||
use think\Db;
|
||
|
||
/**
|
||
* 工单同步 CLI 后台投递
|
||
*
|
||
* Web / Cron 仅负责鉴权与投递，实际 Spider 在独立 CLI 进程中执行，
|
||
* 避免长时间占用 PHP-FPM、Cron 全局锁或 Session 锁。
|
||
*
|
||
* 并行策略：
|
||
* - API 优先类型（Whatshub 等）：独立并行投递，不占 Node Real Browser 槽位
|
||
* - 直连 PHP 类型：独立并行投递
|
||
* - Node 类型：不同 sessionKey（域名）并行；同一 sessionKey 内串行以命中 Browser 温池
|
||
*/
|
||
class SplitTicketSyncDispatchService
|
||
{
|
||
/**
|
||
* 投递手工同步任务到后台 CLI
|
||
*
|
||
* @param int[] $ticketIds 已通过权限校验的工单 ID
|
||
* @return array{queued:int[],skipped:int[],failed:int[]}
|
||
*/
|
||
public function dispatchManual(array $ticketIds): array
|
||
{
|
||
$lockService = new SplitTicketSyncLockService();
|
||
$metaList = $this->loadTicketMetaList($ticketIds);
|
||
|
||
/** @var list<array{id:int,ticket_type:string,ticket_url:string}> $toDispatch */
|
||
$toDispatch = [];
|
||
$skipped = [];
|
||
$failed = [];
|
||
|
||
foreach ($metaList as $meta) {
|
||
$ticketId = (int) $meta['id'];
|
||
if ($lockService->isLocked($ticketId)) {
|
||
$skipped[] = $ticketId;
|
||
continue;
|
||
}
|
||
$toDispatch[] = $meta;
|
||
}
|
||
|
||
$dispatchResult = $this->dispatchTicketMetaList($toDispatch, 'manual');
|
||
|
||
SplitTicketSyncLogger::log('web', 'manual sync dispatched', [
|
||
'queued' => $dispatchResult['queued'],
|
||
'skipped' => $skipped,
|
||
'failed' => $dispatchResult['failed'],
|
||
'phpCli' => $this->resolvePhpBinary(),
|
||
]);
|
||
|
||
return [
|
||
'queued' => $dispatchResult['queued'],
|
||
'skipped' => $skipped,
|
||
'failed' => $dispatchResult['failed'],
|
||
];
|
||
}
|
||
|
||
/**
|
||
* Cron 自动同步：将挑选后的到期工单投递到后台 CLI（Cron 进程立即返回）
|
||
*
|
||
* @param list<array{id:int,ticket_type:string,ticket_url:string}|Ticket|\think\Model> $tickets
|
||
* @return array{
|
||
* queued:list<array<string,mixed>>,
|
||
* skipped:list<array<string,mixed>>,
|
||
* failed:list<array<string,mixed>>,
|
||
* stoppedByNodeBusy:bool
|
||
* }
|
||
*/
|
||
public function dispatchAuto(array $tickets): array
|
||
{
|
||
$lockService = new SplitTicketSyncLockService();
|
||
/** @var list<array{id:int,ticket_type:string,ticket_url:string}> $toDispatch */
|
||
$toDispatch = [];
|
||
/** @var list<array<string,mixed>> $skipped */
|
||
$skipped = [];
|
||
|
||
foreach ($tickets as $ticket) {
|
||
$meta = $this->normalizeTicketMeta($ticket);
|
||
if ($meta === null) {
|
||
continue;
|
||
}
|
||
$ticketId = (int) $meta['id'];
|
||
if ($lockService->isLocked($ticketId)) {
|
||
$skipped[] = $this->buildDispatchEntry(
|
||
$ticketId,
|
||
(string) $meta['ticket_type'],
|
||
(string) $meta['ticket_url'],
|
||
'工单正在同步中'
|
||
);
|
||
continue;
|
||
}
|
||
$toDispatch[] = $meta;
|
||
}
|
||
|
||
$dispatchResult = $this->dispatchTicketMetaList($toDispatch, 'auto');
|
||
|
||
SplitTicketSyncLogger::log('cron', 'auto sync dispatched', [
|
||
'queuedCount' => count($dispatchResult['queued']),
|
||
'skippedCount' => count($skipped) + count($dispatchResult['skipped']),
|
||
'failedCount' => count($dispatchResult['failed']),
|
||
'stoppedByNodeBusy' => $dispatchResult['stoppedByNodeBusy'],
|
||
'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(),
|
||
]);
|
||
|
||
return [
|
||
'queued' => $dispatchResult['queued'],
|
||
'skipped' => array_merge($skipped, $dispatchResult['skipped']),
|
||
'failed' => $dispatchResult['failed'],
|
||
'stoppedByNodeBusy' => $dispatchResult['stoppedByNodeBusy'],
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 查询仍在同步中的工单 ID（依据 runtime 文件锁）
|
||
*
|
||
* @param int[] $ticketIds
|
||
* @return int[]
|
||
*/
|
||
public function filterSyncingIds(array $ticketIds): array
|
||
{
|
||
$lockService = new SplitTicketSyncLockService();
|
||
$syncing = [];
|
||
foreach ($ticketIds as $rawId) {
|
||
$ticketId = (int) $rawId;
|
||
if ($ticketId > 0 && $lockService->isLocked($ticketId)) {
|
||
$syncing[] = $ticketId;
|
||
}
|
||
}
|
||
|
||
return $syncing;
|
||
}
|
||
|
||
/**
|
||
* @param list<array{id:int,ticket_type:string,ticket_url:string}> $metaList
|
||
* @return array{
|
||
* queued:list<array<string,mixed>>,
|
||
* skipped:list<array<string,mixed>>,
|
||
* failed:list<array<string,mixed>>,
|
||
* stoppedByNodeBusy:bool
|
||
* }
|
||
*/
|
||
private function dispatchTicketMetaList(array $metaList, string $source): array
|
||
{
|
||
$php = $this->resolvePhpBinary();
|
||
$think = $this->resolveThinkScript();
|
||
$root = $this->resolveRootPath();
|
||
|
||
/** @var list<array<string,mixed>> $queued */
|
||
$queued = [];
|
||
/** @var list<array<string,mixed>> $skipped */
|
||
$skipped = [];
|
||
/** @var list<array<string,mixed>> $failed */
|
||
$failed = [];
|
||
$stoppedByNodeBusy = false;
|
||
|
||
/** @var list<array{id:int,ticket_type:string,ticket_url:string}> $apiFirstList */
|
||
$apiFirstList = [];
|
||
/** @var list<array{id:int,ticket_type:string,ticket_url:string}> $directList */
|
||
$directList = [];
|
||
/** @var array<string, list<array{id:int,ticket_type:string,ticket_url:string}>> $nodeGroups */
|
||
$nodeGroups = [];
|
||
|
||
foreach ($metaList as $meta) {
|
||
$ticketType = (string) $meta['ticket_type'];
|
||
if (SplitScrmSpiderFactory::isApiFirstSyncType($ticketType)) {
|
||
$apiFirstList[] = $meta;
|
||
continue;
|
||
}
|
||
if (!SplitScrmSpiderFactory::requiresNode($ticketType)) {
|
||
$directList[] = $meta;
|
||
continue;
|
||
}
|
||
$sessionKey = AntiBotConfigBuilder::resolveSessionKey(
|
||
$ticketType,
|
||
(string) $meta['ticket_url']
|
||
);
|
||
$nodeGroups[$sessionKey][] = $meta;
|
||
}
|
||
|
||
foreach ($apiFirstList as $meta) {
|
||
$this->spawnSingleMeta($php, $think, $root, $meta, $source, $queued, $failed);
|
||
}
|
||
|
||
foreach ($directList as $meta) {
|
||
$this->spawnSingleMeta($php, $think, $root, $meta, $source, $queued, $failed);
|
||
}
|
||
|
||
foreach ($nodeGroups as $sessionKey => $group) {
|
||
if ($group === []) {
|
||
continue;
|
||
}
|
||
|
||
/** @var list<array{id:int,ticket_type:string,ticket_url:string}> $nodeReady */
|
||
$nodeReady = [];
|
||
foreach ($group as $meta) {
|
||
$ticketType = (string) $meta['ticket_type'];
|
||
if (!SplitNodeHealthService::canAcceptWorkForTicketType($ticketType)) {
|
||
$stoppedByNodeBusy = true;
|
||
$skipped[] = $this->buildDispatchEntry(
|
||
(int) $meta['id'],
|
||
$ticketType,
|
||
(string) $meta['ticket_url'],
|
||
'Node 队列繁忙，本轮未投递'
|
||
);
|
||
continue;
|
||
}
|
||
$nodeReady[] = $meta;
|
||
}
|
||
|
||
if ($nodeReady === []) {
|
||
continue;
|
||
}
|
||
|
||
if (count($nodeReady) === 1) {
|
||
$this->spawnSingleMeta($php, $think, $root, $nodeReady[0], $source, $queued, $failed);
|
||
continue;
|
||
}
|
||
|
||
$ids = array_map(static function (array $row): int {
|
||
return (int) $row['id'];
|
||
}, $nodeReady);
|
||
|
||
if ($this->spawnCliSequential($php, $think, $root, $ids, $source)) {
|
||
foreach ($nodeReady as $meta) {
|
||
$queued[] = $this->buildQueuedEntry($meta, $sessionKey, 'node-group-serial');
|
||
}
|
||
} else {
|
||
foreach ($nodeReady as $meta) {
|
||
$failed[] = $this->buildDispatchEntry(
|
||
(int) $meta['id'],
|
||
(string) $meta['ticket_type'],
|
||
(string) $meta['ticket_url'],
|
||
'CLI 投递失败'
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
return [
|
||
'queued' => $queued,
|
||
'skipped' => $skipped,
|
||
'failed' => $failed,
|
||
'stoppedByNodeBusy' => $stoppedByNodeBusy,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* @param list<array<string,mixed>> $queued
|
||
* @param list<array<string,mixed>> $failed
|
||
* @param array{id:int,ticket_type:string,ticket_url:string} $meta
|
||
*/
|
||
private function spawnSingleMeta(
|
||
string $php,
|
||
string $think,
|
||
string $root,
|
||
array $meta,
|
||
string $source,
|
||
array &$queued,
|
||
array &$failed
|
||
): void {
|
||
$ticketId = (int) $meta['id'];
|
||
if ($this->spawnCli($php, $think, $root, $ticketId, $source)) {
|
||
$queued[] = $this->buildQueuedEntry($meta, '', 'parallel');
|
||
return;
|
||
}
|
||
|
||
$failed[] = $this->buildDispatchEntry(
|
||
$ticketId,
|
||
(string) $meta['ticket_type'],
|
||
(string) $meta['ticket_url'],
|
||
'CLI 投递失败'
|
||
);
|
||
}
|
||
|
||
/**
|
||
* @param array{id:int,ticket_type:string,ticket_url:string} $meta
|
||
* @return array<string,mixed>
|
||
*/
|
||
private function buildQueuedEntry(array $meta, string $sessionKey, string $mode): array
|
||
{
|
||
return [
|
||
'ticketId' => (int) $meta['id'],
|
||
'ticketType' => (string) $meta['ticket_type'],
|
||
'ticketUrl' => (string) $meta['ticket_url'],
|
||
'sessionKey' => $sessionKey,
|
||
'mode' => $mode,
|
||
'dispatched' => true,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* @return array{ticketId:int,ticketType:string,ticketUrl:string,reason:string}
|
||
*/
|
||
private function buildDispatchEntry(int $ticketId, string $ticketType, string $ticketUrl, string $reason): array
|
||
{
|
||
return [
|
||
'ticketId' => $ticketId,
|
||
'ticketType' => $ticketType,
|
||
'ticketUrl' => $ticketUrl,
|
||
'reason' => $reason,
|
||
];
|
||
}
|
||
|
||
/**
|
||
* @param int[] $ticketIds
|
||
* @return list<array{id:int,ticket_type:string,ticket_url:string}>
|
||
*/
|
||
private function loadTicketMetaList(array $ticketIds): array
|
||
{
|
||
$ids = [];
|
||
foreach ($ticketIds as $rawId) {
|
||
$ticketId = (int) $rawId;
|
||
if ($ticketId > 0) {
|
||
$ids[] = $ticketId;
|
||
}
|
||
}
|
||
if ($ids === []) {
|
||
return [];
|
||
}
|
||
|
||
$rows = Db::name('split_ticket')
|
||
->where('id', 'in', $ids)
|
||
->field('id,ticket_type,ticket_url')
|
||
->select();
|
||
|
||
$list = [];
|
||
foreach ($rows as $row) {
|
||
$list[] = [
|
||
'id' => (int) $row['id'],
|
||
'ticket_type' => (string) ($row['ticket_type'] ?? ''),
|
||
'ticket_url' => (string) ($row['ticket_url'] ?? ''),
|
||
];
|
||
}
|
||
|
||
return $list;
|
||
}
|
||
|
||
/**
|
||
* @param array<string,mixed>|object $ticket
|
||
* @return array{id:int,ticket_type:string,ticket_url:string}|null
|
||
*/
|
||
private function normalizeTicketMeta($ticket): ?array
|
||
{
|
||
if (is_object($ticket) && method_exists($ticket, 'toArray')) {
|
||
$ticket = $ticket->toArray();
|
||
}
|
||
if (!is_array($ticket)) {
|
||
return null;
|
||
}
|
||
$ticketId = (int) ($ticket['id'] ?? 0);
|
||
if ($ticketId <= 0) {
|
||
return null;
|
||
}
|
||
|
||
return [
|
||
'id' => $ticketId,
|
||
'ticket_type' => (string) ($ticket['ticket_type'] ?? ''),
|
||
'ticket_url' => (string) ($ticket['ticket_url'] ?? ''),
|
||
];
|
||
}
|
||
|
||
/**
|
||
* 后台启动 split:sync-tickets CLI（单工单，独立进程）
|
||
*/
|
||
private function spawnCli(string $php, string $think, string $root, int $ticketId, string $source): bool
|
||
{
|
||
$logDir = $this->resolveLogDir();
|
||
$logFile = $logDir . 'cli_' . $ticketId . '_' . date('YmdHis') . '.log';
|
||
|
||
$syncMode = $this->normalizeSyncModeForCli($source);
|
||
$inner = sprintf(
|
||
'%s %s split:sync-tickets --ticket=%d --mode=%s >> %s 2>&1',
|
||
escapeshellarg($php),
|
||
escapeshellarg($think),
|
||
$ticketId,
|
||
$syncMode,
|
||
escapeshellarg($logFile)
|
||
);
|
||
|
||
return $this->runBackgroundCommand($root, $inner, $ticketId, $source . '-parallel');
|
||
}
|
||
|
||
/**
|
||
* 同一 sessionKey 内串行执行多个 CLI（单 nohup 子 shell）
|
||
*
|
||
* @param int[] $ticketIds
|
||
*/
|
||
private function spawnCliSequential(string $php, string $think, string $root, array $ticketIds, string $source): bool
|
||
{
|
||
if ($ticketIds === []) {
|
||
return false;
|
||
}
|
||
|
||
$logDir = $this->resolveLogDir();
|
||
$parts = [];
|
||
foreach ($ticketIds as $ticketId) {
|
||
$ticketId = (int) $ticketId;
|
||
if ($ticketId <= 0) {
|
||
continue;
|
||
}
|
||
$logFile = $logDir . 'cli_' . $ticketId . '_' . date('YmdHis') . '.log';
|
||
$syncMode = $this->normalizeSyncModeForCli($source);
|
||
$parts[] = sprintf(
|
||
'%s %s split:sync-tickets --ticket=%d --mode=%s >> %s 2>&1',
|
||
escapeshellarg($php),
|
||
escapeshellarg($think),
|
||
$ticketId,
|
||
$syncMode,
|
||
escapeshellarg($logFile)
|
||
);
|
||
}
|
||
if ($parts === []) {
|
||
return false;
|
||
}
|
||
|
||
$inner = implode('; ', $parts);
|
||
|
||
return $this->runBackgroundCommand($root, $inner, $ticketIds, $source . '-node-serial');
|
||
}
|
||
|
||
/**
|
||
* @param int|int[] $ticketRef 日志用 ticketId 或 ID 列表
|
||
*/
|
||
private function runBackgroundCommand(string $root, string $inner, $ticketRef, string $mode): bool
|
||
{
|
||
if ($this->isWindows()) {
|
||
$command = sprintf('start /B cmd /C %s', $inner);
|
||
} else {
|
||
$command = sprintf('cd %s && nohup bash -c %s &', escapeshellarg($root), escapeshellarg($inner));
|
||
}
|
||
|
||
if ($this->canUseExec()) {
|
||
@exec($command, $output, $exitCode);
|
||
SplitTicketSyncLogger::log('dispatch', 'cli spawn exec', [
|
||
'ticketRef' => $ticketRef,
|
||
'mode' => $mode,
|
||
'command' => $command,
|
||
'exitCode' => $exitCode,
|
||
]);
|
||
return true;
|
||
}
|
||
|
||
if ($this->canUseShellExec()) {
|
||
@shell_exec($command);
|
||
SplitTicketSyncLogger::log('dispatch', 'cli spawn shell_exec', [
|
||
'ticketRef' => $ticketRef,
|
||
'mode' => $mode,
|
||
'command' => $command,
|
||
]);
|
||
return true;
|
||
}
|
||
|
||
SplitTicketSyncLogger::log('dispatch', 'cli spawn failed: exec disabled', [
|
||
'ticketRef' => $ticketRef,
|
||
'mode' => $mode,
|
||
]);
|
||
|
||
return false;
|
||
}
|
||
|
||
private function resolvePhpBinary(): string
|
||
{
|
||
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();
|
||
return rtrim($root, '/\\') . DIRECTORY_SEPARATOR . 'think';
|
||
}
|
||
|
||
private function resolveRootPath(): string
|
||
{
|
||
if (defined('ROOT_PATH') && ROOT_PATH !== '') {
|
||
return rtrim(ROOT_PATH, '/\\') . DIRECTORY_SEPARATOR;
|
||
}
|
||
|
||
return rtrim(dirname(__DIR__, 3), '/\\') . DIRECTORY_SEPARATOR;
|
||
}
|
||
|
||
private function resolveLogDir(): string
|
||
{
|
||
$runtime = defined('RUNTIME_PATH') ? RUNTIME_PATH : ($this->resolveRootPath() . 'runtime' . DIRECTORY_SEPARATOR);
|
||
$dir = rtrim($runtime, '/\\') . DIRECTORY_SEPARATOR . 'split_ticket_sync' . DIRECTORY_SEPARATOR;
|
||
if (!is_dir($dir)) {
|
||
@mkdir($dir, 0755, true);
|
||
}
|
||
|
||
return $dir;
|
||
}
|
||
|
||
private function isWindows(): bool
|
||
{
|
||
return strtoupper(substr(PHP_OS, 0, 3)) === 'WIN';
|
||
}
|
||
|
||
/**
|
||
* @return string[]
|
||
*/
|
||
private function disabledFunctions(): array
|
||
{
|
||
$raw = (string) ini_get('disable_functions');
|
||
if ($raw === '') {
|
||
return [];
|
||
}
|
||
|
||
return array_filter(array_map('trim', explode(',', $raw)));
|
||
}
|
||
|
||
private function canUseExec(): bool
|
||
{
|
||
return function_exists('exec') && !in_array('exec', $this->disabledFunctions(), true);
|
||
}
|
||
|
||
private function canUseShellExec(): bool
|
||
{
|
||
return function_exists('shell_exec') && !in_array('shell_exec', $this->disabledFunctions(), true);
|
||
}
|
||
|
||
/**
|
||
* 投递 CLI 时携带 --mode，供 sync_success_mode 区分自动/手动
|
||
*/
|
||
private function normalizeSyncModeForCli(string $source): string
|
||
{
|
||
return in_array($source, ['auto', 'manual'], true) ? $source : 'manual';
|
||
}
|
||
}
|