【触客宝】新增同步所有好友方案

2025-05-28 15:43:38 +08:00
parent 8d57dca79e
commit ffbc6d5e77
2 changed files with 137 additions and 0 deletions
--- a/Server/application/command/SyncAllFriendsCommand.php
+++ b/Server/application/command/SyncAllFriendsCommand.php
@@ -0,0 +1,67 @@
+<?php
+namespace app\command;
+
+use think\console\Command;
+use think\console\Input;
+use think\console\Output;
+use think\console\input\Option;
+use think\facade\Log;
+use think\Queue;
+use app\job\SyncAllFriendsJob;
+use think\facade\Cache;
+use think\Db;
+
+class SyncAllFriendsCommand extends Command
+{
+    protected $queueName = 'sync_all_friends';
+
+    protected function configure()
+    {
+        $this->setName('sync:allFriends')
+            ->setDescription('同步所有好友（自动分页队列）')
+            ->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID', date('YmdHis') . rand(1000, 9999));
+    }
+
+    protected function execute(Input $input, Output $output)
+    {
+        $output->writeln('开始同步所有好友...');
+        try {
+            $jobId = $input->getOption('jobId');
+            $queueLockKey = "queue_lock:{$this->queueName}";
+            Cache::rm($queueLockKey);
+            if (Cache::get($queueLockKey)) {
+                $output->writeln("队列 {$this->queueName} 已经在运行中，跳过执行");
+                Log::warning("队列 {$this->queueName} 已经在运行中，跳过执行");
+                return false;
+            }
+            Cache::set($queueLockKey, $jobId, 3600);
+            $output->writeln("已设置队列运行锁，键名:{$queueLockKey}，值:{$jobId}，有效期:1小时");
+
+            $pageSize = 1000;
+            $accounts = Db::table('s2_wechat_account')->where('wechatAlive', 1)->select();
+            foreach ($accounts as $account) {
+                $this->addToQueue($account['wechatId'], 0, $pageSize, '', $jobId, $queueLockKey);
+            }
+
+            $output->writeln('同步所有好友任务已添加到队列');
+        } catch (\Exception $e) {
+            Log::error('同步所有好友任务添加失败：' . $e->getMessage());
+            $output->writeln('同步所有好友任务添加失败：' . $e->getMessage());
+            return false;
+        }
+        return true;
+    }
+
+    public function addToQueue($wechatId, $pageIndex, $pageSize, $preFriendId, $jobId, $queueLockKey)
+    {
+        $data = [
+            'wechatId' => $wechatId,
+            'pageIndex' => $pageIndex,
+            'pageSize' => $pageSize,
+            'preFriendId' => $preFriendId,
+            'jobId' => $jobId,
+            'queueLockKey' => $queueLockKey
+        ];
+        Queue::push(SyncAllFriendsJob::class, $data, $this->queueName);
+    }
+} 
--- a/Server/application/job/SyncAllFriendsJob.php
+++ b/Server/application/job/SyncAllFriendsJob.php
@@ -0,0 +1,70 @@
+<?php
+namespace app\job;
+
+use think\queue\Job;
+use think\facade\Log;
+use think\facade\Cache;
+use think\facade\Config;
+use think\Db;
+
+class SyncAllFriendsJob
+{
+    public function fire(Job $job, $data)
+    {
+        try {
+            $wechatId = $data['wechatId'];
+            $pageIndex = $data['pageIndex'];
+            $pageSize = $data['pageSize'];
+            $preFriendId = isset($data['preFriendId']) ? $data['preFriendId'] : '';
+            $queueLockKey = $data['queueLockKey'];
+            $jobId = $data['jobId'];
+
+            $controller = new \app\api\controller\WechatFriendController();
+            Log::info('开始同步微信id: ' . $wechatId . '，第' . $pageIndex . '页，preFriendId: ' . $preFriendId);
+            $result = $controller->getlist([
+                'wechatAccountKeyword' => $wechatId,
+                'pageIndex' => $pageIndex,
+                'pageSize' => $pageSize,
+                'preFriendId' => $preFriendId
+            ], true);
+
+            if (is_string($result)) {
+                $result = json_decode($result, true);
+            }
+
+            if ($result['code'] == 200) {
+                $friends = $result['data']['list'] ?? $result['data'];
+                if (is_array($friends) && count($friends) == $pageSize) {
+                    $lastFriendId = $friends[count($friends) - 1]['id'];
+                    // 还有下一页，重新入队
+                    $nextPageIndex = $pageIndex + 1;
+                    \think\Queue::push(self::class, [
+                        'wechatId' => $wechatId,
+                        'pageIndex' => $nextPageIndex,
+                        'pageSize' => $pageSize,
+                        'preFriendId' => $lastFriendId,
+                        'jobId' => $jobId,
+                        'queueLockKey' => $queueLockKey
+                    ], $job->getQueue());
+                    Log::info("微信id: {$wechatId} 下一页任务已入队，pageIndex: {$nextPageIndex}，preFriendId: {$lastFriendId}");
+                }
+            }
+
+            $job->delete();
+            Log::info('同步微信id: ' . $wechatId . ' 第' . $pageIndex . '页任务执行成功');
+            // 释放锁逻辑可在所有账号所有分页都完成后处理
+            return true;
+        } catch (\Exception $e) {
+            Log::error('同步所有好友任务异常：' . $e->getMessage());
+            if (!empty($data['queueLockKey'])) {
+                \think\facade\Cache::rm($data['queueLockKey']);
+            }
+            if ($job->attempts() > 3) {
+                $job->delete();
+            } else {
+                $job->release(\think\facade\Config::get('queue.failed_delay', 10));
+            }
+            return false;
+        }
+    }
+}