57 lines
1.7 KiB
PHP
57 lines
1.7 KiB
PHP
<?php
|
||
|
||
namespace app\api\job;
|
||
|
||
use think\queue\Job;
|
||
use app\common\PromotionService;
|
||
use app\common\QueueJob;
|
||
|
||
/**
|
||
* 队列任务：task 级别的 prepare 调度器。
|
||
*
|
||
* 职责：遍历 task 下待准备的 promotion_email_log，
|
||
* 将每封邮件拆成一条 PromotionPrepareEmail 推到 promotion_email 队列，
|
||
* 以便并行调用 LLM 生成个性化描述。
|
||
*
|
||
* 队列名：PromotionPrepareTask
|
||
*/
|
||
class PromotionPrepareTask
|
||
{
|
||
private $oQueueJob;
|
||
|
||
public function __construct()
|
||
{
|
||
$this->oQueueJob = new QueueJob();
|
||
}
|
||
|
||
public function fire(Job $job, $data)
|
||
{
|
||
$this->oQueueJob->init($job);
|
||
|
||
$taskId = isset($data['task_id']) ? intval($data['task_id']) : 0;
|
||
if ($taskId <= 0) {
|
||
$this->oQueueJob->log("PromotionPrepareTask 无效的 task_id，删除任务");
|
||
$job->delete();
|
||
return;
|
||
}
|
||
|
||
try {
|
||
$service = new PromotionService();
|
||
$result = $service->dispatchPrepareEmails($taskId);
|
||
|
||
$dispatched = isset($result['dispatched']) ? $result['dispatched'] : 0;
|
||
$alreadyDone = isset($result['already_done']) ? $result['already_done'] : 0;
|
||
$err = isset($result['error']) ? $result['error'] : '';
|
||
$this->oQueueJob->log("PromotionPrepareTask 完成 | task_id={$taskId} dispatched={$dispatched} already_done={$alreadyDone} error={$err}");
|
||
|
||
$job->delete();
|
||
} catch (\Exception $e) {
|
||
$this->oQueueJob->handleException($e, $job, "task_id={$taskId}");
|
||
} catch (\Throwable $e) {
|
||
$this->oQueueJob->handleException($e, $job, "task_id={$taskId}");
|
||
} finally {
|
||
$this->oQueueJob->finnal();
|
||
}
|
||
}
|
||
}
|