zm/tougao
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改自动推广的相关任务

Browse Source
This commit is contained in:
wangjinlei
2026-05-08 15:38:15 +08:00
parent b1e978ed73
commit 336fa08a18
9 changed files with 116 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
application/api/controller/EmailClient.php
View File

@@ -1761,20 +1761,9 @@ class EmailClient extends Base
$service = new PromotionService();
$taskId = intval($data['id']);
// 调用前快照：用于解释"为什么没入队"
$task = \think\Db::name('promotion_task')->where('task_id', $taskId)->find();
$pending = \think\Db::name('promotion_email_log')
->where('task_id', $taskId)
->where('state', 0)
->where('prepared_at', 0)
->count();
$result = $service->dispatchPrepareEmails($taskId);
return jsonSuccess([
'task_id' => $taskId,
'task_state' => $task ? intval($task['state']) : null, // 0 才能 dispatch；5 已准备完
'pending_before' => intval($pending), // 调用前还能入队的 log 数
'dispatch_result' => $result, // ['dispatched' => N, ...]
]);
}
@@ -1793,6 +1782,72 @@ class EmailClient extends Base
return jsonSuccess($result);
}
public function mytestqqq(){
$data = $this->request->post();
$rule = new Validate([
"id"=>"require"
]);
if(!$rule->check($data)){
return jsonError($rule->getError());
}
$service = new PromotionService();
$service->enqueuePrepareEmail(intval($data['id']));
return jsonSuccess();
}
/**
* 队列写入测试：推一条 myTestQueue，并返回 push 返回值 + 与 queueDebug 相同 Redis 下的 LLEN（便于对照 redis-cli）。
*/
public function mytestQueue()
{
$queueName = 'myTestQueue';
$jobClass = 'app\\api\\job\\myQueue@fire';
$data = ['testData' => 1];
$cfg = Config::get('queue');
$connector = isset($cfg['connector']) ? (string)$cfg['connector'] : '';
if (stripos($connector, 'sync') !== false) {
return jsonError('queue connector is Sync — will not write Redis. Check application/extra/queue.php');
}
if (!extension_loaded('redis')) {
return jsonError('PHP redis extension not loaded — Queue Redis connector cannot run.');
}
try {
$jobId = Queue::push($jobClass, $data, $queueName);
} catch (\Throwable $e) {
return jsonError('Queue::push failed: ' . $e->getMessage());
}
// 独立连一次 Redis，确认与 PHP-FPM 使用的配置一致且 LIST 长度（避免「以为写了其实没有」）
$verify = ['ready_len' => null, 'error' => null];
try {
$redis = new \Redis();
$connectMethod = !empty($cfg['persistent']) ? 'pconnect' : 'connect';
$redis->$connectMethod($cfg['host'] ?? '127.0.0.1', $cfg['port'] ?? 6379);
if (!empty($cfg['password'])) {
$redis->auth($cfg['password']);
}
$redis->select($cfg['select'] ?? 0);
$readyKey = 'queues:' . $queueName;
$verify['ready_len'] = (int)$redis->lLen($readyKey);
$verify['exists'] = (bool)$redis->exists($readyKey);
$verify['ping'] = (string)$redis->ping();
$verify['checked_db'] = (int)($cfg['select'] ?? 0);
$verify['checked_host'] = (string)($cfg['host'] ?? '127.0.0.1');
$verify['checked_port'] = (int)($cfg['port'] ?? 6379);
} catch (\Throwable $e) {
$verify['error'] = $e->getMessage();
}
return jsonSuccess([
'job_id' => $jobId,
'queue' => $queueName,
'connector' => $connector,
'after_push' => $verify,
]);
}
/**
* 队列调试：查看 Redis 里队列长度（不依赖 redis-cli）。
*

18
application/api/job/FetchExperts.php
View File

@@ -4,7 +4,6 @@ namespace app\api\job;
use think\queue\Job;
use app\common\ExpertFinderService;
use app\common\QueueJob;
/**
* 专家抓取队列任务。
@@ -16,25 +15,16 @@ use app\common\QueueJob;
*/
class FetchExperts
{
private $oQueueJob;
public function __construct()
{
$this->oQueueJob = new QueueJob();
}
public function fire(Job $job, $data)
{
$this->oQueueJob->init($job);
$field = isset($data['field']) ? (string)$data['field'] : '';
if ($field === '') {
$this->oQueueJob->log("FetchExperts 无效的 field，删除任务");
$job->delete();
return;
}
try {
$service = new ExpertFinderService();
$service->doFetchForField(
$field,
@@ -42,14 +32,6 @@ class FetchExperts
isset($data['per_page']) ? intval($data['per_page']) : 100,
isset($data['min_year']) ? $data['min_year'] : null
);
$this->oQueueJob->log("FetchExperts 完成 | field={$field}");
$job->delete();
} catch (\Exception $e) {
$this->oQueueJob->handleException($e, $job, "field={$field}");
} catch (\Throwable $e) {
$this->oQueueJob->handleException($e, $job, "field={$field}");
} finally {
$this->oQueueJob->finnal();
}
}
}

