78 lines
2.6 KiB
PHP
78 lines
2.6 KiB
PHP
<?php
|
|
|
|
namespace app\common\command;
|
|
|
|
use app\common\model\ai\AiKbSyncJob;
|
|
use app\common\service\ai\KbService;
|
|
use app\common\service\ai\KbSyncService;
|
|
use think\console\Command;
|
|
use think\console\Input;
|
|
use think\console\Output;
|
|
use think\console\input\Option;
|
|
|
|
class KbConsume extends Command
|
|
{
|
|
protected function configure()
|
|
{
|
|
$this->setName('kb:consume')
|
|
->setDescription('消费AI知识库同步任务')
|
|
->addOption('batch', null, Option::VALUE_OPTIONAL, '批量处理数量', 100);
|
|
}
|
|
|
|
protected function execute(Input $input, Output $output)
|
|
{
|
|
$batch = max(1, (int) $input->getOption('batch'));
|
|
KbSyncService::markMatchBackfillJobs($batch);
|
|
KbSyncService::markArticleBackfillJobs($batch);
|
|
KbSyncService::markLotteryBackfillJobs($batch);
|
|
|
|
$jobs = AiKbSyncJob::where('status', AiKbSyncJob::STATUS_PENDING)
|
|
->where('scheduled_at', '<=', time())
|
|
->order('priority', 'asc')
|
|
->order('id', 'asc')
|
|
->limit($batch)
|
|
->select();
|
|
|
|
$success = 0;
|
|
$failed = 0;
|
|
|
|
foreach ($jobs as $job) {
|
|
$job->status = AiKbSyncJob::STATUS_PROCESSING;
|
|
$job->update_time = time();
|
|
$job->save();
|
|
|
|
try {
|
|
if ($job->action === AiKbSyncJob::ACTION_DELETE) {
|
|
KbService::deleteDocument($job->domain, $job->subtype, (int) $job->source_id);
|
|
$result = ['success' => true];
|
|
} else {
|
|
$result = KbService::upsertDocument($job->domain, $job->subtype, (int) $job->source_id);
|
|
}
|
|
|
|
if (!empty($result['success'])) {
|
|
$job->status = AiKbSyncJob::STATUS_SUCCESS;
|
|
$job->processed_at = time();
|
|
$job->error_msg = '';
|
|
$success++;
|
|
} else {
|
|
$job->status = AiKbSyncJob::STATUS_FAILED;
|
|
$job->retry_count = (int) $job->retry_count + 1;
|
|
$job->error_msg = $result['error'] ?? 'unknown';
|
|
$failed++;
|
|
}
|
|
} catch (\Throwable $e) {
|
|
$job->status = AiKbSyncJob::STATUS_FAILED;
|
|
$job->retry_count = (int) $job->retry_count + 1;
|
|
$job->error_msg = $e->getMessage();
|
|
$failed++;
|
|
}
|
|
|
|
$job->update_time = time();
|
|
$job->save();
|
|
}
|
|
|
|
$output->writeln(sprintf('[kb:consume] success=%d failed=%d jobs=%d', $success, $failed, count($jobs)));
|
|
return 0;
|
|
}
|
|
}
|