1deebf3f5c
- 新增 MediaArchiveJob、MediaOssArchiveService、WechatMediaArchiveService 与回填命令 - Message/DataProcessing 等支持归档调度与持久化下载 URL - WebSocket 控制器整理；朋友圈与文档/定时任务说明更新 Made-with: Cursor
122 lines
4.5 KiB
PHP
122 lines
4.5 KiB
PHP
<?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;
|
||
}
|
||
}
|