16
application/api/job/FillExpertCountry.php
View File

@@ -4,7 +4,6 @@ namespace app\api\job;
use think\queue\Job;
use app\common\ExpertFinderService;
use app\common\QueueJob;
/**
* 队列任务：用本地大模型从 affiliation 推断国家，写入 expert.country_id / country。
@@ -17,16 +16,9 @@ use app\common\QueueJob;
*/
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']) : '';
@@ -35,7 +27,6 @@ class FillExpertCountry
$service = new ExpertFinderService();
try {
if ($expertId && $affiliation !== '') {
$service->fillExpertCountry($expertId, $affiliation, $chatUrl);
}
@@ -43,12 +34,5 @@ class FillExpertCountry
// 链式：处理完当前专家立刻拉下一个进来
$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();
}
}
}

19
application/api/job/PromotionPrepare.php
View File

@@ -4,7 +4,6 @@ namespace app\api\job;
use think\queue\Job;
use app\common\PromotionService;
use app\common\QueueJob;
/**
* 【已废弃 / 兼容保留】
@@ -16,35 +15,17 @@ use app\common\QueueJob;
*/
class PromotionPrepare
{
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("PromotionPrepare[deprecated] 无效的 task_id，删除任务");
$job->delete();
return;
}
try {
$service = new PromotionService();
$service->enqueuePrepareTask($taskId);
$this->oQueueJob->log("PromotionPrepare[deprecated] forwarded task_id={$taskId} -> PromotionPrepareTask");
$job->delete();
} catch (\Exception $e) {
$this->oQueueJob->handleException($e, $job, "[deprecated] task_id={$taskId}");
} catch (\Throwable $e) {
$this->oQueueJob->handleException($e, $job, "[deprecated] task_id={$taskId}");
} finally {
$this->oQueueJob->finnal();
}
}
}

31
application/api/job/PromotionPrepareEmail.php
View File

@@ -4,7 +4,6 @@ namespace app\api\job;
use think\queue\Job;
use app\common\PromotionService;
use app\common\QueueJob;
/**
* 队列任务：单封邮件 prepare（调用 LLM 生成个性化描述 + 渲染模板 + 写入 log）。
@@ -18,41 +17,13 @@ use app\common\QueueJob;
*/
class PromotionPrepareEmail
{
// private $oQueueJob;
// public function __construct()
// {
// $this->oQueueJob = new QueueJob();
// }
public function fire(Job $job, $data)
{
// $this->oQueueJob->init($job);
//
$logId = isset($data['log_id']) ? intval($data['log_id']) : 0;
// if ($logId <= 0) {
// $this->oQueueJob->log("PromotionPrepareEmail 无效的 log_id，删除任务");
// $job->delete();
// return;
// }
//
// try {
$service = new PromotionService();
$service->log("id:".$logId);
// $result = $service->prepareSingleEmail($logId);
//
// $code = isset($result['code']) ? $result['code'] : '';
// $msg = isset($result['msg']) ? $result['msg'] : '';
// $llm = isset($result['llm_status']) ? $result['llm_status'] : '';
// $this->oQueueJob->log("PromotionPrepareEmail 完成 | log_id={$logId} code={$code} llm_status={$llm} msg={$msg}");
$result = $service->prepareSingleEmail($logId);
$job->delete();
// } catch (\Exception $e) {
// $this->oQueueJob->handleException($e, $job, "log_id={$logId}");
// } catch (\Throwable $e) {
// $this->oQueueJob->handleException($e, $job, "log_id={$logId}");
// } finally {
// $this->oQueueJob->finnal();
// }
}
}

