55 lines
1.9 KiB
PHP
55 lines
1.9 KiB
PHP
<?php
|
||
|
||
namespace app\api\job;
|
||
|
||
use think\queue\Job;
|
||
use app\common\ExpertFinderService;
|
||
use app\common\QueueJob;
|
||
|
||
/**
|
||
* 队列任务：用本地大模型从 affiliation 推断国家，写入 expert.country_id / country。
|
||
* 处理完当前专家后，自动找下一个推入同一队列（链式执行），直到全部处理完。
|
||
*
|
||
* 支持多队列并行：通过 $data['queue'] 和 $data['chat_url'] 区分不同的链/模型。
|
||
*
|
||
* 单条任务受本地 LLM 响应时间影响（一般 2-10s），常驻 worker 由 QueueJob
|
||
* 在进程超 6h 或遇致命 DB 错误时主动 exit(1) 让 supervisor 拉起新进程。
|
||
*/
|
||
class FillExpertCountry
|
||
{
|
||
private $oQueueJob;
|
||
|
||
public function __construct()
|
||
{
|
||
$this->oQueueJob = new QueueJob();
|
||
}
|
||
|
||
public function fire(Job $job, $data)
|
||
{
|
||
$this->oQueueJob->init($job);
|
||
|
||
$expertId = intval(isset($data['expert_id']) ? $data['expert_id'] : 0);
|
||
$affiliation = isset($data['affiliation']) ? trim((string)$data['affiliation']) : '';
|
||
$queue = isset($data['queue']) ? (string)$data['queue'] : 'FetchExperts';
|
||
$chatUrl = isset($data['chat_url']) ? (string)$data['chat_url'] : '';
|
||
|
||
$service = new ExpertFinderService();
|
||
|
||
try {
|
||
if ($expertId && $affiliation !== '') {
|
||
$service->fillExpertCountry($expertId, $affiliation, $chatUrl);
|
||
}
|
||
$job->delete();
|
||
|
||
// 链式：处理完当前专家立刻拉下一个进来
|
||
$service->enqueueNextCountryFill(1, $queue, $chatUrl);
|
||
} catch (\Exception $e) {
|
||
$this->oQueueJob->handleException($e, $job, "expert_id={$expertId} queue={$queue}");
|
||
} catch (\Throwable $e) {
|
||
$this->oQueueJob->handleException($e, $job, "expert_id={$expertId} queue={$queue}");
|
||
} finally {
|
||
$this->oQueueJob->finnal();
|
||
}
|
||
}
|
||
}
|