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

新逻辑参考文献相关性整合之前的逻辑

Browse Source
This commit is contained in:
wyn
2026-06-30 09:30:33 +08:00
parent 9c8f7cc3b6
commit da71dfc04e
8 changed files with 162 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
application/common/mq/ReferenceCheckArticleWorker.php
View File

@@ -4,10 +4,11 @@ namespace app\common\mq;
use think\Db;
use app\common\DbReconnectHelper;
use app\common\ReferenceCheckService;
use app\common\ReferenceRelevanceCheckService;
/**
* RabbitMQ 消费：按文章串行，文章内 reference_no 升序逐条校对（含低分同步二轮）
* RabbitMQ 消费（队列 reference_check / ref_check.article）：
* 全局文章串行，文章内 reference_no 升序链式逐条「主题相关性」校对。
*/
class ReferenceCheckArticleWorker
{
@@ -16,12 +17,12 @@ class ReferenceCheckArticleWorker
const BATCH_DONE = 2;
const BATCH_PARTIAL_FAILED = 3;
/** @var ReferenceCheckService */
/** @var ReferenceRelevanceCheckService */
private $svc;
public function __construct()
{
$this->svc = new ReferenceCheckService();
$this->svc = new ReferenceRelevanceCheckService();
}
public function handleMessage(array $payload)
@@ -29,6 +30,7 @@ class ReferenceCheckArticleWorker
DbReconnectHelper::ensure();
$pArticleId = intval(isset($payload['p_article_id']) ? $payload['p_article_id'] : 0);
$batchId = intval(isset($payload['batch_id']) ? $payload['batch_id'] : 0);
$trigger = isset($payload['trigger']) ? (string)$payload['trigger'] : 'enqueue';
if ($pArticleId <= 0 || $batchId <= 0) {
$this->svc->log('ReferenceCheckArticleWorker invalid payload');
return;
@@ -36,7 +38,11 @@ class ReferenceCheckArticleWorker
if (!$this->canStartArticleWork($batchId)) {
$this->svc->log('ReferenceCheckArticleWorker defer batch_id=' . $batchId . ' other article running');
(new ReferenceCheckMqPublisher())->publishArticleStart($pArticleId, $batchId, isset($payload['trigger']) ? $payload['trigger'] : 'enqueue');
(new ReferenceCheckMqPublisher())->publishArticleStart(
$pArticleId,
$batchId,
isset($payload['trigger']) ? $payload['trigger'] : 'enqueue'
);
sleep(3);
return;
}
@@ -48,6 +54,11 @@ class ReferenceCheckArticleWorker
}
}
$this->svc->recoverQueueRowsForArticle($pArticleId);
if ($trigger !== 'recheck_pending_only'
&& ReferenceRelevanceCheckService::PREPARE_LITERATURE_BEFORE_CHECK) {
$this->svc->prepareLiteratureContentByArticle($pArticleId);
}
$this->svc->log('ReferenceCheckArticleWorker start p_article_id=' . $pArticleId . ' batch_id=' . $batchId);
$done = 0;
@@ -61,7 +72,7 @@ class ReferenceCheckArticleWorker
if ($checkId <= 0) {
continue;
}
$result = $this->processOneRow($checkId, $row);
$result = $this->processOneRow($checkId, $row, $trigger === 'recheck_pending_only');
if ($result === 'ok') {
$done++;
} elseif ($result === 'failed') {
@@ -77,7 +88,7 @@ class ReferenceCheckArticleWorker
private function canStartArticleWork($batchId)
{
$running = Db::name('article_reference_check_batch')
$running = Db::name('article_reference_relevance_check_batch')
->where('batch_status', self::BATCH_RUNNING)
->where('id', '<>', intval($batchId))
->count();
@@ -87,7 +98,7 @@ class ReferenceCheckArticleWorker
private function claimBatch($batchId)
{
$now = date('Y-m-d H:i:s');
$affected = Db::name('article_reference_check_batch')
$affected = Db::name('article_reference_relevance_check_batch')
->where('id', intval($batchId))
->whereIn('batch_status', [self::BATCH_WAITING, self::BATCH_RUNNING])
->update([
@@ -99,15 +110,15 @@ class ReferenceCheckArticleWorker
private function getBatch($batchId)
{
return Db::name('article_reference_check_batch')->where('id', intval($batchId))->find();
return Db::name('article_reference_relevance_check_batch')->where('id', intval($batchId))->find();
}
private function fetchNextPendingRow($pArticleId)
{
return Db::name('article_reference_check_result')
return Db::name('article_reference_relevance_check_result')
->where('p_article_id', intval($pArticleId))
->where('queue_status', ReferenceCheckService::QUEUE_PENDING)
->where('status', ReferenceCheckService::RECORD_PENDING)
->where('queue_status', ReferenceRelevanceCheckService::QUEUE_PENDING)
->where('status', ReferenceRelevanceCheckService::RECORD_PENDING)
->order('reference_no asc,am_id asc,text_start asc,id asc')
->find();
}
@@ -115,46 +126,44 @@ class ReferenceCheckArticleWorker
/**
* @return string ok|failed|skip
*/
private function processOneRow($checkId, array $row)
private function processOneRow($checkId, array $row, $skipLiteratureFetch = false)
{
DbReconnectHelper::ensure();
$claimed = Db::name('article_reference_check_result')
$claimed = Db::name('article_reference_relevance_check_result')
->where('id', intval($checkId))
->where('queue_status', ReferenceCheckService::QUEUE_PENDING)
->update(['queue_status' => ReferenceCheckService::QUEUE_RUNNING]);
->where('queue_status', ReferenceRelevanceCheckService::QUEUE_PENDING)
->update(['queue_status' => ReferenceRelevanceCheckService::QUEUE_RUNNING]);
if (intval($claimed) <= 0) {
return 'skip';
}
$retryCount = intval(isset($row['retry_count']) ? $row['retry_count'] : 0);
try {
$this->svc->runReferenceCheckOnce($checkId);
$amId = intval(isset($row['am_id']) ? $row['am_id'] : 0);
if ($amId > 0) {
$this->svc->syncAmRefCheckStatus($amId);
}
$this->svc->markQueueRuntime($checkId, ReferenceCheckService::QUEUE_COMPLETED, $retryCount);
$this->svc->runCheckOnce($checkId, $skipLiteratureFetch);
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_COMPLETED, $retryCount);
return 'ok';
} catch (\Exception $e) {
$this->svc->log('ReferenceCheckArticleWorker check_id=' . $checkId . ' err=' . $e->getMessage());
DbReconnectHelper::ensure();
if ($retryCount < ReferenceCheckService::QUEUE_MAX_RETRY) {
$this->svc->markQueueRuntime($checkId, ReferenceCheckService::QUEUE_PENDING, $retryCount + 1);
return $this->processOneRow($checkId, array_merge($row, ['retry_count' => $retryCount + 1]));
if ($retryCount < ReferenceRelevanceCheckService::QUEUE_MAX_RETRY) {
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_PENDING, $retryCount + 1);
return $this->processOneRow($checkId, array_merge($row, ['retry_count' => $retryCount + 1]), $skipLiteratureFetch);
}
try {
$this->svc->updateCheckResult($checkId, [
'status' => ReferenceCheckService::RECORD_FAILED,
'error_msg' => $e->getMessage(),
]);
$this->svc->markQueueRuntime($checkId, ReferenceCheckService::QUEUE_FAILED, $retryCount);
$fresh = Db::name('article_reference_relevance_check_result')->where('id', intval($checkId))->find();
$groupRows = !empty($fresh) ? $this->svc->findCitationGroupRowsForWorker($fresh) : [];
if (!empty($groupRows)) {
$this->svc->failGroupWithQueue($groupRows, $e->getMessage(), $retryCount);
} else {
$this->svc->updateCheckResult($checkId, [
'status' => ReferenceRelevanceCheckService::RECORD_FAILED,
'error_msg' => $e->getMessage(),
]);
$this->svc->markQueueRuntime($checkId, ReferenceRelevanceCheckService::QUEUE_FAILED, $retryCount);
}
} catch (\Exception $e2) {
\think\Log::error('ReferenceCheckArticleWorker markFailed: ' . $e2->getMessage());
}
$amId = intval(isset($row['am_id']) ? $row['am_id'] : 0);
if ($amId > 0) {
$this->svc->syncAmRefCheckStatus($amId);
}
return 'failed';
}
}
@@ -170,7 +179,7 @@ class ReferenceCheckArticleWorker
if ($failed > 0) {
$status = self::BATCH_PARTIAL_FAILED;
}
Db::name('article_reference_check_batch')->where('id', intval($batchId))->update([
Db::name('article_reference_relevance_check_batch')->where('id', intval($batchId))->update([
'batch_status' => $status,
'done_count' => intval($done),
'failed_count' => intval($failed),
@@ -183,7 +192,7 @@ class ReferenceCheckArticleWorker
private function publishNextWaitingBatch()
{
$next = Db::name('article_reference_check_batch')
$next = Db::name('article_reference_relevance_check_batch')
->where('batch_status', self::BATCH_WAITING)
->order('id asc')
->find();
@@ -197,8 +206,8 @@ class ReferenceCheckArticleWorker
isset($next['trigger']) ? $next['trigger'] : 'enqueue'
);
} catch (\Exception $e) {
$this->svc->log('publishNextWaitingBatch failed: ' . $e->getMessage());
\think\Log::error('publishNextWaitingBatch: ' . $e->getMessage());
$this->svc->log('ReferenceCheck publishNextWaitingBatch failed: ' . $e->getMessage());
\think\Log::error('ReferenceCheck publishNextWaitingBatch: ' . $e->getMessage());
}
}
}