23
application/api/job/PromotionPrepareTask.php
View File

@@ -17,40 +17,17 @@ use app\common\QueueJob;
*/
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();
}
}
}

22
application/api/job/PromotionSend.php
View File

@@ -4,7 +4,6 @@ namespace app\api\job;
use think\queue\Job;
use app\common\PromotionService;
use app\common\QueueJob;
/**
* 队列任务：发送 task 下"已 prepare"的邮件，按 min/max_interval 控制节奏。
@@ -16,39 +15,18 @@ use app\common\QueueJob;
*/
class PromotionSend
{
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("PromotionSend 无效的 task_id，删除任务");
$job->delete();
return;
}
try {
$service = new PromotionService();
$result = $service->processNextEmail($taskId);
$done = !empty($result['done']) ? 1 : 0;
$reason = isset($result['reason']) ? $result['reason'] : '';
$this->oQueueJob->log("PromotionSend 完成 | task_id={$taskId} done={$done} reason={$reason}");
$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();
}
}
}

31
application/api/job/myQueue.php Normal file
View File

@@ -0,0 +1,31 @@
<?php
namespace app\api\job;
use think\queue\Job;
class myQueue
{
public function fire(Job $job, $data)
{
try {
// 你的业务逻辑（哪怕是空的）
// 这里写任何代码
// 执行成功，删除任务
$job->delete();
} catch (\Throwable $e) {
// 如果你不想重试，直接删除
$job->delete();
}
}
// 可选：任务失败时执行
public function failed($data)
{
// 失败后不做处理，直接跳过
}
}

30
application/common/PromotionService.php
View File

@@ -319,19 +319,27 @@ class PromotionService
$this->log("dispatchPrepareEmails task_id={$taskId} no_logs -> state=5");
return ['dispatched' => 0, 'already_done' => true, 'error' => null];
}
// return $logIds;
foreach ($logIds as $logId) {
echo $logId."----";
$this->enqueuePrepareEmail(intval($logId));
}
$this->log("dispatchPrepareEmails task_id={$taskId} dispatched=" . count($logIds));
return ['dispatched' => count($logIds), 'already_done' => false, 'error' => null];
}
public function prepareSingleEmailTest($logId){
$log = Db::name('promotion_email_log')->where('log_id', $logId)->find();
// $task = Db::name('promotion_task')->where('task_id', $log['task_id'])->find();
// if (!$task) {
// Db::name('promotion_email_log')->where('log_id', $logId)->update([
// 'state' => 2,
// 'error_msg' => 'Task not found',
// 'send_time' => time(),
// ]);
// return ['code' => 1, 'msg' => 'task_not_found', 'llm_status' => 0];
// }
return jsonSuccess($log);
}
/**
* 对单封邮件执行准备：拉取 expert / journal，调 LLM 生成描述，渲染模板，写回 log。
*
@@ -768,16 +776,14 @@ class PromotionService
* 队列名：promotion_email
* 启动 worker：php think queue:listen --queue promotion_email
*/
public function enqueuePrepareEmail($logId, $delay = 0)
public function enqueuePrepareEmail($logId)
{
$jobClass = 'app\api\job\PromotionPrepareEmail@fire';
$data = ['log_id' => intval($logId)];
if ($delay > 0) {
Queue::later($delay, $jobClass, $data, 'PromotionPrepareEmail');
} else {
Queue::push($jobClass, $data, 'PromotionPrepareEmail');
}
$res =Queue::push($jobClass, $data, 'PromotionPrepareEmail');
}
/**