feat(media): 微信消息媒体 OSS 归档与下载地址回写

- 新增 MediaArchiveJob、MediaOssArchiveService、WechatMediaArchiveService 与回填命令 - Message/DataProcessing 等支持归档调度与持久化下载 URL - WebSocket 控制器整理；朋友圈与文档/定时任务说明更新 Made-with: Cursor
2026-04-13 17:23:02 +08:00
parent 3bf1b2aee9
commit b1bfcec35b
12 changed files with 970 additions and 134 deletions
--- a/application/command/BackfillMediaOssCommand.php
+++ b/application/command/BackfillMediaOssCommand.php
@@ -0,0 +1,121 @@
+<?php
+
+namespace app\command;
+
+use app\common\service\WechatMediaArchiveService;
+use app\job\MediaArchiveJob;
+use think\console\Command;
+use think\console\Input;
+use think\console\Output;
+use think\console\input\Option;
+use think\Db;
+
+class BackfillMediaOssCommand extends Command
+{
+    protected function configure()
+    {
+        $this->setName('media:archive')
+            ->setDescription('扫描聊天记录和朋友圈媒体资源并加入 OSS 归档队列')
+            ->addOption('scope', null, Option::VALUE_OPTIONAL, '归档范围：chat|moments|all', 'all')
+            ->addOption('limit', null, Option::VALUE_OPTIONAL, '单次扫描数量', 100)
+            ->addOption('startId', null, Option::VALUE_OPTIONAL, '仅扫描大于该ID的记录', 0)
+            ->addOption('messageId', null, Option::VALUE_OPTIONAL, '仅同步归档单条聊天记录（本地调试用，不走队列）', 0);
+    }
+
+    protected function execute(Input $input, Output $output)
+    {
+        $singleMessageId = (int)$input->getOption('messageId');
+        if ($singleMessageId > 0) {
+            return $this->archiveOneChatMessage($singleMessageId, $output);
+        }
+
+        $scope = strtolower((string)$input->getOption('scope'));
+        $limit = max(1, (int)$input->getOption('limit'));
+        $startId = max(0, (int)$input->getOption('startId'));
+
+        $messageCount = 0;
+        $momentCount = 0;
+
+        if (in_array($scope, ['all', 'chat'])) {
+            $messageIds = Db::table('s2_wechat_message')
+                ->where('id', '>', $startId)
+                ->whereIn('msgType', [3, 34, 43, 47, 49])
+                ->where(function ($query) {
+                    $query->whereLike('content', 'http%')
+                        ->whereOrLike('content', '%"url"%')
+                        ->whereOrLike('content', '%"previewImage"%')
+                        ->whereOrLike('content', '%"tencentUrl"%');
+                })
+                ->order('id', 'asc')
+                ->limit($limit)
+                ->column('id');
+
+            foreach ($messageIds as $id) {
+                MediaArchiveJob::dispatch('message', $id, ['source' => 'backfill_command']);
+                $messageCount++;
+            }
+        }
+
+        if (in_array($scope, ['all', 'moments'])) {
+            $momentIds = Db::table('s2_wechat_moments')
+                ->where('id', '>', $startId)
+                ->where(function ($query) {
+                    $query->where('isOssUploaded', 0)
+                        ->whereOr('ossUrls', 'null')
+                        ->whereOr('ossUrls', '')
+                        ->whereOr('ossUrls', '[]');
+                })
+                ->where(function ($query) {
+                    $query->where('resUrls', '<>', '')
+                        ->whereOr('urls', '<>', '');
+                })
+                ->order('id', 'asc')
+                ->limit($limit)
+                ->column('id');
+
+            foreach ($momentIds as $id) {
+                MediaArchiveJob::dispatch('moment', $id, ['source' => 'backfill_command']);
+                $momentCount++;
+            }
+        }
+
+        $output->writeln(sprintf(
+            '已加入 OSS 归档队列：聊天记录 %d 条，朋友圈 %d 条。',
+            $messageCount,
+            $momentCount
+        ));
+
+        return 0;
+    }
+
+    /**
+     * 单条聊天消息同步归档（用于本地验证 AliyunOSS 与下载链路）
+     */
+    protected function archiveOneChatMessage($messageId, Output $output)
+    {
+        $row = Db::table('s2_wechat_message')->where('id', $messageId)->find();
+        if (empty($row)) {
+            $output->writeln('错误：未找到消息 id=' . $messageId);
+            return 1;
+        }
+
+        $preview = function ($s, $len) {
+            $s = (string)$s;
+            return function_exists('mb_substr') ? mb_substr($s, 0, $len) : substr($s, 0, $len);
+        };
+
+        $output->writeln('msgType=' . ($row['msgType'] ?? '') . ' content(前200字): ' . $preview($row['content'] ?? '', 200));
+
+        $ok = WechatMediaArchiveService::archiveMessageById($messageId);
+        if (!$ok) {
+            $output->writeln('失败：归档未执行或无需更新（请确认 msgType 为 3/34/43/47/49 且 content 含可下载 http 资源）');
+            return 1;
+        }
+
+        $after = Db::table('s2_wechat_message')->where('id', $messageId)->find();
+        $output->writeln('成功：归档完成。更新后 content(前300字):');
+        $output->writeln($preview($after['content'] ?? '', 300));
+
+        return 0;
+    }
+}
--- a/application/command/SyncWechatDataToCkbTask.php
+++ b/application/command/SyncWechatDataToCkbTask.php
@@ -9,7 +9,7 @@ use think\console\Command;
 use think\facade\App;
 use WeChatDeviceApi\Adapters\ChuKeBao\Adapter as ChuKeBaoAdapter;

-// */7 * * * * cd /www/wwwroot/mckb_quwanzhi_com/Server && php think sync:wechatData >> /www/wwwroot/mckb_quwanzhi_com/Server/runtime/log/sync_wechat_data.log 2>&1
+// */7 * * * * cd /www/wwwroot/ckbapi.quwanzhi.com/Server && php think sync:wechatData >> /www/wwwroot/ckbapi.quwanzhi.com/Server/runtime/log/sync_wechat_data.log 2>&1
 class SyncWechatDataToCkbTask extends Command
 {
    protected $lockFile;