添加whatshub工单

2026-06-20 04:47:34 +08:00
parent a6c87e8e25
commit d5a0ffa6db
271 changed files with 9377 additions and 303 deletions
@@ -0,0 +1,64 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\common\service;
+
+/**
+ * 定时同步全局互斥锁，防止每分钟 cron 重叠执行打满 Node 队列
+ */
+class SplitCronLockService
+{
+    private const LOCK_FILE = 'cron.lock';
+
+    /**
+     * 尝试获取全局 cron 锁
+     */
+    public function acquire(): bool
+    {
+        $path = $this->lockPath();
+        if ($this->isStaleLock($path)) {
+            @unlink($path);
+        }
+        if (is_file($path)) {
+            return false;
+        }
+        $payload = json_encode([
+            'pid'  => getmypid(),
+            'time' => time(),
+        ], JSON_UNESCAPED_UNICODE);
+        $written = @file_put_contents($path, $payload, LOCK_EX);
+        return $written !== false;
+    }
+
+    /**
+     * 释放全局 cron 锁
+     */
+    public function release(): void
+    {
+        $path = $this->lockPath();
+        if (is_file($path)) {
+            @unlink($path);
+        }
+    }
+
+    private function lockPath(): string
+    {
+        $runtime = defined('RUNTIME_PATH') ? RUNTIME_PATH : (dirname(__DIR__, 3) . '/runtime/');
+        $dir = $runtime . 'split_ticket_sync/';
+        if (!is_dir($dir)) {
+            @mkdir($dir, 0755, true);
+        }
+        return $dir . self::LOCK_FILE;
+    }
+
+    private function isStaleLock(string $path): bool
+    {
+        if (!is_file($path)) {
+            return false;
+        }
+        $mtime = (int) @filemtime($path);
+        $ttl = SplitSyncConfigService::getCronLockTtlSeconds();
+        return $mtime > 0 && (time() - $mtime) > $ttl;
+    }
+}
@@ -12,13 +12,13 @@ class SplitFriendUrlBuilder
    /**
     * 构建跳转 URL；无法构建时返回空字符串
     *
-     * @param string $whatsAppReplyText 仅 WhatsApp 类型使用，预填消息文案（urlencode 在内部处理）
+     * @param string $replyText WhatsApp / Telegram 预填消息文案（内部 rawurlencode）
     */
    public static function build(
        string $numberType,
        string $number,
        string $numberTypeCustom = '',
-        string $whatsAppReplyText = ''
+        string $replyText = ''
    ): string {
        $number = trim($number);
        if ($number === '') {
@@ -27,9 +27,9 @@ class SplitFriendUrlBuilder

        switch ($numberType) {
            case 'whatsapp':
-                return self::buildWhatsApp($number, $whatsAppReplyText);
+                return self::buildWhatsApp($number, $replyText);
            case 'telegram':
-                return self::buildTelegram($number);
+                return self::buildTelegram($number, $replyText);
            case 'line':
                return self::buildLine($number);
            case 'custom':
@@ -59,16 +59,22 @@ class SplitFriendUrlBuilder
    }

    /**
-     * Telegram：https://t.me/+ 号码（去掉前导 +）
+     * Telegram：https://t.me/+ 号码（去掉前导 +），可选 ?text= 预填消息（须 URL 编码）
     */
-    private static function buildTelegram(string $number): string
+    private static function buildTelegram(string $number, string $replyText = ''): string
    {
        $id = ltrim($number, '+');
        if ($id === '') {
            return '';
        }

-        return 'https://t.me/+' . $id;
+        $url = 'https://t.me/+' . $id;
+        $replyText = trim($replyText);
+        if ($replyText !== '') {
+            $url .= '?text=' . rawurlencode($replyText);
+        }
+
+        return $url;
    }

    /**
@@ -0,0 +1,96 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\common\service;
+
+/**
+ * Node Puppeteer 服务健康与队列负载探测
+ */
+class SplitNodeHealthService
+{
+    /**
+     * 拉取 /api/stats，失败返回 null（不阻断同步，由后续请求快速失败）
+     *
+     * @return array<string, mixed>|null
+     */
+    public static function fetchStats(): ?array
+    {
+        $url = SplitSyncConfigService::getNodeHost() . '/api/stats';
+        $ch = curl_init($url);
+        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
+        curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, 2);
+        curl_setopt($ch, CURLOPT_TIMEOUT, 3);
+        $response = curl_exec($ch);
+        if (curl_errno($ch)) {
+            curl_close($ch);
+            return null;
+        }
+        $httpCode = (int) curl_getinfo($ch, CURLINFO_HTTP_CODE);
+        curl_close($ch);
+        if ($httpCode !== 200) {
+            return null;
+        }
+        $decoded = json_decode((string) $response, true);
+        return is_array($decoded) ? $decoded : null;
+    }
+
+    /**
+     * 当前是否适合向 Node 提交新 Browser 任务
+     */
+    public static function canAcceptWork(): bool
+    {
+        $stats = self::fetchStats();
+        if ($stats === null) {
+            return true;
+        }
+        if (!empty($stats['shuttingDown'])) {
+            return false;
+        }
+        $queue = is_array($stats['queue'] ?? null) ? $stats['queue'] : [];
+        $config = is_array($stats['config'] ?? null) ? $stats['config'] : [];
+        $queued = (int) ($queue['queued'] ?? 0);
+        $active = (int) ($queue['active'] ?? 0);
+        $max = max(1, (int) ($config['maxConcurrentBrowsers'] ?? ($queue['max'] ?? 4)));
+        $threshold = SplitSyncConfigService::getNodeQueueSkipThreshold();
+        if ($queued >= $threshold) {
+            return false;
+        }
+        return $active < $max;
+    }
+
+    /**
+     * @return array{queued:int,active:int,max:int}
+     */
+    public static function getQueueSnapshot(): array
+    {
+        $stats = self::fetchStats();
+        if ($stats === null) {
+            return ['queued' => 0, 'active' => 0, 'max' => 0];
+        }
+        $queue = is_array($stats['queue'] ?? null) ? $stats['queue'] : [];
+        $config = is_array($stats['config'] ?? null) ? $stats['config'] : [];
+        return [
+            'queued' => (int) ($queue['queued'] ?? 0),
+            'active' => (int) ($queue['active'] ?? 0),
+            'max'    => max(1, (int) ($config['maxConcurrentBrowsers'] ?? ($queue['max'] ?? 4))),
+        ];
+    }
+
+    /**
+     * Node 侧 Real Browser / Captcha 是否就绪（供同步失败日志可读性）
+     *
+     * @return array{realBrowserReady:bool,captchaConfigured:bool}
+     */
+    public static function getAntiBotSnapshot(): array
+    {
+        $stats = self::fetchStats();
+        if ($stats === null) {
+            return ['realBrowserReady' => false, 'captchaConfigured' => false];
+        }
+        return [
+            'realBrowserReady'  => !empty($stats['realBrowserReady']),
+            'captchaConfigured' => !empty($stats['captchaConfigured']),
+        ];
+    }
+}
@@ -72,16 +72,16 @@ class SplitRedirectService
            ? (string) ($picked['number_type_custom'] ?? '')
            : (string) $picked->getAttr('number_type_custom');

-        $whatsAppReplyText = '';
-        if ($numberType === 'whatsapp') {
-            $whatsAppReplyText = SplitAutoReplyService::pickRandomLine((string) $link->getAttr('auto_reply'));
+        $replyText = '';
+        if (in_array($numberType, ['whatsapp', 'telegram'], true)) {
+            $replyText = SplitAutoReplyService::pickRandomLine((string) $link->getAttr('auto_reply'));
        }

        $redirectUrl = SplitFriendUrlBuilder::build(
            $numberType,
            $numberValue,
            $numberCustom,
-            $whatsAppReplyText
+            $replyText
        );
        if ($redirectUrl === '') {
            return null;
@@ -6,9 +6,11 @@ namespace app\common\service;

 use app\common\library\scrm\ScrmSpiderInterface;
 use app\common\library\scrm\spider\A2cSpider;
+use app\common\library\scrm\spider\ChatknowSpider;
 use app\common\library\scrm\spider\HaiwangSpider;
 use app\common\library\scrm\spider\HuojianSpider;
 use app\common\library\scrm\spider\SsCustomerSpider;
+use app\common\library\scrm\spider\WhatshubSpider;
 use app\common\library\scrm\spider\XingheSpider;

 /**
@@ -25,6 +27,8 @@ class SplitScrmSpiderFactory
        'huojian'     => HuojianSpider::class,
        'xinghe'      => XingheSpider::class,
        'ss_customer' => SsCustomerSpider::class,
+        'chatknow'    => ChatknowSpider::class,
+        'whatshub'    => WhatshubSpider::class,
        // ceo_scrm 等未实现类型：新增 spider 类后在此注册
    ];

@@ -45,6 +49,16 @@ class SplitScrmSpiderFactory
        return self::resolveClass($ticketType) !== null;
    }

+    /**
+     * 已注册且已实现蜘蛛的类型列表
+     *
+     * @return list<string>
+     */
+    public static function listSupportedTypes(): array
+    {
+        return array_keys(self::MAP);
+    }
+
    /**
     * @return ScrmSpiderInterface|null
     */
@@ -50,6 +50,42 @@ class SplitSyncConfigService
        return max(0, (int) $value);
    }

+    /**
+     * 单次 cron 最多处理几条到期工单，避免一分钟内打满 Node Browser 槽位
+     */
+    public static function getMaxTicketsPerCronRun(): int
+    {
+        $value = self::getConfigValue('split_sync_max_per_cron');
+        if ($value === '') {
+            return 2;
+        }
+        return max(1, min(20, (int) $value));
+    }
+
+    /**
+     * 全局 cron 锁过期秒数，防止异常退出后永久占锁
+     */
+    public static function getCronLockTtlSeconds(): int
+    {
+        $value = self::getConfigValue('split_sync_cron_lock_ttl');
+        if ($value === '') {
+            return 1800;
+        }
+        return max(300, (int) $value);
+    }
+
+    /**
+     * Node 排队数达到该阈值时，本轮 cron 不再提交新任务
+     */
+    public static function getNodeQueueSkipThreshold(): int
+    {
+        $value = self::getConfigValue('split_sync_node_queue_threshold');
+        if ($value === '') {
+            return 2;
+        }
+        return max(0, (int) $value);
+    }
+
    private static function getConfigValue(string $name): string
    {
        $site = Config::get('site.' . $name);
@@ -30,9 +30,10 @@ class SplitTicketSyncService
    /**
     * 同步单条工单
     *
+     * @param bool $dueValidated 为 true 时跳过 shouldSkip（由 syncDueTickets 预筛后传入）
     * @return array{success:bool,message:string,skipped?:bool}
     */
-    public function syncOne(int $ticketId, bool $force = false): array
+    public function syncOne(int $ticketId, bool $force = false, bool $dueValidated = false): array
    {
        $ticket = Ticket::get($ticketId);
        if (!$ticket) {
@@ -50,7 +51,7 @@ class SplitTicketSyncService
            'nodeHost'      => SplitSyncConfigService::getNodeHost(),
        ]);

-        if (!$force) {
+        if (!$force && !$dueValidated) {
            $skip = $this->shouldSkip($ticket);
            if ($skip !== null) {
                SplitTicketSyncLogger::log('sync', 'skipped', ['reason' => $skip]);
@@ -76,23 +77,47 @@ class SplitTicketSyncService
    }

    /**
-     * 扫描到期工单并同步
+     * 扫描到期工单并同步（限流 + Node 队列感知）
     */
    public function syncDueTickets(): int
    {
-        $count = 0;
+        $maxPerRun = SplitSyncConfigService::getMaxTicketsPerCronRun();
+        $autoSyncTypes = $this->resolveAutoSyncTicketTypes();
+        if ($autoSyncTypes === []) {
+            SplitTicketSyncLogger::log('cron', 'scan skip', ['reason' => 'no auto-sync ticket types']);
+            return 0;
+        }
+
+        if (!SplitNodeHealthService::canAcceptWork()) {
+            SplitTicketSyncLogger::log('cron', 'scan skip', [
+                'reason' => 'node queue busy',
+                'queue'  => SplitNodeHealthService::getQueueSnapshot(),
+            ]);
+            return 0;
+        }
+
        $failThreshold = SplitSyncConfigService::getFailPauseThreshold();
-        $query = Ticket::where('status', 'normal');
+        $query = Ticket::where('status', 'normal')->whereIn('ticket_type', $autoSyncTypes);
        if ($failThreshold > 0) {
            $query->where('sync_fail_count', '<', $failThreshold);
        }
-        $list = $query->select();
+        // 多取候选：间隔过滤在 PHP 完成，优先同步最久未更新的工单
+        $list = $query->order('sync_time', 'asc')->order('id', 'asc')
+            ->limit($maxPerRun * 5)
+            ->select();

        SplitTicketSyncLogger::log('cron', 'scan start', [
            'candidateCount' => count($list),
+            'maxPerRun'      => $maxPerRun,
+            'autoSyncTypes'  => $autoSyncTypes,
+            'nodeQueue'      => SplitNodeHealthService::getQueueSnapshot(),
        ]);

+        $count = 0;
        foreach ($list as $ticket) {
+            if ($count >= $maxPerRun) {
+                break;
+            }
            $skip = $this->shouldSkip($ticket);
            if ($skip !== null) {
                SplitTicketSyncLogger::log('cron', 'candidate skipped', [
@@ -102,7 +127,14 @@ class SplitTicketSyncService
                ]);
                continue;
            }
-            $result = $this->syncOne((int) $ticket['id'], false);
+            if (!SplitNodeHealthService::canAcceptWork()) {
+                SplitTicketSyncLogger::log('cron', 'node busy, stop batch', [
+                    'processed' => $count,
+                    'queue'     => SplitNodeHealthService::getQueueSnapshot(),
+                ]);
+                break;
+            }
+            $result = $this->syncOne((int) $ticket['id'], false, true);
            if (!empty($result['skipped'])) {
                continue;
            }
@@ -113,6 +145,22 @@ class SplitTicketSyncService
        return $count;
    }

+    /**
+     * 已配置自动同步周期且已实现蜘蛛的工单类型
+     *
+     * @return list<string>
+     */
+    private function resolveAutoSyncTicketTypes(): array
+    {
+        $types = [];
+        foreach (SplitScrmSpiderFactory::listSupportedTypes() as $ticketType) {
+            if (SplitSyncConfigService::getIntervalMinutes($ticketType) > 0) {
+                $types[] = $ticketType;
+            }
+        }
+        return $types;
+    }
+
    /**
     * @return array{success:bool,message:string}
     */