damon/links
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复随机号码前先按权重随机工单，并修复降权号码筛选与清理

Browse Source
This commit is contained in:
root
2026-07-04 22:09:39 +08:00
parent 9b57b0c400
commit 6b72cd7b67
37 changed files with 2201 additions and 473 deletions
Download Patch File Download Diff File Expand all files Collapse all files
patches/application/common/service/SplitNodeHealthService.php
+14
View File
@@ -42,6 +42,20 @@ class SplitNodeHealthService
*/
public static function canAcceptWork(): bool
{
return self::canAcceptWorkForTicketType('');
}
/**
* 指定工单类型是否可向 Node 提交新任务
*
* API 优先类型（如 Whatshub）常态不占 Real Browser，调度层应跳过队列门禁。
*/
public static function canAcceptWorkForTicketType(string $ticketType): bool
{
if ($ticketType !== '' && SplitScrmSpiderFactory::isApiFirstSyncType($ticketType)) {
return true;
}
$stats = self::fetchStats();
if ($stats === null) {
return true;
patches/application/common/service/SplitNumberRatioDeferredService.php
+184
View File
@@ -0,0 +1,184 @@
<?php
declare(strict_types=1);
namespace app\common\service;
use app\admin\model\split\Number;
use think\Db;
/**
* 下号比率触线降权（ratio_deferred）规则与降权池清理
*
* 降权前提：已有落地页访问（visit_count > last_sync_visit_count），
* 且自上次同步以来进线未增长时累计 streak，触线才标记降权。
*/
class SplitNumberRatioDeferredService
{
/**
* 是否具备计入 streak 的落地页访问基数
*
* @param array<string, mixed>|Number $number
*/
public function hasLandingVisitBaseline($number): bool
{
$visitCount = $this->readInt($number, 'visit_count');
if ($visitCount <= 0) {
return false;
}
$lastVisit = $this->readInt($number, 'last_sync_visit_count');
return $visitCount > $lastVisit;
}
/**
* 根据访问与进线快照计算新的 streak
*
* @param array<string, mixed>|Number $number
*/
public function computeStreakForVisit($number, int $currentStreak): int
{
if (!$this->hasLandingVisitBaseline($number)) {
return $currentStreak;
}
$inboundCount = $this->readInt($number, 'inbound_count');
$lastInbound = $this->readInt($number, 'last_sync_inbound_count');
if ($inboundCount <= $lastInbound) {
return $currentStreak + 1;
}
return 0;
}
/**
* 同步侧：根据 visit / inbound 与上次同步检查点重算 streak
*
* @param array<string, mixed>|Number $number
*/
public function computeStreakForSync($number, int $currentStreak): int
{
$visitCount = $this->readInt($number, 'visit_count');
if ($visitCount <= 0) {
return 0;
}
$lastVisit = $this->readInt($number, 'last_sync_visit_count');
$inboundCount = $this->readInt($number, 'inbound_count');
$lastInbound = $this->readInt($number, 'last_sync_inbound_count');
if ($inboundCount > $lastInbound) {
return 0;
}
if ($visitCount > $lastVisit) {
return $currentStreak + ($visitCount - $lastVisit);
}
return $currentStreak;
}
/**
* 是否应标记 ratio_deferred（访问侧触线降权，不关 status）
*/
public function shouldMarkDeferred(int $streak, int $assignRatio, bool $hasLandingBaseline): bool
{
return $hasLandingBaseline && $assignRatio > 0 && $streak >= $assignRatio;
}
/**
* 是否为应清理的无效降权记录
*
* @param array<string, mixed>|Number $number
*/
public function isInvalidDeferredRecord($number): bool
{
if ($this->readInt($number, 'ratio_deferred') !== 1) {
return false;
}
return !$this->hasLandingVisitBaseline($number);
}
/**
* 清理无效降权标记（ratio_deferred=1 但无有效落地页访问基线）
*
* @param int[]|null $adminIds 数据权限管理员 ID 列表，null 表示不限制
* @return int 清理条数
*/
public function cleanupInvalidDeferred(?array $adminIds = null): int
{
$query = Number::where('ratio_deferred', 1);
if ($adminIds !== null && $adminIds !== []) {
$query->where('admin_id', 'in', $adminIds);
}
$rows = $query->field('id,visit_count,last_sync_visit_count,ratio_deferred,no_inbound_click_streak')->select();
if ($rows === null || $rows === []) {
return 0;
}
$now = time();
$count = 0;
Db::startTrans();
try {
foreach ($rows as $row) {
if (!$this->isInvalidDeferredRecord($row)) {
continue;
}
$update = [
'ratio_deferred' => 0,
'ratio_deferred_at' => null,
'updatetime' => $now,
];
if ($this->readInt($row, 'visit_count') <= 0) {
$update['no_inbound_click_streak'] = 0;
}
Number::where('id', (int) $row['id'])->update($update);
$count++;
}
Db::commit();
} catch (\Throwable $e) {
Db::rollback();
throw $e;
}
return $count;
}
/**
* 统计当前权限范围内无效降权条数（清理前预览）
*
* @param int[]|null $adminIds
*/
public function countInvalidDeferred(?array $adminIds = null): int
{
$query = Number::where('ratio_deferred', 1);
if ($adminIds !== null && $adminIds !== []) {
$query->where('admin_id', 'in', $adminIds);
}
$total = 0;
foreach ($query->field('id,visit_count,last_sync_visit_count,ratio_deferred')->select() as $row) {
if ($this->isInvalidDeferredRecord($row)) {
$total++;
}
}
return $total;
}
/**
* @param array<string, mixed>|Number $number
*/
private function readInt($number, string $field): int
{
if (is_array($number)) {
return (int) ($number[$field] ?? 0);
}
return (int) $number->getAttr($field);
}
}
patches/application/common/service/SplitNumberVisitRuleService.php
+13 -9
View File
@@ -14,6 +14,13 @@ use app\admin\model\split\Ticket;
*/
class SplitNumberVisitRuleService
{
private SplitNumberRatioDeferredService $ratioDeferredService;
public function __construct(?SplitNumberRatioDeferredService $ratioDeferredService = null)
{
$this->ratioDeferredService = $ratioDeferredService ?? new SplitNumberRatioDeferredService();
}
/**
* 访问计数已递增后调用：累加 streak，达到 assign_ratio 时标记 ratio_deferred=1
*/
@@ -39,15 +46,11 @@ class SplitNumberVisitRuleService
}
$assignRatio = (int) ($ticket['assign_ratio'] ?? 0);
$inboundCount = (int) $number['inbound_count'];
$lastInbound = (int) ($number['last_sync_inbound_count'] ?? 0);
$streak = (int) ($number['no_inbound_click_streak'] ?? 0);
$hasBaseline = $this->ratioDeferredService->hasLandingVisitBaseline($number);
// 自上次同步检查点以来进线未增长，则本次访问计入 streak
if ($inboundCount <= $lastInbound) {
$streak++;
} else {
$streak = 0;
if ($hasBaseline) {
$streak = $this->ratioDeferredService->computeStreakForVisit($number, $streak);
}
$update = [
@@ -55,8 +58,9 @@ class SplitNumberVisitRuleService
'updatetime' => time(),
];
// 非手动号码且达到下号比率：触线降权，不关 status，等待同步确认
if ((int) $number['manual_manage'] === 0 && $assignRatio > 0 && $streak >= $assignRatio) {
if ((int) $number['manual_manage'] === 0
&& $this->ratioDeferredService->shouldMarkDeferred($streak, $assignRatio, $hasBaseline)
) {
$update['ratio_deferred'] = 1;
$update['ratio_deferred_at'] = time();
}
patches/application/common/service/SplitNumberWeighService.php
+2 -2
View File
@@ -9,8 +9,8 @@ use app\admin\model\split\Link;
/**
* 分流链接随机打乱配置与落地页选号
*
* - random_shuffle=0：跨工单合并号码池，按 id 顺序严格轮转
* - random_shuffle=1：跨工单合并号码池，每次访问完全随机选号
* - random_shuffle=0：跨工单合并号码池，按 id 顺序全局严格轮转
* - random_shuffle=1：先按各工单开启号码数比例随机选工单，再在该工单内随机选号
* - 同步写入时 random_shuffle=1 还会打乱新号码 insert 顺序（影响 id 分布）
*/
class SplitNumberWeighService
patches/application/common/service/SplitRedirectService.php
+122 -15
View File
@@ -6,14 +6,17 @@ namespace app\common\service;
use app\admin\model\split\Link;
use app\admin\model\split\Number;
use app\admin\model\split\Ticket;
use think\Collection;
/**
* 分流链接落地页：查链、跨工单合并选号、拼接跳转 URL、访问计数
* 分流链接落地页：查链、选号、拼接跳转 URL、访问计数
*
* 号码池为同一 split_link_id 下全部 status=normal 的号码；
* 优先选用 ratio_deferred=0 的号码，全部触线降权时回退至降权池仍可轮转。
* random_shuffle 关闭时严格轮转，开启时每次访问随机选号。
* 号码池：同一 split_link_id 下 status=normal，且所属工单未关闭（ticket.status≠hidden）的号码；
* 优先选用 ratio_deferred=0，全部触线降权时回退至降权池。
*
* random_shuffle=0：跨工单合并池，全局严格轮转；
* random_shuffle=1：先按各工单开启号码数量比例随机选工单，再在该工单内随机选号。
*/
class SplitRedirectService
{
@@ -55,20 +58,17 @@ class SplitRedirectService
return null;
}
/** @var Collection<int, Number>|array<int, Number> $numbers */
$numbers = Number::where('split_link_id', (int) $link['id'])
->where('status', 'normal')
->order('id', 'asc')
->field('id,number,number_type,number_type_custom,ratio_deferred')
->select();
$list = $numbers instanceof Collection ? $numbers->all() : (array) $numbers;
$linkId = (int) $link->getAttr('id');
$list = $this->loadEligibleNumbers($linkId);
if ($list === []) {
return null;
}
$linkId = (int) $link->getAttr('id');
$picked = $this->pickNumberRow($linkId, $list);
if (SplitNumberWeighService::isRandomShuffleEnabled($linkId)) {
$picked = $this->pickNumberRowRandomByTicketWeight($linkId, $list);
} else {
$picked = $this->pickNumberRow($linkId, $list);
}
if ($picked === null) {
return null;
}
@@ -103,10 +103,117 @@ class SplitRedirectService
return $redirectUrl;
}
/**
* 可参与选号的号码：status=normal，且 ticket_name 不属于已关闭工单
*
* @return list<Number|array<string, mixed>>
*/
private function loadEligibleNumbers(int $linkId): array
{
if ($linkId <= 0) {
return [];
}
/** @var list<string> $hiddenTicketNames */
$hiddenTicketNames = Ticket::where('split_link_id', $linkId)
->where('status', 'hidden')
->column('ticket_name');
$hiddenTicketNames = array_values(array_unique(array_filter(array_map(
static function ($name): string {
return trim((string) $name);
},
$hiddenTicketNames
))));
$query = Number::where('split_link_id', $linkId)
->where('status', 'normal')
->order('id', 'asc')
->field('id,number,number_type,number_type_custom,ratio_deferred,ticket_name');
if ($hiddenTicketNames !== []) {
$query->where('ticket_name', 'not in', $hiddenTicketNames);
}
$numbers = $query->select();
return $numbers instanceof Collection ? $numbers->all() : (array) $numbers;
}
/**
* random_shuffle=1：先按工单开启号码数加权随机选工单，再在该工单内选号
*
* @param list<Number|array<string, mixed>> $rows
* @return array<string, mixed>|Number|null
*/
private function pickNumberRowRandomByTicketWeight(int $linkId, array $rows)
{
$groups = $this->groupNumbersByTicket($rows);
if ($groups === []) {
return null;
}
$ticketKey = $this->pickTicketGroupKeyWeighted($groups);
if ($ticketKey === null || !isset($groups[$ticketKey])) {
return null;
}
return $this->pickNumberRow($linkId, $groups[$ticketKey]);
}
/**
* @param list<Number|array<string, mixed>> $rows
* @return array<string, list<Number|array<string, mixed>>>
*/
private function groupNumbersByTicket(array $rows): array
{
/** @var array<string, list<Number|array<string, mixed>>> $groups */
$groups = [];
foreach ($rows as $row) {
$name = is_array($row)
? trim((string) ($row['ticket_name'] ?? ''))
: trim((string) $row->getAttr('ticket_name'));
$key = $name !== '' ? $name : '__manual__';
$groups[$key][] = $row;
}
return $groups;
}
/**
* 按各分组号码数量加权随机选一个工单（分组 key）
*
* @param array<string, list<Number|array<string, mixed>>> $groups
*/
private function pickTicketGroupKeyWeighted(array $groups): ?string
{
$total = 0;
foreach ($groups as $rows) {
$total += count($rows);
}
if ($total <= 0) {
return null;
}
$keys = array_keys($groups);
if (count($keys) === 1) {
return $keys[0];
}
$r = random_int(1, $total);
$acc = 0;
foreach ($groups as $key => $rows) {
$acc += count($rows);
if ($r <= $acc) {
return $key;
}
}
return $keys[0];
}
/**
* 双层选号：优先 ratio_deferred=0，否则回退至降权池
*
* @param array<int, array<string, mixed>|Number> $rows
* @param list<Number|array<string, mixed>> $rows
* @return array<string, mixed>|Number|null
*/
private function pickNumberRow(int $linkId, array $rows)
patches/application/common/service/SplitScrmSpiderFactory.php
+27
View File
@@ -33,6 +33,17 @@ class SplitScrmSpiderFactory
// ceo_scrm 等未实现类型：新增 spider 类后在此注册
];
/**
* API 优先同步类型：常态走 PHP cURL，不占 Node Real Browser 槽位
*
* 调度层据此跳过 Node 队列门禁并允许并行投递；蜘蛛实现本身不在此修改。
*
* @var list<string>
*/
private const API_FIRST_SYNC_TYPES = [
'whatshub',
];
/**
* @return class-string<ScrmSpiderInterface>|null
*/
@@ -60,6 +71,22 @@ class SplitScrmSpiderFactory
return array_keys(self::MAP);
}
/**
* 是否 API 优先类型（调度层可并行投递且不占 Node 队列）
*/
public static function isApiFirstSyncType(string $ticketType): bool
{
return in_array(trim($ticketType), self::API_FIRST_SYNC_TYPES, true);
}
/**
* @return list<string>
*/
public static function listApiFirstSyncTypes(): array
{
return self::API_FIRST_SYNC_TYPES;
}
/**
* 同步是否依赖 Node Headless（纯 PHP cURL 等为 false）
*/
patches/application/common/service/SplitSyncConfigService.php
+1 -1
View File
@@ -59,7 +59,7 @@ class SplitSyncConfigService
if ($value === '') {
return 5;
}
return max(1, min(20, (int) $value));
return max(1, min(50, (int) $value));
}
/**
patches/application/common/service/SplitTicketRuleService.php
+24 -13
View File
@@ -12,6 +12,13 @@ use app\admin\model\split\Ticket;
*/
class SplitTicketRuleService
{
private SplitNumberRatioDeferredService $ratioDeferredService;
public function __construct(?SplitNumberRatioDeferredService $ratioDeferredService = null)
{
$this->ratioDeferredService = $ratioDeferredService ?? new SplitNumberRatioDeferredService();
}
/**
* 同步后应用工单开关规则；号码开关由 applyNumberRules 综合裁决（平台在线 + 单号上限 + 下号比率）
*/
@@ -108,14 +115,13 @@ class SplitTicketRuleService
foreach ($numbers as $number) {
$visitCount = (int) $number['visit_count'];
$inboundCount = (int) $number['inbound_count'];
$lastVisit = (int) ($number['last_sync_visit_count'] ?? 0);
$lastInbound = (int) ($number['last_sync_inbound_count'] ?? 0);
$streak = (int) ($number['no_inbound_click_streak'] ?? 0);
if ($visitCount > $lastVisit && $inboundCount <= $lastInbound) {
$streak += ($visitCount - $lastVisit);
} elseif ($inboundCount > $lastInbound) {
if ($visitCount <= 0) {
$streak = 0;
} else {
$streak = $this->ratioDeferredService->computeStreakForSync($number, $streak);
}
$update = [
@@ -125,20 +131,25 @@ class SplitTicketRuleService
'updatetime' => time(),
];
// 手动管理（含用户手动关闭）的号码：仅更新统计字段，不改状态与降权标记
$clearDeferred = [
'ratio_deferred' => 0,
'ratio_deferred_at' => null,
];
// 手动管理：更新统计；无效降权（无落地页访问）一并清除
if ((int) $number['manual_manage'] === 1) {
if ($this->ratioDeferredService->isInvalidDeferredRecord($number) || $visitCount <= 0) {
$update = array_merge($update, $clearDeferred);
if ($visitCount <= 0) {
$update['no_inbound_click_streak'] = 0;
}
}
Number::where('id', (int) $number['id'])->update($update);
continue;
}
// 同步裁决 ratio_deferred：进线增长则解除降权；触线且无进线增长则关号并清除降权
if ($inboundCount > $lastInbound) {
$update['ratio_deferred'] = 0;
$update['ratio_deferred_at'] = null;
} else {
$update['ratio_deferred'] = 0;
$update['ratio_deferred_at'] = null;
}
// 同步裁决 ratio_deferred：同步时清除访问侧降权，由 streak 决定是否关号
$update = array_merge($update, $clearDeferred);
$update['status'] = $this->resolveAutomatedStatus(
$ticket,
patches/application/common/service/SplitTicketSyncDispatchService.php
+307 -76
View File
@@ -4,15 +4,19 @@ declare(strict_types=1);
namespace app\common\service;
use app\common\library\scrm\AntiBotConfigBuilder;
use think\Db;
/**
* 工单手工同步 CLI 后台投递
* 工单同步 CLI 后台投递
*
* Web 请求仅负责鉴权与投递，实际 Spider 在独立 CLI 进程中执行，
* 避免长时间占用 PHP-FPM 与 Session 锁导致后台其它页面无法打开。
* Web / Cron 仅负责鉴权与投递，实际 Spider 在独立 CLI 进程中执行，
* 避免长时间占用 PHP-FPM、Cron 全局锁或 Session 锁。
*
* Node 类型（A2C / Whatshub 等）在同一 nohup 子 shell 内串行执行，避免并行 Chrome 压垮温池。
* 并行策略：
* - API 优先类型（Whatshub 等）：独立并行投递，不占 Node Real Browser 槽位
* - 直连 PHP 类型：独立并行投递
* - Node 类型：不同 sessionKey（域名）并行；同一 sessionKey 内串行以命中 Browser 温池
*/
class SplitTicketSyncDispatchService
{
@@ -24,75 +28,91 @@ class SplitTicketSyncDispatchService
*/
public function dispatchManual(array $ticketIds): array
{
$queued = [];
$lockService = new SplitTicketSyncLockService();
$metaList = $this->loadTicketMetaList($ticketIds);
/** @var list<array{id:int,ticket_type:string,ticket_url:string}> $toDispatch */
$toDispatch = [];
$skipped = [];
$failed = [];
$php = $this->resolvePhpBinary();
$think = $this->resolveThinkScript();
$root = $this->resolveRootPath();
$lockService = new SplitTicketSyncLockService();
$nodeTypeMap = array_flip(SplitScrmSpiderFactory::listNodeSyncTypes());
$typeMap = $this->loadTicketTypeMap($ticketIds);
/** @var int[] $nodeTicketIds */
$nodeTicketIds = [];
/** @var int[] $directTicketIds */
$directTicketIds = [];
foreach ($ticketIds as $rawId) {
$ticketId = (int) $rawId;
if ($ticketId <= 0) {
continue;
}
foreach ($metaList as $meta) {
$ticketId = (int) $meta['id'];
if ($lockService->isLocked($ticketId)) {
$skipped[] = $ticketId;
continue;
}
$ticketType = (string) ($typeMap[$ticketId] ?? '');
if ($ticketType !== '' && isset($nodeTypeMap[$ticketType])) {
$nodeTicketIds[] = $ticketId;
} else {
$directTicketIds[] = $ticketId;
}
$toDispatch[] = $meta;
}
foreach ($directTicketIds as $ticketId) {
if ($this->spawnCli($php, $think, $root, $ticketId)) {
$queued[] = $ticketId;
} else {
$failed[] = $ticketId;
}
}
if ($nodeTicketIds !== []) {
$spawned = $this->spawnCliSequential($php, $think, $root, $nodeTicketIds);
if ($spawned) {
foreach ($nodeTicketIds as $ticketId) {
$queued[] = $ticketId;
}
} else {
foreach ($nodeTicketIds as $ticketId) {
$failed[] = $ticketId;
}
}
}
$dispatchResult = $this->dispatchTicketMetaList($toDispatch, 'manual');
SplitTicketSyncLogger::log('web', 'manual sync dispatched', [
'queued' => $queued,
'skipped' => $skipped,
'failed' => $failed,
'nodeSerial' => $nodeTicketIds,
'directParallel' => $directTicketIds,
'phpCli' => $php,
'queued' => $dispatchResult['queued'],
'skipped' => $skipped,
'failed' => $dispatchResult['failed'],
'phpCli' => $this->resolvePhpBinary(),
]);
return [
'queued' => $queued,
'queued' => $dispatchResult['queued'],
'skipped' => $skipped,
'failed' => $failed,
'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'],
];
}
@@ -117,10 +137,181 @@ class SplitTicketSyncDispatchService
}
/**
* @param int[] $ticketIds
* @return array<int, string> ticketId => ticket_type
* @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 loadTicketTypeMap(array $ticketIds): array
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) {
@@ -135,42 +326,72 @@ class SplitTicketSyncDispatchService
$rows = Db::name('split_ticket')
->where('id', 'in', $ids)
->field('id,ticket_type')
->field('id,ticket_type,ticket_url')
->select();
$map = [];
$list = [];
foreach ($rows as $row) {
$map[(int) $row['id']] = (string) ($row['ticket_type'] ?? '');
$list[] = [
'id' => (int) $row['id'],
'ticket_type' => (string) ($row['ticket_type'] ?? ''),
'ticket_url' => (string) ($row['ticket_url'] ?? ''),
];
}
return $map;
return $list;
}
/**
* 后台启动 split:sync-tickets CLI（单工单，并行投递）
* @param array<string,mixed>|object $ticket
* @return array{id:int,ticket_type:string,ticket_url:string}|null
*/
private function spawnCli(string $php, string $think, string $root, int $ticketId): bool
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 >> %s 2>&1',
'%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, 'parallel');
return $this->runBackgroundCommand($root, $inner, $ticketId, $source . '-parallel');
}
/**
* Node 类型工单：单条 nohup 内串行执行多个 CLI，避免并行 Chrome
* 同一 sessionKey 内串行执行多个 CLI（单 nohup 子 shell）
*
* @param int[] $ticketIds
*/
private function spawnCliSequential(string $php, string $think, string $root, array $ticketIds): bool
private function spawnCliSequential(string $php, string $think, string $root, array $ticketIds, string $source): bool
{
if ($ticketIds === []) {
return false;
@@ -184,11 +405,13 @@ class SplitTicketSyncDispatchService
continue;
}
$logFile = $logDir . 'cli_' . $ticketId . '_' . date('YmdHis') . '.log';
$syncMode = $this->normalizeSyncModeForCli($source);
$parts[] = sprintf(
'%s %s split:sync-tickets --ticket=%d >> %s 2>&1',
'%s %s split:sync-tickets --ticket=%d --mode=%s >> %s 2>&1',
escapeshellarg($php),
escapeshellarg($think),
$ticketId,
$syncMode,
escapeshellarg($logFile)
);
}
@@ -198,7 +421,7 @@ class SplitTicketSyncDispatchService
$inner = implode('; ', $parts);
return $this->runBackgroundCommand($root, $inner, $ticketIds, 'node-serial');
return $this->runBackgroundCommand($root, $inner, $ticketIds, $source . '-node-serial');
}
/**
@@ -214,7 +437,7 @@ class SplitTicketSyncDispatchService
if ($this->canUseExec()) {
@exec($command, $output, $exitCode);
SplitTicketSyncLogger::log('web', 'cli spawn exec', [
SplitTicketSyncLogger::log('dispatch', 'cli spawn exec', [
'ticketRef' => $ticketRef,
'mode' => $mode,
'command' => $command,
@@ -225,7 +448,7 @@ class SplitTicketSyncDispatchService
if ($this->canUseShellExec()) {
@shell_exec($command);
SplitTicketSyncLogger::log('web', 'cli spawn shell_exec', [
SplitTicketSyncLogger::log('dispatch', 'cli spawn shell_exec', [
'ticketRef' => $ticketRef,
'mode' => $mode,
'command' => $command,
@@ -233,7 +456,7 @@ class SplitTicketSyncDispatchService
return true;
}
SplitTicketSyncLogger::log('web', 'cli spawn failed: exec disabled', [
SplitTicketSyncLogger::log('dispatch', 'cli spawn failed: exec disabled', [
'ticketRef' => $ticketRef,
'mode' => $mode,
]);
@@ -333,4 +556,12 @@ class SplitTicketSyncDispatchService
{
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';
}
}
patches/application/common/service/SplitTicketSyncService.php
+109 -116
View File
@@ -21,11 +21,14 @@ class SplitTicketSyncService
private SplitTicketSyncLockService $lockService;
private SplitTicketSyncDispatchService $dispatchService;
public function __construct()
{
$this->numberSync = new SplitTicketNumberSyncService();
$this->ruleService = new SplitTicketRuleService();
$this->lockService = new SplitTicketSyncLockService();
$this->dispatchService = new SplitTicketSyncDispatchService();
}
/**
@@ -126,7 +129,7 @@ class SplitTicketSyncService
);
$successCount = count(array_filter($processed, static function (array $row): bool {
return !empty($row['success']);
return !empty($row['dispatched']);
}));
$failedCount = count($processed) - $successCount;
@@ -151,7 +154,7 @@ class SplitTicketSyncService
}
/**
* 直连 PHP 通道：按周期同步，不占 Node 名额
* 直连 PHP 通道：挑选到期工单并投递后台 CLI，不占 Node 名额
*
* @param list<string> $directTypes
* @return array{
@@ -171,16 +174,10 @@ class SplitTicketSyncService
$list = $this->loadOpenTicketsForTypes($directTypes, $failThreshold, 0);
$candidateIds = $this->extractTicketIds($list);
SplitTicketSyncLogger::logCron('direct channel start', [
'types' => $directTypes,
'candidateCount' => count($list),
]);
/** @var list<array<string, mixed>> $processed */
$processed = [];
/** @var list<array<string, mixed>> $skipped */
/** @var list<array<string,mixed>> $dueList */
$dueList = [];
/** @var list<array<string,mixed>> $skipped */
$skipped = [];
$count = 0;
foreach ($list as $ticket) {
$ticketId = (int) $ticket['id'];
@@ -193,42 +190,60 @@ class SplitTicketSyncService
$this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, $skip, 'direct');
continue;
}
$dueList[] = $ticket;
}
$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;
}
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<array<string,mixed>> $processed */
$processed = [];
foreach ($dispatchResult['queued'] as $row) {
$processed[] = [
'ticketId' => $ticketId,
'ticketType' => $ticketType,
'ticketUrl' => $ticketUrl,
'ticketId' => (int) ($row['ticketId'] ?? 0),
'ticketType' => (string) ($row['ticketType'] ?? ''),
'ticketUrl' => (string) ($row['ticketUrl'] ?? ''),
'channel' => 'direct',
'success' => !empty($result['success']),
'message' => (string) ($result['message'] ?? ''),
'dispatched' => true,
'message' => '已投递后台同步',
];
$count++;
}
return [
'processedCount' => $count,
'processedCount' => count($processed),
'processed' => $processed,
'skipped' => $skipped,
'candidateIds' => $candidateIds,
'summary' => [
'channel' => 'direct',
'processedCount' => $count,
'processedCount' => count($processed),
'skippedCount' => count($skipped),
'dispatchMode' => 'async-cli',
],
'stoppedByNodeBusy' => false,
];
}
/**
* Node 通道：限流 + 类型公平轮转
* Node 通道：限流 + 类型公平轮转 + 异步 CLI 投递
*
* @param list<string> $nodeTypes
* @return array{
@@ -269,123 +284,101 @@ class SplitTicketSyncService
'dueCount' => count($duePool),
'pickedCount' => count($picked),
'nodeQueue' => SplitNodeHealthService::getQueueSnapshot(),
'dispatchMode' => 'async-cli',
]);
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<array<string, mixed>> $processed */
$processed = [];
/** @var list<array<string, mixed>> $skipped */
/** @var list<array<string,mixed>> $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)
);
if (isset($pickedIdMap[$ticketId])) {
continue;
}
if (!SplitNodeHealthService::canAcceptWork()) {
$stoppedByNodeBusy = true;
$skipped[] = $this->buildCronTicketEntry($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙，停止本轮后续同步');
$this->logCronCandidateSkipped($ticketId, $ticketType, $ticketUrl, 'Node 队列繁忙，停止本轮后续同步', 'node');
$skip = $this->shouldSkip($ticket);
if ($skip !== null) {
continue;
}
$skipped[] = $this->buildCronTicketEntry(
$ticketId,
(string) $ticket['ticket_type'],
(string) $ticket['ticket_url'],
sprintf('未进入本轮 Node 候选(前%d条到期工单公平轮转)', $candidateLimit)
);
}
$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;
}
$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<array<string,mixed>> $processed */
$processed = [];
foreach ($dispatchResult['queued'] as $row) {
$processed[] = [
'ticketId' => $ticketId,
'ticketType' => $ticketType,
'ticketUrl' => $ticketUrl,
'sessionKey' => $sessionKey,
'ticketId' => (int) ($row['ticketId'] ?? 0),
'ticketType' => (string) ($row['ticketType'] ?? ''),
'ticketUrl' => (string) ($row['ticketUrl'] ?? ''),
'sessionKey' => (string) ($row['sessionKey'] ?? ''),
'channel' => 'node',
'success' => !empty($result['success']),
'message' => (string) ($result['message'] ?? ''),
'dispatched' => true,
'message' => '已投递后台同步',
];
$count++;
}
return [
'processedCount' => $count,
'processedCount' => count($processed),
'processed' => $processed,
'skipped' => $skipped,
'candidateIds' => $candidateIds,
'summary' => [
'channel' => 'node',
'processedCount' => $count,
'processedCount' => count($processed),
'skippedCount' => count($skipped),
'maxPerRun' => $maxPerRun,
'dispatchMode' => 'async-cli',
],
'stoppedByNodeBusy' => $stoppedByNodeBusy,
'stoppedByNodeBusy' => $dispatchResult['stoppedByNodeBusy'],
];
}
/**
* @param list<array<string,mixed>> $rows
* @return list<array{ticketId:int,ticketType:string,ticketUrl:string,reason:string}>
*/
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<array<string,mixed>> $rows
* @return list<array{ticketId:int,ticketType:string,ticketUrl:string,reason:string}>
*/
private function mapDispatchFailedToCron(array $rows): array
{
return $this->mapDispatchSkippedToCron($rows);
}
/**
* @param list<string> $types
* @return list<Ticket>